diff --git a/backend/tests/test_auth.py b/backend/tests/test_auth.py new file mode 100644 index 0000000..c15a52f --- /dev/null +++ b/backend/tests/test_auth.py @@ -0,0 +1,323 @@ +"""Tests for Kithara Studio auth endpoints and mode switching.""" + +import os +import sys +from pathlib import Path + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient + +# Path setup +BACKEND_SRC = Path(__file__).resolve().parent.parent / "src" +sys.path.insert(0, str(BACKEND_SRC)) + + +# ── Fixtures ─────────────────────────────────────────────────── + +@pytest_asyncio.fixture +async def single_mode_client(tmp_path): + """Client with KITHARA_MODE=single (no auth required).""" + data_dir = tmp_path / "data" + data_dir.mkdir() + clips_dir = data_dir / "clips" + clips_dir.mkdir() + waveforms_dir = data_dir / "waveforms" + waveforms_dir.mkdir() + artwork_dir = data_dir / "artwork" + artwork_dir.mkdir() + test_db = data_dir / "kithara.db" + + os.environ["KITHARA_MODE"] = "single" + os.environ["KITHARA_ADMIN_USER"] = "testadmin" + os.environ["KITHARA_ADMIN_PASS"] = "testpass" + os.environ["ARTWORK_DIR"] = str(artwork_dir) + + import database as db_mod + import waveform as wf_mod + import image_gen as ig_mod + import main + import auth as auth_mod + + # Re-init auth module with new env + auth_mod.KITHARA_MODE = os.environ.get("KITHARA_MODE", "single") + + # Patch paths + orig_db_path = db_mod.DB_PATH + orig_clips_dir = main.CLIPS_DIR + orig_artwork_dir = main.ARTWORK_DIR + orig_wf_clips = wf_mod.CLIPS_DIR + orig_wf_wave = wf_mod.WAVEFORM_DIR + orig_ig_artwork = ig_mod.ARTWORK_DIR + + db_mod.DB_PATH = test_db + main.CLIPS_DIR = clips_dir + main.ARTWORK_DIR = artwork_dir + wf_mod.CLIPS_DIR = clips_dir + wf_mod.WAVEFORM_DIR = waveforms_dir + ig_mod.ARTWORK_DIR = artwork_dir + + await db_mod.init_db() + await auth_mod.ensure_users_table() + await auth_mod.bootstrap_admin() + + transport = ASGITransport(app=main.app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield {"client": c, "db": test_db} + + db_mod.DB_PATH = orig_db_path + main.CLIPS_DIR = orig_clips_dir + main.ARTWORK_DIR = orig_artwork_dir + wf_mod.CLIPS_DIR = orig_wf_clips + wf_mod.WAVEFORM_DIR = orig_wf_wave + ig_mod.ARTWORK_DIR = orig_ig_artwork + + for key in ["KITHARA_MODE", "KITHARA_ADMIN_USER", "KITHARA_ADMIN_PASS", "ARTWORK_DIR"]: + os.environ.pop(key, None) + + +@pytest_asyncio.fixture +async def multi_mode_client(tmp_path): + """Client with KITHARA_MODE=multi (auth required).""" + data_dir = tmp_path / "data" + data_dir.mkdir() + clips_dir = data_dir / "clips" + clips_dir.mkdir() + waveforms_dir = data_dir / "waveforms" + waveforms_dir.mkdir() + artwork_dir = data_dir / "artwork" + artwork_dir.mkdir() + test_db = data_dir / "kithara.db" + + os.environ["KITHARA_MODE"] = "multi" + os.environ["KITHARA_ADMIN_USER"] = "testadmin" + os.environ["KITHARA_ADMIN_PASS"] = "testpass" + os.environ["ARTWORK_DIR"] = str(artwork_dir) + + import database as db_mod + import waveform as wf_mod + import image_gen as ig_mod + import main + import auth as auth_mod + + auth_mod.KITHARA_MODE = "multi" + + orig_db_path = db_mod.DB_PATH + orig_clips_dir = main.CLIPS_DIR + orig_artwork_dir = main.ARTWORK_DIR + orig_wf_clips = wf_mod.CLIPS_DIR + orig_wf_wave = wf_mod.WAVEFORM_DIR + orig_ig_artwork = ig_mod.ARTWORK_DIR + + db_mod.DB_PATH = test_db + main.CLIPS_DIR = clips_dir + main.ARTWORK_DIR = artwork_dir + wf_mod.CLIPS_DIR = clips_dir + wf_mod.WAVEFORM_DIR = waveforms_dir + ig_mod.ARTWORK_DIR = artwork_dir + + await db_mod.init_db() + await auth_mod.ensure_users_table() + await auth_mod.bootstrap_admin() + + transport = ASGITransport(app=main.app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield {"client": c, "db": test_db} + + db_mod.DB_PATH = orig_db_path + main.CLIPS_DIR = orig_clips_dir + main.ARTWORK_DIR = orig_artwork_dir + wf_mod.CLIPS_DIR = orig_wf_clips + wf_mod.WAVEFORM_DIR = orig_wf_wave + ig_mod.ARTWORK_DIR = orig_ig_artwork + + for key in ["KITHARA_MODE", "KITHARA_ADMIN_USER", "KITHARA_ADMIN_PASS", "ARTWORK_DIR"]: + os.environ.pop(key, None) + + +# ── Single Mode Tests ─────────────────────────────────────────── + +class TestSingleMode: + """In single-user mode, no auth is required.""" + + @pytest.mark.asyncio + async def test_health_shows_single_mode(self, single_mode_client): + response = await single_mode_client["client"].get("/api/health") + assert response.status_code == 200 + data = response.json() + assert data["mode"] == "single" + + @pytest.mark.asyncio + async def test_me_returns_null_user_single_mode(self, single_mode_client): + response = await single_mode_client["client"].get("/api/auth/me") + assert response.status_code == 200 + data = response.json() + assert data["mode"] == "single" + assert data["user"] is None + + @pytest.mark.asyncio + async def test_clips_accessible_without_auth_in_single_mode(self, single_mode_client): + response = await single_mode_client["client"].get("/api/clips") + assert response.status_code == 200 + + +# ── Multi Mode Tests ─────────────────────────────────────────── + +class TestMultiModeAuth: + """In multi-user mode, auth is required for protected endpoints.""" + + @pytest.mark.asyncio + async def test_health_shows_multi_mode(self, multi_mode_client): + response = await multi_mode_client["client"].get("/api/health") + assert response.status_code == 200 + data = response.json() + assert data["mode"] == "multi" + + @pytest.mark.asyncio + async def test_me_requires_auth_in_multi_mode(self, multi_mode_client): + response = await multi_mode_client["client"].get("/api/auth/me") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_clips_require_auth_in_multi_mode(self, multi_mode_client): + response = await multi_mode_client["client"].get("/api/clips") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_admin_login(self, multi_mode_client): + response = await multi_mode_client["client"].post( + "/api/auth/login", + json={"username": "testadmin", "password": "testpass"}, + ) + assert response.status_code == 200 + data = response.json() + assert "access_token" in data + assert data["token_type"] == "bearer" + assert data["user"]["username"] == "testadmin" + assert data["user"]["role"] == "admin" + + @pytest.mark.asyncio + async def test_login_wrong_password(self, multi_mode_client): + response = await multi_mode_client["client"].post( + "/api/auth/login", + json={"username": "testadmin", "password": "wrong"}, + ) + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_login_nonexistent_user(self, multi_mode_client): + response = await multi_mode_client["client"].post( + "/api/auth/login", + json={"username": "nonexistent", "password": "pass"}, + ) + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_me_with_valid_token(self, multi_mode_client): + # Login first + login_response = await multi_mode_client["client"].post( + "/api/auth/login", + json={"username": "testadmin", "password": "testpass"}, + ) + token = login_response.json()["access_token"] + + # Use token to access /me + response = await multi_mode_client["client"].get( + "/api/auth/me", + headers={"Authorization": f"Bearer {token}"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["mode"] == "multi" + assert data["user"]["username"] == "testadmin" + assert data["user"]["role"] == "admin" + + @pytest.mark.asyncio + async def test_clips_with_valid_token(self, multi_mode_client): + # Login + login_response = await multi_mode_client["client"].post( + "/api/auth/login", + json={"username": "testadmin", "password": "testpass"}, + ) + token = login_response.json()["access_token"] + + # Access clips with token + response = await multi_mode_client["client"].get( + "/api/clips", + headers={"Authorization": f"Bearer {token}"}, + ) + assert response.status_code == 200 + + @pytest.mark.asyncio + async def test_register_with_invite_code(self, multi_mode_client): + # First, admin creates an invite code via the DB + import aiosqlite + + db_path = multi_mode_client["db"] + async with aiosqlite.connect(db_path) as db: + await db.execute( + "INSERT INTO invite_codes (code, created_by, max_uses, current_uses) VALUES (?, ?, ?, ?)", + ("test-invite-code", "testadmin", 5, 0), + ) + await db.commit() + + # Register a new user + response = await multi_mode_client["client"].post( + "/api/auth/register", + json={ + "username": "newuser", + "password": "newpass123", + "invite_code": "test-invite-code", + "display_name": "New User", + }, + ) + assert response.status_code == 200 + data = response.json() + assert "access_token" in data + assert data["user"]["username"] == "newuser" + assert data["user"]["role"] == "user" + + @pytest.mark.asyncio + async def test_register_with_invalid_invite_code(self, multi_mode_client): + response = await multi_mode_client["client"].post( + "/api/auth/register", + json={ + "username": "baduser", + "password": "pass123", + "invite_code": "invalid-code", + }, + ) + assert response.status_code == 400 + + @pytest.mark.asyncio + async def test_register_duplicate_username(self, multi_mode_client): + # Create invite code + import aiosqlite + + db_path = multi_mode_client["db"] + async with aiosqlite.connect(db_path) as db: + await db.execute( + "INSERT INTO invite_codes (code, created_by, max_uses, current_uses) VALUES (?, ?, ?, ?)", + ("dupe-invite", "testadmin", 5, 0), + ) + await db.commit() + + # Register first user + await multi_mode_client["client"].post( + "/api/auth/register", + json={"username": "dupeuser", "password": "pass123", "invite_code": "dupe-invite"}, + ) + + # Try to register same username again (need another invite code) + async with aiosqlite.connect(db_path) as db: + await db.execute( + "INSERT INTO invite_codes (code, created_by, max_uses, current_uses) VALUES (?, ?, ?, ?)", + ("dupe-invite-2", "testadmin", 5, 0), + ) + await db.commit() + + response = await multi_mode_client["client"].post( + "/api/auth/register", + json={"username": "dupeuser", "password": "pass456", "invite_code": "dupe-invite-2"}, + ) + assert response.status_code == 409 \ No newline at end of file