# 返利 Oracle 仓库：.env 填充、领取 SQL、落表一条连接。不连真库。

from __future__ import annotations

from contextlib import contextmanager
import os

import pytest

from rpa_oracle import RepoUnavailable, load_job_env
from persist import USER_NAME, GroupSave
from repo import (
    LEASE_MINUTES,
    OracleRebateRepo,
    _SELECT_DRAFTS,
    _SELECT_DRAFTS_LIMIT,
    _SELECT_PROTOCOLS,
    _SELECT_PROTOCOLS_LIMIT,
)


# 返利 .env 空缺时补 oms .env，已有非空环境变量不覆盖。
def test_load_runtime_env_fills_blank_from_oms_env(tmp_path, monkeypatch):
    monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
    monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
    monkeypatch.setenv("RPA_ORACLE_USER", "keep-me")
    rebate_env = tmp_path / "rebate.env"
    oms_env = tmp_path / "oms.env"
    rebate_env.write_text(
        "RPA_ORACLE_HOST=192.168.12.244\nRPA_ORACLE_PASSWORD=\nRPA_ORACLE_USER=from-rebate\n",
        encoding="utf-8",
    )
    oms_env.write_text(
        "RPA_ORACLE_PASSWORD=not-for-git\nRPA_ORACLE_USER=from-oms\n",
        encoding="utf-8",
    )
    load_job_env(tmp_path, [rebate_env, oms_env])
    assert os.environ["RPA_ORACLE_HOST"] == "192.168.12.244"
    assert os.environ["RPA_ORACLE_PASSWORD"] == "not-for-git"
    assert os.environ["RPA_ORACLE_USER"] == "keep-me"


# 缺 HOST/用户/密码时 from_env 抛数据库不可用。
def test_from_env_without_user_raises(monkeypatch):
    monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
    monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
    monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
    with pytest.raises(RepoUnavailable):
        OracleRebateRepo.from_env()


# 录 SQL 的假游标，按表名返回预设领取行。
class _FakeCursor:
    # 挂上共享的假连接。
    def __init__(self, conn):
        self._conn = conn
        self._rows: list = []
        self.description = None
        self.rowcount = 1

    # 进入 with，返回自身。
    def __enter__(self):
        return self

    # 退出 with，不吞异常。
    def __exit__(self, *args):
        return False

    # 记下一次批量 SQL 和全部 bind。
    def executemany(self, sql, seq):
        self._conn.many.append((sql, [dict(item) for item in seq]))

    # 记下 SQL 和 bind；SELECT 按表切领取结果。一条 IN 超过 1000 个 fid 时抛错。
    def execute(self, sql, params=None):
        params = params or {}
        _reject_oversized_in(params)
        self._conn.executed.append((sql, params))
        text = sql.lstrip().lower()
        if text.startswith("select") and "rpa_contract_agreement_result" in text:
            self._rows = self._conn.draft_rows
            self.description = self._conn.draft_description
        elif text.startswith("select") and "rpa_purchase_contract_data" in text:
            self._rows = self._conn.protocol_rows
            self.description = self._conn.protocol_description
        else:
            self._rows = []

    # 返回本次 SELECT 预设行的副本。
    def fetchall(self):
        return list(self._rows)


# 假连接：同一实例上的 cursor 共享 executed 列表。
class _FakeConn:
    # 可选带上表 4 起草行和表 3 只录协议行。
    def __init__(self, *, draft_rows=None, protocol_rows=None):
        self.executed: list[tuple[str, dict]] = []
        self.many: list[tuple[str, list]] = []
        self.committed = False
        self.commits = 0
        self.draft_rows = draft_rows or []
        self.protocol_rows = protocol_rows or []
        self.draft_description = _DRAFT_DESCRIPTION
        self.protocol_description = _PROTOCOL_DESCRIPTION

    # 新开一个录 SQL 的游标。
    def cursor(self):
        return _FakeCursor(self)

    # 记下已提交。
    def commit(self):
        self.committed = True
        self.commits += 1

    # 本套用例不应回滚。
    def rollback(self):
        raise AssertionError("不应 rollback")

    # 归还时由池调用，假连接不计数。
    def close(self):
        return None


# 只交出一条连接，并记下借了几次。
class _FakePool:
    # 挂上要借出的那条假连接。
    def __init__(self, conn):
        self._conn = conn
        self.acquires = 0

    # 借出连接并给 acquires 加一。
    @contextmanager
    def acquire(self):
        self.acquires += 1
        yield self._conn


_DRAFT_DESCRIPTION = (
    ("FID",),
    ("OMS_ID",),
    ("ACTION_CODE",),
    ("CONTRACT_STATUS",),
    ("CONTRACT_FAIL_REASON",),
    ("CONTRACT_NO",),
    ("AGREEMENT_STATUS",),
    ("AGREEMENT_FAIL_REASON",),
    ("AGREEMENT_ID",),
    ("AGREEMENT_DTL_ID",),
    ("ORG_CODE",),
    ("PUSH_STATUS",),
    ("PROCESS_STATUS",),
)

_PROTOCOL_DESCRIPTION = (
    ("FID",),
    ("OMS_ID",),
    ("CONTRACT_NO",),
    ("CONTRACT_ENTRY_STATUS",),
    ("PROCESS_STATUS",),
    ("ORG_CODE",),
    ("PROCESS_RESULT",),
)


# 构造表 4 领取查询返回的一行元组。
def _draft_fetched(fid: int = 11, oms_id: int = 2026100501, contract_no: str = "HT1"):
    """构造表 4 领取查询返回的一行元组。"""
    return (
        fid,
        oms_id,
        1,
        1,
        None,
        contract_no,
        None,
        None,
        None,
        None,
        "901",
        0,
        0,
    )


# 构造表 3 只录协议领取查询返回的一行元组。
def _protocol_fetched(fid: int = 21, oms_id: int = 2026100503):
    return (fid, oms_id, "HT-OLD", 0, 0, "901", "")


# 领取 SQL 带 FOR UPDATE SKIP LOCKED，过期 process_status=3 可再领，同号未过租约整组排除。
def test_claim_sql_uses_skip_locked_and_lease():
    draft_sql = " ".join(_SELECT_DRAFTS.split()).lower()
    assert "for update skip locked" in draft_sql
    assert "process_status = 0" in draft_sql
    assert "process_status = 3" in draft_sql
    assert "not exists" in draft_sql
    assert "numtodsinterval(:lease_minutes" in draft_sql
    assert "rownum" not in draft_sql
    assert "skip locked" in " ".join(_SELECT_DRAFTS_LIMIT.split()).lower()
    protocol_sql = " ".join(_SELECT_PROTOCOLS.split()).lower()
    assert "for update skip locked" in protocol_sql
    assert "process_status = 0" in protocol_sql
    assert "process_status = 3" in protocol_sql
    assert "process_time" in protocol_sql
    assert "numtodsinterval(:lease_minutes" in protocol_sql
    assert "skip locked" in " ".join(_SELECT_PROTOCOLS_LIMIT.split()).lower()
    assert LEASE_MINUTES == 30


# 领取一次借连接：SKIP LOCKED 选出后表 4、表 3 都改成 process_status=3。
def test_claim_pending_locks_and_stamps_in_one_acquire():
    conn = _FakeConn(
        draft_rows=[_draft_fetched()],
        protocol_rows=[_protocol_fetched()],
    )
    pool = _FakePool(conn)
    repo = OracleRebateRepo(pool)
    claimed = repo.claim_pending(limit=5)
    assert pool.acquires == 1
    assert [row["fid"] for row in claimed.drafts] == [11]
    assert claimed.drafts[0]["process_status"] == 3
    assert claimed.drafts[0]["update_user_name"] == USER_NAME
    assert [row["fid"] for row in claimed.protocols] == [21]
    assert claimed.protocols[0]["process_status"] == 3
    assert conn.committed is True
    selects = [item for item in conn.executed if item[0].lstrip().lower().startswith("select")]
    updates = [item for item in conn.executed if item[0].lstrip().lower().startswith("update")]
    assert len(selects) == 2
    assert len(updates) == 2
    assert "skip locked" in selects[0][0].lower()
    assert selects[0][1]["lease_minutes"] == 30
    assert "limit" not in selects[0][1]
    assert selects[1][1]["limit"] == 4
    assert "skip locked" in selects[1][0].lower()
    draft_update = " ".join(updates[0][0].split()).lower()
    assert "process_status = 3" in draft_update
    assert "系统管理员" in updates[0][0]
    protocol_update = " ".join(updates[1][0].split()).lower()
    assert "process_status = 3" in protocol_update


# 表 4 更新/释放/插入和表 3 更新各一次 executemany，同一连接同一事务。
def test_save_group_writes_in_one_acquire():
    conn = _FakeConn()
    pool = _FakePool(conn)
    repo = OracleRebateRepo(pool)
    payload = GroupSave(
        table4_updates=[
            {"fid": 1, "contract_status": 2, "process_status": 1},
            {
                "fid": 3,
                "contract_status": 1,
                "process_status": 0,
                "agreement_fail_reason": "送审失败",
            },
        ],
        table4_inserts=[
            {
                "oms_id": 2026100501,
                "action_code": 2,
                "contract_status": None,
                "contract_fail_reason": None,
                "contract_no": "HT1",
                "agreement_status": 1,
                "agreement_fail_reason": None,
                "agreement_id": "AGR1",
                "agreement_dtl_id": "AGRD1",
                "process_status": 1,
                "org_code": "901",
            },
            {
                "oms_id": 2026100502,
                "action_code": 2,
                "contract_status": None,
                "contract_fail_reason": None,
                "contract_no": "HT1",
                "agreement_status": 1,
                "agreement_fail_reason": None,
                "agreement_id": "AGR1",
                "agreement_dtl_id": "AGRD2",
                "process_status": 1,
                "org_code": "901",
            },
        ],
        table4_release_fids=[2, 4],
        table3_updates=[
            {"fid": 10, "process_status": 1, "process_result": "协议已录入"},
            {"fid": 11, "process_status": 0, "process_result": "下一班再领"},
        ],
    )
    repo.save_group(payload)
    assert pool.acquires == 1
    assert conn.committed is True
    assert conn.executed == []
    assert len(conn.many) == 4
    draft_sql, draft_binds = conn.many[0]
    release_sql, release_binds = conn.many[1]
    insert_sql, insert_binds = conn.many[2]
    protocol_sql, protocol_binds = conn.many[3]
    assert "agreement_fail_reason" in draft_sql.lower()
    assert [row["fid"] for row in draft_binds] == [1, 3]
    assert draft_binds[0]["agreement_fail_reason"] is None
    assert draft_binds[1]["agreement_fail_reason"] == "送审失败"
    assert [row["fid"] for row in release_binds] == [2, 4]
    assert "process_status = 0" in " ".join(release_sql.split()).lower()
    assert "insert into rpa_contract_agreement_result" in insert_sql.lower()
    assert [row["oms_id"] for row in insert_binds] == [2026100501, 2026100502]
    assert "rpa_purchase_contract_data" in protocol_sql.lower()
    assert [row["fid"] for row in protocol_binds] == [10, 11]
    assert [row["process_status"] for row in protocol_binds] == [1, 0]


# 班次 session 内领取加落表只从池里借一次连接。
def test_session_claim_and_save_acquire_once():
    conn = _FakeConn(draft_rows=[_draft_fetched()])
    pool = _FakePool(conn)
    repo = OracleRebateRepo(pool)
    payload = GroupSave(
        table4_updates=[
            {"fid": 11, "contract_status": 2, "process_status": 1},
        ],
        table4_inserts=[],
        table4_release_fids=[],
        table3_updates=[],
    )
    with repo.session():
        claimed = repo.claim_pending()
        assert [row["fid"] for row in claimed.drafts] == [11]
        repo.save_group(payload)
    assert pool.acquires == 1
    assert conn.commits >= 2


def _fid_bind_count(params: dict) -> int:
    """数出本段 bind 里的 fid 个数。"""
    return sum(
        1
        for key in params
        if isinstance(key, str) and key.startswith("id") and key[2:].isdigit()
    )


def _reject_oversized_in(params: dict) -> None:
    """一条 IN 超过 1000 个 fid 时抛错，模拟 ORA-01795。"""
    if _fid_bind_count(params) > 1000:
        raise RuntimeError("ORA-01795")


def test_claim_one_contract_over_1000_rows_keeps_the_whole_group():
    """一个合同号超过 1000 行仍整组领走，更新按 500 分段，不报数据库不可用。"""
    rows = [
        _draft_fetched(fid, oms_id=2026100500 + fid, contract_no="HT-BIG")
        for fid in range(1, 1002)
    ]
    conn = _FakeConn(draft_rows=rows)
    pool = _FakePool(conn)
    repo = OracleRebateRepo(pool)
    claimed = repo.claim_pending(limit=10)
    assert pool.acquires == 1
    assert len(claimed.drafts) == 1001
    assert {row["contract_no"] for row in claimed.drafts} == {"HT-BIG"}
    assert all(row["process_status"] == 3 for row in claimed.drafts)
    assert claimed.protocols == []
    updates = [
        item for item in conn.executed if item[0].lstrip().lower().startswith("update")
    ]
    assert [_fid_bind_count(params) for _, params in updates] == [500, 500, 1]
    assert conn.committed is True