132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
# ruff: noqa: PLR2004, PLW0108
|
|
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from routes import users
|
|
|
|
|
|
def auth_context(subscription=None):
|
|
return SimpleNamespace(
|
|
user=SimpleNamespace(
|
|
id=1,
|
|
username="alice",
|
|
telegram_id=12345,
|
|
referal_code="ref-code",
|
|
balance=12.5,
|
|
subscription=subscription,
|
|
)
|
|
)
|
|
|
|
|
|
def test_protected_user_endpoint_requires_authorization(client):
|
|
response = client.get("/users/me")
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_get_me_serializes_auth_context(client):
|
|
client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
|
|
|
|
response = client.get("/users/me")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"username": "alice",
|
|
"telegram_id": 12345,
|
|
"referal_code": "ref-code",
|
|
"bonus_balance": 12.5,
|
|
}
|
|
|
|
|
|
def test_subscription_without_local_subscription_has_empty_details(client):
|
|
client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
|
|
|
|
response = client.get("/users/subscription")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"has_subscription": False,
|
|
"devices": None,
|
|
"expires_at": None,
|
|
"addon_ids": [],
|
|
"subscription_link": None,
|
|
"duration_days": None,
|
|
}
|
|
|
|
|
|
def test_subscription_keeps_local_data_when_external_user_is_missing(client):
|
|
subscription = SimpleNamespace(
|
|
id=2,
|
|
devices=3,
|
|
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
|
|
duration_days=30,
|
|
addons=[SimpleNamespace(addon_id="a"), SimpleNamespace(addon_id="b")],
|
|
)
|
|
client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context(subscription)
|
|
with (
|
|
patch("routes.users.get_sdk", return_value="sdk"),
|
|
patch("routes.users.get_rw_user", new=AsyncMock(return_value=None)),
|
|
):
|
|
response = client.get("/users/subscription")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["has_subscription"] is True
|
|
assert response.json()["addon_ids"] == ["a", "b"]
|
|
assert response.json()["subscription_link"] is None
|
|
|
|
|
|
def test_hwid_returns_empty_list_and_rejects_missing_external_user(client):
|
|
client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
|
|
with (
|
|
patch("routes.users.get_sdk", return_value="sdk"),
|
|
patch("routes.users.get_rw_user", new=AsyncMock(return_value=None)),
|
|
):
|
|
missing_user = client.get("/users/subscription/hwid")
|
|
|
|
rw_user = SimpleNamespace(uuid="uuid")
|
|
with (
|
|
patch("routes.users.get_sdk", return_value="sdk"),
|
|
patch("routes.users.get_rw_user", new=AsyncMock(return_value=rw_user)),
|
|
patch("routes.users.get_hwid_list", new=AsyncMock(return_value=None)),
|
|
):
|
|
empty_list = client.get("/users/subscription/hwid")
|
|
|
|
assert missing_user.status_code == 403
|
|
assert empty_list.status_code == 200
|
|
assert empty_list.json() == []
|
|
|
|
|
|
def test_hwid_maps_client_prefix_and_delete_preserves_external_result(client):
|
|
client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
|
|
rw_user = SimpleNamespace(uuid="uuid")
|
|
device = SimpleNamespace(
|
|
platform="Windows", device_model="PC", user_agent="v2ray/6.0", hwid="abc"
|
|
)
|
|
with (
|
|
patch("routes.users.get_sdk", return_value="sdk"),
|
|
patch("routes.users.get_rw_user", new=AsyncMock(return_value=rw_user)),
|
|
patch("routes.users.get_hwid_list", new=AsyncMock(return_value=[device])),
|
|
patch("routes.users.delete_hwid", new=AsyncMock(return_value=False)),
|
|
):
|
|
devices_response = client.get("/users/subscription/hwid")
|
|
delete_response = client.delete("/users/subscription/hwid?hwid=abc")
|
|
missing_hwid_response = client.delete("/users/subscription/hwid")
|
|
|
|
assert devices_response.json() == [
|
|
{"os": "Windows", "model": "PC", "client": "v2ray", "hwid": "abc"}
|
|
]
|
|
assert delete_response.status_code == 200
|
|
assert delete_response.json() == {"success": False}
|
|
assert missing_hwid_response.status_code == 422
|
|
|
|
|
|
def test_plans_returns_pricing_service_result(client):
|
|
expected = {"device_price": 100, "addons": []}
|
|
with patch("routes.plans.get_pricing_model", new=AsyncMock(return_value=expected)):
|
|
response = client.get("/plans/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == expected
|