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

from __future__ import annotations

import json
import socket
import threading
import time
from uuid import UUID

from playwright.sync_api import sync_playwright
import uvicorn

from app import app
from core.auth.models import CurrentUser
from schemas.shared.user import UserContext
from v1.users.dependencies import get_current_user, get_user_service
from v1.users.schemas import UserUpdateRequest


class FakeUserService:
    """Fake service for E2E testing."""

    def __init__(self, user: UserContext) -> None:
        self._user = user

    async def get_me(self) -> UserContext:
        return self._user

    async def update_me(self, update: UserUpdateRequest) -> UserContext:
        return UserContext(
            id=self._user.id,
            username=(
                update.username if update.username is not None else self._user.username
            ),
            avatar_url=(
                update.avatar_url
                if update.avatar_url is not None
                else self._user.avatar_url
            ),
            bio=update.bio if update.bio is not None else self._user.bio,
        )


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_profile_flow_e2e() -> None:
    user_id = UUID("00000000-0000-0000-0000-000000000001")
    user = UserContext(
        id=str(user_id),
        username="demo",
        avatar_url=None,
        bio=None,
    )
    app.dependency_overrides[get_user_service] = lambda: FakeUserService(user)  # type: ignore[return-value]
    app.dependency_overrides[get_current_user] = lambda: CurrentUser(id=user_id)

    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:
                me = request_context.get("/api/v1/users/me")
                assert me.status == 200
                assert me.json()["username"] == "demo"

                updated = request_context.patch(
                    "/api/v1/users/me",
                    data=json.dumps({"username": "updated"}),
                    headers={"Content-Type": "application/json"},
                )
                assert updated.status == 200
                assert updated.json()["username"] == "updated"
            finally:
                request_context.dispose()
    finally:
        app.dependency_overrides = {}
        server.should_exit = True
        thread.join(timeout=5)