from pathlib import Path import pytest from fastapi.testclient import TestClient from sqlmodel import SQLModel, create_engine @pytest.fixture def api_client(tmp_path, monkeypatch): db_path = tmp_path / "plex_test.db" monkeypatch.setenv("APPOPS_DB_PATH", str(db_path)) monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token") from backend.app.core.config import settings from backend.app.db import session from backend.app.db.init_db import init_db from backend.app.main import app previous_apis = dict(settings.apis) for key in ["PLEX_URL", "PLEX_TOKEN", "PLEX_VERIFY_TLS", "PLEX_TIMEOUT_SECONDS", "PLEX_DEFAULT_LIBRARY_ID", "PLEX_MACHINE_IDENTIFIER"]: settings.apis[key] = None engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False}) session.engine = engine SQLModel.metadata.drop_all(engine) SQLModel.metadata.create_all(engine) init_db() with TestClient(app) as client: yield client settings.apis = previous_apis def _set_plex_config(url: str | None = "http://plex:32400", token: str | None = "PLEX_SECRET_TOKEN"): from backend.app.core.config import settings previous = dict(settings.apis) settings.apis["PLEX_URL"] = url settings.apis["PLEX_TOKEN"] = token settings.apis["PLEX_VERIFY_TLS"] = False settings.apis["PLEX_TIMEOUT_SECONDS"] = "10" settings.apis["PLEX_DEFAULT_LIBRARY_ID"] = None settings.apis["PLEX_MACHINE_IDENTIFIER"] = None return previous def _restore_config(previous): from backend.app.core.config import settings settings.apis.clear() settings.apis.update(previous) def _server_payload(): return { "MediaContainer": { "friendlyName": "Widefrog Plex", "machineIdentifier": "machine-123456789", "version": "1.40.2", "platform": "Linux", "updatedAt": 1700000000, "myPlex": True, } } def _libraries_payload(): return { "MediaContainer": { "Directory": [ {"key": "1", "title": "Movies", "type": "movie", "agent": "tv.plex.agents.movie", "scanner": "Plex Movie", "Location": [{"path": "/srv/example/media/movies"}]}, {"key": "2", "title": "Shows", "type": "show", "agent": "tv.plex.agents.series", "scanner": "Plex Series", "Location": [{"path": "/srv/example/media/tv"}]}, ] } } def _sessions_payload(): return { "MediaContainer": { "size": 2, "Metadata": [ { "sessionKey": "11", "title": "Direct Movie", "type": "movie", "User": {"title": "Jens"}, "Player": {"title": "Apple TV", "state": "playing", "product": "Plex", "platform": "tvOS", "address": "198.18.10.55"}, "Session": {"bandwidth": 12000}, "Media": [{"Part": [{"decision": "directplay"}], "videoDecision": "directplay", "audioDecision": "directplay"}], "viewOffset": 120000, "duration": 240000, }, { "sessionKey": "12", "title": "Transcoded Episode", "type": "episode", "User": {"title": "Remote"}, "Player": {"title": "Chrome", "state": "playing", "product": "Plex Web", "platform": "Windows", "address": "203.0.113.9"}, "Session": {"bandwidth": 4500}, "TranscodeSession": {"videoDecision": "transcode", "audioDecision": "copy"}, "Media": [{"videoDecision": "transcode", "audioDecision": "copy"}], "viewOffset": 50000, "duration": 100000, }, ], } } def _recent_payload(): return { "MediaContainer": { "Metadata": [ {"title": "Arrival", "type": "movie", "year": 2016, "librarySectionTitle": "Movies", "addedAt": 1700000000, "thumb": "/library/metadata/1/thumb", "summary": "A quiet first contact story."}, {"title": "Pilot", "grandparentTitle": "Example Show", "type": "episode", "year": 2024, "librarySectionTitle": "Shows", "addedAt": 1700000500}, ] } } def test_plex_connector_missing_config_reports_required_env(api_client): response = api_client.post("/api/connectors/plex/test") assert response.status_code == 200 payload = response.json() assert payload["success"] is False assert payload["status"] == "missing_config" assert payload["missing_config"] is True assert payload["required_env_vars"] == ["PLEX_URL", "PLEX_TOKEN"] assert payload["secretRef"] == "PLEX_TOKEN" def test_plex_health_success_normalizes_response(monkeypatch): from backend.app.connectors.plex import PlexConnector previous = _set_plex_config() monkeypatch.setattr(PlexConnector, "_request", lambda self, path, params=None: _server_payload(), raising=True) payload = PlexConnector().health() assert payload["success"] is True assert payload["status"] == "connected" assert payload["friendlyName"] == "Widefrog Plex" assert payload["version"] == "1.40.2" assert payload["machineIdentifier"] == "machine-123456789" assert payload["url"] == "http://plex:32400" assert "PLEX_SECRET_TOKEN" not in str(payload) _restore_config(previous) def test_plex_auth_failure_redacts_token(monkeypatch): from backend.app.connectors.plex import PlexConnector previous = _set_plex_config() def fail_request(self, path, params=None): raise RuntimeError("auth_failed token=PLEX_SECRET_TOKEN") monkeypatch.setattr(PlexConnector, "_request", fail_request, raising=True) payload = PlexConnector().health() assert payload["success"] is False assert payload["status"] == "auth_failed" assert "PLEX_SECRET_TOKEN" not in str(payload) assert "[REDACTED]" in str(payload) _restore_config(previous) def test_plex_library_status_normalizes_sections(monkeypatch): from backend.app.connectors.plex import PlexConnector previous = _set_plex_config() def fake_request(self, path, params=None): if path == "/library/sections": return _libraries_payload() if path == "/library/sections/1/all": return {"MediaContainer": {"size": 14}} if path == "/library/sections/2/all": return {"MediaContainer": {"size": 8}} return _server_payload() monkeypatch.setattr(PlexConnector, "_request", fake_request, raising=True) payload = PlexConnector().library_status() assert payload["success"] is True assert payload["total"] == 2 assert payload["items"][0]["id"] == "1" assert payload["items"][0]["type"] == "movie" assert payload["items"][0]["count"] == 14 assert payload["items"][0]["locations"] == [".../movies"] _restore_config(previous) def test_plex_sessions_and_stream_summary(monkeypatch): from backend.app.connectors.plex import PlexConnector previous = _set_plex_config() monkeypatch.setattr(PlexConnector, "_request", lambda self, path, params=None: _sessions_payload(), raising=True) sessions = PlexConnector().active_sessions() summary = PlexConnector().stream_summary() assert sessions["success"] is True assert sessions["total"] == 2 assert sessions["items"][0]["user"] == "Jens" assert sessions["items"][0]["ip"] == "192.168.10.x" assert sessions["items"][1]["transcodeDecision"] == "transcode" assert summary["activeStreams"] == 2 assert summary["directPlay"] == 1 assert summary["transcoding"] == 1 assert summary["users"] == 2 _restore_config(previous) def test_plex_recent_media_normalizes_items(monkeypatch): from backend.app.connectors.plex import PlexConnector previous = _set_plex_config() monkeypatch.setattr(PlexConnector, "_request", lambda self, path, params=None: _recent_payload(), raising=True) payload = PlexConnector().recent_media() assert payload["success"] is True assert payload["total"] == 2 assert payload["items"][0]["title"] == "Arrival" assert payload["items"][1]["title"] == "Example Show - Pilot" _restore_config(previous) def test_plex_scan_library_allowed_creates_job_and_audit(api_client, monkeypatch): monkeypatch.setattr( "backend.app.connectors.plex.PlexConnector.scan_library", lambda self, library_id: {"success": True, "status": "ok", "message": "scan_started", "data": {"libraryId": library_id}}, raising=True, ) response = api_client.post("/api/apps/plex/libraries/1/scan", json={"source": "manual"}) payload = response.json() assert response.status_code == 200 assert payload["success"] is True assert payload["decision"] == "allowed" assert payload["jobId"].startswith("job-") assert payload["auditId"].startswith("audit-") assert payload["data"]["libraryId"] == "1" def test_plex_scan_library_voice_requires_approval_without_connector_call(api_client, monkeypatch): def fail_scan(*_args, **_kwargs): raise AssertionError("connector should not run before approval") monkeypatch.setattr("backend.app.connectors.plex.PlexConnector.scan_library", fail_scan, raising=True) payload = api_client.post("/api/apps/plex/libraries/1/scan", json={"source": "voice"}).json() assert payload["success"] is False assert payload["decision"] == "approval_required" assert payload["approvalId"].startswith("appr-") assert payload["auditId"].startswith("audit-") def test_plex_scan_all_requires_approval_or_blocks_without_connector_call(api_client, monkeypatch): def fail_scan_all(*_args, **_kwargs): raise AssertionError("scan all should wait for approval") monkeypatch.setattr("backend.app.connectors.plex.PlexConnector.scan_all_libraries", fail_scan_all, raising=True) manual = api_client.post("/api/apps/plex/libraries/scan-all", json={"source": "manual"}).json() voice = api_client.post("/api/apps/plex/libraries/scan-all", json={"source": "voice"}).json() assert manual["decision"] == "approval_required" assert voice["decision"] == "blocked" def test_plex_scan_all_executes_after_approval(api_client, monkeypatch): calls = [] def scan_all(_self): calls.append('scan') return {'success': True, 'status': 'accepted', 'message': 'scan_all_started'} monkeypatch.setattr("backend.app.connectors.plex.PlexConnector.scan_all_libraries", scan_all, raising=True) requested = api_client.post("/api/apps/plex/libraries/scan-all", json={"source": "manual"}).json() api_client.patch(f"/api/approvals/{requested['approvalId']}", json={"status": "approved"}) executed = api_client.post(f"/api/approvals/{requested['approvalId']}/execute").json() assert executed['success'] is True assert executed['approvalStatus'] == 'executed' assert calls == ['scan'] def test_plex_destructive_actions_are_blocked(api_client): for action in [ "plex.delete_media", "plex.delete_library", "plex.empty_trash", "plex.clean_bundles", "plex.optimize_database", "plex.kill_session", "plex.user_management", "plex.force_refresh_all_metadata", "plex.mass_edit", "plex.unknown_mutation", ]: payload = api_client.post("/api/policies/evaluate", json={"action": action, "target": "plex", "source": "manual"}).json() assert payload["decision"] == "blocked" def test_plex_inventory_classification_and_candidate(api_client, monkeypatch): from backend.app.services.app_classifier import classify_container result = classify_container({"name": "binhex-plexpass", "image": "binhex/arch-plexpass:latest", "ports": ["32400:32400"]}) assert result["app_type"] == "plex" assert result["category"] == "media" assert result["subcategory"] == "media_server" assert result["connector"] == "plex" assert result["policy_profile"] == "media_server" assert result["healthcheck_suggestion"] == {"port": 32400} monkeypatch.setattr("backend.app.connectors.portainer.PortainerConnector.test_connection", lambda self: {"status": "connected"}, raising=True) monkeypatch.setattr( "backend.app.connectors.portainer.PortainerConnector.list_containers", lambda self: {"status": "ok", "containers": [{"id": "plex-container", "name": "binhex-plexpass", "image": "binhex/arch-plexpass", "state": "running", "status": "running", "ports": ["32400:32400"]}]}, raising=True, ) api_client.post("/api/inventory/discover") candidates = api_client.get("/api/connectors/plex/candidates").json() assert len(candidates) == 1 assert candidates[0]["connectorId"] == "plex" assert candidates[0]["appType"] == "plex" assert candidates[0]["missingSecrets"] == ["PLEX_URL", "PLEX_TOKEN"]