# -*- coding: utf-8 -*- """Unit tests for recent/former endpoint routing.""" import sys from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import pytest PROJECT_ROOT = Path(__file__).resolve().parents[2] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from api.endpoint_routing import plan_calls, recent_boundary TZ = ZoneInfo("Asia/Shanghai") def _now(): return datetime(2025, 12, 18, 10, 0, 0, tzinfo=TZ) def test_recent_boundary_month_start(): b = recent_boundary(_now()) assert b.isoformat() == "2025-09-01T00:00:00+08:00" def test_paylog_routes_to_former_when_old_window(): params = {"siteId": 1, "StartPayTime": "2025-08-01 00:00:00", "EndPayTime": "2025-08-02 00:00:00"} calls = plan_calls("/PayLog/GetPayLogListPage", params, now=_now(), tz=TZ) assert [c.endpoint for c in calls] == ["/PayLog/GetFormerPayLogListPage"] def test_coupon_usage_stays_same_path_even_when_old(): params = {"siteId": 1, "startTime": "2025-08-01 00:00:00", "endTime": "2025-08-02 00:00:00"} calls = plan_calls("/Promotion/GetOfflineCouponConsumePageList", params, now=_now(), tz=TZ) assert [c.endpoint for c in calls] == ["/Promotion/GetOfflineCouponConsumePageList"] def test_goods_outbound_routes_to_queryformer_when_old(): params = {"siteId": 1, "startTime": "2025-08-01 00:00:00", "endTime": "2025-08-02 00:00:00"} calls = plan_calls("/GoodsStockManage/QueryGoodsOutboundReceipt", params, now=_now(), tz=TZ) assert [c.endpoint for c in calls] == ["/GoodsStockManage/QueryFormerGoodsOutboundReceipt"] def test_settlement_records_split_when_crossing_boundary(): params = {"siteId": 1, "rangeStartTime": "2025-08-15 00:00:00", "rangeEndTime": "2025-09-10 00:00:00"} calls = plan_calls("/Site/GetAllOrderSettleList", params, now=_now(), tz=TZ) assert [c.endpoint for c in calls] == ["/Site/GetFormerOrderSettleList", "/Site/GetAllOrderSettleList"] assert calls[0].params["rangeEndTime"] == "2025-09-01 00:00:00" assert calls[1].params["rangeStartTime"] == "2025-09-01 00:00:00" @pytest.mark.parametrize( "endpoint", [ "/PayLog/GetFormerPayLogListPage", "/Site/GetFormerOrderSettleList", "/GoodsStockManage/QueryFormerGoodsOutboundReceipt", ], ) def test_explicit_former_endpoint_not_rerouted(endpoint): params = {"siteId": 1, "startTime": "2025-08-01 00:00:00", "endTime": "2025-08-02 00:00:00"} calls = plan_calls(endpoint, params, now=_now(), tz=TZ) assert [c.endpoint for c in calls] == [endpoint]