tests/worker/test_worker_api.py

"""
Tests for worker API endpoints.

torch is mocked using autouse fixture at session scope to avoid
loading broken .dylib in the local test environment.
"""
import pytest
import sys
from unittest.mock import MagicMock, patch


# ---------------------------------------------------------------------------
# torch mock — must wrap the entire test session because create_app and
# the TestClient both access the application lazily during requests.
# ---------------------------------------------------------------------------
_TORCH_MOCK = MagicMock()
# Patch at import time so the mock is in place for the whole module lifetime.
sys.modules.setdefault("torch", _TORCH_MOCK)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def settings_with_auth():
    from openmusic_analysis.settings import Settings
    s = Settings.from_env()
    object.__setattr__(s, "worker_tokens", ["test-secret-token"])
    object.__setattr__(s, "allow_unpinned_models", True)
    object.__setattr__(s, "enable_legacy_api", False)
    return s


@pytest.fixture(scope="module")
def settings_no_auth():
    from openmusic_analysis.settings import Settings
    s = Settings.from_env()
    object.__setattr__(s, "worker_tokens", [])
    object.__setattr__(s, "allow_unpinned_models", True)
    object.__setattr__(s, "enable_legacy_api", False)
    return s


@pytest.fixture(scope="module")
def client_with_auth(settings_with_auth):
    from openmusic_analysis.api import create_app
    from fastapi.testclient import TestClient
    app = create_app(settings=settings_with_auth)
    return TestClient(app, raise_server_exceptions=False)


@pytest.fixture(scope="module")
def client_no_auth(settings_no_auth):
    from openmusic_analysis.api import create_app
    from fastapi.testclient import TestClient
    app = create_app(settings=settings_no_auth)
    return TestClient(app, raise_server_exceptions=False)


# ---------------------------------------------------------------------------
# Authentication tests
# ---------------------------------------------------------------------------
def test_missing_auth_returns_401(client_with_auth):
    """Request without Authorization header must return 401."""
    response = client_with_auth.get("/v1/capabilities")
    assert response.status_code == 401
    assert response.json()["error"]["code"] == "UNAUTHORIZED"


def test_wrong_token_returns_401(client_with_auth):
    """Request with wrong Bearer token must return 401."""
    response = client_with_auth.get(
        "/v1/capabilities",
        headers={"Authorization": "Bearer wrong-token"},
    )
    assert response.status_code == 401


def test_valid_token_passes(client_with_auth):
    """Request with valid Bearer token must not return 401."""
    response = client_with_auth.get(
        "/v1/capabilities",
        headers={"Authorization": "Bearer test-secret-token"},
    )
    assert response.status_code != 401


def test_no_token_configured_allows_all(client_no_auth):
    """When no tokens are configured, all requests pass auth."""
    response = client_no_auth.get("/v1/capabilities")
    assert response.status_code != 401


# ---------------------------------------------------------------------------
# Health endpoints
# ---------------------------------------------------------------------------
def test_liveness_no_auth_required(client_with_auth):
    """Health liveness endpoint must not require auth."""
    response = client_with_auth.get("/health/live")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"


def test_readiness_returns_valid_status(client_with_auth):
    """Readiness endpoint must return 200 or 503 with correct schema."""
    response = client_with_auth.get("/health/ready")
    assert response.status_code in (200, 503)
    if response.status_code == 200:
        assert response.json()["status"] == "ready"
    else:
        assert response.json()["status"] == "not_ready"


# ---------------------------------------------------------------------------
# Legacy route isolation
# ---------------------------------------------------------------------------
def test_legacy_routes_not_registered(client_with_auth):
    """When legacy API is disabled, legacy routes must return 404."""
    assert client_with_auth.post("/v1/tracks/analyze").status_code == 404
    assert client_with_auth.post("/v1/tracks/rank-similar").status_code == 404
    assert client_with_auth.get("/v1/models").status_code == 404