# 连接池：thick client 只初始化一次；缺配置抛固定文案；acquire 会归还。

from __future__ import annotations

import sys
from types import SimpleNamespace

import pytest

from rpa_oracle.errors import RepoUnavailable


# 配齐建池所需的三项环境变量。
def _creds(monkeypatch) -> None:
    monkeypatch.setenv("RPA_ORACLE_USER", "rpainit")
    monkeypatch.setenv("RPA_ORACLE_PASSWORD", "not-for-git")
    monkeypatch.setenv("RPA_ORACLE_HOST", "secret-host.example")
    monkeypatch.setenv("RPA_ORACLE_PORT", "1521")
    monkeypatch.setenv("RPA_ORACLE_SERVICE", "rpa")
    monkeypatch.setenv("RPA_ORACLE_THICK", "1")


# 把假 oracledb 塞进 sys.modules，供 pool 函数内 import。
def _install_fake_oracledb(monkeypatch, *, init=None, create=None):
    fake = SimpleNamespace(
        init_oracle_client=init or (lambda **kwargs: None),
        create_pool=create or (lambda **kwargs: SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))),
    )
    monkeypatch.setitem(sys.modules, "oracledb", fake)
    return fake


# 每个用例把 thick client 已初始化标记清掉。
@pytest.fixture
def reset_client(monkeypatch):
    import rpa_oracle.pool as pool_mod

    monkeypatch.setattr(pool_mod, "_client_ready", False)
    yield pool_mod
    monkeypatch.setattr(pool_mod, "_client_ready", False)


# OMS 角色默认池大小 1–4。
def test_pool_limits_oms_defaults(monkeypatch):
    monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
    monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
    from rpa_oracle.pool import pool_limits

    assert pool_limits("oms") == (1, 4)


# 班次角色默认池大小 1–1。
def test_pool_limits_job_defaults(monkeypatch):
    monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
    monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
    from rpa_oracle.pool import pool_limits

    assert pool_limits("job") == (1, 1)


# RPA_ORACLE_POOL_MIN / MAX 覆盖角色默认值。
def test_pool_limits_env_overrides(monkeypatch):
    monkeypatch.setenv("RPA_ORACLE_POOL_MIN", "2")
    monkeypatch.setenv("RPA_ORACLE_POOL_MAX", "8")
    from rpa_oracle.pool import pool_limits

    assert pool_limits("oms") == (2, 8)
    assert pool_limits("job") == (2, 8)


# 缺用户时抛「数据库不可用」，文案不含 host。
def test_missing_user_raises_fixed_message(monkeypatch):
    monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
    monkeypatch.setenv("RPA_ORACLE_PASSWORD", "not-for-git")
    monkeypatch.setenv("RPA_ORACLE_HOST", "secret-host.example")
    from rpa_oracle.pool import create_pool

    with pytest.raises(RepoUnavailable) as caught:
        create_pool()
    assert caught.value.message == "数据库不可用"
    assert "secret-host" not in str(caught.value)
    assert "ORA-" not in str(caught.value)


# 缺密码时抛「数据库不可用」，文案不含 DSN。
def test_missing_password_raises_fixed_message(monkeypatch):
    monkeypatch.setenv("RPA_ORACLE_USER", "rpainit")
    monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
    monkeypatch.setenv("RPA_ORACLE_HOST", "secret-host.example")
    from rpa_oracle.pool import create_pool

    with pytest.raises(RepoUnavailable) as caught:
        create_pool()
    assert caught.value.message == "数据库不可用"
    assert "secret-host" not in str(caught.value)


# create_pool 调用两次时 init_oracle_client 只跑一次。
def test_thick_init_runs_once(monkeypatch, reset_client):
    _creds(monkeypatch)
    calls: list[dict] = []

    # 记下 thick client 初始化参数。
    def init_oracle_client(**kwargs):
        calls.append(kwargs)

    created: list[dict] = []

    # 记下建池参数并返回空池替身。
    def create_pool_raw(**kwargs):
        created.append(kwargs)
        return SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))

    _install_fake_oracledb(monkeypatch, init=init_oracle_client, create=create_pool_raw)
    from rpa_oracle.pool import create_pool

    create_pool()
    create_pool()
    assert len(calls) == 1
    assert len(created) == 2


# thick client 初始化失败时对外仍是「数据库不可用」，不含库路径。
def test_thick_init_failure_hides_lib_path(monkeypatch, reset_client):
    _creds(monkeypatch)
    monkeypatch.setenv("RPA_ORACLE_CLIENT_LIB", "/secret/instantclient")

    # 假装 Instant Client 加载失败，并把路径写进异常。
    def init_oracle_client(**kwargs):
        raise OSError(f"cannot load {kwargs.get('lib_dir')}")

    _install_fake_oracledb(monkeypatch, init=init_oracle_client)
    from rpa_oracle.pool import create_pool

    with pytest.raises(RepoUnavailable) as caught:
        create_pool()
    assert caught.value.message == "数据库不可用"
    assert "instantclient" not in str(caught.value)
    assert "ORA-" not in str(caught.value)


# OMS 建池把 min=1 max=4 传给 oracledb.create_pool。
def test_create_pool_passes_oms_limits(monkeypatch, reset_client):
    _creds(monkeypatch)
    monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
    monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
    seen: dict = {}

    # 记下 OMS 建池参数。
    def create_pool_raw(**kwargs):
        seen.update(kwargs)
        return SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))

    _install_fake_oracledb(monkeypatch, create=create_pool_raw)
    from rpa_oracle.pool import create_pool

    create_pool(role="oms")
    assert seen["min"] == 1
    assert seen["max"] == 4
    assert "secret-host.example" in seen["dsn"]
    assert seen["user"] == "rpainit"


def test_create_pool_passes_job_limits(monkeypatch, reset_client):
    """班次建池保持 min=1 max=1，并带 60 秒探活。"""
    _creds(monkeypatch)
    monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
    monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
    seen: dict = {}

    # 记下班次建池参数。
    def create_pool_raw(**kwargs):
        seen.update(kwargs)
        return SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))

    _install_fake_oracledb(monkeypatch, create=create_pool_raw)
    from rpa_oracle.pool import create_pool

    create_pool(role="job")
    assert seen["min"] == 1
    assert seen["max"] == 1
    assert seen["ping_interval"] == 60


# with acquire() 结束时把连接 close 归还。
def test_acquire_releases_on_exit(monkeypatch, reset_client):
    _creds(monkeypatch)

    # 记下 close 是否被调用。
    class FakeConn:
        # 连接尚未归还。
        def __init__(self):
            self.closed = False

        # 标记已归还。
        def close(self):
            self.closed = True

    conn = FakeConn()

    # 每次 acquire 都给出同一条假连接。
    class FakeRawPool:
        # 返回预先准备的假连接。
        def acquire(self):
            return conn

    _install_fake_oracledb(monkeypatch, create=lambda **kwargs: FakeRawPool())
    from rpa_oracle.pool import create_pool

    pool = create_pool()
    with pool.acquire() as held:
        assert held is conn
        assert conn.closed is False
    assert conn.closed is True


# with 块里抛错时连接仍归还。
def test_acquire_releases_after_error(monkeypatch, reset_client):
    _creds(monkeypatch)

    # 记下 close 是否被调用。
    class FakeConn:
        # 连接尚未归还。
        def __init__(self):
            self.closed = False

        # 标记已归还。
        def close(self):
            self.closed = True

    conn = FakeConn()

    # 每次 acquire 都给出同一条假连接。
    class FakeRawPool:
        # 返回预先准备的假连接。
        def acquire(self):
            return conn

    _install_fake_oracledb(monkeypatch, create=lambda **kwargs: FakeRawPool())
    from rpa_oracle.pool import create_pool

    pool = create_pool()
    with pytest.raises(RuntimeError):
        with pool.acquire():
            raise RuntimeError("boom")
    assert conn.closed is True


def test_failed_ping_or_dead_connection_borrows_a_new_one():
    """探活失败或执行前已死时，下一次借到新连接。"""
    from rpa_oracle.pool import OraclePool

    class Conn:
        """假连接。ping 失败或 dead 表示执行前不可用。"""

        def __init__(self, name, *, alive=True, dead=False):
            """记下名字和是否还活着。"""
            self.name = name
            self._alive = alive
            self.dead = dead
            self.closed = False

        def ping(self):
            """探活。不可用时抛错。"""
            if not self._alive:
                raise OSError("ping failed")

        def close(self):
            """标记已归还。"""
            self.closed = True

    live = Conn("live")
    live_after_dead = Conn("live-after-dead")

    class Raw:
        """按队列发连接，并记下被丢掉的名字。"""

        def __init__(self):
            """先给出坏连接，再给出活连接。"""
            self.queue = [
                Conn("ping-dead", alive=False),
                live,
                Conn("flag-dead", dead=True),
                live_after_dead,
            ]
            self.dropped: list[str] = []

        def acquire(self):
            """取出下一条连接。"""
            return self.queue.pop(0)

        def drop(self, conn):
            """记下丢掉的连接。"""
            self.dropped.append(conn.name)

    raw = Raw()
    pool = OraclePool(raw)
    with pool.acquire() as held:
        assert held.name == "live"
    with pool.acquire() as held:
        assert held.name == "live-after-dead"
    assert raw.dropped == ["ping-dead", "flag-dead"]
    assert live.closed is True
    assert live_after_dead.closed is True


def test_uncommitted_failure_retries_that_statement_only():
    """未提交失败再借一次只重试这一笔。已经 commit 的不重做。"""
    from rpa_oracle.pool import OraclePool

    class Conn:
        """假连接，记下 commit 次数。"""

        def __init__(self, name):
            """按借出顺序命名。"""
            self.name = name
            self.closed = False
            self.commits = 0

        def commit(self):
            """记一次提交。"""
            self.commits += 1

        def close(self):
            """标记已归还。"""
            self.closed = True

    class Raw:
        """每次借出一条新连接，并记下被丢掉的名字。"""

        def __init__(self):
            """从 1 开始编号。"""
            self.n = 0
            self.dropped: list[int] = []
            self.conns: list[Conn] = []

        def acquire(self):
            """借出下一条新连接。"""
            self.n += 1
            conn = Conn(self.n)
            self.conns.append(conn)
            return conn

        def drop(self, conn):
            """记下丢掉的连接。"""
            self.dropped.append(conn.name)

    raw = Raw()
    pool = OraclePool(raw)
    seen: list[int] = []

    def work(conn):
        """第一笔未提交就失败，重试的那笔才提交。"""
        seen.append(conn.name)
        if conn.name == 1:
            raise RuntimeError("not committed")
        conn.commit()

    pool.run(work)
    assert seen == [1, 2]
    assert raw.dropped == [1]
    assert raw.conns[0].commits == 0
    assert raw.conns[1].commits == 1
    assert raw.conns[1].closed is True

    committed_seen: list[str] = []

    def committed_then_fail(conn):
        """先提交再失败。这一笔不能重做。"""
        committed_seen.append(f"run-{conn.name}")
        conn.commit()
        committed_seen.append(f"commit-{conn.name}")
        raise RuntimeError("after commit")

    before = raw.n
    with pytest.raises(RuntimeError, match="after commit"):
        pool.run(committed_then_fail)
    name = before + 1
    assert committed_seen == [f"run-{name}", f"commit-{name}"]
    assert raw.dropped == [1, name]
    assert raw.n == name