# 加购仓库:领取戳、租约、假连接录 SQL。不连真库。 from __future__ import annotations import ast from contextlib import contextmanager from datetime import datetime, timedelta from pathlib import Path import os import pytest from rpa_oracle import RepoUnavailable, load_job_env from cart_repo import ( LEASE_MINUTES, STAMP_CLAIMING, STAMP_DONE, STAMP_SELECTION, MemoryCartRepo, OracleCartRepo, _pending_select_sql, ) # 构造一条待加购表 2 行。 def _row(**overrides) -> dict: data = { "fid": 1, "oms_id": 2026100201, "platform_type": 66, "supplier_name": "广东壹号药业有限公司", "bar_code": "06901234567890", "purchase_qty": 7, "status": 1, "fail_reason": None, "push_status": 0, "update_user_name": STAMP_SELECTION, "operate_type": 1, } data.update(overrides) return data # 系统管理员成功行领走盖成加购中,已完成和失败行不领。 def test_claim_pending_stamps_processing_and_skips_done(): repo = MemoryCartRepo( [ _row(), _row(fid=2, oms_id=2026100202, update_user_name=STAMP_DONE), _row(fid=3, oms_id=2026100203, status=2, update_user_name=STAMP_SELECTION), _row(fid=4, oms_id=2026100204, operate_type=2), ] ) claimed = repo.claim_pending() assert [row.fid for row in claimed] == [1] assert claimed[0].update_user_name == STAMP_CLAIMING assert repo.rows[1]["update_user_name"] == STAMP_CLAIMING assert repo.rows[2]["update_user_name"] == STAMP_DONE assert repo.rows[3]["status"] == 2 # 戳加购中且 update_time 超过租约的行可以再领。 def test_claim_reclaims_expired_claiming_stamp(): now = datetime(2026, 10, 22, 12, 0, 0) repo = MemoryCartRepo( [_row(update_user_name=STAMP_CLAIMING, update_time=now - timedelta(minutes=31))] ) claimed = repo.claim_pending(now=now) assert [row.fid for row in claimed] == [1] assert repo.rows[1]["update_user_name"] == STAMP_CLAIMING # 戳加购中且未过租约的行不领。 def test_claim_skips_fresh_claiming_stamp(): now = datetime(2026, 10, 22, 12, 0, 0) repo = MemoryCartRepo( [_row(update_user_name=STAMP_CLAIMING, update_time=now - timedelta(minutes=1))] ) claimed = repo.claim_pending(now=now) assert claimed == [] assert repo.rows[1]["update_user_name"] == STAMP_CLAIMING # 成功只盖加购班,不插行、不改 status。 def test_mark_success_does_not_insert_or_change_status(): repo = MemoryCartRepo([_row()]) repo.claim_pending() repo.mark_success(1) assert list(repo.rows) == [1] assert repo.rows[1]["status"] == 1 assert repo.rows[1]["fail_reason"] is None assert repo.rows[1]["update_user_name"] == STAMP_DONE assert repo.rows[1]["push_status"] == 0 # 失败改 status=2 并写下原因。 def test_mark_fail_sets_status_and_reason(): repo = MemoryCartRepo([_row()]) repo.claim_pending() repo.mark_fail(1, "库存不足") assert list(repo.rows) == [1] assert repo.rows[1]["status"] == 2 assert repo.rows[1]["fail_reason"] == "库存不足" assert repo.rows[1]["update_user_name"] == STAMP_DONE # 加购 .env 空缺时补选品和 oms .env,已有非空环境变量不覆盖。 def test_load_runtime_env_fills_blank_from_other_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") cart_env = tmp_path / "cart.env" select_env = tmp_path / "select.env" oms_env = tmp_path / "oms.env" cart_env.write_text("RPA_ORACLE_HOST=192.168.12.244\nRPA_ORACLE_PASSWORD=\n", encoding="utf-8") select_env.write_text("RPA_ORACLE_PASSWORD=from-select\nRPA_ORACLE_USER=from-select\n", encoding="utf-8") oms_env.write_text("RPA_ORACLE_PASSWORD=from-oms\n", encoding="utf-8") load_job_env(tmp_path, [cart_env, select_env, oms_env]) assert os.environ["RPA_ORACLE_HOST"] == "192.168.12.244" assert os.environ["RPA_ORACLE_PASSWORD"] == "from-select" 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): OracleCartRepo.from_env() # 录 SQL 的假游标,fetchall 返回预设行。 class _FakeCursor: # 挂上共享的 executed 列表和预设查询结果。 def __init__(self, executed: list, rows: list | None = None, description=None): self._executed = executed self._rows = rows or [] self.description = description self.rowcount = 1 # 进入 with,返回自身。 def __enter__(self): return self # 退出 with,不吞异常。 def __exit__(self, *args): return False # 记下 SQL 和 bind。一条 IN 超过 1000 个 fid 时抛错,模拟 ORA-01795。 def execute(self, sql, params=None): params = params or {} _reject_oversized_in(params) self._executed.append((sql, params)) # 返回预设行的副本。 def fetchall(self): return list(self._rows) # 假连接:同一实例上的 cursor 共享 executed 列表。 class _FakeConn: # 可选带上领取查询要返回的行。 def __init__(self, rows: list | None = None, description=None): self.executed: list[tuple[str, dict]] = [] self.committed = False self.commits = 0 self._rows = rows or [] self._description = description # 新开一个录 SQL 的游标。 def cursor(self): return _FakeCursor(self.executed, self._rows, self._description) # 记下已提交。 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 _CLAIM_DESCRIPTION = ( ("FID",), ("OMS_ID",), ("PLATFORM_TYPE",), ("SUPPLIER_NAME",), ("BAR_CODE",), ("PURCHASE_QTY",), ("STATUS",), ("FAIL_REASON",), ("PUSH_STATUS",), ("UPDATE_USER_NAME",), ("OPERATE_TYPE",), ) # 构造领取查询返回的一行元组。 def _fetched_row(fid: int = 11): return ( fid, 2026100201, 66, "广东壹号药业有限公司", "0690", 7, 1, None, 0, STAMP_SELECTION, 1, ) # 领取 SQL 带 FOR UPDATE SKIP LOCKED,过期加购中可再领。 def test_claim_sql_uses_skip_locked_and_lease(): sql, params = _pending_select_sql(limit=None, platform_types=None) text = " ".join(sql.split()).lower() assert "for update skip locked" in text assert "numtodsinterval(:lease_minutes" in text assert params["lease_minutes"] == LEASE_MINUTES assert LEASE_MINUTES == 30 limited, limited_params = _pending_select_sql(limit=5, platform_types=None) assert "skip locked" in " ".join(limited.split()).lower() assert limited_params["limit"] == 5 # 领取一次借连接:SKIP LOCKED 选出后盖成加购中。 def test_oracle_claim_updates_stamp(): conn = _FakeConn(rows=[_fetched_row()], description=_CLAIM_DESCRIPTION) pool = _FakePool(conn) repo = OracleCartRepo(pool) claimed = repo.claim_pending() assert pool.acquires == 1 assert conn.committed is True assert len(claimed) == 1 assert claimed[0].fid == 11 assert claimed[0].update_user_name == STAMP_CLAIMING select_sql, select_binds = conn.executed[0] assert "skip locked" in select_sql.lower() assert select_binds["lease_minutes"] == 30 update_sql = " ".join(conn.executed[1][0].split()).lower() assert "update_user_name" in update_sql assert conn.executed[1][1]["stamp"] == STAMP_CLAIMING assert 11 in conn.executed[1][1].values() # 班次 session 内领取加两行盖章只从池里借一次连接。 def test_session_claim_and_marks_acquire_once(): conn = _FakeConn( rows=[_fetched_row(11), _fetched_row(12)], description=_CLAIM_DESCRIPTION, ) pool = _FakePool(conn) repo = OracleCartRepo(pool) with repo.session(): claimed = repo.claim_pending() assert [row.fid for row in claimed] == [11, 12] repo.mark_success(11) repo.mark_fail(12, "库存不足") assert pool.acquires == 1 assert conn.commits >= 3 mark_sqls = [item[0].lower() for item in conn.executed[2:]] assert any("fail_reason" in sql for sql in mark_sqls) 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_1001_rows_updates_first_chunk_without_unavailable(): """领取 1001 条时第一段 500 能领走,同一事务提交,不报数据库不可用。""" conn = _FakeConn( rows=[_fetched_row(fid) for fid in range(1, 1002)], description=_CLAIM_DESCRIPTION, ) pool = _FakePool(conn) repo = OracleCartRepo(pool) claimed = repo.claim_pending() assert pool.acquires == 1 assert len(claimed) == 1001 assert claimed[0].update_user_name == STAMP_CLAIMING 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 updates[0][1]["id0"] == 1 assert updates[0][1]["stamp"] == STAMP_CLAIMING assert conn.committed is True # 登录和搜索不再往 sys.path 贴选品。 def test_login_and_search_do_not_append_sys_path(): root = Path(__file__).resolve().parent.parent for name in ("yaoex_login.py", "cart_search.py"): tree = ast.parse((root / name).read_text(encoding="utf-8")) for node in ast.walk(tree): if not isinstance(node, ast.Call): continue func = node.func if isinstance(func, ast.Attribute) and func.attr in {"append", "insert"}: value = func.value if isinstance(value, ast.Attribute) and value.attr == "path": raise AssertionError(f"{name} 仍在改 sys.path") if isinstance(value, ast.Name) and value.id == "path": raise AssertionError(f"{name} 仍在改 path")