sovereign_ai/SovereignAI-Starter/tests/test_auth.py

import os
import unittest
from pathlib import Path
from uuid import uuid4

_PROJECT_ROOT = Path(__file__).resolve().parents[1]
os.environ.setdefault(
    "SOVEREIGNAI_DATA_DIR", str(_PROJECT_ROOT / ".test-runtime" / "auth-tests")
)

from fastapi.testclient import TestClient
from starlette.requests import Request

from app import auth, database
from app.main import app, create_local_session


class AuthenticationTests(unittest.TestCase):
    def setUp(self) -> None:
        self.client = TestClient(app)
        self.created_users: list[str] = []

    def tearDown(self) -> None:
        with database._connect() as connection:
            connection.executemany(
                "DELETE FROM users WHERE id=?",
                [(user_id,) for user_id in self.created_users],
            )

    def _register(self, email: str | None = None) -> tuple[str, str]:
        account_email = email or f"{uuid4().hex}@example.test"
        response = self.client.post(
            "/v1/auth/register",
            json={"email": account_email, "password": "a long secure passphrase"},
        )
        self.assertEqual(response.status_code, 201, response.text)
        data = response.json()
        self.created_users.append(data["user"]["id"])
        return data["user"]["id"], data["access_token"]

    def test_conversation_api_rejects_missing_or_forged_development_identity(self) -> None:
        response = self.client.get(
            "/v1/conversations",
            headers={"X-User-ID": database.LOCAL_USER_ID},
        )

        self.assertEqual(response.status_code, 401, response.text)
        self.assertEqual(response.headers["www-authenticate"], "Bearer")

    def test_register_login_me_and_logout_use_revocable_sessions(self) -> None:
        user_id, token = self._register("User@Example.Test")
        me = self.client.get(
            "/v1/auth/me", headers={"Authorization": f"Bearer {token}"}
        )
        self.assertEqual(me.status_code, 200, me.text)
        self.assertEqual(me.json()["user"], {
            "id": user_id,
            "email": "user@example.test",
            "mode": "account",
        })

        with database._connect() as connection:
            stored = connection.execute(
                "SELECT password_hash FROM user_identities WHERE user_id=?", (user_id,)
            ).fetchone()
            self.assertNotIn("a long secure passphrase", stored["password_hash"])
            self.assertEqual(
                connection.execute(
                    "SELECT COUNT(*) FROM auth_sessions WHERE user_id=?", (user_id,)
                ).fetchone()[0],
                1,
            )

        login = self.client.post(
            "/v1/auth/login",
            json={"email": "USER@example.test", "password": "a long secure passphrase"},
        )
        self.assertEqual(login.status_code, 200, login.text)
        wrong_password = self.client.post(
            "/v1/auth/login",
            json={"email": "user@example.test", "password": "a different wrong phrase"},
        )
        self.assertEqual(wrong_password.status_code, 401)

        logout = self.client.post(
            "/v1/auth/logout", headers={"Authorization": f"Bearer {token}"}
        )
        self.assertEqual(logout.status_code, 200, logout.text)
        expired = self.client.get(
            "/v1/auth/me", headers={"Authorization": f"Bearer {token}"}
        )
        self.assertEqual(expired.status_code, 401)

    def test_conversations_are_isolated_between_accounts(self) -> None:
        first_user, first_token = self._register()
        _, second_token = self._register()
        conversation_id = str(uuid4())
        saved = self.client.put(
            f"/v1/conversations/{conversation_id}",
            headers={"Authorization": f"Bearer {first_token}"},
            json={
                "title": "private",
                "messages": [{"role": "user", "content": "private message"}],
            },
        )
        self.assertEqual(saved.status_code, 200, saved.text)
        read = self.client.get(
            f"/v1/conversations/{conversation_id}",
            headers={"Authorization": f"Bearer {second_token}"},
        )
        self.assertEqual(read.status_code, 404)
        first_list = self.client.get(
            "/v1/conversations", headers={"Authorization": f"Bearer {first_token}"}
        )
        second_list = self.client.get(
            "/v1/conversations", headers={"Authorization": f"Bearer {second_token}"}
        )
        self.assertEqual(first_list.json()[0]["id"], conversation_id)
        self.assertEqual(second_list.json(), [])
        self.assertTrue(first_user)

    def test_password_account_rejects_duplicate_email(self) -> None:
        email = f"{uuid4().hex}@example.test"
        self._register(email)
        duplicate = self.client.post(
            "/v1/auth/register",
            json={"email": email.upper(), "password": "a long secure passphrase"},
        )
        self.assertEqual(duplicate.status_code, 409, duplicate.text)

    def test_local_bootstrap_is_restricted_to_loopback_clients(self) -> None:
        remote = self.client.post("/v1/auth/local-session", json={})
        self.assertEqual(remote.status_code, 403, remote.text)

        request = Request(
            {
                "type": "http",
                "method": "POST",
                "path": "/v1/auth/local-session",
                "headers": [],
                "client": ("127.0.0.1", 43210),
                "server": ("127.0.0.1", 8000),
                "scheme": "http",
                "query_string": b"",
            }
        )
        token_response = create_local_session(request)
        self.assertEqual(token_response["user"]["id"], database.LOCAL_USER_ID)
        authenticated = self.client.get(
            "/v1/auth/me",
            headers={"Authorization": f"Bearer {token_response['access_token']}"},
        )
        self.assertEqual(authenticated.status_code, 200, authenticated.text)


if __name__ == "__main__":
    unittest.main()