harden telegram webapp production readiness

tests/conftest.py

@@ -0,0 +1,71 @@
+import hashlib
+import hmac
+import json
+import time
+from collections.abc import AsyncGenerator
+from urllib.parse import urlencode
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.api.deps import get_current_telegram_user
+from app.core.config import settings
+from app.db.base import Base
+from app.db.session import get_session
+from app.main import app
+from app.models import car, expense, push, user  # noqa: F401
+
+TEST_BOT_TOKEN = "123456:test-token"
+TEST_INTERNAL_TOKEN = "internal-test-token"
+
+
+def make_init_data(telegram_id: int, first_name: str = "Test") -> str:
+    user_payload = json.dumps(
+        {"id": telegram_id, "first_name": first_name, "username": str(telegram_id)},
+        separators=(",", ":"),
+    )
+    values = {"auth_date": str(int(time.time())), "user": user_payload}
+    data_check_string = "\n".join(f"{key}={values[key]}" for key in sorted(values))
+    secret = hmac.new(b"WebAppData", TEST_BOT_TOKEN.encode(), hashlib.sha256).digest()
+    values["hash"] = hmac.new(secret, data_check_string.encode(), hashlib.sha256).hexdigest()
+    return urlencode(values)
+
+
+@pytest.fixture(autouse=True)
+def configure_settings() -> None:
+    settings.bot_token = TEST_BOT_TOKEN
+    settings.internal_api_token = TEST_INTERNAL_TOKEN
+    settings.app_env = "test"
+    settings.allow_dev_auth = False
+    yield
+
+
+@pytest.fixture()
+async def client() -> AsyncGenerator[AsyncClient, None]:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    session_factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    async def override_session() -> AsyncGenerator:
+        async with session_factory() as session:
+            yield session
+
+    app.dependency_overrides[get_session] = override_session
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as test_client:
+        yield test_client
+    app.dependency_overrides.pop(get_session, None)
+    app.dependency_overrides.pop(get_current_telegram_user, None)
+    await engine.dispose()
+
+
+@pytest.fixture()
+def auth_headers() -> dict[str, str]:
+    return {"X-Telegram-Init-Data": make_init_data(1001)}
+
+
+@pytest.fixture()
+def other_auth_headers() -> dict[str, str]:
+    return {"X-Telegram-Init-Data": make_init_data(2002)}

tests/test_auth.py

@@ -0,0 +1,30 @@
+import pytest
+from conftest import TEST_BOT_TOKEN, make_init_data
+
+from app.core.config import Settings
+from app.services.telegram_auth import verify_webapp_init_data
+
+
+def test_telegram_init_data_auth() -> None:
+    values = verify_webapp_init_data(make_init_data(42), TEST_BOT_TOKEN)
+
+    assert values["id"] == 42
+
+
+def test_cors_config_reads_csv() -> None:
+    settings = Settings(
+        bot_token="token",
+        cors_origins="https://drivers.smartsoltech.kr,https://t.me",
+    )
+
+    assert settings.cors_origin_list == ["https://drivers.smartsoltech.kr", "https://t.me"]
+
+
+@pytest.mark.asyncio
+async def test_user_cannot_get_foreign_car(client, auth_headers, other_auth_headers) -> None:
+    created = await client.post("/api/cars", headers=auth_headers, json={"name": "Owner car"})
+    car_id = created.json()["id"]
+
+    response = await client.get(f"/api/cars/{car_id}", headers=other_auth_headers)
+
+    assert response.status_code == 403

tests/test_entries.py

@@ -0,0 +1,89 @@
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_user_cannot_add_fuel_to_foreign_car(client, auth_headers, other_auth_headers) -> None:
+    created = await client.post("/api/cars", headers=auth_headers, json={"name": "Owner car"})
+    car_id = created.json()["id"]
+
+    response = await client.post(
+        "/api/fuel",
+        headers=other_auth_headers,
+        json={
+            "car_id": car_id,
+            "entry_date": "2026-05-12",
+            "odometer": 1000,
+            "liters": 30,
+            "price_per_liter": 2,
+        },
+    )
+
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_fuel_crud(client, auth_headers) -> None:
+    car = (await client.post("/api/cars", headers=auth_headers, json={"name": "Fuel car"})).json()
+    created = await client.post(
+        "/api/fuel",
+        headers=auth_headers,
+        json={
+            "car_id": car["id"],
+            "entry_date": "2026-05-12",
+            "odometer": 1000,
+            "liters": 30,
+            "price_per_liter": 2,
+        },
+    )
+    assert created.status_code == 201
+    entry_id = created.json()["id"]
+
+    patched = await client.patch(
+        f"/api/fuel/{entry_id}",
+        headers=auth_headers,
+        json={"liters": 35, "price_per_liter": 3},
+    )
+
+    assert patched.status_code == 200
+    assert patched.json()["total_cost"] == "105.00"
+    deleted = await client.delete(f"/api/fuel/{entry_id}", headers=auth_headers)
+    assert deleted.status_code == 204
+
+
+@pytest.mark.asyncio
+async def test_service_crud(client, auth_headers) -> None:
+    car = (await client.post("/api/cars", headers=auth_headers, json={"name": "Service car"})).json()
+    created = await client.post(
+        "/api/service",
+        headers=auth_headers,
+        json={
+            "car_id": car["id"],
+            "entry_date": "2026-05-12",
+            "service_type": "maintenance",
+            "title": "Oil",
+            "total_cost": 100,
+        },
+    )
+    assert created.status_code == 201
+    entry_id = created.json()["id"]
+
+    patched = await client.patch(
+        f"/api/service/{entry_id}",
+        headers=auth_headers,
+        json={"title": "Oil and filter", "next_due_odometer": 2000},
+    )
+
+    assert patched.status_code == 200
+    assert patched.json()["title"] == "Oil and filter"
+    deleted = await client.delete(f"/api/service/{entry_id}", headers=auth_headers)
+    assert deleted.status_code == 204
+
+
+@pytest.mark.asyncio
+async def test_stats_do_not_fail_with_insufficient_data(client, auth_headers) -> None:
+    car = (await client.post("/api/cars", headers=auth_headers, json={"name": "Stats car"})).json()
+
+    response = await client.get(f"/api/cars/{car['id']}/stats", headers=auth_headers)
+
+    assert response.status_code == 200
+    assert response.json()["avg_consumption_l_per_100km"] is None