social-app/backend/tests/e2e/test_auth_flow.py

from __future__ import annotations

import json
import socket
import threading
import time

from playwright.sync_api import sync_playwright
import uvicorn

from app import app
from v1.auth.dependencies import get_auth_service
from v1.auth.schemas import (
    AuthUser,
    SessionCreateRequest,
    SessionRefreshRequest,
    SessionResponse,
    VerificationCreateRequest,
    VerificationCreateResponse,
    VerificationResendRequest,
    VerificationVerifyRequest,
)
from v1.auth.service import AuthService


class FakeE2EAuthService(AuthService):
    def __init__(self) -> None:
        self._user = AuthUser(id="user-1", email="user@example.com")

    async def create_verification(
        self, request: VerificationCreateRequest
    ) -> VerificationCreateResponse:
        return VerificationCreateResponse(email=request.email)

    async def verify_verification(
        self, request: VerificationVerifyRequest
    ) -> SessionResponse:
        return SessionResponse(
            access_token="access-1",
            refresh_token="refresh-1",
            expires_in=3600,
            token_type="bearer",
            user=self._user,
        )

    async def resend_verification(self, request: VerificationResendRequest) -> None:
        return None

    async def create_session(self, request: SessionCreateRequest) -> SessionResponse:
        return SessionResponse(
            access_token="access-2",
            refresh_token="refresh-2",
            expires_in=3600,
            token_type="bearer",
            user=self._user,
        )

    async def refresh_session(self, request: SessionRefreshRequest) -> SessionResponse:
        return SessionResponse(
            access_token="access-3",
            refresh_token="refresh-3",
            expires_in=3600,
            token_type="bearer",
            user=self._user,
        )

    async def delete_session(self, refresh_token: str | None) -> None:
        return None


def _find_free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def _wait_for_port(host: str, port: int, timeout: float = 5.0) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            if sock.connect_ex((host, port)) == 0:
                return
        time.sleep(0.05)
    raise RuntimeError("Server did not start in time")


def _start_server(host: str, port: int):
    config = uvicorn.Config(app, host=host, port=port, log_level="info")
    server = uvicorn.Server(config)
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    _wait_for_port(host, port)
    return server, thread


def test_auth_flow_e2e() -> None:
    app.dependency_overrides[get_auth_service] = lambda: FakeE2EAuthService()
    host = "127.0.0.1"
    port = _find_free_port()
    server, thread = _start_server(host, port)

    try:
        with sync_playwright() as playwright:
            request_context = playwright.request.new_context(
                base_url=f"http://{host}:{port}"
            )
            try:
                verification = request_context.post(
                    "/api/v1/auth/verifications",
                    data=json.dumps(
                        {
                            "username": "demo",
                            "email": "user@example.com",
                            "password": "secret123",
                        }
                    ),
                    headers={"Content-Type": "application/json"},
                )
                assert verification.status == 202

                verify = request_context.post(
                    "/api/v1/auth/verifications/verify",
                    data=json.dumps(
                        {
                            "email": "user@example.com",
                            "token": "123456",
                        }
                    ),
                    headers={"Content-Type": "application/json"},
                )
                assert verify.status == 200
                assert verify.json()["access_token"] == "access-1"

                login = request_context.post(
                    "/api/v1/auth/sessions",
                    data=json.dumps(
                        {"email": "user@example.com", "password": "secret123"}
                    ),
                    headers={"Content-Type": "application/json"},
                )
                assert login.status == 200
                assert login.json()["access_token"] == "access-2"

                refresh = request_context.post(
                    "/api/v1/auth/sessions/refresh",
                    data=json.dumps({"refresh_token": "refresh-2"}),
                    headers={"Content-Type": "application/json"},
                )
                assert refresh.status == 200
                assert refresh.json()["access_token"] == "access-3"

                logout = request_context.delete(
                    "/api/v1/auth/sessions",
                    data=json.dumps({"refresh_token": "refresh-3"}),
                    headers={"Content-Type": "application/json"},
                )
                assert logout.status == 204
            finally:
                request_context.dispose()
    finally:
        app.dependency_overrides = {}
        server.should_exit = True
        thread.join(timeout=5)