cmroubao/erp-connector/tests/test_client.py

from __future__ import annotations

import base64
import json
from datetime import date
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_created_range_query_reproduces_har_contract_and_deduplicates() -> None:
    stock = {"id": 99001122, "code": "FREIGHT-001"}
    detail = {"id": 99001122, "details": []}
    session = FakeSession(
        [
            FakeResponse(envelope(2)),
            FakeResponse(envelope({"total": 2, "list": [stock, stock]})),
            FakeResponse(envelope({"total": 1, "list": [detail]})),
        ]
    )
    client = ERPClient(session=session)

    result = client.get_freight_details_by_created_range(
        date(2026, 7, 22),
        date(2026, 7, 28),
    )

    assert result["count"] == 1
    assert result["query"] == {
        "mode": "CREATED_RANGE",
        "createdFrom": "2026-07-22",
        "createdTo": "2026-07-28",
    }
    assert session.calls[0]["json"]["queries"] == [
        {
            "dvalue": "2026-07-22,2026-07-28",
            "tableName": "t_stock",
            "colName": "created",
            "op": 0,
            "type": 3,
            "tableAlias": "t",
            "optType": 0,
        }
    ]


def test_created_range_rejects_more_than_seven_inclusive_days() -> None:
    client = ERPClient(session=FakeSession([]))

    with pytest.raises(ValueError, match="7 天"):
        client.query_stock_by_created_range("2026-07-21", "2026-07-28")


def test_empty_created_range_does_not_request_details() -> None:
    session = FakeSession([FakeResponse(envelope(0))])
    client = ERPClient(session=session)

    result = client.get_freight_details_by_created_range(
        "2026-07-28",
        "2026-07-28",
    )

    assert result["records"] == []
    assert len(session.calls) == 1


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")