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 / "tautulli_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 ["TAUTULLI_URL", "TAUTULLI_API_KEY", "TAUTULLI_VERIFY_TLS", "TAUTULLI_TIMEOUT_SECONDS"]: 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_tautulli_config(url: str | None = "http://tautulli:8181", api_key: str | None = "TAUTULLI_SECRET_KEY"): from backend.app.core.config import settings previous = dict(settings.apis) settings.apis["TAUTULLI_URL"] = url settings.apis["TAUTULLI_API_KEY"] = api_key settings.apis["TAUTULLI_VERIFY_TLS"] = False settings.apis["TAUTULLI_TIMEOUT_SECONDS"] = "10" return previous def _restore_config(previous): from backend.app.core.config import settings settings.apis.clear() settings.apis.update(previous) def _api_payload(data, result="success", message="ok"): return {"response": {"result": result, "message": message, "data": data}} def _server_info(): return _api_payload({ "tautulli_version": "2.14.0", "pms_name": "Widefrog Plex", "pms_version": "1.41.0", "pms_platform": "Linux", "pms_identifier": "machine-abcdef123456", "pms_is_plexpass": True, }) def _activity(): return _api_payload({ "stream_count": 2, "lan_bandwidth": 12000, "wan_bandwidth": 4500, "total_bandwidth": 16500, "sessions": [ { "session_key": "1", "friendly_name": "Jens", "full_title": "Direct Movie", "media_type": "movie", "player": "Apple TV", "platform": "tvOS", "state": "playing", "progress_percent": 55, "quality_profile": "1080p", "transcode_decision": "direct play", "video_decision": "direct play", "audio_decision": "direct play", "bandwidth": 12000, "ip_address": "198.18.10.55", "location": "lan", }, { "session_key": "2", "username": "Remote", "full_title": "Transcoded Episode", "media_type": "episode", "player": "Chrome", "platform": "Windows", "state": "playing", "progress_percent": 10, "quality_profile": "720p", "transcode_decision": "transcode", "video_decision": "transcode", "audio_decision": "copy", "bandwidth": 4500, "ip_address": "203.0.113.9", "location": "wan", }, ], }) def _history(): return _api_payload({"data": [ {"row_id": 10, "friendly_name": "Jens", "full_title": "Arrival", "media_type": "movie", "watched_status": 1, "date": 1700000000, "duration": 7200, "platform": "tvOS", "transcode_decision": "direct play"}, {"row_id": 11, "username": "Remote", "full_title": "Example Show - Pilot", "media_type": "episode", "watched_status": 0, "date": 1700000500, "duration": 1800, "platform": "Web", "transcode_decision": "transcode"}, ]}) def _users(): return _api_payload([ {"user_id": 1, "username": "jens", "friendly_name": "Jens", "email": "secret@example.test", "thumb": "http://avatar", "last_seen": 1700000000, "plays": 42, "duration": 15000}, {"user_id": 2, "username": "remote", "friendly_name": "Remote", "last_seen": 1700000500, "plays": 7, "duration": 2400}, ]) def _libraries(): return _api_payload([ {"section_id": 1, "section_name": "Movies", "section_type": "movie", "count": 1200, "parent_count": 0, "child_count": 0, "last_accessed": 1700000000}, {"section_id": 2, "section_name": "Shows", "section_type": "show", "count": 120, "parent_count": 120, "child_count": 1200}, ]) def _recently_added(): return _api_payload({"recently_added": [ {"title": "Dune", "media_type": "movie", "section_name": "Movies", "added_at": 1700000000, "year": 2024, "thumb": "/library/metadata/1/thumb"}, {"title": "Pilot", "grandparent_title": "Example Show", "media_type": "episode", "section_name": "Shows", "added_at": 1700000500, "year": 2025}, ]}) def _plays_by_date(): return _api_payload({"data": [{"date": "2026-06-24", "plays": 3, "duration": 3000}, {"date": "2026-06-25", "plays": 5, "duration": 6000}]}) def _top_users(): return _api_payload({"data": [{"friendly_name": "Jens", "plays": 8, "duration": 9000}]}) def _top_platforms(): return _api_payload({"data": [{"platform": "tvOS", "plays": 6}, {"platform": "Web", "plays": 2}]}) def test_tautulli_connector_missing_config_reports_required_env(api_client): response = api_client.post("/api/connectors/tautulli/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"] == ["TAUTULLI_URL", "TAUTULLI_API_KEY"] assert payload["secretRef"] == "TAUTULLI_API_KEY" def test_tautulli_health_success_normalizes_response(monkeypatch): from backend.app.connectors.tautulli import TautulliConnector previous = _set_tautulli_config() monkeypatch.setattr(TautulliConnector, "_command", lambda self, cmd, params=None: _server_info(), raising=True) payload = TautulliConnector().health() assert payload["success"] is True assert payload["status"] == "connected" assert payload["tautulliVersion"] == "2.14.0" assert payload["plexServerName"] == "Widefrog Plex" assert payload["url"] == "http://tautulli:8181" assert "TAUTULLI_SECRET_KEY" not in str(payload) _restore_config(previous) def test_tautulli_auth_failure_redacts_api_key(monkeypatch): from backend.app.connectors.tautulli import TautulliConnector previous = _set_tautulli_config() def fail_command(self, cmd, params=None): raise RuntimeError("auth_failed apikey=TAUTULLI_SECRET_KEY") monkeypatch.setattr(TautulliConnector, "_command", fail_command, raising=True) payload = TautulliConnector().health() assert payload["success"] is False assert payload["status"] == "auth_failed" assert "TAUTULLI_SECRET_KEY" not in str(payload) assert "[REDACTED]" in str(payload) _restore_config(previous) def test_tautulli_activity_and_stream_summary(monkeypatch): from backend.app.connectors.tautulli import TautulliConnector previous = _set_tautulli_config() monkeypatch.setattr(TautulliConnector, "_command", lambda self, cmd, params=None: _activity(), raising=True) activity = TautulliConnector().activity() summary = TautulliConnector().stream_summary() assert activity["success"] is True assert activity["streamCount"] == 2 assert activity["sessions"][0]["ipAddress"] == "192.168.10.x" assert activity["sessions"][1]["transcodeDecision"] == "transcode" assert summary["activeStreams"] == 2 assert summary["directPlay"] == 1 assert summary["transcoding"] == 1 assert summary["lanStreams"] == 1 assert summary["wanStreams"] == 1 _restore_config(previous) def test_tautulli_history_users_libraries_recent_and_stats(monkeypatch): from backend.app.connectors.tautulli import TautulliConnector previous = _set_tautulli_config() def fake_command(self, cmd, params=None): return { "get_history": _history(), "get_users": _users(), "get_libraries": _libraries(), "get_recently_added": _recently_added(), "get_plays_by_date": _plays_by_date(), "get_plays_by_top_10_users": _top_users(), "get_plays_by_top_10_platforms": _top_platforms(), }[cmd] monkeypatch.setattr(TautulliConnector, "_command", fake_command, raising=True) connector = TautulliConnector() history = connector.history() users = connector.users() libraries = connector.libraries() recent = connector.recently_added() stats = connector.watch_stats() assert history["items"][0]["rowId"] == 10 assert history["items"][1]["transcodeDecision"] == "transcode" assert users["items"][0]["friendlyName"] == "Jens" assert "email" not in users["items"][0] assert "secret@example.test" not in str(users) assert libraries["items"][0]["sectionId"] == "1" assert recent["items"][1]["title"] == "Example Show - Pilot" assert stats["totalPlays"] == 8 assert stats["topUsers"][0]["name"] == "Jens" assert stats["topPlatforms"][0]["platform"] == "tvOS" _restore_config(previous) def test_tautulli_plex_info_normalizes_server_info(monkeypatch): from backend.app.connectors.tautulli import TautulliConnector previous = _set_tautulli_config() monkeypatch.setattr(TautulliConnector, "_command", lambda self, cmd, params=None: _server_info(), raising=True) payload = TautulliConnector().plex_info() assert payload["success"] is True assert payload["name"] == "Widefrog Plex" assert payload["version"] == "1.41.0" assert payload["machineIdentifier"] == "machine-..." assert payload["plexPass"] is True _restore_config(previous) def test_tautulli_privacy_policies_and_destructive_actions_are_blocked(api_client): for action in ["tautulli.history", "tautulli.users", "tautulli.watch_stats"]: voice = api_client.post("/api/policies/evaluate", json={"action": action, "target": "tautulli", "source": "voice"}).json() assert voice["decision"] in {"approval_required", "blocked"} for action in [ "tautulli.terminate_session", "tautulli.download_database", "tautulli.download_config", "tautulli.delete_notifier", "tautulli.notify", "tautulli.user_management", "tautulli.mass_edit", "tautulli.unknown_mutation", ]: payload = api_client.post("/api/policies/evaluate", json={"action": action, "target": "tautulli", "source": "manual"}).json() assert payload["decision"] == "blocked" def test_tautulli_inventory_classification_and_candidate(api_client, monkeypatch): from backend.app.services.app_classifier import classify_container result = classify_container({"name": "binhex-tautulli", "image": "linuxserver/tautulli:latest", "ports": ["8181:8181"]}) assert result["app_type"] == "tautulli" assert result["category"] == "media" assert result["subcategory"] == "plex_analytics" assert result["connector"] == "tautulli" assert result["policy_profile"] == "media_analytics" assert result["privacy_sensitive"] is True 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": "tautulli-container", "name": "binhex-tautulli", "image": "linuxserver/tautulli", "state": "running", "status": "running", "ports": ["8181:8181"]}]}, raising=True, ) api_client.post("/api/inventory/discover") candidates = api_client.get("/api/connectors/tautulli/candidates").json() assert len(candidates) == 1 assert candidates[0]["connectorId"] == "tautulli" assert candidates[0]["appType"] == "tautulli" assert candidates[0]["missingSecrets"] == ["TAUTULLI_URL", "TAUTULLI_API_KEY"]