diff --git a/tests/test_link_code_routes.py b/tests/test_link_code_routes.py new file mode 100644 index 0000000..537d873 --- /dev/null +++ b/tests/test_link_code_routes.py @@ -0,0 +1,137 @@ +# 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) + 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) + 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)