headquarter/apps/api/tests/integration/test_users_api.py

import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import io

from fastapi.testclient import TestClient
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.user.user import User


def _postgres_available() -> bool:
    """Check whether a PostgreSQL server is reachable for integration tests."""
    import asyncpg

    async def _check() -> bool:
        try:
            conn = await asyncpg.connect(
                host="localhost", port=5432, user="headquarter", password="headquarter", database="headquarter"
            )
            await conn.close()
            return True
        except Exception:
            return False

    return asyncio.run(_check())


pytestmark = pytest.mark.skipif(
    not _postgres_available(),
    reason="PostgreSQL not available on localhost:5432",
)


def _prepare_users_test_db() -> None:
    async def _run() -> None:
        engine = create_async_engine(
            build_database_url(
                user="headquarter",
                password="headquarter",
                host="localhost",
                port=5432,
                database="headquarter",
            )
        )
        async with engine.begin() as connection:
            await connection.run_sync(Base.metadata.create_all)
            await connection.execute(text("TRUNCATE TABLE refresh_tokens, users RESTART IDENTITY CASCADE"))
        await engine.dispose()

    asyncio.run(_run())


def _load_app():
    import importlib
    import src.database as database_module
    import src.api.user.users as users_module
    import src.main as main_module

    importlib.reload(database_module)
    importlib.reload(users_module)
    importlib.reload(main_module)
    return main_module.app


def _insert_test_user(user_id: str) -> None:
    async def _run() -> None:
        engine = create_async_engine(
            build_database_url(
                user="headquarter",
                password="headquarter",
                host="localhost",
                port=5432,
                database="headquarter",
            )
        )
        async with engine.begin() as connection:
            await connection.run_sync(Base.metadata.create_all)
        from sqlalchemy.ext.asyncio import async_sessionmaker

        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        async with session_factory() as session:
            user = User(
                id=uuid.UUID(user_id),
                email="test@headquarter.local",
                name="Test User",
                authentik_id="test-user",
                avatar_url=None,
            )
            await session.merge(user)
            await session.commit()
        await engine.dispose()

    asyncio.run(_run())


def _create_auth_cookie(user_id: str) -> str:
    settings = Settings()
    return create_session_cookie(
        settings=settings,
        user_id=user_id,
    )


@pytest.mark.integration
def test_get_profile_returns_401_without_cookie() -> None:
    _prepare_users_test_db()
    app = _load_app()

    client = TestClient(app)
    response = client.get("/users/me")

    assert response.status_code == 401


@pytest.mark.integration
def test_get_profile_returns_user_data() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))
    response = client.get("/users/me")

    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "test@headquarter.local"
    assert data["name"] == "Test User"
    assert data["avatar_url"] is None


@pytest.mark.integration
def test_update_profile_changes_name_and_email() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))
    response = client.put("/users/me", json={"name": "Updated Name", "email": "updated@headquarter.local"})

    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Updated Name"
    assert data["email"] == "updated@headquarter.local"


@pytest.mark.integration
def test_update_profile_rejects_empty_name() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))
    response = client.put("/users/me", json={"name": "   "})

    assert response.status_code == 400


@pytest.mark.integration
def test_update_profile_rejects_invalid_email() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))
    response = client.put("/users/me", json={"email": "not-an-email"})

    assert response.status_code == 400


@pytest.mark.integration
def test_upload_avatar_updates_avatar_url() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))

    # Create a simple 1x1 PNG
    png_data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
    response = client.post(
        "/users/me/avatar",
        files={"file": ("test.png", io.BytesIO(png_data), "image/png")},
    )

    assert response.status_code == 200
    data = response.json()
    assert data["avatar_url"] is not None
    assert data["avatar_url"].startswith("/uploads/avatars/")


@pytest.mark.integration
def test_upload_avatar_rejects_invalid_file_type() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))

    response = client.post(
        "/users/me/avatar",
        files={"file": ("test.txt", io.BytesIO(b"not an image"), "text/plain")},
    )

    assert response.status_code == 400


@pytest.mark.integration
def test_upload_avatar_rejects_oversized_file() -> None:
    _prepare_users_test_db()
    user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
    _insert_test_user(user_id)
    app = _load_app()

    client = TestClient(app)
    client.cookies.set("access_token", _create_auth_cookie(user_id))

    large_png = b"\x89PNG\r\n\x1a\n" + b"\x00" * (3 * 1024 * 1024)  # 3MB
    response = client.post(
        "/users/me/avatar",
        files={"file": ("large.png", io.BytesIO(large_png), "image/png")},
    )

    assert response.status_code == 400