# 接口 3 收件契约。假连接录 SQL，不连 Oracle。按对方 fid 幂等。 from __future__ import annotations import os import pytest from fastapi.testclient import TestClient os.environ["OMS_INBOUND_TOKEN"] = "test-oms-token" from app.main import app, get_pool from tests.fake_oracle import FakePool, ora_error TOKEN = "test-oms-token" PATH = "/api/oms/v1/purchase-contracts" # 构造一条合法的接口 3 明细，overrides 覆盖字段。 def _item(**overrides): row = { "fid": 200001, "omsId": 20260928000001, "supplierId": "SUP001", "supplierName": "示例供应商", "contractSignDate": "2026-09-28", "contractStartDate": "2026-10-01", "contractEndDate": "2027-09-30", "paymentTerm": 90, "goodsId": "G000123456789", "goodsName": "阿莫西林胶囊", "regNo": "国药准字H20201234", "spec": "0.25g*24粒", "unit": "盒", "qty": "10.0000", "purchaseTaxRate": "0.13", "unitPrice": "100.0000", "goodsSourceId": "GS001", "purchaserId": "P001", "purchaserName": "张三", "plannerId": "PL001", "plannerName": "李四", "contractEntryStatus": None, "contractNo": None, "orgCode": "901", "customerType": 1, "goodsMdmCode": "QX0100167583", "supplierMdmCode": "DW0100046361", } row.update(overrides) return row # 注入 FakePool，整次请求走同一块假库。 @pytest.fixture def client(): pool = FakePool() app.dependency_overrides[get_pool] = lambda: pool with TestClient(app) as test_client: yield test_client, pool app.dependency_overrides.clear() # 带或不带 X-Oms-Token 发 POST。 def _post(client, body, token=TOKEN): headers = {"X-Oms-Token": token} if token is not None else {} return client.post(PATH, json=body, headers=headers) # 缺 X-Oms-Token 时接口返回 401。 def test_missing_token_returns_401(client): test_client, pool = client resp = _post(test_client, {"data": [_item()]}, token=None) assert resp.status_code == 401 assert resp.json()["results"] == [] assert pool.db.audits # 缺 data 节点时整包 422，一条都不落。 def test_missing_data_returns_422(client): test_client, pool = client resp = _post(test_client, {}) assert resp.status_code == 422 assert pool.db.contracts == {} # 超过 100 条时整包 422，假库仍空。 def test_over_100_rows_returns_422(client): test_client, pool = client rows = [_item(fid=i, omsId=i) for i in range(1, 102)] resp = _post(test_client, {"data": rows}) assert resp.status_code == 422 assert pool.db.contracts == {} # 合法一行 ACCEPTED，process_status=0，fid 原样落库。 def test_accepts_valid_row(client): test_client, pool = client resp = _post(test_client, {"data": [_item()]}) assert resp.status_code == 200 body = resp.json() assert body["code"] == 200 assert body["results"] == [ { "fid": 200001, "resultCode": "ACCEPTED", "message": "采购合同数据已接收，等待处理", } ] stored = pool.db.contracts[200001] assert stored["process_status"] == 0 assert stored["oms_id"] == 20260928000001 assert stored["supplier_id"] == "SUP001" assert stored["goods_mdm_code"] == "QX0100167583" assert stored["contract_entry_status"] is None assert stored["contract_no"] is None # 跨请求同 fid：DUPLICATE，库里品名不改。 def test_existing_fid_is_duplicate_and_not_updated(client): test_client, pool = client _post(test_client, {"data": [_item()]}) resp = _post(test_client, {"data": [_item(goodsName="不应覆盖")]}) assert resp.status_code == 200 result = resp.json()["results"][0] assert result["resultCode"] == "DUPLICATE" assert result["message"] == "序号已存在，未重复入库" assert pool.db.contracts[200001]["goods_name"] == "阿莫西林胶囊" # 同一 POST 里两条相同 fid：后一条 CONFLICT，只落第一条。 def test_duplicate_fid_in_same_batch_is_conflict(client): test_client, pool = client resp = _post( test_client, {"data": [_item(), _item(omsId=20260928000002)]}, ) assert resp.status_code == 200 results = resp.json()["results"] assert results[0]["resultCode"] == "ACCEPTED" assert results[1]["resultCode"] == "CONFLICT" assert results[1]["fid"] == 200001 assert len(pool.db.contracts) == 1 # contractEntryStatus=0 却没有合同号时 REJECTED。 def test_retry_protocol_without_contract_no_is_rejected(client): test_client, pool = client resp = _post(test_client, {"data": [_item(contractEntryStatus=0, contractNo=None)]}) assert resp.status_code == 200 result = resp.json()["results"][0] assert result["resultCode"] == "REJECTED" assert "contractNo" in result["message"] assert pool.db.contracts == {} # contractEntryStatus=0 且带合同号时 ACCEPTED。 def test_retry_protocol_with_contract_no_is_accepted(client): test_client, pool = client resp = _post( test_client, {"data": [_item(fid=200002, contractEntryStatus=0, contractNo="HT20260928002")]}, ) assert resp.status_code == 200 assert resp.json()["results"][0]["resultCode"] == "ACCEPTED" stored = pool.db.contracts[200002] assert stored["contract_entry_status"] == 0 assert stored["contract_no"] == "HT20260928002" # 结束日早于开始日时 REJECTED。 def test_end_date_before_start_is_rejected(client): test_client, pool = client resp = _post( test_client, {"data": [_item(contractStartDate="2027-09-30", contractEndDate="2026-10-01")]}, ) assert resp.status_code == 200 assert resp.json()["results"][0]["resultCode"] == "REJECTED" assert pool.db.contracts == {} # JSON 字符串 omsId 不转整数，标 REJECTED。 def test_string_oms_id_is_rejected(client): test_client, pool = client resp = _post(test_client, {"data": [_item(omsId="20260928000001")]}) assert resp.status_code == 200 assert resp.json()["results"][0]["resultCode"] == "REJECTED" assert pool.db.contracts == {} # 缺货品 MDM 码 REJECTED，不入库。 def test_missing_mdm_is_rejected_and_not_stored(client): test_client, pool = client resp = _post(test_client, {"data": [_item(goodsMdmCode="")]}) assert resp.status_code == 200 result = resp.json()["results"][0] assert result["resultCode"] == "REJECTED" assert "goodsMdmCode" in result["message"] assert pool.db.contracts == {} # 混批 DUPLICATE + ACCEPTED + REJECTED 时 HTTP 仍 200。 def test_mixed_batch_still_http_200(client): test_client, pool = client _post(test_client, {"data": [_item(fid=1, omsId=1)]}) resp = _post( test_client, { "data": [ _item(fid=1, omsId=1), _item(fid=2, omsId=2), _item(fid=3, omsId=3, orgCode=""), ] }, ) assert resp.status_code == 200 codes = [r["resultCode"] for r in resp.json()["results"]] assert codes == ["DUPLICATE", "ACCEPTED", "REJECTED"] assert set(pool.db.contracts) == {1, 2} # 库故障时 HTTP 503，body 不含 ORA。 def test_repo_unavailable_returns_503(client): test_client, pool = client pool.db.raise_execute = ora_error( 12170, "ORA-12170: TNS:Connect timeout occurred at rpa_purchase_contract_data", ) resp = _post(test_client, {"data": [_item()]}) assert resp.status_code == 503 assert resp.json()["code"] == 503 assert resp.json()["msg"] == "数据库不可用" assert resp.json()["results"] == [] assert "ORA-" not in resp.text assert "rpa_purchase_contract_data" not in resp.text # 一批两条合法明细只从池里借一次连接。 def test_batch_two_rows_acquires_once(client): test_client, pool = client resp = _post( test_client, {"data": [_item(fid=1, omsId=1), _item(fid=2, omsId=2)]}, ) assert resp.status_code == 200 assert [r["resultCode"] for r in resp.json()["results"]] == ["ACCEPTED", "ACCEPTED"] assert pool.acquires == 1 many = [c for c in pool.db.calls if c[0] == "executemany"] assert len(many) == 1