rin.kagamine
674abbbe8f
- 14 tests covering auth flow in both modes:
- TestSingleMode: health shows single, /me returns null user, clips accessible without auth
- TestMultiModeAuth: health shows multi, /me requires auth, clips require auth,
admin login, wrong password, nonexistent user, valid token /me,
clips with valid token, register with invite, invalid invite, duplicate username
- Tests require librosa (same as existing test_api.py) — pass in Docker CI
- Conftest fixture pattern matches existing app_client fixture
323 lines
11 KiB
Python
323 lines
11 KiB
Python
"""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 |