feat(t221): add exact-order erp connector

erp-connector/tests/test_client.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import base64
+import json
+from urllib.parse import urlparse
+
+import pytest
+import requests
+
+from shunyunbaoerp import (
+    ERPAuthenticationError,
+    ERPClient,
+    ERPNotFoundError,
+    ERPProtocolError,
+)
+
+
+class FakeResponse:
+    def __init__(
+        self,
+        payload=None,
+        *,
+        status_code: int = 200,
+        content: bytes = b"",
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        self._payload = payload
+        self.status_code = status_code
+        self.content = content
+        self.headers = headers or {"Content-Type": "application/json"}
+
+    def json(self):
+        return self._payload
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise requests.HTTPError(str(self.status_code))
+
+
+class FakeSession:
+    def __init__(self, responses: list[FakeResponse]) -> None:
+        self.responses = list(responses)
+        self.headers: dict[str, str] = {}
+        self.calls: list[dict[str, object]] = []
+        self.closed = False
+
+    def get(self, url: str, **kwargs):
+        self.calls.append({"method": "GET", "url": url, **kwargs})
+        return self.responses.pop(0)
+
+    def request(self, method: str, url: str, **kwargs):
+        self.calls.append({"method": method, "url": url, **kwargs})
+        return self.responses.pop(0)
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def envelope(data, *, status=True, msg="成功", code=None):
+    return {"status": status, "msg": msg, "data": data, "code": code}
+
+
+def fake_jwt(exp: int = 2_000_000_000) -> str:
+    encode = lambda value: base64.urlsafe_b64encode(
+        json.dumps(value, separators=(",", ":")).encode()
+    ).decode().rstrip("=")
+    return f"{encode({'alg': 'none'})}.{encode({'exp': exp, 'username': 'demo'})}.signature"
+
+
+def test_login_keeps_captcha_and_login_in_same_session() -> None:
+    session = FakeSession(
+        [
+            FakeResponse(
+                content=b"image",
+                headers={"Content-Type": "image/png", "Set-Cookie": "omitted"},
+            ),
+            FakeResponse(
+                envelope(
+                    {
+                        "user": {"id": 1, "username": "demo"},
+                        "token": fake_jwt(),
+                    }
+                )
+            ),
+        ]
+    )
+    client = ERPClient(session=session)
+
+    assert client.fetch_captcha() == b"image"
+    user = client.login("demo", "secret", "abcd")
+
+    assert user["id"] == 1
+    assert client.is_logged_in is True
+    assert client.token_expires_at is not None
+    assert urlparse(session.calls[0]["url"]).path == "/api/p/code1"
+    assert urlparse(session.calls[1]["url"]).path == "/am/auth/login"
+    assert session.calls[1]["json"] == {
+        "username": "demo",
+        "password": "secret",
+        "code": "abcd",
+    }
+
+
+def test_login_requires_captcha_from_same_client() -> None:
+    client = ERPClient(session=FakeSession([]))
+
+    with pytest.raises(ERPProtocolError):
+        client.login("demo", "secret", "abcd")
+
+
+def test_query_and_detail_reproduce_har_contract() -> None:
+    stock = {
+        "id": 99001122,
+        "code": "FREIGHT-001",
+        "orderCode": "PLATFORM-001",
+    }
+    detail = {
+        "id": 99001122,
+        "code": "FREIGHT-001",
+        "details": [{"id": 2, "productQty": 1}],
+    }
+    session = FakeSession(
+        [
+            FakeResponse(envelope(1)),
+            FakeResponse(envelope({"total": 1, "list": [stock]})),
+            FakeResponse(envelope({"total": 1, "list": [detail]})),
+        ]
+    )
+    client = ERPClient(session=session)
+
+    result = client.get_freight_details("FREIGHT-001")
+
+    assert result["count"] == 1
+    assert result["records"][0] == {"stock": stock, "detail": detail}
+    assert [urlparse(call["url"]).path for call in session.calls] == [
+        "/am/stock/listTotal",
+        "/am/stock/list",
+        "/am/stock/detail/listByStock",
+    ]
+
+    query = session.calls[0]["json"]["queries"][0]
+    assert query == {
+        "dvalue": "FREIGHT-001",
+        "tableName": "t_stock",
+        "colName": "allcode",
+        "op": 6,
+        "type": 0,
+        "tableAlias": "t",
+        "optType": 1,
+    }
+    assert len(session.calls[0]["json"]["columns"]) == 72
+    assert session.calls[2]["params"] == {"hist": 0}
+    assert session.calls[2]["json"] == {"ids": [99001122]}
+
+
+def test_not_found_stops_before_detail_request() -> None:
+    session = FakeSession([FakeResponse(envelope(0))])
+    client = ERPClient(session=session)
+
+    with pytest.raises(ERPNotFoundError):
+        client.get_freight_details("missing")
+
+    assert len(session.calls) == 1
+
+
+def test_expired_session_becomes_authentication_error() -> None:
+    session = FakeSession(
+        [
+            FakeResponse(
+                envelope(None, status=False, msg="未登录或登录过期", code="-2")
+            )
+        ]
+    )
+    client = ERPClient(session=session)
+
+    with pytest.raises(ERPAuthenticationError):
+        client.query_stock_by_order_number("FREIGHT-001")
+
+
+def test_rejects_unexpected_bulk_match() -> None:
+    session = FakeSession([FakeResponse(envelope(101))])
+    client = ERPClient(session=session, max_matches=100)
+
+    with pytest.raises(ERPProtocolError, match="安全上限"):
+        client.query_stock_by_order_number("too-broad")