bambuddy/backend/tests/integration/test_auth_api.py

"""Integration tests for Authentication API endpoints.

Tests the full request/response cycle for /api/v1/auth/ and /api/v1/users/ endpoints.
"""

import pytest
from httpx import AsyncClient


class TestAuthStatusAPI:
    """Integration tests for /api/v1/auth/status endpoint."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_auth_status_disabled(self, async_client: AsyncClient):
        """Verify auth status returns disabled when not configured."""
        response = await async_client.get("/api/v1/auth/status")

        assert response.status_code == 200
        result = response.json()
        assert "auth_enabled" in result
        assert result["auth_enabled"] is False
        assert result["requires_setup"] is True


class TestAuthSetupAPI:
    """Integration tests for /api/v1/auth/setup endpoint."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_setup_auth_disabled(self, async_client: AsyncClient):
        """Verify auth can be set up with auth disabled (no password required)."""
        response = await async_client.post(
            "/api/v1/auth/setup",
            json={"auth_enabled": False},
        )

        assert response.status_code == 200
        result = response.json()
        assert result["auth_enabled"] is False
        assert result["admin_created"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_setup_auth_enabled_requires_credentials(self, async_client: AsyncClient):
        """Verify enabling auth requires admin username and password."""
        response = await async_client.post(
            "/api/v1/auth/setup",
            json={"auth_enabled": True},
        )

        assert response.status_code == 400
        assert "Admin username and password are required" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_setup_auth_enabled_with_credentials(self, async_client: AsyncClient):
        """Verify auth can be enabled with admin credentials."""
        response = await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "testadmin",
                "admin_password": "TestPass1!",
            },
        )

        assert response.status_code == 200
        result = response.json()
        assert result["auth_enabled"] is True
        assert result["admin_created"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_setup_weak_password_rejected_when_creating_new_admin(self, async_client: AsyncClient):
        """Complexity is enforced only when a new admin is being created."""
        response = await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "weakpw_admin",
                "admin_password": "NoSpecial1",
            },
        )

        assert response.status_code == 400
        assert "special character" in response.json()["detail"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_setup_reenable_with_existing_admin_ignores_password(self, async_client: AsyncClient, db_session):
        """Re-enabling auth when an admin already exists must not reject the placeholder
        password the frontend still sends. Regression for the LDAP re-enable flow that
        previously 422'd because the Pydantic schema enforced complexity unconditionally.
        """
        from backend.app.core.auth import get_password_hash
        from backend.app.models.user import User

        existing = User(
            username="existing_admin",
            # pragma: allowlist secret — test fixture only, not a real credential
            password_hash=get_password_hash("DoesNotMatter1!"),  # noqa: S106
            role="admin",
            is_active=True,
        )
        db_session.add(existing)
        await db_session.commit()

        response = await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "irrelevant",
                "admin_password": "NoSpecial1",
            },
        )

        assert response.status_code == 200
        result = response.json()
        assert result["auth_enabled"] is True
        assert result["admin_created"] is False


class TestAuthLoginAPI:
    """Integration tests for /api/v1/auth/login endpoint."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_login_auth_disabled(self, async_client: AsyncClient):
        """Verify login fails when auth is not enabled."""
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "admin", "password": "password"},
        )

        assert response.status_code == 400
        assert "Authentication is not enabled" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_login_success(self, async_client: AsyncClient):
        """Verify login succeeds with valid credentials after setup."""
        # First enable auth
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "logintest",
                "admin_password": "LoginPass1!",
            },
        )

        # Now login
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "logintest", "password": "LoginPass1!"},
        )

        assert response.status_code == 200
        result = response.json()
        assert "access_token" in result
        assert result["token_type"] == "bearer"
        assert result["user"]["username"] == "logintest"
        assert result["user"]["role"] == "admin"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_login_invalid_credentials(self, async_client: AsyncClient):
        """Verify login fails with invalid credentials."""
        # First enable auth
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "invalidtest",
                "admin_password": "CorrectPass1!",
            },
        )

        # Try login with wrong password
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "invalidtest", "password": "wrongpassword"},
        )

        assert response.status_code == 401
        assert "Incorrect username or password" in response.json()["detail"]


class TestAuthMeAPI:
    """Integration tests for /api/v1/auth/me endpoint."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_without_token(self, async_client: AsyncClient):
        """Verify /me fails without authentication token."""
        response = await async_client.get("/api/v1/auth/me")

        assert response.status_code == 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_with_valid_token(self, async_client: AsyncClient):
        """Verify /me returns user info with valid token."""
        # Setup and login
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "metest",
                "admin_password": "MePass1!",
            },
        )

        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "metest", "password": "MePass1!"},
        )
        token = login_response.json()["access_token"]

        # Get current user
        response = await async_client.get(
            "/api/v1/auth/me",
            headers={"Authorization": f"Bearer {token}"},
        )

        assert response.status_code == 200
        result = response.json()
        assert result["username"] == "metest"
        assert result["role"] == "admin"
        assert result["is_active"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_with_ownerless_api_key_bearer(self, async_client: AsyncClient, db_session):
        """A legacy key has no identity to report, but no longer claims admin (#1894)."""
        from backend.app.core.auth import generate_api_key
        from backend.app.models.api_key import APIKey

        # Create an API key directly in the database
        full_key, key_hash, key_prefix = generate_api_key()
        api_key = APIKey(name="test-kiosk", key_hash=key_hash, key_prefix=key_prefix, enabled=True)
        db_session.add(api_key)
        await db_session.commit()

        # Call /me with the API key as Bearer token
        response = await async_client.get(
            "/api/v1/auth/me",
            headers={"Authorization": f"Bearer {full_key}"},
        )

        assert response.status_code == 200
        result = response.json()
        assert result["id"] == 0
        assert result["username"].startswith("api-key:")
        assert result["role"] != "admin"
        assert result["is_admin"] is False
        assert result["is_active"] is True
        # can_read_status defaults True, so the scope-derived set is non-empty
        # -- but it is a set, not "every permission there is".
        assert len(result["permissions"]) > 0
        assert "users:create" not in result["permissions"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_with_ownerless_api_key_header(self, async_client: AsyncClient, db_session):
        """Same as above via the X-API-Key header rather than Bearer."""
        from backend.app.core.auth import generate_api_key
        from backend.app.models.api_key import APIKey

        full_key, key_hash, key_prefix = generate_api_key()
        api_key = APIKey(name="test-kiosk-header", key_hash=key_hash, key_prefix=key_prefix, enabled=True)
        db_session.add(api_key)
        await db_session.commit()

        response = await async_client.get(
            "/api/v1/auth/me",
            headers={"X-API-Key": full_key},
        )

        assert response.status_code == 200
        result = response.json()
        assert result["id"] == 0
        assert result["username"].startswith("api-key:")
        assert result["is_admin"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_with_invalid_api_key(self, async_client: AsyncClient):
        """Verify /me rejects invalid API key."""
        response = await async_client.get(
            "/api/v1/auth/me",
            headers={"Authorization": "Bearer bb_invalid_key_value"},
        )

        assert response.status_code == 401

    async def _owned_key(self, async_client: AsyncClient, db_session, **scopes):
        """Set up auth and return (owner, full_key) for a key with ``scopes``.

        The owner is given an email and a group explicitly rather than relying
        on what /auth/setup happens to seed, so the assertions about what /me
        withholds cannot pass vacuously.
        """
        from sqlalchemy import select
        from sqlalchemy.orm import selectinload

        from backend.app.core.auth import generate_api_key
        from backend.app.models.api_key import APIKey
        from backend.app.models.group import Group
        from backend.app.models.user import User

        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "keyowner",
                "admin_password": "KeyPass1!",
            },
        )
        owner = (
            await db_session.execute(select(User).where(User.username == "keyowner").options(selectinload(User.groups)))
        ).scalar_one()
        owner.email = "keyowner@example.invalid"
        group = Group(name="key-owner-group", description="t", permissions=["printers:read"], is_system=False)
        db_session.add(group)
        await db_session.flush()
        owner.groups.append(group)

        full_key, key_hash, key_prefix = generate_api_key()
        db_session.add(
            APIKey(name="owned", key_hash=key_hash, key_prefix=key_prefix, enabled=True, user_id=owner.id, **scopes)
        )
        await db_session.commit()
        return owner, full_key

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_reports_the_key_owner_not_a_synthetic_admin(self, async_client: AsyncClient, db_session):
        """The id is the point of #1894 -- it is what created_by_id filters on."""
        owner, full_key = await self._owned_key(async_client, db_session)

        response = await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})

        assert response.status_code == 200
        result = response.json()
        assert result["id"] == owner.id
        assert result["username"] == "keyowner"
        # The owner is an admin; the key still is not, because no key reaches
        # an administrative route regardless of who owns it.
        assert result["is_admin"] is False
        assert result["role"] != "admin"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_withholds_owner_email_and_groups(self, async_client: AsyncClient, db_session):
        """Identity, not the owner's profile -- anyone holding the key sees this."""
        owner, full_key = await self._owned_key(async_client, db_session)
        assert owner.email is not None and owner.groups  # the helper made both non-empty

        result = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})).json()

        assert result["email"] is None
        assert result["groups"] == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_permissions_track_the_key_scopes_not_the_owner(self, async_client: AsyncClient, db_session):
        """A key owned by an admin still reports only what its flags allow."""
        _, full_key = await self._owned_key(
            async_client,
            db_session,
            can_read_status=True,
            can_control_printer=False,
            can_queue=False,
        )

        perms = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})).json()["permissions"]

        assert "printers:read" in perms  # can_read_status
        assert "printers:control" not in perms  # can_control_printer is off
        assert "queue:create" not in perms  # can_queue is off
        assert "users:create" not in perms  # administrative: unmapped for keys

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_me_permissions_are_exactly_what_the_gate_admits(self, async_client: AsyncClient, db_session):
        """/me must not drift from _check_apikey_permissions.

        The whole defect in #1894 was a /me response that described a different
        credential than the one the gate enforces, so pin them to each other
        rather than to a hand-written list that can rot. The owner is threaded
        through both sides for the same reason -- the gate narrows to the
        owner's permissions, so a check that skipped the owner would stop
        catching drift the moment the owner is not an administrator.
        """
        from fastapi import HTTPException
        from sqlalchemy import select

        from backend.app.core.auth import _check_apikey_permissions, resolve_apikey_owner
        from backend.app.core.permissions import ALL_PERMISSIONS
        from backend.app.models.api_key import APIKey

        _, full_key = await self._owned_key(async_client, db_session, can_read_status=True, can_control_printer=False)

        response = await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})
        reported = set(response.json()["permissions"])

        key = (await db_session.execute(select(APIKey).where(APIKey.name == "owned"))).scalar_one()
        owner = await resolve_apikey_owner(db_session, key)
        for perm in ALL_PERMISSIONS:
            try:
                _check_apikey_permissions(key, [perm], owner=owner)
            except HTTPException:
                assert perm not in reported, f"/me reports '{perm}' but the gate denies it"
            else:
                assert perm in reported, f"the gate admits '{perm}' but /me omits it"


class TestUsersAPI:
    """Integration tests for /api/v1/users/ endpoints."""

    @pytest.fixture
    async def auth_token(self, async_client: AsyncClient):
        """Setup auth and return admin token."""
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "usersadmin",
                "admin_password": "AdminPass1!",
            },
        )

        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "usersadmin", "password": "AdminPass1!"},
        )
        return login_response.json()["access_token"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_users_requires_auth(self, async_client: AsyncClient):
        """Verify listing users requires authentication when auth is enabled."""
        # First enable auth
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "authreqadmin",
                "admin_password": "AdminPass1!",
            },
        )

        # Now try to list users without a token
        response = await async_client.get("/api/v1/users/")

        assert response.status_code == 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_users_as_admin(self, async_client: AsyncClient, auth_token: str):
        """Verify admin can list users."""
        response = await async_client.get(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 200
        result = response.json()
        assert isinstance(result, list)
        assert len(result) >= 1  # At least the admin user

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_user(self, async_client: AsyncClient, auth_token: str):
        """Verify admin can create a new user."""
        response = await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "username": "newuser",
                "password": "Newuserpass1!",
                "role": "user",
            },
        )

        assert response.status_code == 201
        result = response.json()
        assert result["username"] == "newuser"
        assert result["role"] == "user"
        assert result["is_active"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_user_duplicate_username(self, async_client: AsyncClient, auth_token: str):
        """Verify creating user with duplicate username fails."""
        # Create first user
        await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "username": "duplicateuser",
                "password": "Password123!",
                "role": "user",
            },
        )

        # Try to create duplicate
        response = await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "username": "duplicateuser",
                "password": "Password456!",
                "role": "user",
            },
        )

        assert response.status_code == 400
        assert "Username already exists" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_user(self, async_client: AsyncClient, auth_token: str):
        """Verify admin can update a user."""
        # Create user
        create_response = await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "username": "updateuser",
                "password": "Password123!",
                "role": "user",
            },
        )
        user_id = create_response.json()["id"]

        # Update user
        response = await async_client.patch(
            f"/api/v1/users/{user_id}",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={"role": "admin"},
        )

        assert response.status_code == 200
        assert response.json()["role"] == "admin"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_user(self, async_client: AsyncClient, auth_token: str, db_session):
        """Verify admin can delete a user and that all auth-table side effects cascade.

        The auth-cleanup side effects matter on SQLite (FK enforcement off by default):
        without explicit DELETEs in the endpoint, deleting a user leaves orphan rows
        in user_oidc_links / user_totp / user_otp_codes / api_keys — which would
        block SSO re-login and leak MFA secrets (#1285).
        """
        from sqlalchemy import select

        from backend.app.models.api_key import APIKey
        from backend.app.models.long_lived_token import LongLivedToken
        from backend.app.models.oidc_provider import UserOIDCLink
        from backend.app.models.user import User
        from backend.app.models.user_otp_code import UserOTPCode
        from backend.app.models.user_totp import UserTOTP

        # Create user
        create_response = await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "username": "deleteuser",
                "password": "Password123!",
                "role": "user",
            },
        )
        user_id = create_response.json()["id"]

        # Delete user
        response = await async_client.delete(
            f"/api/v1/users/{user_id}",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 204

        # All auth-related rows for this user must be gone — see #1285.
        await db_session.commit()
        user_row = await db_session.execute(select(User).where(User.id == user_id))
        assert user_row.scalar_one_or_none() is None, "User row not deleted"

        for model in (UserOIDCLink, UserTOTP, UserOTPCode, APIKey, LongLivedToken):
            rows = await db_session.execute(select(model).where(model.user_id == user_id))
            assert rows.scalars().all() == [], f"Orphan {model.__name__} rows left after user delete"


class TestAuthDisableAPI:
    """Integration tests for /api/v1/auth/disable endpoint."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_disable_auth(self, async_client: AsyncClient):
        """Verify admin can disable authentication."""
        # Setup auth
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "disableadmin",
                "admin_password": "AdminPass1!",
            },
        )

        # Login to get token
        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "disableadmin", "password": "AdminPass1!"},
        )
        token = login_response.json()["access_token"]

        # Disable auth
        response = await async_client.post(
            "/api/v1/auth/disable",
            headers={"Authorization": f"Bearer {token}"},
        )

        assert response.status_code == 200
        assert response.json()["auth_enabled"] is False

        # Verify auth is now disabled
        status_response = await async_client.get("/api/v1/auth/status")
        assert status_response.json()["auth_enabled"] is False


class TestGroupsAPI:
    """Integration tests for /api/v1/groups/ endpoints."""

    @pytest.fixture
    async def auth_token(self, async_client: AsyncClient):
        """Setup auth and return admin token."""
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "groupsadmin",
                "admin_password": "AdminPass1!",
            },
        )

        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "groupsadmin", "password": "AdminPass1!"},
        )
        return login_response.json()["access_token"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_groups(self, async_client: AsyncClient, auth_token: str):
        """Verify listing groups returns default groups."""
        response = await async_client.get(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 200
        groups = response.json()
        assert isinstance(groups, list)
        # Should have default groups: Administrators, Operators, Viewers
        group_names = [g["name"] for g in groups]
        assert "Administrators" in group_names
        assert "Operators" in group_names
        assert "Viewers" in group_names

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_permissions(self, async_client: AsyncClient, auth_token: str):
        """Verify getting available permissions."""
        response = await async_client.get(
            "/api/v1/groups/permissions",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 200
        permissions = response.json()
        assert isinstance(permissions, dict)
        # Should have permission categories
        assert "Printers" in permissions or len(permissions) > 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_group(self, async_client: AsyncClient, auth_token: str):
        """Verify creating a new group."""
        response = await async_client.post(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "name": "Custom Group",
                "description": "A custom test group",
                "permissions": ["printers:read", "archives:read"],
            },
        )

        assert response.status_code == 201
        group = response.json()
        assert group["name"] == "Custom Group"
        assert group["description"] == "A custom test group"
        assert "printers:read" in group["permissions"]
        assert group["is_system"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_group(self, async_client: AsyncClient, auth_token: str):
        """Verify updating a group."""
        # Create a group first
        create_response = await async_client.post(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "name": "Update Test Group",
                "permissions": ["printers:read"],
            },
        )
        group_id = create_response.json()["id"]

        # Update the group
        response = await async_client.patch(
            f"/api/v1/groups/{group_id}",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "description": "Updated description",
                "permissions": ["printers:read", "printers:control"],
            },
        )

        assert response.status_code == 200
        group = response.json()
        assert group["description"] == "Updated description"
        assert "printers:control" in group["permissions"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cannot_delete_system_group(self, async_client: AsyncClient, auth_token: str):
        """Verify system groups cannot be deleted."""
        # Get the Administrators group
        list_response = await async_client.get(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
        )
        admin_group = next(g for g in list_response.json() if g["name"] == "Administrators")

        # Try to delete it
        response = await async_client.delete(
            f"/api/v1/groups/{admin_group['id']}",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 400
        assert "system group" in response.json()["detail"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_custom_group(self, async_client: AsyncClient, auth_token: str):
        """Verify custom groups can be deleted."""
        # Create a group
        create_response = await async_client.post(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={"name": "Delete Test Group"},
        )
        group_id = create_response.json()["id"]

        # Delete it
        response = await async_client.delete(
            f"/api/v1/groups/{group_id}",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 204


class TestUserGroupsAPI:
    """Integration tests for user-group assignments."""

    @pytest.fixture
    async def auth_token(self, async_client: AsyncClient):
        """Setup auth and return admin token."""
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "usergroupadmin",
                "admin_password": "AdminPass1!",
            },
        )

        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "usergroupadmin", "password": "AdminPass1!"},
        )
        return login_response.json()["access_token"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_user_with_groups(self, async_client: AsyncClient, auth_token: str):
        """Verify creating a user with group assignments."""
        # Get Operators group ID
        groups_response = await async_client.get(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
        )
        operators_group = next(g for g in groups_response.json() if g["name"] == "Operators")

        # Create user with group
        response = await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={
                "username": "groupuser",
                "password": "Password123!",
                "group_ids": [operators_group["id"]],
            },
        )

        assert response.status_code == 201
        user = response.json()
        assert any(g["name"] == "Operators" for g in user["groups"])

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_add_user_to_group(self, async_client: AsyncClient, auth_token: str):
        """Verify adding a user to a group."""
        # Create a user
        user_response = await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {auth_token}"},
            json={"username": "addtogroup", "password": "Password123!"},
        )
        user_id = user_response.json()["id"]

        # Get Viewers group
        groups_response = await async_client.get(
            "/api/v1/groups/",
            headers={"Authorization": f"Bearer {auth_token}"},
        )
        viewers_group = next(g for g in groups_response.json() if g["name"] == "Viewers")

        # Add user to group
        response = await async_client.post(
            f"/api/v1/groups/{viewers_group['id']}/users/{user_id}",
            headers={"Authorization": f"Bearer {auth_token}"},
        )

        assert response.status_code == 204

        # Verify user is in group
        user_check = await async_client.get(
            f"/api/v1/users/{user_id}",
            headers={"Authorization": f"Bearer {auth_token}"},
        )
        assert any(g["name"] == "Viewers" for g in user_check.json()["groups"])


class TestChangePasswordAPI:
    """Integration tests for /api/v1/users/me/change-password endpoint."""

    @pytest.fixture
    async def user_token(self, async_client: AsyncClient):
        """Setup auth and return regular user token."""
        # Enable auth with admin
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "pwchangeadmin",
                "admin_password": "AdminPass1!",
            },
        )

        admin_login = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "pwchangeadmin", "password": "AdminPass1!"},
        )
        admin_token = admin_login.json()["access_token"]

        # Create a regular user
        await async_client.post(
            "/api/v1/users/",
            headers={"Authorization": f"Bearer {admin_token}"},
            json={"username": "pwchangeuser", "password": "Oldpassword123!"},
        )

        # Login as regular user
        user_login = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "pwchangeuser", "password": "Oldpassword123!"},
        )
        return user_login.json()["access_token"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_change_password_success(self, async_client: AsyncClient, user_token: str):
        """Verify user can change their own password."""
        response = await async_client.post(
            "/api/v1/users/me/change-password",
            headers={"Authorization": f"Bearer {user_token}"},
            json={
                "current_password": "Oldpassword123!",
                "new_password": "Newpassword456!",
            },
        )

        assert response.status_code == 200
        assert "success" in response.json()["message"].lower()

        # Verify can login with new password
        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "pwchangeuser", "password": "Newpassword456!"},
        )
        assert login_response.status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_change_password_wrong_current(self, async_client: AsyncClient, user_token: str):
        """Verify changing password fails with wrong current password."""
        response = await async_client.post(
            "/api/v1/users/me/change-password",
            headers={"Authorization": f"Bearer {user_token}"},
            json={
                "current_password": "wrongpassword",
                "new_password": "Newpassword456!",
            },
        )

        assert response.status_code == 400
        assert "incorrect" in response.json()["detail"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_change_password_requires_auth(self, async_client: AsyncClient):
        """Verify changing password requires authentication."""
        response = await async_client.post(
            "/api/v1/users/me/change-password",
            json={
                "current_password": "oldpassword",
                "new_password": "Strongpass456!",
            },
        )

        assert response.status_code == 401


class TestAuthMiddlewarePublicRoutes:
    """Tests for auth middleware public route configuration.

    These routes must be accessible without authentication, even when auth is enabled,
    because browser elements like <img src> and <video src> don't send Authorization headers.
    """

    @pytest.fixture
    async def enabled_auth(self, async_client: AsyncClient):
        """Enable auth for testing middleware behavior."""
        await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "middlewareadmin",
                "admin_password": "AdminPass1!",
            },
        )

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_auth_status_is_public(self, async_client: AsyncClient, enabled_auth):
        """Verify /api/v1/auth/status is accessible without auth."""
        response = await async_client.get("/api/v1/auth/status")
        assert response.status_code == 200
        assert "auth_enabled" in response.json()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_system_appliance_is_public(self, async_client: AsyncClient, enabled_auth):
        """Verify /api/v1/system/appliance is reachable without a JWT.

        The SPA's i18n bootstrap fetches this BEFORE login to seed locale,
        hostname, timezone, and NTP-gate state. The route handler has no
        auth dependency, but the global auth_middleware blocks every
        /api/ path not in PUBLIC_API_ROUTES — so without an explicit
        allowlist entry the user sees a 401 in the browser console on
        every page load.
        """
        response = await async_client.get("/api/v1/system/appliance")
        assert response.status_code == 200, response.text
        body = response.json()
        # Shape contract (no-auth surface):
        for key in ("hostname", "timezone", "locale", "time_synced"):
            assert key in body

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_auth_login_is_public(self, async_client: AsyncClient, enabled_auth):
        """Verify /api/v1/auth/login is accessible without auth."""
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "middlewareadmin", "password": "AdminPass1!"},
        )
        # Should not return 401 (unauthorized) - it should either succeed or return
        # a different error (like 400 for wrong credentials)
        assert response.status_code != 401 or "token" in response.json()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_auth_setup_is_public(self, async_client: AsyncClient):
        """Verify /api/v1/auth/setup is accessible without auth (needed for setup/recovery)."""
        # Don't enable auth first - test that setup endpoint itself is accessible
        response = await async_client.post(
            "/api/v1/auth/setup",
            json={"auth_enabled": False},
        )
        # Should not be 401
        assert response.status_code != 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_updates_version_is_public(self, async_client: AsyncClient, enabled_auth):
        """Verify /api/v1/updates/version is accessible without auth."""
        response = await async_client.get("/api/v1/updates/version")
        # Should not be 401
        assert response.status_code != 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_protected_route_requires_auth(self, async_client: AsyncClient, enabled_auth):
        """Verify non-public routes return 401 without token."""
        response = await async_client.get("/api/v1/printers/")
        assert response.status_code == 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_protected_route_works_with_token(self, async_client: AsyncClient, enabled_auth):
        """Verify non-public routes work with valid token."""
        # Login to get token
        login_response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "middlewareadmin", "password": "AdminPass1!"},
        )
        token = login_response.json()["access_token"]

        # Access protected route
        response = await async_client.get(
            "/api/v1/printers/",
            headers={"Authorization": f"Bearer {token}"},
        )
        assert response.status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_advanced_auth_status_is_public(self, async_client: AsyncClient, enabled_auth):
        """Verify /api/v1/auth/advanced-auth/status is accessible without auth."""
        response = await async_client.get("/api/v1/auth/advanced-auth/status")
        # Should not be 401 (must be accessible for login page)
        assert response.status_code != 401
        # Should return valid response (200 with auth status)
        if response.status_code == 200:
            result = response.json()
            assert "advanced_auth_enabled" in result
            assert "smtp_configured" in result

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_forgot_password_is_public(self, async_client: AsyncClient, enabled_auth):
        """Verify /api/v1/auth/forgot-password is accessible without auth."""
        response = await async_client.post(
            "/api/v1/auth/forgot-password",
            json={"email": "test@example.com"},
        )
        # Should not be 401 (must be accessible for password reset from login page)
        assert response.status_code != 401
        # Will likely be 400 (advanced auth not enabled) but that's okay -
        # the important thing is it's not blocked by auth middleware
        assert response.status_code in [200, 400]


# ===========================================================================
# H-1: Input length validation
# ===========================================================================


class TestInputLengthValidation:
    """LoginRequest and SetupRequest must reject oversized inputs (H-1)."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_login_password_too_long_rejected(self, async_client: AsyncClient):
        """Password exceeding 256 characters must be rejected with 422."""
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "admin", "password": "x" * 257},
        )
        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_login_username_too_long_rejected(self, async_client: AsyncClient):
        """Username exceeding 150 characters must be rejected with 422."""
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "u" * 151, "password": "password"},
        )
        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_setup_password_too_long_rejected(self, async_client: AsyncClient):
        """SetupRequest admin_password exceeding 256 characters must be rejected with 422."""
        response = await async_client.post(
            "/api/v1/auth/setup",
            json={
                "auth_enabled": True,
                "admin_username": "admin",
                "admin_password": "x" * 257,
            },
        )
        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_login_password_at_limit_accepted(self, async_client: AsyncClient):
        """Password of exactly 256 characters must pass schema validation (may fail auth)."""
        response = await async_client.post(
            "/api/v1/auth/login",
            json={"username": "admin", "password": "x" * 256},
        )
        # Schema accepts it; auth may reject with 401 (auth disabled) or 400
        assert response.status_code != 422