326 lines
12 KiB
Python
326 lines
12 KiB
Python
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"]
|