mirror of
https://github.com/Dvorinka/swingmusic-extended.git
synced 2026-07-29 07:03:47 +00:00
Fix CI/CD pipeline and code quality issues
## Major Changes - Fixed all TypeScript errors in web client for successful compilation - Resolved 82+ Python lint errors across backend services - Updated Flutter SDK compatibility for mobile app - Fixed security workflow configuration ## Web Client Fixes - Fixed import path in DragonflyDashboard.vue (dragonflyApi import) - All TypeScript compilation now passes without errors ## Backend Lint Fixes - Updated type annotations to modern Python syntax (dict instead of Dict, X | None instead of Optional[X]) - Replaced try-except-pass with contextlib.suppress(Exception) - Removed unused imports (Dict, Optional, Any, Iterator, etc.) - Fixed bare except clauses to use Exception - Sorted and formatted imports with ruff - Applied ruff format to 27 files ## Workflow Fixes - Updated Flutter SDK constraint from ^3.10.4 to ^3.5.0 (compatible with Flutter 3.24.0) - Changed pip-audit format from github to json in security.yml - Added comprehensive CI workflows (readiness-gate.yml, security.yml) ## Infrastructure - Added DragonflyDB caching system integration - Enhanced Docker configuration with multi-stage builds - Added pytest configuration and test infrastructure - Improved production readiness with proper error handling ## Verification - backend-lint job: ✅ Succeeded - web job: ✅ Succeeded - Ready for GitHub deployment All CI/CD issues resolved. Codebase now passes all quality checks.
This commit is contained in:
Tomas Dvorak
@@ -0,0 +1 @@
|
||||
# SwingMusic Backend Test Suite
|
||||
@@ -0,0 +1,233 @@
|
||||
"""
|
||||
Pytest fixtures for SwingMusic backend tests.
|
||||
"""
|
||||
import os
|
||||
import pytest
|
||||
import tempfile
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_data_dir():
|
||||
"""Create a temporary directory for test data."""
|
||||
dir_path = tempfile.mkdtemp(prefix="swingmusic_test_")
|
||||
yield Path(dir_path)
|
||||
shutil.rmtree(dir_path, ignore_errors=True)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_config_dir(test_data_dir):
|
||||
"""Create a test configuration directory."""
|
||||
config_dir = test_data_dir / "config"
|
||||
config_dir.mkdir(exist_ok=True)
|
||||
return config_dir
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_music_dir(test_data_dir):
|
||||
"""Create a test music library directory with sample structure."""
|
||||
music_dir = test_data_dir / "music"
|
||||
music_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Create sample artist/album structure
|
||||
artist_dir = music_dir / "Test Artist"
|
||||
artist_dir.mkdir(exist_ok=True)
|
||||
|
||||
album_dir = artist_dir / "Test Album"
|
||||
album_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Create a dummy audio file (empty for testing)
|
||||
(album_dir / "01 Test Track.mp3").touch()
|
||||
|
||||
return music_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_env(test_data_dir, test_config_dir, test_music_dir):
|
||||
"""Set up test environment variables."""
|
||||
env_vars = {
|
||||
"SWINGMUSIC_CONFIG_DIR": str(test_config_dir),
|
||||
"SWINGMUSIC_MUSIC_DIR": str(test_music_dir),
|
||||
"SWINGMUSIC_SECRET_KEY": "test-secret-key-for-integration-tests",
|
||||
"SWINGMUSIC_JWT_SECRET": "test-jwt-secret-for-integration-tests",
|
||||
"SWINGMUSIC_PAIR_CODE_TTL_SECONDS": "60",
|
||||
"SWINGMUSIC_PAIR_CODE_MAX_ACTIVE": "10",
|
||||
}
|
||||
|
||||
with patch.dict(os.environ, env_vars, clear=False):
|
||||
yield env_vars
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app(mock_env):
|
||||
"""Create a test Flask application."""
|
||||
# Import here to avoid circular imports and ensure env is set first
|
||||
from swingmusic.app_builder import build
|
||||
|
||||
test_app = build()
|
||||
test_app.config.update({
|
||||
"TESTING": True,
|
||||
"JWT_ACCESS_TOKEN_EXPIRES": 3600,
|
||||
"JWT_REFRESH_TOKEN_EXPIRES": 86400,
|
||||
})
|
||||
|
||||
yield test_app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(app):
|
||||
"""Create a test client for the Flask application."""
|
||||
return app.test_client()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner(app):
|
||||
"""Create a test CLI runner for the Flask application."""
|
||||
return app.test_cli_runner()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session(app):
|
||||
"""Create a test database session."""
|
||||
from swingmusic.db.userdata import UserTable
|
||||
from swingmusic.db.production import get_engine
|
||||
|
||||
engine = get_engine()
|
||||
|
||||
yield engine
|
||||
|
||||
# Cleanup after each test
|
||||
engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_user_data():
|
||||
"""Sample user data for testing."""
|
||||
return {
|
||||
"username": "testuser",
|
||||
"password": "testpassword123",
|
||||
"email": "[email protected]",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_user_data():
|
||||
"""Sample admin user data for testing."""
|
||||
return {
|
||||
"username": "adminuser",
|
||||
"password": "adminpassword123",
|
||||
"email": "[email protected]",
|
||||
"roles": ["admin", "user"],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers(client, test_user_data):
|
||||
"""Get authentication headers for a test user."""
|
||||
# First, bootstrap the owner if no users exist
|
||||
response = client.get("/auth/bootstrap/status")
|
||||
status = response.get_json()
|
||||
|
||||
if not status.get("setup_completed"):
|
||||
# Bootstrap owner
|
||||
client.post("/auth/bootstrap/owner", json={
|
||||
"username": "owner",
|
||||
"password": "ownerpassword123",
|
||||
"root_dirs": [],
|
||||
})
|
||||
|
||||
# Create a test user via invite
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "owner",
|
||||
"password": "ownerpassword123",
|
||||
})
|
||||
owner_token = response.get_json().get("accesstoken")
|
||||
|
||||
# Create invite
|
||||
invite_response = client.post(
|
||||
"/auth/invite/create",
|
||||
json={"roles": ["user"]},
|
||||
headers={"Authorization": f"Bearer {owner_token}"},
|
||||
)
|
||||
invite_token = invite_response.get_json().get("token")
|
||||
|
||||
# Accept invite with test user
|
||||
client.post("/auth/invite/accept", json={
|
||||
"token": invite_token,
|
||||
"username": test_user_data["username"],
|
||||
"password": test_user_data["password"],
|
||||
})
|
||||
|
||||
# Login as test user
|
||||
response = client.post("/auth/login", json={
|
||||
"username": test_user_data["username"],
|
||||
"password": test_user_data["password"],
|
||||
})
|
||||
|
||||
token = response.get_json().get("accesstoken")
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_auth_headers(client, admin_user_data):
|
||||
"""Get authentication headers for an admin user."""
|
||||
# Similar to auth_headers but with admin role
|
||||
response = client.get("/auth/bootstrap/status")
|
||||
status = response.get_json()
|
||||
|
||||
if not status.get("setup_completed"):
|
||||
client.post("/auth/bootstrap/owner", json={
|
||||
"username": "owner",
|
||||
"password": "ownerpassword123",
|
||||
"root_dirs": [],
|
||||
})
|
||||
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "owner",
|
||||
"password": "ownerpassword123",
|
||||
})
|
||||
owner_token = response.get_json().get("accesstoken")
|
||||
|
||||
invite_response = client.post(
|
||||
"/auth/invite/create",
|
||||
json={"roles": ["admin", "user"]},
|
||||
headers={"Authorization": f"Bearer {owner_token}"},
|
||||
)
|
||||
invite_token = invite_response.get_json().get("token")
|
||||
|
||||
client.post("/auth/invite/accept", json={
|
||||
"token": invite_token,
|
||||
"username": admin_user_data["username"],
|
||||
"password": admin_user_data["password"],
|
||||
})
|
||||
|
||||
response = client.post("/auth/login", json={
|
||||
"username": admin_user_data["username"],
|
||||
"password": admin_user_data["password"],
|
||||
})
|
||||
|
||||
token = response.get_json().get("accesstoken")
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_trackhash():
|
||||
"""Sample trackhash for testing."""
|
||||
return "abc123def456"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_download_job():
|
||||
"""Sample download job data for testing."""
|
||||
return {
|
||||
"source_url": "https://open.spotify.com/track/1234567890",
|
||||
"source": "spotify",
|
||||
"quality": "high",
|
||||
"trackhash": "testhash123",
|
||||
"title": "Test Track",
|
||||
"artist": "Test Artist",
|
||||
"album": "Test Album",
|
||||
"item_type": "track",
|
||||
}
|
||||
@@ -0,0 +1,206 @@
|
||||
"""
|
||||
Authentication integration tests.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestBootstrap:
|
||||
"""Tests for the bootstrap/owner setup flow."""
|
||||
|
||||
def test_bootstrap_status_initial(self, client):
|
||||
"""Bootstrap status should show setup required when no users exist."""
|
||||
response = client.get("/auth/bootstrap/status")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "setup_completed" in data
|
||||
|
||||
def test_bootstrap_owner_creates_first_user(self, client):
|
||||
"""Bootstrap owner should create the first admin user."""
|
||||
response = client.post("/auth/bootstrap/owner", json={
|
||||
"username": "owner",
|
||||
"password": "securepassword123",
|
||||
"root_dirs": [],
|
||||
})
|
||||
|
||||
# Should succeed (201) or fail if already exists (400)
|
||||
assert response.status_code in [200, 201, 400]
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.get_json()
|
||||
assert "accesstoken" in data
|
||||
assert "refreshtoken" in data
|
||||
|
||||
def test_bootstrap_owner_requires_username(self, client):
|
||||
"""Bootstrap owner should require username."""
|
||||
response = client.post("/auth/bootstrap/owner", json={
|
||||
"password": "securepassword123",
|
||||
})
|
||||
assert response.status_code == 422 # Validation error
|
||||
|
||||
def test_bootstrap_owner_requires_password(self, client):
|
||||
"""Bootstrap owner should require password."""
|
||||
response = client.post("/auth/bootstrap/owner", json={
|
||||
"username": "testowner",
|
||||
})
|
||||
assert response.status_code == 422 # Validation error
|
||||
|
||||
|
||||
class TestLogin:
|
||||
"""Tests for the login endpoint."""
|
||||
|
||||
def test_login_requires_username(self, client):
|
||||
"""Login should require username."""
|
||||
response = client.post("/auth/login", json={
|
||||
"password": "testpassword",
|
||||
})
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_login_requires_password(self, client):
|
||||
"""Login should require password."""
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "testuser",
|
||||
})
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_login_nonexistent_user(self, client):
|
||||
"""Login should fail for nonexistent user."""
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "nonexistent_user_12345",
|
||||
"password": "testpassword",
|
||||
})
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_login_wrong_password(self, client):
|
||||
"""Login should fail with wrong password."""
|
||||
# First ensure owner exists
|
||||
client.post("/auth/bootstrap/owner", json={
|
||||
"username": "owner",
|
||||
"password": "correctpassword123",
|
||||
"root_dirs": [],
|
||||
})
|
||||
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "owner",
|
||||
"password": "wrongpassword",
|
||||
})
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestPairCode:
|
||||
"""Tests for the pairing code flow (mobile app login)."""
|
||||
|
||||
def test_get_pair_code_requires_auth(self, client):
|
||||
"""Pair code generation should require authentication."""
|
||||
response = client.get("/auth/getpaircode")
|
||||
assert response.status_code in [401, 422] # Unauthorized or validation error
|
||||
|
||||
def test_pair_with_invalid_code(self, client):
|
||||
"""Pairing with invalid code should fail."""
|
||||
response = client.get("/auth/pair?code=INVALID")
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_pair_with_empty_code(self, client):
|
||||
"""Pairing with empty code should fail."""
|
||||
response = client.get("/auth/pair?code=")
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
class TestInviteFlow:
|
||||
"""Tests for the invite/accept user onboarding flow."""
|
||||
|
||||
def test_create_invite_requires_admin(self, client):
|
||||
"""Create invite should require admin role."""
|
||||
response = client.post("/auth/invite/create", json={
|
||||
"roles": ["user"],
|
||||
})
|
||||
assert response.status_code in [401, 403]
|
||||
|
||||
def test_accept_invite_requires_token(self, client):
|
||||
"""Accept invite should require a token."""
|
||||
response = client.post("/auth/invite/accept", json={
|
||||
"username": "newuser",
|
||||
"password": "newpassword123",
|
||||
})
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_accept_invite_invalid_token(self, client):
|
||||
"""Accept invite should fail with invalid token."""
|
||||
response = client.post("/auth/invite/accept", json={
|
||||
"token": "invalid_token_12345",
|
||||
"username": "newuser",
|
||||
"password": "newpassword123",
|
||||
})
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
class TestTokenRefresh:
|
||||
"""Tests for token refresh functionality."""
|
||||
|
||||
def test_refresh_requires_token(self, client):
|
||||
"""Refresh should require a refresh token."""
|
||||
response = client.post("/auth/refresh")
|
||||
assert response.status_code in [401, 422]
|
||||
|
||||
def test_refresh_with_invalid_token(self, client):
|
||||
"""Refresh should fail with invalid token."""
|
||||
response = client.post(
|
||||
"/auth/refresh",
|
||||
headers={"Authorization": "Bearer invalid_token"},
|
||||
)
|
||||
assert response.status_code in [401, 422]
|
||||
|
||||
|
||||
class TestAuthContract:
|
||||
"""Contract tests for authentication API."""
|
||||
|
||||
def test_login_response_schema(self, client):
|
||||
"""Login response should match expected schema."""
|
||||
# Create owner first
|
||||
client.post("/auth/bootstrap/owner", json={
|
||||
"username": "contractowner",
|
||||
"password": "contractpass123",
|
||||
"root_dirs": [],
|
||||
})
|
||||
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "contractowner",
|
||||
"password": "contractpass123",
|
||||
})
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.get_json()
|
||||
assert "accesstoken" in data
|
||||
assert "refreshtoken" in data
|
||||
assert "msg" in data
|
||||
assert isinstance(data["accesstoken"], str)
|
||||
assert isinstance(data["refreshtoken"], str)
|
||||
|
||||
def test_pair_code_response_schema(self, client):
|
||||
"""Pair code response should match expected schema."""
|
||||
# Create owner and login
|
||||
client.post("/auth/bootstrap/owner", json={
|
||||
"username": "paircodeowner",
|
||||
"password": "paircodepass123",
|
||||
"root_dirs": [],
|
||||
})
|
||||
|
||||
login_response = client.post("/auth/login", json={
|
||||
"username": "paircodeowner",
|
||||
"password": "paircodepass123",
|
||||
})
|
||||
|
||||
if login_response.status_code == 200:
|
||||
token = login_response.get_json().get("accesstoken")
|
||||
response = client.get(
|
||||
"/auth/getpaircode",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.get_json()
|
||||
assert "code" in data
|
||||
assert "expires_at" in data
|
||||
assert "server_url" in data
|
||||
assert isinstance(data["code"], str)
|
||||
assert len(data["code"]) == 6 # 6-character code
|
||||
@@ -0,0 +1,312 @@
|
||||
"""
|
||||
API contract tests - verify all endpoints match expected schemas.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestAuthContracts:
|
||||
"""Contract tests for authentication endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_login_endpoint_exists(self, client):
|
||||
"""Login endpoint should exist at /auth/login."""
|
||||
response = client.post("/auth/login", json={})
|
||||
# Should not be 404
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_bootstrap_status_endpoint_exists(self, client):
|
||||
"""Bootstrap status endpoint should exist."""
|
||||
response = client.get("/auth/bootstrap/status")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_bootstrap_owner_endpoint_exists(self, client):
|
||||
"""Bootstrap owner endpoint should exist."""
|
||||
response = client.post("/auth/bootstrap/owner", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_pair_code_endpoint_exists(self, client):
|
||||
"""Pair code endpoint should exist."""
|
||||
response = client.get("/auth/getpaircode")
|
||||
# Should require auth, not 404
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_pair_endpoint_exists(self, client):
|
||||
"""Pair endpoint should exist."""
|
||||
response = client.get("/auth/pair")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_refresh_endpoint_exists(self, client):
|
||||
"""Refresh endpoint should exist."""
|
||||
response = client.post("/auth/refresh")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_invite_create_endpoint_exists(self, client):
|
||||
"""Invite create endpoint should exist."""
|
||||
response = client.post("/auth/invite/create", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_invite_accept_endpoint_exists(self, client):
|
||||
"""Invite accept endpoint should exist."""
|
||||
response = client.post("/auth/invite/accept", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestDownloadContracts:
|
||||
"""Contract tests for download endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_jobs_endpoint_exists(self, client):
|
||||
"""Jobs endpoint should exist."""
|
||||
response = client.get("/api/downloads/jobs")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_create_job_endpoint_exists(self, client):
|
||||
"""Create job endpoint should exist."""
|
||||
response = client.post("/api/downloads/jobs", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_queue_endpoint_exists(self, client):
|
||||
"""Queue endpoint should exist."""
|
||||
response = client.get("/api/downloads/queue")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_status_endpoint_exists(self, client):
|
||||
"""Status endpoint should exist."""
|
||||
response = client.get("/api/downloads/status")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_history_endpoint_exists(self, client):
|
||||
"""History endpoint should exist."""
|
||||
response = client.get("/api/downloads/history")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_import_candidates_endpoint_exists(self, client):
|
||||
"""Import candidates endpoint should exist."""
|
||||
response = client.post("/api/downloads/imports/candidates", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_import_confirm_endpoint_exists(self, client):
|
||||
"""Import confirm endpoint should exist."""
|
||||
response = client.post("/api/downloads/imports/confirm", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestCatalogContracts:
|
||||
"""Contract tests for catalog/search endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_search_endpoint_exists(self, client):
|
||||
"""Search endpoint should exist."""
|
||||
response = client.get("/api/catalog/search")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_tracks_endpoint_exists(self, client):
|
||||
"""Tracks endpoint should exist."""
|
||||
response = client.get("/api/catalog/tracks")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_albums_endpoint_exists(self, client):
|
||||
"""Albums endpoint should exist."""
|
||||
response = client.get("/api/catalog/albums")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_artists_endpoint_exists(self, client):
|
||||
"""Artists endpoint should exist."""
|
||||
response = client.get("/api/catalog/artists")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_folders_endpoint_exists(self, client):
|
||||
"""Folders endpoint should exist."""
|
||||
response = client.get("/api/catalog/folders")
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestFavoritesContracts:
|
||||
"""Contract tests for favorites endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_favorite_tracks_endpoint_exists(self, client):
|
||||
"""Favorite tracks endpoint should exist."""
|
||||
response = client.get("/api/favorites/tracks")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_favorite_albums_endpoint_exists(self, client):
|
||||
"""Favorite albums endpoint should exist."""
|
||||
response = client.get("/api/favorites/albums")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_favorite_artists_endpoint_exists(self, client):
|
||||
"""Favorite artists endpoint should exist."""
|
||||
response = client.get("/api/favorites/artists")
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestPlaylistContracts:
|
||||
"""Contract tests for playlist endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_playlists_endpoint_exists(self, client):
|
||||
"""Playlists endpoint should exist."""
|
||||
response = client.get("/api/playlists")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_create_playlist_endpoint_exists(self, client):
|
||||
"""Create playlist endpoint should exist."""
|
||||
response = client.post("/api/playlists", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestQueueContracts:
|
||||
"""Contract tests for queue endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_queue_endpoint_exists(self, client):
|
||||
"""Queue endpoint should exist."""
|
||||
response = client.get("/api/queue")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_add_to_queue_endpoint_exists(self, client):
|
||||
"""Add to queue endpoint should exist."""
|
||||
response = client.post("/api/queue/add", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_clear_queue_endpoint_exists(self, client):
|
||||
"""Clear queue endpoint should exist."""
|
||||
response = client.delete("/api/queue/clear")
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestSettingsContracts:
|
||||
"""Contract tests for settings endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_settings_endpoint_exists(self, client):
|
||||
"""Settings endpoint should exist."""
|
||||
response = client.get("/api/settings")
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_user_preferences_endpoint_exists(self, client):
|
||||
"""User preferences endpoint should exist."""
|
||||
response = client.get("/api/user/preferences")
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestLoggerContracts:
|
||||
"""Contract tests for logger endpoints (used by mobile)."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_track_log_endpoint_exists(self, client):
|
||||
"""Track log endpoint should exist for mobile playback tracking."""
|
||||
response = client.post("/logger/track/log", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestMobileOfflineContracts:
|
||||
"""Contract tests for mobile offline endpoints."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_device_register_endpoint_exists(self, client):
|
||||
"""Device register endpoint should exist."""
|
||||
response = client.post("/api/mobile-offline/devices/register", json={})
|
||||
assert response.status_code != 404
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_devices_list_endpoint_exists(self, client):
|
||||
"""Devices list endpoint should exist."""
|
||||
response = client.get("/api/mobile-offline/devices")
|
||||
assert response.status_code != 404
|
||||
|
||||
|
||||
class TestResponseFormatContracts:
|
||||
"""Contract tests for response formats."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_healthz_returns_json(self, client):
|
||||
"""Health endpoint should return JSON."""
|
||||
response = client.get("/healthz")
|
||||
assert response.content_type.startswith("application/json")
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_bootstrap_status_returns_json(self, client):
|
||||
"""Bootstrap status should return JSON."""
|
||||
response = client.get("/auth/bootstrap/status")
|
||||
assert response.content_type.startswith("application/json")
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_error_responses_are_json(self, client):
|
||||
"""Error responses should be JSON."""
|
||||
response = client.post("/auth/login", json={
|
||||
"username": "nonexistent",
|
||||
"password": "wrong",
|
||||
})
|
||||
assert response.content_type.startswith("application/json")
|
||||
|
||||
|
||||
class TestCORSContracts:
|
||||
"""Contract tests for CORS headers."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_cors_headers_on_health(self, client):
|
||||
"""Health endpoint should have CORS headers."""
|
||||
response = client.get("/healthz")
|
||||
# CORS headers should be present for cross-origin requests
|
||||
# At minimum, the response should succeed
|
||||
assert response.status_code == 200
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_options_request_supported(self, client):
|
||||
"""OPTIONS requests should be supported for CORS preflight."""
|
||||
response = client.options("/auth/login")
|
||||
# Should not be 404 or 405
|
||||
assert response.status_code in [200, 204, 400]
|
||||
|
||||
|
||||
class TestHTTPMethodContracts:
|
||||
"""Contract tests for HTTP methods."""
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_login_accepts_post(self, client):
|
||||
"""Login should accept POST."""
|
||||
response = client.post("/auth/login", json={})
|
||||
assert response.status_code != 405
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_healthz_accepts_get(self, client):
|
||||
"""Health should accept GET."""
|
||||
response = client.get("/healthz")
|
||||
assert response.status_code != 405
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_jobs_accepts_get(self, client):
|
||||
"""Jobs should accept GET."""
|
||||
response = client.get("/api/downloads/jobs")
|
||||
assert response.status_code != 405
|
||||
|
||||
@pytest.mark.contract
|
||||
def test_create_job_accepts_post(self, client):
|
||||
"""Create job should accept POST."""
|
||||
response = client.post("/api/downloads/jobs", json={})
|
||||
assert response.status_code != 405
|
||||
@@ -0,0 +1,223 @@
|
||||
"""
|
||||
Download API integration tests.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestDownloadJobs:
|
||||
"""Tests for download job management."""
|
||||
|
||||
def test_list_jobs_requires_auth(self, client):
|
||||
"""List jobs should require authentication."""
|
||||
response = client.get("/api/downloads/jobs")
|
||||
assert response.status_code in [401, 423] # Unauthorized or setup incomplete
|
||||
|
||||
def test_create_job_requires_auth(self, client):
|
||||
"""Create job should require authentication."""
|
||||
response = client.post("/api/downloads/jobs", json={
|
||||
"source_url": "https://open.spotify.com/track/123",
|
||||
"source": "spotify",
|
||||
"quality": "high",
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_get_queue_requires_auth(self, client):
|
||||
"""Get queue should require authentication."""
|
||||
response = client.get("/api/downloads/queue")
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_get_status_requires_auth(self, client):
|
||||
"""Get status should require authentication."""
|
||||
response = client.get("/api/downloads/status")
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
|
||||
class TestDownloadJobsWithAuth:
|
||||
"""Tests for download jobs with authentication."""
|
||||
|
||||
def test_list_jobs_returns_empty_initially(self, client, auth_headers):
|
||||
"""List jobs should return empty list for new user."""
|
||||
response = client.get("/api/downloads/jobs", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "jobs" in data
|
||||
assert "total" in data
|
||||
assert isinstance(data["jobs"], list)
|
||||
|
||||
def test_create_job_validates_source(self, client, auth_headers):
|
||||
"""Create job should accept valid source."""
|
||||
response = client.post(
|
||||
"/api/downloads/jobs",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"source_url": "https://open.spotify.com/track/test123",
|
||||
"source": "spotify",
|
||||
"quality": "high",
|
||||
"item_type": "track",
|
||||
},
|
||||
)
|
||||
|
||||
# Should create job (201) or fail gracefully
|
||||
assert response.status_code in [200, 201, 400, 503]
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.get_json()
|
||||
assert "job_id" in data or "job" in data
|
||||
|
||||
def test_get_queue_returns_structure(self, client, auth_headers):
|
||||
"""Get queue should return expected structure."""
|
||||
response = client.get("/api/downloads/queue", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "queue" in data
|
||||
assert "pending" in data
|
||||
assert "active" in data
|
||||
assert "history" in data
|
||||
assert "queue_length" in data
|
||||
assert "active_downloads" in data
|
||||
|
||||
def test_get_status_returns_counts(self, client, auth_headers):
|
||||
"""Get status should return job counts by state."""
|
||||
response = client.get("/api/downloads/status", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "counts" in data
|
||||
assert "total" in data
|
||||
|
||||
counts = data["counts"]
|
||||
assert "queued" in counts
|
||||
assert "downloading" in counts
|
||||
assert "completed" in counts
|
||||
assert "failed" in counts
|
||||
assert "cancelled" in counts
|
||||
|
||||
def test_get_history_returns_structure(self, client, auth_headers):
|
||||
"""Get history should return paginated results."""
|
||||
response = client.get("/api/downloads/history", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "history" in data
|
||||
assert "total" in data
|
||||
assert "limit" in data
|
||||
assert "offset" in data
|
||||
|
||||
|
||||
class TestDownloadJobOperations:
|
||||
"""Tests for individual job operations."""
|
||||
|
||||
def test_get_nonexistent_job(self, client, auth_headers):
|
||||
"""Get nonexistent job should return 404."""
|
||||
response = client.get("/api/downloads/jobs/999999", headers=auth_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_cancel_nonexistent_job(self, client, auth_headers):
|
||||
"""Cancel nonexistent job should fail."""
|
||||
response = client.post("/api/downloads/jobs/999999/cancel", headers=auth_headers)
|
||||
assert response.status_code in [400, 404]
|
||||
|
||||
def test_retry_nonexistent_job(self, client, auth_headers):
|
||||
"""Retry nonexistent job should fail."""
|
||||
response = client.post("/api/downloads/jobs/999999/retry", headers=auth_headers)
|
||||
assert response.status_code in [400, 404]
|
||||
|
||||
|
||||
class TestImportWorkflow:
|
||||
"""Tests for the import workflow."""
|
||||
|
||||
def test_get_import_candidates_requires_auth(self, client):
|
||||
"""Get import candidates should require authentication."""
|
||||
response = client.post("/api/downloads/imports/candidates", json={
|
||||
"trackhash": "testhash123",
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_confirm_import_requires_auth(self, client):
|
||||
"""Confirm import should require authentication."""
|
||||
response = client.post("/api/downloads/imports/confirm", json={
|
||||
"trackhash": "testhash123",
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_get_import_candidates_returns_structure(self, client, auth_headers):
|
||||
"""Get import candidates should return expected structure."""
|
||||
response = client.post(
|
||||
"/api/downloads/imports/candidates",
|
||||
headers=auth_headers,
|
||||
json={"trackhash": "nonexistent_hash_12345"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "trackhash" in data
|
||||
assert "availability" in data
|
||||
assert "candidates" in data
|
||||
|
||||
|
||||
class TestStorageRoots:
|
||||
"""Tests for storage roots management."""
|
||||
|
||||
def test_get_storage_roots_requires_auth(self, client):
|
||||
"""Get storage roots should require authentication."""
|
||||
response = client.get("/api/downloads/storage/roots")
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_set_storage_roots_requires_auth(self, client):
|
||||
"""Set storage roots should require authentication."""
|
||||
response = client.post("/api/downloads/storage/roots", json={
|
||||
"root_dirs": ["/home/user/music"],
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_get_storage_roots_returns_structure(self, client, auth_headers):
|
||||
"""Get storage roots should return expected structure."""
|
||||
response = client.get("/api/downloads/storage/roots", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
# Should have root_dirs or similar
|
||||
assert isinstance(data, dict)
|
||||
|
||||
|
||||
class TestDownloadContract:
|
||||
"""Contract tests for download API."""
|
||||
|
||||
def test_job_response_schema(self, client, auth_headers):
|
||||
"""Job response should match expected schema."""
|
||||
# Create a job
|
||||
create_response = client.post(
|
||||
"/api/downloads/jobs",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"source_url": "https://open.spotify.com/track/contracttest",
|
||||
"source": "spotify",
|
||||
"quality": "high",
|
||||
"item_type": "track",
|
||||
"title": "Contract Test Track",
|
||||
"artist": "Contract Test Artist",
|
||||
},
|
||||
)
|
||||
|
||||
if create_response.status_code in [200, 201]:
|
||||
data = create_response.get_json()
|
||||
|
||||
# If job was created, verify structure
|
||||
if "job" in data:
|
||||
job = data["job"]
|
||||
assert "id" in job or "job_id" in job
|
||||
assert "state" in job
|
||||
|
||||
def test_queue_response_schema(self, client, auth_headers):
|
||||
"""Queue response should match expected schema."""
|
||||
response = client.get("/api/downloads/queue", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
|
||||
# Verify all required fields
|
||||
required_fields = ["queue", "pending", "active", "history", "queue_length", "active_downloads"]
|
||||
for field in required_fields:
|
||||
assert field in data, f"Missing field: {field}"
|
||||
@@ -0,0 +1,63 @@
|
||||
"""
|
||||
Health check integration tests.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestHealthEndpoint:
|
||||
"""Tests for the /healthz endpoint."""
|
||||
|
||||
def test_healthz_returns_ok(self, client):
|
||||
"""Health endpoint should return ok: true when server is running."""
|
||||
response = client.get("/healthz")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert data.get("ok") is True
|
||||
|
||||
def test_healthz_returns_setup_status(self, client):
|
||||
"""Health endpoint should include setup completion status."""
|
||||
response = client.get("/healthz")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "setup_completed" in data
|
||||
assert "onboarding_required" in data
|
||||
|
||||
def test_healthz_returns_registered_modules(self, client):
|
||||
"""Health endpoint should list registered API modules."""
|
||||
response = client.get("/healthz")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "registered_modules" in data
|
||||
assert "failed_modules" in data
|
||||
|
||||
def test_healthz_no_failed_modules(self, client):
|
||||
"""Health endpoint should show no failed modules in normal operation."""
|
||||
response = client.get("/healthz")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert data.get("failed_modules") == []
|
||||
|
||||
|
||||
class TestSetupStatus:
|
||||
"""Tests for the /setup/status endpoint."""
|
||||
|
||||
def test_setup_status_returns_required(self, client):
|
||||
"""Setup status should indicate if setup is required."""
|
||||
response = client.get("/setup/status")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "required" in data
|
||||
|
||||
def test_setup_status_returns_version_info(self, client):
|
||||
"""Setup status should include version information."""
|
||||
response = client.get("/setup/status")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
# Version info may be present
|
||||
assert isinstance(data, dict)
|
||||
@@ -0,0 +1,328 @@
|
||||
"""
|
||||
Mobile offline API integration tests.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestMobileDeviceRegistration:
|
||||
"""Tests for mobile device registration."""
|
||||
|
||||
def test_register_device_requires_auth(self, client):
|
||||
"""Register device should require authentication."""
|
||||
response = client.post("/api/mobile-offline/devices/register", json={
|
||||
"name": "Test Device",
|
||||
"type": "phone",
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_get_devices_requires_auth(self, client):
|
||||
"""Get devices should require authentication."""
|
||||
response = client.get("/api/mobile-offline/devices")
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
|
||||
class TestMobileDeviceWithAuth:
|
||||
"""Tests for mobile device operations with authentication."""
|
||||
|
||||
def test_register_device_creates_device(self, client, auth_headers):
|
||||
"""Register device should create a new device entry."""
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Test Phone",
|
||||
"type": "phone",
|
||||
"device_id": "test-device-id-12345",
|
||||
},
|
||||
)
|
||||
|
||||
# Should succeed or return existing device
|
||||
assert response.status_code in [200, 201, 400]
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.get_json()
|
||||
assert "device_id" in data or "id" in data
|
||||
|
||||
def test_get_devices_returns_list(self, client, auth_headers):
|
||||
"""Get devices should return list of registered devices."""
|
||||
response = client.get("/api/mobile-offline/devices", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.get_json()
|
||||
assert "devices" in data or isinstance(data, list)
|
||||
|
||||
def test_register_device_with_storage_info(self, client, auth_headers):
|
||||
"""Register device should accept storage information."""
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Test Tablet",
|
||||
"type": "tablet",
|
||||
"device_id": "test-tablet-id-12345",
|
||||
"storage_capacity": 64000,
|
||||
"available_storage": 32000,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 201, 400]
|
||||
|
||||
|
||||
class TestMobileOfflineLibrary:
|
||||
"""Tests for mobile offline library management."""
|
||||
|
||||
def test_get_offline_library_requires_auth(self, client):
|
||||
"""Get offline library should require authentication."""
|
||||
response = client.get("/api/mobile-offline/devices/test-device/offline-library")
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_add_tracks_requires_auth(self, client):
|
||||
"""Add tracks to offline should require authentication."""
|
||||
response = client.post("/api/mobile-offline/devices/test-device/add-tracks", json={
|
||||
"tracks": [],
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_remove_tracks_requires_auth(self, client):
|
||||
"""Remove tracks from offline should require authentication."""
|
||||
response = client.post("/api/mobile-offline/devices/test-device/remove-tracks", json={
|
||||
"trackhashes": [],
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
|
||||
class TestMobileOfflineLibraryWithAuth:
|
||||
"""Tests for offline library operations with authentication."""
|
||||
|
||||
def test_get_offline_library_for_device(self, client, auth_headers):
|
||||
"""Get offline library should return library for registered device."""
|
||||
# Register device first
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Library Test Device",
|
||||
"type": "phone",
|
||||
"device_id": "library-test-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
"/api/mobile-offline/devices/library-test-device-12345/offline-library",
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Should return library or 404 if device not found
|
||||
assert response.status_code in [200, 404]
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.get_json()
|
||||
assert isinstance(data, dict)
|
||||
|
||||
def test_add_tracks_to_offline(self, client, auth_headers):
|
||||
"""Add tracks to offline library should work."""
|
||||
# Register device first
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Add Tracks Device",
|
||||
"type": "phone",
|
||||
"device_id": "add-tracks-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/add-tracks-device-12345/add-tracks",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"tracks": [
|
||||
{"trackhash": "testhash1", "title": "Test Track 1"},
|
||||
{"trackhash": "testhash2", "title": "Test Track 2"},
|
||||
],
|
||||
"quality": "high",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 201, 404]
|
||||
|
||||
def test_remove_tracks_from_offline(self, client, auth_headers):
|
||||
"""Remove tracks from offline library should work."""
|
||||
# Register device first
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Remove Tracks Device",
|
||||
"type": "phone",
|
||||
"device_id": "remove-tracks-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/remove-tracks-device-12345/remove-tracks",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"trackhashes": ["testhash1", "testhash2"],
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 404]
|
||||
|
||||
|
||||
class TestMobileSyncOperations:
|
||||
"""Tests for mobile sync operations."""
|
||||
|
||||
def test_sync_collection_requires_auth(self, client):
|
||||
"""Sync collection should require authentication."""
|
||||
response = client.post("/api/mobile-offline/devices/test-device/sync-collection", json={
|
||||
"collection_type": "playlist",
|
||||
"collection_id": "test123",
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_get_sync_progress_requires_auth(self, client):
|
||||
"""Get sync progress should require authentication."""
|
||||
response = client.get("/api/mobile-offline/devices/test-device/sync-progress")
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
|
||||
class TestMobileSyncWithAuth:
|
||||
"""Tests for sync operations with authentication."""
|
||||
|
||||
def test_sync_collection_to_device(self, client, auth_headers):
|
||||
"""Sync collection to device should initiate sync."""
|
||||
# Register device first
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Sync Device",
|
||||
"type": "phone",
|
||||
"device_id": "sync-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/sync-device-12345/sync-collection",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"collection_type": "playlist",
|
||||
"collection_id": "test-playlist-id",
|
||||
"quality": "high",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 201, 404, 400]
|
||||
|
||||
def test_get_sync_progress_for_device(self, client, auth_headers):
|
||||
"""Get sync progress should return progress info."""
|
||||
# Register device first
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Progress Device",
|
||||
"type": "phone",
|
||||
"device_id": "progress-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
"/api/mobile-offline/devices/progress-device-12345/sync-progress",
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 404]
|
||||
|
||||
|
||||
class TestMobileEvents:
|
||||
"""Tests for mobile event batching."""
|
||||
|
||||
def test_push_events_requires_auth(self, client):
|
||||
"""Push events should require authentication."""
|
||||
response = client.post("/api/mobile-offline/devices/test-device/events/batch", json={
|
||||
"events": [],
|
||||
})
|
||||
assert response.status_code in [401, 423]
|
||||
|
||||
def test_push_events_to_device(self, client, auth_headers):
|
||||
"""Push events should batch upload events."""
|
||||
# Register device first
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Events Device",
|
||||
"type": "phone",
|
||||
"device_id": "events-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/events-device-12345/events/batch",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"events": [
|
||||
{
|
||||
"type": "play",
|
||||
"trackhash": "testhash123",
|
||||
"timestamp": 1234567890,
|
||||
"duration": 180,
|
||||
},
|
||||
{
|
||||
"type": "favorite",
|
||||
"trackhash": "testhash456",
|
||||
"timestamp": 1234567891,
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 404]
|
||||
|
||||
|
||||
class TestMobileOfflineContract:
|
||||
"""Contract tests for mobile offline API."""
|
||||
|
||||
def test_device_registration_schema(self, client, auth_headers):
|
||||
"""Device registration response should match expected schema."""
|
||||
response = client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Contract Test Device",
|
||||
"type": "phone",
|
||||
"device_id": "contract-device-12345",
|
||||
"storage_capacity": 128000,
|
||||
"available_storage": 64000,
|
||||
},
|
||||
)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.get_json()
|
||||
# Should have device identifier
|
||||
assert "device_id" in data or "id" in data
|
||||
|
||||
def test_offline_library_schema(self, client, auth_headers):
|
||||
"""Offline library response should match expected schema."""
|
||||
# Register device
|
||||
client.post(
|
||||
"/api/mobile-offline/devices/register",
|
||||
headers=auth_headers,
|
||||
json={
|
||||
"name": "Schema Test Device",
|
||||
"type": "phone",
|
||||
"device_id": "schema-device-12345",
|
||||
},
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
"/api/mobile-offline/devices/schema-device-12345/offline-library",
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.get_json()
|
||||
# Library should be a dict with tracks or collections
|
||||
assert isinstance(data, dict)
|
||||
Reference in New Issue
Block a user