LiuMangMang/SmartUp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a1e2c059b92e84c81a41800a8d4d0b42ea28b15f
SmartUp/backend/test_finance_daily_summary.py
T

380 lines
14 KiB
Python
Raw Blame History

"""Tests for finance daily summary service."""
from __future__ import annotations
import json
from datetime import date
from unittest.mock import MagicMock
import httpx
from app.services.finance_service import (
classify_upstream,
date_to_shanghai_timestamps,
fetch_upstream_cost_sub2api,
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
class FakeWebsiteClientCtx:
def __init__(self, response: dict | None = None, raise_exc: Exception | None = None):
self.response = response or {"data": {"total_actual_cost": 123.45}}
self.raise_exc = raise_exc
self.calls: list[dict] = []
def __enter__(self): return self
def __exit__(self, *a): pass
def _request(self, method, path, body=None, params=None):
self.calls.append({"method": method, "path": path, "body": body, "params": params})
if self.raise_exc:
raise self.raise_exc
return self.response
class FakeUpstreamClientCtx:
"""Context manager mock for UpstreamClient."""
def __init__(
self,
token="tok",
user_id="u1",
quota_per_unit=500000,
response: dict | None = None,
raise_exc: Exception | None = None,
):
self._token = token
self._new_api_user = user_id
self._quota_per_unit = quota_per_unit
self.response = response or {"data": {"quota": 5_000_000}}
self.raise_exc = raise_exc
self.calls: list[dict] = []
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 _url(self, path): return f"http://up.test/{path.lstrip('/')}"
def _send_request(self, method, url, **kwargs):
self.calls.append({"method": method, "url": url, **kwargs})
if self.raise_exc:
raise self.raise_exc
return _make_mock_response(self.response)
# ─────────────────────────────────────────────
# 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_cookie_mode():
u = FakeUpstream(
api_prefix="",
auth_type="cookie",
auth_config_json=json.dumps({"cookie_string": "sid=x", "user_id": "7"}),
)
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()
client = FakeWebsiteClientCtx({"data": {"total_actual_cost": 123.45}})
monkeypatch.setattr("app.services.finance_service.Sub2ApiWebsiteClient", lambda **kw: client)
amount, err = fetch_website_revenue(w, date(2026, 7, 2))
assert err is None
assert abs(amount - 123.45) < 1e-6
assert client.calls == [{
"method": "GET",
"path": "/usage/stats",
"body": None,
"params": {
"start_date": "2026-07-02",
"end_date": "2026-07-02",
"timezone": "Asia/Shanghai",
"nocache": "true",
},
}]
def test_website_revenue_nested_data(monkeypatch):
"""total_actual_cost directly in response root (no wrapping data key)."""
w = FakeWebsite()
monkeypatch.setattr(
"app.services.finance_service.Sub2ApiWebsiteClient",
lambda **kw: FakeWebsiteClientCtx({"total_actual_cost": 55.5}),
)
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()
monkeypatch.setattr(
"app.services.finance_service.Sub2ApiWebsiteClient",
lambda **kw: FakeWebsiteClientCtx({"data": {"some_other_field": 99}}),
)
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()
monkeypatch.setattr(
"app.services.finance_service.Sub2ApiWebsiteClient",
lambda **kw: FakeWebsiteClientCtx(raise_exc=Exception("HTTP 403: Forbidden")),
)
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(
"app.services.finance_service.Sub2ApiWebsiteClient",
lambda **kw: FakeWebsiteClientCtx(raise_exc=Exception("connection refused")),
)
amount, err = fetch_website_revenue(w, date(2026, 7, 2))
assert amount == 0.0
assert err is not None
# ─────────────────────────────────────────────
# upstream cost tests
# ─────────────────────────────────────────────
def test_upstream_cost_sub2api_uses_upstream_client(monkeypatch):
u = FakeUpstream(api_prefix="api/v1", auth_type="bearer")
client = FakeUpstreamClientCtx(response={"data": {"total_actual_cost": 42.0}})
monkeypatch.setattr("app.services.finance_service.UpstreamClient", lambda **kw: client)
amount, err = fetch_upstream_cost_sub2api(u, date(2026, 7, 2))
assert err is None
assert abs(amount - 42.0) < 1e-6
assert client.calls[0]["method"] == "GET"
assert client.calls[0]["url"] == "http://up.test/usage/stats"
assert client.calls[0]["params"] == {
"start_date": "2026-07-02",
"end_date": "2026-07-02",
"timezone": "Asia/Shanghai",
}
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)
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
assert mock_client_instance.calls[0]["method"] == "GET"
assert mock_client_instance.calls[0]["url"] == "http://up.test/api/log/stat"
assert mock_client_instance.calls[0]["params"] == {
"type": 2,
"start_timestamp": 1782921600,
"end_timestamp": 1783007999,
}
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(response={"data": {}}),
)
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")
bad_resp = MagicMock()
bad_resp.status_code = 401
bad_resp.text = "Unauthorized"
exc = httpx.HTTPStatusError("401", request=MagicMock(), response=bad_resp)
monkeypatch.setattr(
"app.services.finance_service.UpstreamClient",
lambda **kw: FakeUpstreamClientCtx(raise_exc=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")
def mock_fetch_website(row, _date):
return (50.0, None) if row.id == 1 else (0.0, "timeout")
monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", mock_fetch_website)
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
monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", lambda *_: (100.0, None))
monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", lambda *_: (40.0, None))
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
Reference in New Issue View Git Blame Copy Permalink