138 lines
5.1 KiB
Python
138 lines
5.1 KiB
Python
# ruff: noqa: PLR2004
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from config import cfg
|
|
from routes import link_codes
|
|
from schemas.enums import LinkCodeStatus
|
|
|
|
|
|
def auth_context(user_id=42):
|
|
return SimpleNamespace(user=SimpleNamespace(id=user_id))
|
|
|
|
|
|
def service_identity():
|
|
return SimpleNamespace(service="telegram-bot")
|
|
|
|
|
|
def test_generate_link_code_requires_authorization(client):
|
|
response = client.post("/link-codes")
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "No authorization provided."
|
|
|
|
|
|
def test_generate_link_code_creates_active_code_with_expiry_and_deep_link(client):
|
|
client.app.dependency_overrides[link_codes.get_auth_context] = auth_context
|
|
created_codes = []
|
|
|
|
async def create_code(session, *, code, user_id, status, expires_at):
|
|
created_codes.append(
|
|
SimpleNamespace(
|
|
session=session,
|
|
code=code,
|
|
user_id=user_id,
|
|
status=status,
|
|
expires_at=expires_at,
|
|
)
|
|
)
|
|
return created_codes[-1]
|
|
|
|
before = datetime.now(UTC)
|
|
with (
|
|
patch(
|
|
"routes.link_codes.secrets.token_urlsafe", return_value="one-time-code"
|
|
) as token_urlsafe,
|
|
patch("routes.link_codes.create_link_code", side_effect=create_code),
|
|
):
|
|
response = client.post("/link-codes")
|
|
after = datetime.now(UTC)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json() == {
|
|
"code": "one-time-code",
|
|
"expires_at": created_codes[0].expires_at.isoformat().replace("+00:00", "Z"),
|
|
"deep_link": f"{cfg.bot_url}?start=one-time-code",
|
|
}
|
|
assert created_codes[0].user_id == 42
|
|
assert created_codes[0].status is LinkCodeStatus.ACTIVE
|
|
token_urlsafe.assert_called_once_with(cfg.link_code_length)
|
|
assert before + timedelta(minutes=cfg.link_code_ttl) <= created_codes[0].expires_at
|
|
assert created_codes[0].expires_at <= after + timedelta(minutes=cfg.link_code_ttl)
|
|
|
|
|
|
def test_consume_link_code_requires_code_and_telegram_id(client):
|
|
client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
|
|
|
|
missing_code = client.post("/link-codes/consume", json={"telegram_id": 12345})
|
|
missing_telegram_id = client.post("/link-codes/consume", json={"code": "link-code"})
|
|
invalid_telegram_id = client.post(
|
|
"/link-codes/consume", json={"code": "link-code", "telegram_id": "not-an-id"}
|
|
)
|
|
|
|
assert missing_code.status_code == 422
|
|
assert missing_telegram_id.status_code == 422
|
|
assert invalid_telegram_id.status_code == 422
|
|
|
|
|
|
def test_consume_link_code_rejects_unknown_code(client):
|
|
client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
|
|
lookup = AsyncMock(return_value=None)
|
|
with patch("routes.link_codes.get_link_code_by_code", lookup):
|
|
response = client.post(
|
|
"/link-codes/consume", json={"code": "missing", "telegram_id": 12345}
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"] == "Code not found"
|
|
lookup.assert_awaited_once()
|
|
assert lookup.await_args.args[1] == "missing"
|
|
|
|
|
|
def test_consume_link_code_rejects_code_for_deleted_user(client):
|
|
client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
|
|
link_code = SimpleNamespace(user_id=42, status=LinkCodeStatus.ACTIVE)
|
|
repository = SimpleNamespace(get_user_by_id=AsyncMock(return_value=None))
|
|
with (
|
|
patch("routes.link_codes.get_link_code_by_code", new=AsyncMock(return_value=link_code)),
|
|
patch("routes.link_codes.UserRepository", return_value=repository),
|
|
):
|
|
response = client.post(
|
|
"/link-codes/consume", json={"code": "orphaned", "telegram_id": 12345}
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"] == "User not found"
|
|
repository.get_user_by_id.assert_awaited_once_with(42)
|
|
|
|
|
|
def test_consume_link_code_updates_telegram_id_and_returns_user(client):
|
|
client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
|
|
link_code = SimpleNamespace(user_id=42, status=LinkCodeStatus.ACTIVE)
|
|
user = SimpleNamespace(
|
|
username="alice", telegram_id=12345, referal_code="ref-code", balance=100
|
|
)
|
|
repository = SimpleNamespace(
|
|
get_user_by_id=AsyncMock(return_value=user),
|
|
update_telegram_id=AsyncMock(return_value=user),
|
|
)
|
|
with (
|
|
patch("routes.link_codes.get_link_code_by_code", new=AsyncMock(return_value=link_code)),
|
|
patch("routes.link_codes.UserRepository", return_value=repository),
|
|
):
|
|
response = client.post(
|
|
"/link-codes/consume", json={"code": "link-code", "telegram_id": 12345}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"username": "alice",
|
|
"telegram_id": 12345,
|
|
"referal_code": "ref-code",
|
|
"bonus_balance": 100,
|
|
}
|
|
repository.get_user_by_id.assert_awaited_once_with(42)
|
|
repository.update_telegram_id.assert_awaited_once_with(user, 12345)
|