AutoRpa/rpa_oms/tests/test_oracle_repo.py

# 接口 1 Oracle 仓库：缺配置 503；按约束名认重复；一批 executemany。不连真库。

from __future__ import annotations

from datetime import date
from decimal import Decimal

import pytest

from app.inbound.repo import RepoUnavailable
from tests.fake_oracle import FakeDb, ora_error


# 构造一条表 1 落库记录。
def _record(**overrides):
    row = {
        "oms_id": 2026092801,
        "bar_code": "06901234567890",
        "order_date": date(2026, 9, 28),
        "sale_price": Decimal("128.5000"),
        "sale_qty": 10,
        "plan_purchase_qty": 10,
        "goods_id": "G000123456789",
        "goods_name": "阿莫西林胶囊",
        "goods_spec": "0.25g*24粒",
        "manufacturer": "示例制药有限公司",
        "approval_no": "国药准字H20201234",
        "unit": "盒",
        "customer_id": "C00010001",
        "customer_name": "示例医院",
        "org_code": "907",
        "goods_mdm_code": "QX0100167583",
        "customer_mdm_code": "DW0100079727",
        "deal_flag": 0,
    }
    row.update(overrides)
    return row


# 并发首个请求只建一个池。
def test_get_pool_creates_once_under_concurrency(monkeypatch):
    import threading
    import time

    import app.main as main

    created: list[object] = []

    # 拖住建池，让多个线程同时穿过「还没有池」的窗口。
    def fake_create_pool(*, role: str = "oms"):
        time.sleep(0.05)
        pool = object()
        created.append(pool)
        return pool

    monkeypatch.setattr("rpa_oracle.pool.create_pool", fake_create_pool)
    main._pool = None
    barrier = threading.Barrier(8)
    errors: list[BaseException] = []

    # 同时打 get_pool。
    def worker() -> None:
        try:
            barrier.wait()
            main.get_pool()
        except BaseException as exc:
            errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    try:
        assert errors == []
        assert len(created) == 1
        assert main._pool is created[0]
    finally:
        main._pool = None


# 缺用户时 get_pool 抛「数据库不可用」，文案不含缺项名单。
def test_from_env_without_user_raises(monkeypatch):
    monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
    monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
    import app.main as main

    main._pool = None
    with pytest.raises(RepoUnavailable) as caught:
        main.get_pool()
    assert caught.value.message == "数据库不可用"
    assert "RPA_ORACLE" not in str(caught.value)


# uk_rpa_sgmd_oms_id 冲突时 insert_many 把该 oms_id 放进重复集合。
def test_insert_unique_violation_is_duplicate():
    from app.inbound.oracle_repo import OracleSalesGoodsRepo

    db = FakeDb()
    db.collide_sales.add(2026092801)
    repo = OracleSalesGoodsRepo(db.connection())
    collided = repo.insert_many([_record()])
    assert collided == {2026092801}


# 未知唯一约束不当成 oms_id 重复，抛「数据库不可用」。
def test_unknown_unique_constraint_raises_unavailable():
    from app.inbound.oracle_repo import OracleSalesGoodsRepo

    db = FakeDb()
    db.collide_sales.add(2026092801)
    db.collide_constraint = "UK_SOMETHING_ELSE"
    repo = OracleSalesGoodsRepo(db.connection())
    with pytest.raises(RepoUnavailable) as caught:
        repo.insert_many([_record()])
    assert caught.value.message == "数据库不可用"


# 待插入多行时先 SELECT IN 再一次 executemany。
def test_insert_many_uses_executemany_and_select_in():
    from app.inbound.oracle_repo import OracleSalesGoodsRepo

    db = FakeDb()
    conn = db.connection()
    repo = OracleSalesGoodsRepo(conn)
    existing = repo.existing_oms_ids([1, 2])
    assert existing == set()
    collided = repo.insert_many([_record(oms_id=1), _record(oms_id=2)])
    assert collided == set()
    assert set(db.sales) == {1, 2}
    selects = [c for c in db.calls if c[0] == "execute" and "SELECT" in c[1].upper()]
    assert any("IN" in c[1].upper() for c in selects)
    many = [c for c in db.calls if c[0] == "executemany"]
    assert len(many) == 1
    assert many[0][3] is True


# INSERT 仍包含选品班要读的列。
def test_insert_sql_includes_select_job_columns():
    from app.inbound.oracle_repo import _BIND_KEYS, _INSERT_SQL

    sql = " ".join(_INSERT_SQL.split())
    for column in (
        "oms_id",
        "bar_code",
        "sale_price",
        "plan_purchase_qty",
        "goods_name",
        "goods_spec",
        "manufacturer",
        "approval_no",
        "org_code",
        "deal_flag",
    ):
        assert column in sql
    for key in (
        "oms_id",
        "bar_code",
        "sale_price",
        "plan_purchase_qty",
        "org_code",
    ):
        assert key in _BIND_KEYS
    assert "deal_flag" in sql


# 创建人/更新人固定 0 / 系统管理员，时间用 SYSDATE。
def test_insert_sql_sets_system_auditor():
    from app.inbound.oracle_repo import _INSERT_SQL

    sql = " ".join(_INSERT_SQL.split())
    assert "create_user_id" in sql
    assert "create_user_name" in sql
    assert "update_time" in sql
    assert "update_user_id" in sql
    assert "update_user_name" in sql
    assert "SYSDATE, 0, '系统管理员'" in sql
    assert sql.count("SYSDATE") == 2


# 非唯一约束的 ORA 对外仍是「数据库不可用」，不带原文。
def test_other_ora_error_message_is_fixed():
    from app.inbound.oracle_repo import OracleSalesGoodsRepo

    db = FakeDb()
    db.raise_execute = ora_error(12170, "ORA-12170: TNS:Connect timeout occurred")
    repo = OracleSalesGoodsRepo(db.connection())
    with pytest.raises(RepoUnavailable) as caught:
        repo.existing_oms_ids([1])
    assert caught.value.message == "数据库不可用"
    assert "ORA-" not in str(caught.value)