122 lines
3.2 KiB
Python
122 lines
3.2 KiB
Python
"""Oracle 仓库：缺配置 503；ORA-00001 映射成 DuplicateOmsId。不连真库。"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("OMS_INBOUND_TOKEN", "test-oms-token")
|
||
|
||
from app.inbound.repo import DuplicateOmsId, RepoUnavailable
|
||
|
||
|
||
def _record():
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
return {
|
||
"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,
|
||
}
|
||
|
||
|
||
def test_from_env_without_user_raises(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
|
||
from app.inbound.oracle_repo import OracleSalesGoodsRepo
|
||
|
||
with pytest.raises(RepoUnavailable):
|
||
OracleSalesGoodsRepo.from_env()
|
||
|
||
|
||
def test_insert_unique_violation_is_duplicate():
|
||
from app.inbound.oracle_repo import OracleSalesGoodsRepo
|
||
|
||
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 = OracleSalesGoodsRepo(connect=lambda: FakeConn())
|
||
with pytest.raises(DuplicateOmsId):
|
||
repo.insert(_record())
|
||
|
||
|
||
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
|
||
|
||
|
||
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
|