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 / "deluge_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 [ "DELUGE_URL", "DELUGE_PASSWORD", "DELUGE_VERIFY_TLS", "DELUGE_TIMEOUT_SECONDS", "DELUGE_DEFAULT_HOST_ID", "DELUGE_WEB_PATH", ]: 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_deluge_config(url: str | None = "http://deluge:8112", password: str | None = "DELUGE_SECRET_PASSWORD"): from backend.app.core.config import settings previous = dict(settings.apis) settings.apis["DELUGE_URL"] = url settings.apis["DELUGE_PASSWORD"] = password settings.apis["DELUGE_VERIFY_TLS"] = False settings.apis["DELUGE_TIMEOUT_SECONDS"] = "10" settings.apis["DELUGE_WEB_PATH"] = "/json" return previous def _restore_config(previous): from backend.app.core.config import settings settings.apis.clear() settings.apis.update(previous) def _torrent_payload(): return { "active-id": { "name": "Active Linux ISO", "state": "Downloading", "progress": 42.5, "download_payload_rate": 2048, "upload_payload_rate": 512, "eta": 120, "ratio": 0.4, "total_size": 1000000, "tracker_host": "tracker.example", "label": "linux", "save_path": "/srv/example/downloads/linux", "is_finished": False, }, "paused-id": { "name": "Paused Media", "state": "Paused", "progress": 12, "download_payload_rate": 0, "upload_payload_rate": 0, "eta": 0, "ratio": 0, "total_size": 2000000, "tracker_host": "tracker.example", "label": "tv", "save_path": "/srv/example/downloads/tv", "is_finished": False, }, "error-id": { "name": "Broken Torrent", "state": "Error", "progress": 50, "download_payload_rate": 0, "upload_payload_rate": 0, "eta": 0, "ratio": 0.1, "total_size": 3000000, "tracker_host": "bad.example", "label": "", "save_path": "/srv/example/downloads/broken", "error": "Tracker failed", "is_finished": False, }, } def test_deluge_connector_missing_config_reports_required_env(api_client): response = api_client.post("/api/connectors/deluge/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"] == ["DELUGE_URL", "DELUGE_PASSWORD"] assert payload["secretRef"] == "DELUGE_PASSWORD" def test_deluge_health_success_normalizes_response(monkeypatch): from backend.app.connectors.deluge import DelugeConnector previous = _set_deluge_config() monkeypatch.setattr(DelugeConnector, "_login", lambda self: True, raising=True) monkeypatch.setattr( DelugeConnector, "_rpc", lambda self, method, params=None, authed=True: { "web.connected": True, "daemon.connected": True, }.get(method, "2.1.1"), raising=True, ) payload = DelugeConnector().health() assert payload["success"] is True assert payload["status"] == "connected" assert payload["webConnected"] is True assert payload["daemonConnected"] is True assert payload["version"] == "2.1.1" assert payload["url"] == "http://deluge:8112" assert "DELUGE_SECRET_PASSWORD" not in str(payload) _restore_config(previous) def test_deluge_auth_failure_redacts_password(monkeypatch): from backend.app.connectors.deluge import DelugeConnector previous = _set_deluge_config() def fail_login(self): raise RuntimeError("auth_failed password=DELUGE_SECRET_PASSWORD") monkeypatch.setattr(DelugeConnector, "_login", fail_login, raising=True) payload = DelugeConnector().health() assert payload["success"] is False assert payload["status"] == "auth_failed" assert "DELUGE_SECRET_PASSWORD" not in str(payload) assert "[REDACTED]" in str(payload) _restore_config(previous) def test_deluge_health_uses_session_fallback_when_optional_methods_unknown(monkeypatch): from backend.app.connectors.deluge import DelugeConnector previous = _set_deluge_config() monkeypatch.setattr(DelugeConnector, "_login", lambda self: True, raising=True) def fake_rpc(self, method, params=None, authed=True): if method in {"web.connected", "daemon.connected", "daemon.get_version"}: raise RuntimeError("Unknown method") if method == "core.get_session_status": return {"num_torrents": 57} raise AssertionError(f"unexpected method {method}") monkeypatch.setattr(DelugeConnector, "_rpc", fake_rpc, raising=True) payload = DelugeConnector().health() assert payload["success"] is True assert payload["status"] == "connected" assert payload["message"] == "Deluge connector ready." assert payload["diagnostics"]["healthFallback"] == "core.get_session_status" assert payload["diagnostics"]["optionalMethodWarnings"] _restore_config(previous) def test_deluge_session_status_normalizes_rates_and_counts(monkeypatch): from backend.app.connectors.deluge import DelugeConnector previous = _set_deluge_config() monkeypatch.setattr(DelugeConnector, "_login", lambda self: True, raising=True) def fake_rpc(self, method, params=None, authed=True): if method == "web.connected": return True if method == "core.get_session_status": return {"download_rate": 4096, "upload_rate": 1024, "num_torrents": 3} if method == "web.update_ui": return {"torrents": _torrent_payload()} return None monkeypatch.setattr(DelugeConnector, "_rpc", fake_rpc, raising=True) payload = DelugeConnector().session_status() assert payload["success"] is True assert payload["connected"] is True assert payload["downloadRate"] == 4096 assert payload["uploadRate"] == 1024 assert payload["numTorrents"] == 3 assert payload["active"] == 1 assert payload["paused"] == 1 assert payload["stalled"] == 1 assert payload["errors"] == 1 _restore_config(previous) def test_deluge_torrent_status_and_filters(monkeypatch): from backend.app.connectors.deluge import DelugeConnector previous = _set_deluge_config() monkeypatch.setattr(DelugeConnector, "_login", lambda self: True, raising=True) monkeypatch.setattr( DelugeConnector, "_rpc", lambda self, method, params=None, authed=True: {"torrents": _torrent_payload()} if method == "web.update_ui" else True, raising=True, ) torrents = DelugeConnector().torrent_status() active = DelugeConnector().active_downloads() stalled = DelugeConnector().stalled_downloads() paused = DelugeConnector().paused_downloads() assert torrents["success"] is True assert torrents["total"] == 3 assert torrents["items"][0]["id"] == "active-id" assert torrents["items"][0]["category"] == "linux" assert torrents["items"][0]["savePath"] == ".../linux" assert "DELUGE_SECRET_PASSWORD" not in str(torrents) assert active["total"] == 1 assert active["items"][0]["id"] == "active-id" assert stalled["total"] == 1 assert stalled["items"][0]["id"] == "error-id" assert paused["total"] == 1 assert paused["items"][0]["id"] == "paused-id" _restore_config(previous) def test_deluge_speed_status_uses_session_rates(monkeypatch): from backend.app.connectors.deluge import DelugeConnector previous = _set_deluge_config() monkeypatch.setattr( DelugeConnector, "session_status", lambda self: {"success": True, "connected": True, "downloadRate": 100, "uploadRate": 25, "numTorrents": 2, "active": 1, "paused": 0, "stalled": 0, "errors": 0}, raising=True, ) payload = DelugeConnector().speed_status() assert payload["success"] is True assert payload["downloadRate"] == 100 assert payload["uploadRate"] == 25 assert payload["active"] == 1 _restore_config(previous) def test_deluge_pause_torrent_allowed_creates_job_and_audit(api_client, monkeypatch): monkeypatch.setattr( "backend.app.connectors.deluge.DelugeConnector.pause_torrent", lambda self, torrent_id: {"success": True, "status": "ok", "message": "torrent_paused", "data": {"torrentId": torrent_id}}, raising=True, ) response = api_client.post("/api/apps/deluge/torrents/active-id/pause", 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"]["torrentId"] == "active-id" def test_deluge_resume_torrent_voice_requires_approval_without_connector_call(api_client, monkeypatch): def fail_resume(*_args, **_kwargs): raise AssertionError("connector should not run before approval") monkeypatch.setattr("backend.app.connectors.deluge.DelugeConnector.resume_torrent", fail_resume, raising=True) response = api_client.post("/api/apps/deluge/torrents/paused-id/resume", json={"source": "voice"}) payload = response.json() assert payload["success"] is False assert payload["decision"] == "approval_required" assert payload["approvalId"].startswith("appr-") assert payload["auditId"].startswith("audit-") def test_deluge_pause_all_resume_all_require_approval_or_block_without_connector_call(api_client, monkeypatch): def fail_bulk(*_args, **_kwargs): raise AssertionError("bulk connector call should wait for approval") monkeypatch.setattr("backend.app.connectors.deluge.DelugeConnector.pause_all", fail_bulk, raising=True) monkeypatch.setattr("backend.app.connectors.deluge.DelugeConnector.resume_all", fail_bulk, raising=True) manual_pause = api_client.post("/api/apps/deluge/pause-all", json={"source": "manual"}).json() manual_resume = api_client.post("/api/apps/deluge/resume-all", json={"source": "manual"}).json() voice_pause = api_client.post("/api/apps/deluge/pause-all", json={"source": "voice"}).json() voice_resume = api_client.post("/api/apps/deluge/resume-all", json={"source": "voice"}).json() assert manual_pause["decision"] == "approval_required" assert manual_resume["decision"] == "approval_required" assert voice_pause["decision"] == "blocked" assert voice_resume["decision"] == "blocked" def test_deluge_pause_all_executes_only_after_separate_approval_step(api_client, monkeypatch): calls = [] def pause_all(_self): calls.append('pause') return {'success': True, 'status': 'ok', 'message': 'all_torrents_paused'} monkeypatch.setattr("backend.app.connectors.deluge.DelugeConnector.pause_all", pause_all, raising=True) requested = api_client.post("/api/apps/deluge/pause-all", json={"source": "manual"}).json() assert calls == [] approved = api_client.patch(f"/api/approvals/{requested['approvalId']}", json={"status": "approved"}).json() assert approved['status'] == 'approved' assert calls == [] executed = api_client.post(f"/api/approvals/{requested['approvalId']}/execute").json() assert executed['success'] is True assert executed['approvalStatus'] == 'executed' assert executed['jobId'].startswith('job-') assert executed['auditId'].startswith('audit-') assert calls == ['pause'] approvals = api_client.get('/api/approvals').json() approval = next(item for item in approvals if item['id'] == requested['approvalId']) assert approval['status'] == 'executed' assert approval['job_id'] == executed['jobId'] def test_deluge_destructive_actions_are_blocked(api_client): for action in [ "deluge.remove_torrent", "deluge.remove_torrent_data", "deluge.delete_data", "deluge.force_recheck", "deluge.move_storage", "deluge.add_torrent", "deluge.add_magnet", "deluge.mass_edit", "deluge.set_options", "deluge.unknown_mutation", ]: payload = api_client.post("/api/policies/evaluate", json={"action": action, "target": "deluge", "source": "manual"}).json() assert payload["decision"] == "blocked" def test_deluge_inventory_classification_and_candidate(api_client, monkeypatch): from backend.app.services.app_classifier import classify_container result = classify_container({"name": "binhex-delugevpn", "image": "binhex/arch-delugevpn:latest", "ports": ["8112:8112"]}) assert result["app_type"] == "deluge" assert result["category"] == "downloads" assert result["subcategory"] == "torrent_client" assert result["connector"] == "deluge" assert result["policy_profile"] == "download_client" assert result["criticality"] == "medium" assert result["healthcheck_suggestion"] == {"port": 8112} 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": "deluge-container", "name": "binhex-delugevpn", "image": "binhex/arch-delugevpn", "state": "running", "status": "running", "ports": ["8112:8112"]}]}, raising=True, ) api_client.post("/api/inventory/discover") candidates = api_client.get("/api/connectors/deluge/candidates").json() assert len(candidates) == 1 assert candidates[0]["connectorId"] == "deluge" assert candidates[0]["appType"] == "deluge" assert candidates[0]["missingSecrets"] == ["DELUGE_URL", "DELUGE_PASSWORD"]