100 lines
2.8 KiB
Python
100 lines
2.8 KiB
Python
"""接口 3 Oracle 仓库：INSERT 绑 fid；ORA-00001 映射成 DuplicateFid。不连真库。"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
|
||
from app.inbound.repo import DuplicateFid, RepoUnavailable
|
||
|
||
|
||
def _record():
|
||
return {
|
||
"fid": 200001,
|
||
"oms_id": 20260928000001,
|
||
"supplier_id": "SUP001",
|
||
"supplier_name": "示例供应商",
|
||
"contract_sign_date": date(2026, 9, 28),
|
||
"contract_start_date": date(2026, 10, 1),
|
||
"contract_end_date": date(2027, 9, 30),
|
||
"payment_term": 90,
|
||
"goods_id": "G000123456789",
|
||
"goods_name": "阿莫西林胶囊",
|
||
"reg_no": "国药准字H20201234",
|
||
"spec": "0.25g*24粒",
|
||
"unit": "盒",
|
||
"qty": Decimal("10.0000"),
|
||
"purchase_tax_rate": Decimal("0.1300"),
|
||
"unit_price": Decimal("100.0000"),
|
||
"goods_source_id": "GS001",
|
||
"purchaser_id": "P001",
|
||
"purchaser_name": "张三",
|
||
"planner_id": "PL001",
|
||
"planner_name": "李四",
|
||
"contract_entry_status": None,
|
||
"contract_no": None,
|
||
"org_code": "901",
|
||
"customer_type": 1,
|
||
"goods_mdm_code": "QX0100167583",
|
||
"supplier_mdm_code": "DW0100046361",
|
||
"process_status": 0,
|
||
}
|
||
|
||
|
||
def test_from_env_without_host_raises(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
|
||
from app.inbound.oracle_contract_repo import OraclePurchaseContractRepo
|
||
|
||
with pytest.raises(RepoUnavailable):
|
||
OraclePurchaseContractRepo.from_env()
|
||
|
||
|
||
def test_insert_unique_violation_is_duplicate_fid():
|
||
from app.inbound.oracle_contract_repo import OraclePurchaseContractRepo
|
||
|
||
class FakeError(Exception):
|
||
def __init__(self):
|
||
self.args = (type("Info", (), {"code": 1})(),)
|
||
|
||
class FakeCursor:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *args):
|
||
return False
|
||
|
||
def execute(self, sql, params=None):
|
||
raise FakeError()
|
||
|
||
class FakeConn:
|
||
def cursor(self):
|
||
return FakeCursor()
|
||
|
||
def commit(self):
|
||
raise AssertionError("unique 冲突不应 commit")
|
||
|
||
def rollback(self):
|
||
self.rolled_back = True
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
repo = OraclePurchaseContractRepo(connect=lambda: FakeConn())
|
||
with pytest.raises(DuplicateFid):
|
||
repo.insert(_record())
|
||
|
||
|
||
def test_insert_sql_binds_source_fid_and_pending_status():
|
||
from app.inbound.oracle_contract_repo import _BIND_KEYS, _INSERT_SQL
|
||
|
||
sql = " ".join(_INSERT_SQL.split())
|
||
assert "fid" in sql
|
||
assert ":fid" in sql
|
||
assert "process_status" in sql
|
||
assert "SYSTIMESTAMP" in sql
|
||
assert "fid" in _BIND_KEYS
|
||
assert "oms_id" in _BIND_KEYS
|
||
assert "goods_mdm_code" in _BIND_KEYS
|
||
assert "supplier_mdm_code" in _BIND_KEYS
|