fb82e3947d
Some checks failed
CICD / Build and Push CICD Images (push) Successful in 37m42s
CICD / Build CICD Image Failure Postmortem (push) Has been skipped
CICD / Dependency Audits (Informational) (push) Successful in 15m24s
CICD / Source Checks (push) Failing after 18m19s
CICD / Build Release Images (push) Has been skipped
CICD / Build Tester Images (push) Has been skipped
CICD / Production Image Failures Postmortem (push) Has been skipped
CICD / Production Images Complete (push) Failing after 11s
CICD / Runtime Black-Box Integration Tests (push) Has been skipped
CICD / End-to-End Tests (push) Has been skipped
CICD / Integration Tests Failure Postmortem (push) Has been skipped
CICD / E2E Tests Failure Postmortem (push) Has been skipped
CICD / Promote Staging Images To Release (push) Has been skipped
CICD / Source Lanes Failure Postmortem (push) Successful in 16s
CICD / CICD Tests Complete (push) Failing after 28s
134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
"""Integration tests for API endpoints."""
|
|
|
|
from importlib import metadata
|
|
from typing import cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from backend.main import app, compatibility_status, get_api_session
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
class TestAPIIntegration:
|
|
"""Integration tests for API endpoints."""
|
|
|
|
def test_health_check(self) -> None:
|
|
"""Test API health check endpoint."""
|
|
healthy_session = cast("AsyncSession", AsyncMock(spec=AsyncSession))
|
|
healthy_session.execute = AsyncMock(return_value=1)
|
|
|
|
async def override_get_session():
|
|
"""Provide a healthy session dependency override for tests."""
|
|
yield healthy_session
|
|
|
|
app.dependency_overrides[get_api_session] = override_get_session
|
|
try:
|
|
with TestClient(app) as local_client:
|
|
response = local_client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "healthy", "database": "connected"}
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
def test_health_check_db_unavailable(self) -> None:
|
|
"""Health endpoint should return unavailable when DB probe fails."""
|
|
unhealthy_session = cast("AsyncSession", AsyncMock(spec=AsyncSession))
|
|
unhealthy_session.execute = AsyncMock(
|
|
side_effect=SQLAlchemyError("database unavailable")
|
|
)
|
|
|
|
async def override_get_session():
|
|
"""Provide an unhealthy session dependency override for tests."""
|
|
yield unhealthy_session
|
|
|
|
app.dependency_overrides[get_api_session] = override_get_session
|
|
try:
|
|
with TestClient(app) as local_client:
|
|
response = local_client.get("/health")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {
|
|
"status": "unhealthy",
|
|
"database": "disconnected",
|
|
}
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
def test_root_endpoint(self) -> None:
|
|
"""Test root endpoint."""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"message": "Plex Playlist Backend API"}
|
|
|
|
def test_startup_rejects_invalid_runtime_policy(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Startup should fail fast when compatibility policy is invalid."""
|
|
monkeypatch.setenv("BACKEND_REQUIRED_PYTHON", "99.0")
|
|
|
|
with pytest.raises(RuntimeError), TestClient(app):
|
|
pass
|
|
|
|
monkeypatch.delenv("BACKEND_REQUIRED_PYTHON", raising=False)
|
|
|
|
def test_compatibility_endpoint_reports_policy_status(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Compatibility endpoint should expose runtime and pinning policy status."""
|
|
monkeypatch.setenv("BACKEND_REQUIRED_PYTHON", "3.14")
|
|
expected_required_packages = compatibility_status()["required_packages"]
|
|
|
|
with TestClient(app) as local_client:
|
|
response = local_client.get("/compatibility")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["ok"] is True
|
|
assert payload["required_python"] == "3.14"
|
|
assert "current_python" in payload
|
|
assert payload["python_policy_valid"] is True
|
|
assert payload["required_packages"] == expected_required_packages
|
|
assert payload["package_errors"] == {}
|
|
|
|
monkeypatch.delenv("BACKEND_REQUIRED_PYTHON", raising=False)
|
|
|
|
def test_startup_rejects_malformed_required_python(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Startup should fail fast when Python policy is not strict major.minor."""
|
|
monkeypatch.setenv("BACKEND_REQUIRED_PYTHON", "3.14.1")
|
|
|
|
with pytest.raises(RuntimeError), TestClient(app):
|
|
pass
|
|
|
|
monkeypatch.delenv("BACKEND_REQUIRED_PYTHON", raising=False)
|
|
|
|
def test_compatibility_status_handles_missing_package_metadata(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Compatibility checks should handle missing package metadata gracefully."""
|
|
|
|
def raise_not_found(_: str) -> str:
|
|
raise metadata.PackageNotFoundError("fake")
|
|
|
|
monkeypatch.setattr("backend.main._installed_version", raise_not_found)
|
|
status = compatibility_status()
|
|
package_checks = status["package_checks"]
|
|
package_errors = status["package_errors"]
|
|
|
|
assert status["ok"] is False
|
|
assert isinstance(package_checks, dict)
|
|
assert isinstance(package_errors, dict)
|
|
assert package_checks["fastapi"] is False
|
|
assert package_checks["uvicorn"] is False
|
|
assert "fastapi" in package_errors
|
|
assert "uvicorn" in package_errors
|