"""接口 3 收件契约。内存仓库，不连 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.inbound.memory_repo import MemoryPurchaseContractRepo from app.inbound.repo import RepoUnavailable from app.main import app, get_contract_repo TOKEN = "test-oms-token" PATH = "/api/oms/v1/purchase-contracts" 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 @pytest.fixture def client(): repo = MemoryPurchaseContractRepo() app.dependency_overrides[get_contract_repo] = lambda: repo with TestClient(app) as test_client: yield test_client, repo app.dependency_overrides.clear() 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) def test_missing_token_returns_401(client): test_client, _ = client resp = _post(test_client, {"data": [_item()]}, token=None) assert resp.status_code == 401 assert resp.json()["results"] == [] def test_missing_data_returns_422(client): test_client, repo = client resp = _post(test_client, {}) assert resp.status_code == 422 assert repo.count() == 0 def test_over_100_rows_returns_422(client): test_client, repo = 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 repo.count() == 0 def test_accepts_valid_row(client): test_client, repo = 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 = repo.get(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 def test_existing_fid_is_duplicate_and_not_updated(client): test_client, repo = 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 repo.get(200001)["goods_name"] == "阿莫西林胶囊" def test_duplicate_fid_in_same_batch_is_conflict(client): test_client, repo = 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 repo.count() == 1 def test_retry_protocol_without_contract_no_is_rejected(client): test_client, repo = 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 repo.count() == 0 def test_retry_protocol_with_contract_no_is_accepted(client): test_client, repo = 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 = repo.get(200002) assert stored["contract_entry_status"] == 0 assert stored["contract_no"] == "HT20260928002" def test_end_date_before_start_is_rejected(client): test_client, repo = 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 repo.count() == 0 def test_string_oms_id_is_rejected(client): test_client, repo = client resp = _post(test_client, {"data": [_item(omsId="20260928000001")]}) assert resp.status_code == 200 assert resp.json()["results"][0]["resultCode"] == "REJECTED" assert repo.count() == 0 def test_missing_mdm_is_rejected_and_not_stored(client): test_client, repo = 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 repo.count() == 0 def test_mixed_batch_still_http_200(client): test_client, repo = 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 repo.count() == 2 def test_repo_unavailable_returns_503(client): test_client, _ = client class Boom: def insert(self, record): raise RepoUnavailable("数据库不可用") app.dependency_overrides[get_contract_repo] = lambda: Boom() resp = _post(test_client, {"data": [_item()]}) assert resp.status_code == 503 assert resp.json()["code"] == 503 assert resp.json()["results"] == []