"""Tests for finance daily summary service.""" from __future__ import annotations import json from datetime import date from unittest.mock import MagicMock, patch import httpx import pytest from app.services.finance_service import ( classify_upstream, date_to_shanghai_timestamps, fetch_upstream_cost_new_api, fetch_website_revenue, get_daily_summary, ) # ───────────────────────────────────────────── # Fake model helpers # ───────────────────────────────────────────── class FakeUpstream: def __init__(self, **kw): self.id = kw.get("id", 1) self.name = kw.get("name", "U1") self.base_url = kw.get("base_url", "http://up.test") self.api_prefix = kw.get("api_prefix", "api/v1") self.auth_type = kw.get("auth_type", "bearer") self.auth_config_json = kw.get("auth_config_json", json.dumps({"token": "tok"})) self.timeout_seconds = kw.get("timeout_seconds", 30) self.enabled = True class FakeWebsite: def __init__(self, **kw): self.id = kw.get("id", 1) self.name = kw.get("name", "W1") self.base_url = kw.get("base_url", "http://web.test") self.api_prefix = kw.get("api_prefix", "api/v1/admin") self.auth_type = kw.get("auth_type", "api_key") self.auth_config_json = kw.get("auth_config_json", json.dumps({"key": "k1", "header": "x-api-key"})) self.timeout_seconds = kw.get("timeout_seconds", 30) self.enabled = True # ───────────────────────────────────────────── # classify_upstream tests # ───────────────────────────────────────────── def test_classify_upstream_sub2api(): u = FakeUpstream(api_prefix="api/v1", auth_type="bearer") assert classify_upstream(u) == "sub2api" def test_classify_upstream_new_api_by_token_type(): u = FakeUpstream(api_prefix="", auth_type="new_api_token") assert classify_upstream(u) == "new_api" def test_classify_upstream_nox_token_type(): u = FakeUpstream(api_prefix="", auth_type="nox_token") assert classify_upstream(u) == "new_api" def test_classify_upstream_new_api_by_login_path(): u = FakeUpstream( api_prefix="", auth_type="login_password", auth_config_json=json.dumps({"email": "a@b.com", "password": "p", "login_path": "/api/user/login"}), ) assert classify_upstream(u) == "new_api" def test_classify_upstream_unknown(): u = FakeUpstream(api_prefix="", auth_type="bearer") assert classify_upstream(u) == "unknown" def test_classify_upstream_login_password_without_new_api_path(): u = FakeUpstream( api_prefix="", auth_type="login_password", auth_config_json=json.dumps({"email": "a@b.com", "password": "p"}), ) assert classify_upstream(u) == "unknown" # ───────────────────────────────────────────── # Date/timestamp tests # ───────────────────────────────────────────── def test_date_to_shanghai_timestamps(): start_ts, end_ts = date_to_shanghai_timestamps(date(2026, 7, 2)) # 2026-07-02 00:00:00 CST = 2026-07-01 16:00:00 UTC = 1782921600 # 2026-07-02 23:59:59 CST = 2026-07-02 15:59:59 UTC = 1783007999 assert start_ts == 1782921600 assert end_ts == 1783007999 assert end_ts - start_ts == 86399 # exactly one day minus one second # ───────────────────────────────────────────── # fetch_website_revenue tests # ───────────────────────────────────────────── def _make_mock_response(json_body: dict, status_code: int = 200): mock_resp = MagicMock() mock_resp.status_code = status_code mock_resp.json.return_value = json_body mock_resp.text = json.dumps(json_body) mock_resp.raise_for_status = MagicMock() if status_code >= 400: mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError( "error", request=MagicMock(), response=mock_resp ) return mock_resp def test_website_revenue_success(monkeypatch): w = FakeWebsite() mock_resp = _make_mock_response({"data": {"total_actual_cost": 123.45}}) monkeypatch.setattr("httpx.get", lambda *a, **kw: mock_resp) amount, err = fetch_website_revenue(w, date(2026, 7, 2)) assert err is None assert abs(amount - 123.45) < 1e-6 def test_website_revenue_nested_data(monkeypatch): """total_actual_cost directly in response root (no wrapping data key).""" w = FakeWebsite() mock_resp = _make_mock_response({"total_actual_cost": 55.5}) monkeypatch.setattr("httpx.get", lambda *a, **kw: mock_resp) amount, err = fetch_website_revenue(w, date(2026, 7, 2)) assert err is None assert abs(amount - 55.5) < 1e-6 def test_website_revenue_no_field(monkeypatch): w = FakeWebsite() mock_resp = _make_mock_response({"data": {"some_other_field": 99}}) monkeypatch.setattr("httpx.get", lambda *a, **kw: mock_resp) amount, err = fetch_website_revenue(w, date(2026, 7, 2)) assert amount == 0.0 assert err is not None assert "total_actual_cost" in err def test_website_revenue_http_error(monkeypatch): w = FakeWebsite() bad_resp = MagicMock() bad_resp.status_code = 403 bad_resp.text = "Forbidden" exc = httpx.HTTPStatusError("403 Forbidden", request=MagicMock(), response=bad_resp) monkeypatch.setattr("httpx.get", lambda *a, **kw: (_ for _ in ()).throw(exc)) amount, err = fetch_website_revenue(w, date(2026, 7, 2)) assert amount == 0.0 assert err is not None assert "403" in err def test_website_revenue_connection_error(monkeypatch): w = FakeWebsite() monkeypatch.setattr("httpx.get", lambda *a, **kw: (_ for _ in ()).throw(Exception("connection refused"))) amount, err = fetch_website_revenue(w, date(2026, 7, 2)) assert amount == 0.0 assert err is not None # ───────────────────────────────────────────── # fetch_upstream_cost_new_api tests # ───────────────────────────────────────────── class FakeUpstreamClientCtx: """Context manager mock for UpstreamClient.""" def __init__(self, token="tok", user_id="u1", quota_per_unit=500000): self._token = token self._new_api_user = user_id self._quota_per_unit = quota_per_unit def __enter__(self): return self def __exit__(self, *a): pass def ensure_authenticated(self): pass def _new_api_quota_per_unit(self): return self._quota_per_unit def test_upstream_cost_new_api_success(monkeypatch): u = FakeUpstream(api_prefix="", auth_type="new_api_token") mock_client_instance = FakeUpstreamClientCtx(quota_per_unit=500000) monkeypatch.setattr("app.services.finance_service.UpstreamClient", lambda **kw: mock_client_instance) mock_resp = _make_mock_response({"data": {"quota": 5_000_000}}) monkeypatch.setattr("httpx.get", lambda *a, **kw: mock_resp) amount, err = fetch_upstream_cost_new_api(u, date(2026, 7, 2)) assert err is None assert abs(amount - 10.0) < 1e-6 # 5_000_000 / 500_000 = 10.0 def test_upstream_cost_new_api_missing_quota(monkeypatch): u = FakeUpstream(api_prefix="", auth_type="new_api_token") monkeypatch.setattr("app.services.finance_service.UpstreamClient", lambda **kw: FakeUpstreamClientCtx()) mock_resp = _make_mock_response({"data": {}}) monkeypatch.setattr("httpx.get", lambda *a, **kw: mock_resp) amount, err = fetch_upstream_cost_new_api(u, date(2026, 7, 2)) assert amount == 0.0 assert err is not None assert "quota" in err def test_upstream_cost_new_api_http_error(monkeypatch): u = FakeUpstream(api_prefix="", auth_type="new_api_token") monkeypatch.setattr("app.services.finance_service.UpstreamClient", lambda **kw: FakeUpstreamClientCtx()) bad_resp = MagicMock() bad_resp.status_code = 401 bad_resp.text = "Unauthorized" exc = httpx.HTTPStatusError("401", request=MagicMock(), response=bad_resp) monkeypatch.setattr("httpx.get", lambda *a, **kw: (_ for _ in ()).throw(exc)) amount, err = fetch_upstream_cost_new_api(u, date(2026, 7, 2)) assert amount == 0.0 assert err is not None assert "401" in err # ───────────────────────────────────────────── # get_daily_summary integration-style tests # ───────────────────────────────────────────── def test_get_daily_summary_website_failure_excluded_from_total(monkeypatch): """Failed websites must NOT be counted in total_revenue.""" w_ok = FakeWebsite(id=1, name="OK") w_fail = FakeWebsite(id=2, name="Fail") call_count = {"n": 0} def mock_get(*a, **kw): call_count["n"] += 1 url = a[0] if a else kw.get("url", "") # First call (OK website) returns valid data if call_count["n"] == 1: return _make_mock_response({"data": {"total_actual_cost": 50.0}}) # Second call (Fail website) raises raise Exception("timeout") monkeypatch.setattr("httpx.get", mock_get) result = get_daily_summary([w_ok, w_fail], [], date(2026, 7, 2)) assert abs(result["total_revenue"] - 50.0) < 1e-6 assert result["total_cost"] == 0.0 assert result["failed_count"] == 1 assert result["success"] is False failed = [i for i in result["website_items"] if i["status"] == "failed"] ok = [i for i in result["website_items"] if i["status"] == "success"] assert len(failed) == 1 assert len(ok) == 1 def test_get_daily_summary_unknown_upstream_is_failed(): """Unknown upstream type must produce status='failed', not silently contribute 0.""" u = FakeUpstream(api_prefix="", auth_type="bearer") # unknown type result = get_daily_summary([], [u], date(2026, 7, 2)) assert result["failed_count"] == 1 assert result["success"] is False assert result["upstream_items"][0]["status"] == "failed" assert "未知上游类型" in result["upstream_items"][0]["error"] assert result["total_cost"] == 0.0 # unknown upstreams don't contribute def test_get_daily_summary_all_success(monkeypatch): """When all items succeed, success=True, totals are correct.""" w = FakeWebsite() u = FakeUpstream(api_prefix="api/v1", auth_type="bearer") # sub2api call_count = {"n": 0} def mock_get(*a, **kw): call_count["n"] += 1 if call_count["n"] == 1: return _make_mock_response({"data": {"total_actual_cost": 100.0}}) return _make_mock_response({"data": {"total_actual_cost": 40.0}}) monkeypatch.setattr("httpx.get", mock_get) result = get_daily_summary([w], [u], date(2026, 7, 2)) assert result["success"] is True assert result["failed_count"] == 0 assert abs(result["total_revenue"] - 100.0) < 1e-6 assert abs(result["total_cost"] - 40.0) < 1e-6 assert abs(result["net_income"] - 60.0) < 1e-6 assert abs(result["margin_percent"] - 60.0) < 1e-6