406 lines
15 KiB
Python
406 lines
15 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 / "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"]
|