import os
import asyncio
import json
import time
from datetime import datetime, timedelta, timezone

os.environ.setdefault("BUTLER_TOKEN", "test-token")

from fastapi.testclient import TestClient

import app


def test_redact_response_recursively():
    source = {
        "id": 6,
        "hawserToken": "secret-value",
        "nested": [{"webhookSecret": "also-secret", "name": "tdarr"}],
    }
    result = app._redact_response(source)
    assert result["hawserToken"] == "[REDACTED]"
    assert result["nested"][0]["webhookSecret"] == "[REDACTED]"
    assert result["nested"][0]["name"] == "tdarr"


def test_inventory_defaults_node7_and_explicit_user():
    inventory = """
[node]
node2 ansible_host=10.5.85.12
[apps]
tdarr ansible_host=10.2.1.104
emby-chris ansible_host=10.7.1.106
special ansible_host=10.7.1.200 ansible_user=operator
"""
    hosts = {item["name"]: item for item in app._inventory_hosts(inventory)}
    assert hosts["node2"]["user"] == "root"
    assert hosts["tdarr"]["user"] == "sascha"
    assert hosts["emby-chris"]["user"] == "chris"
    assert hosts["special"]["user"] == "operator"


def test_health_exposes_current_version():
    with TestClient(app.app) as client:
        response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["version"] == app.VERSION == "2.3.5"


def test_paperless_import_queues_pdf_through_butler(monkeypatch):
    captured = {}

    def fake_upload(host, command, payload, timeout):
        captured.update(host=host, command=command, payload=payload, timeout=timeout)
        return 0, "", ""

    monkeypatch.setattr(app, "_ssh_bytes", fake_upload)
    pdf = b"%PDF-1.7\nvalid-test-payload"
    with TestClient(app.app) as client:
        response = client.post(
            "/paperless/import?filename=Hausordnung%20Stand%2009.05.2022.pdf",
            content=pdf,
            headers={"Authorization": "Bearer test-token", "Content-Type": "application/pdf"},
        )
    assert response.status_code == 200
    body = response.json()
    assert body["status"] == "queued"
    assert body["filename"].startswith("Hausordnung_Stand_09.05.2022-")
    assert captured["host"] == app.PAPERLESS_SSH
    assert captured["payload"] == pdf
    assert app.PAPERLESS_CONSUME_DIR in captured["command"]


def test_paperless_import_rejects_non_pdf(monkeypatch):
    monkeypatch.setattr(app, "_ssh_bytes", lambda *args: (_ for _ in ()).throw(AssertionError("must not upload")))
    with TestClient(app.app) as client:
        response = client.post(
            "/paperless/import?filename=bad.pdf",
            content=b"not a pdf",
            headers={"Authorization": "Bearer test-token"},
        )
    assert response.status_code == 400


def test_ui_serves_self_contained_operator_console():
    with TestClient(app.app) as client:
        response = client.get("/ui")
    assert response.status_code == 200
    assert "Pfannkuchen Butler" in response.text
    assert 'id="operations-grid"' in response.text
    assert 'id="doctor-form"' in response.text
    assert 'id="preflight-form"' in response.text
    assert 'id="sharing-form"' in response.text
    assert 'id="sharing-events"' in response.text
    assert "/emby/account-sharing" in response.text
    assert 'id="login"' not in response.text
    assert "Butler-Token" not in response.text
    assert "localStorage" not in response.text


def test_ui_asset_is_mounted_read_only_in_compose():
    from pathlib import Path
    import yaml
    compose = yaml.safe_load(Path(app.__file__).with_name("compose.yaml").read_text(encoding="utf-8"))
    mounts = compose["services"]["homelab-butler"]["volumes"]
    assert "./ui.html:/app/ui.html:ro" in mounts


def test_ui_session_login_uses_httponly_cookie_and_csrf():
    app._ui_sessions.clear()
    with TestClient(app.app) as client:
        denied = client.post("/ui/login", json={"token": "wrong"})
        assert denied.status_code == 401

        login = client.post("/ui/login", json={"token": "test-token"})
        assert login.status_code == 200
        assert "HttpOnly" in login.headers.get("set-cookie", "")
        csrf = client.cookies.get("butler_csrf")
        assert csrf

        capabilities = client.get("/capabilities")
        assert capabilities.status_code == 200

        blocked = client.post("/config/reload")
        assert blocked.status_code == 403
        allowed = client.post("/config/reload", headers={"X-CSRF-Token": csrf})
        assert allowed.status_code == 200


def test_ui_anonymous_session_is_automatic_and_strictly_read_only():
    app._ui_sessions.clear()
    with TestClient(app.app) as client:
        session = client.get("/ui/session")
        assert session.status_code == 200
        assert session.json()["authenticated"] is True
        assert session.json()["read_only"] is True
        assert "HttpOnly" in session.headers.get("set-cookie", "")

        capabilities = client.get("/capabilities")
        assert capabilities.status_code == 200

        csrf = client.cookies.get("butler_csrf")
        mutation = client.post("/config/reload", headers={"X-CSRF-Token": csrf})
        assert mutation.status_code == 403
        assert "read-only" in mutation.json()["detail"].lower()


def test_emby_network_identity_normalizes_ipv4_and_ipv6_privacy_addresses():
    ipv4 = app._emby_network_identity("203.0.113.9:443")
    assert ipv4["ip"] == "203.0.113.9"
    assert ipv4["network"] == "203.0.113.9/32"
    assert ipv4["identity"] == "203.0.113.9/32"

    first = app._emby_network_identity("2003:abcd:1234:5678::1")
    privacy_peer = app._emby_network_identity("[2003:abcd:1234:5678:ffff::99]:443")
    sibling_subnet = app._emby_network_identity("2003:abcd:1234:9999::1")
    assert first["network"] == privacy_peer["network"] == "2003:abcd:1234:5678::/64"
    assert first["parent"] == sibling_subnet["parent"] == "2003:abcd:1234::/48"
    assert first["identity"] == sibling_subnet["identity"] == "2003:abcd:1234::/48"


def test_emby_sharing_flags_concurrent_distinct_networks_but_not_sibling_ipv6_subnets():
    def series(endpoint, values, city):
        return {
            "metric": {
                "job": "emby-sascha", "username": "Alice", "remoteEndPoint": endpoint,
                "city": city, "region": "Test", "countryCode": "DE",
                "latitude": "48.1", "longitude": "11.5",
            },
            "values": [[timestamp, "1"] for timestamp in values],
        }

    payload = [
        series("2606:4700:1234:1000::1", [100, 160, 220, 280, 340, 400], "Home"),
        series("2606:4700:1234:2000::2", [100, 160, 220, 280, 340, 400], "Home privacy subnet"),
        series("2001:4860:9999:1000::1", [100, 160, 220, 280, 340, 400], "Away"),
    ]

    result = app._analyze_emby_sharing(payload, step_seconds=60)

    assert result["summary"]["concurrent_events"] == 1
    event = result["events"][0]
    assert event["type"] == "concurrent_networks"
    assert event["username"] == "Alice"
    assert len(event["evidence"]) == 2
    assert {item["identity"] for item in event["evidence"]} == {
        "2606:4700:1234::/48", "2001:4860:9999::/48"
    }


def test_emby_sharing_ignores_short_overlap_inside_prometheus_staleness_window():
    def series(endpoint):
        return {"metric": {"job": "emby-sascha", "username": "Alice", "remoteEndPoint": endpoint,
                           "city": "Munich", "region": "Bavaria", "countryCode": "DE",
                           "latitude": "48.1", "longitude": "11.5"},
                "values": [[100, "1"], [160, "1"]]}

    result = app._analyze_emby_sharing([series("8.8.8.8"), series("1.1.1.1")], step_seconds=60)

    assert result["summary"]["concurrent_events"] == 0


def test_emby_sharing_flags_geographically_impossible_network_change():
    payload = [
        {
            "metric": {"job": "emby-chris", "username": "Bob", "remoteEndPoint": "8.8.8.8",
                       "city": "Berlin", "region": "Berlin", "countryCode": "DE",
                       "latitude": "52.5200", "longitude": "13.4050"},
            "values": [[100, "1"], [160, "1"]],
        },
        {
            "metric": {"job": "emby-chris", "username": "Bob", "remoteEndPoint": "1.1.1.1",
                       "city": "New York", "region": "New York", "countryCode": "US",
                       "latitude": "40.7128", "longitude": "-74.0060"},
            "values": [[400, "1"], [460, "1"]],
        },
    ]

    result = app._analyze_emby_sharing(payload, step_seconds=60)

    assert result["summary"]["concurrent_events"] == 0
    assert result["summary"]["impossible_travel_events"] == 1
    event = result["events"][0]
    assert event["type"] == "impossible_travel"
    assert event["distance_km"] > 6000
    assert event["required_speed_kmh"] > 1000


def test_emby_account_sharing_endpoint_is_read_only_and_filterable(monkeypatch):
    async def history(days, server):
        assert days == 7
        assert server == "all"
        return ([{
            "metric": {"job": "emby-sascha", "username": "Alice", "remoteEndPoint": "8.8.8.8",
                       "city": "Munich", "region": "Bavaria", "countryCode": "DE",
                       "latitude": "48.1", "longitude": "11.5"},
            "values": [[100, "1"], [160, "1"]],
        }], 60)

    monkeypatch.setattr(app, "_fetch_emby_session_history", history)
    with TestClient(app.app) as client:
        response = client.get(
            "/emby/account-sharing?days=7&username=alice",
            headers={"Authorization": "Bearer test-token"},
        )

    assert response.status_code == 200
    payload = response.json()
    assert payload["policy"]["mode"] == "conservative"
    assert payload["policy"]["ipv6_detection_identity"] == "/48"
    assert payload["summary"]["users_analyzed"] == 1
    assert payload["users"][0]["username"] == "Alice"


def test_emby_history_step_stays_below_prometheus_resolution_limit():
    assert app._emby_history_step(7) == 60
    for days in (7, 30, 90):
        step = app._emby_history_step(days)
        assert step % 60 == 0
        assert (days * 86400) / step <= 10_500


def test_emby_history_fetch_chunks_90_days_and_merges_equal_series(monkeypatch):
    calls = []

    class FakeResponse:
        def __init__(self, params):
            self.params = params

        def raise_for_status(self):
            return None

        def json(self):
            start = self.params["start"]
            end = self.params["end"]
            return {"status": "success", "data": {"result": [{
                "metric": {"job": "emby-sascha", "username": "Alice", "remoteEndPoint": "8.8.8.8"},
                "values": [[start, "1"], [end, "1"]],
            }]}}

    class FakeClient:
        def __init__(self, **_kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def get(self, _url, params, headers, cookies):
            calls.append(params)
            return FakeResponse(params)

    monkeypatch.setattr(app, "SERVICES", {"grafana": {"url": "http://grafana", "auth": "none"}})
    monkeypatch.setattr(app.httpx, "AsyncClient", FakeClient)
    monkeypatch.setattr(app.time, "time", lambda: 10_000_000)

    series, step = asyncio.run(app._fetch_emby_session_history(90, "all"))

    assert len(calls) == 3
    assert all(call["end"] - call["start"] <= 30 * 86400 for call in calls)
    assert step == 780
    assert len(series) == 1
    timestamps = [value[0] for value in series[0]["values"]]
    assert timestamps == sorted(set(timestamps))


def test_capabilities_is_live_machine_readable_safety_map():
    with TestClient(app.app) as client:
        response = client.get(
            "/capabilities",
            headers={"Authorization": "Bearer test-token"},
        )
    assert response.status_code == 200
    payload = response.json()
    by_operation = {(item["method"], item["path"]): item for item in payload["operations"]}
    assert ("GET", "/network/wireguard/{host}") in by_operation
    assert by_operation[("GET", "/network/wireguard/{host}")]["mode"] == "read_only"
    removal = by_operation[("DELETE", "/network/wireguard/{host}/peer")]
    assert removal["mode"] == "mutation"
    assert removal["dry_run"] is True
    assert removal["critical"] is True
    assert by_operation[("DELETE", "/vm/destroy/{vmid}")]["dry_run"] is True
    assert all(item["path"] != "/{service}/{path}" for item in payload["operations"])
    assert payload["model_contract"]["instruction"].startswith("Prefer read_only")


def test_info_advertises_capabilities_endpoint():
    with TestClient(app.app) as client:
        response = client.get("/info", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["endpoints"]["capabilities"] == "/capabilities"
    assert response.json()["endpoints"]["doctor"] == "/doctor/{target}"
    assert response.json()["endpoints"]["drift"] == "/drift"
    assert response.json()["endpoints"]["maintenance_preflight"] == "/maintenance/preflight"
    assert response.json()["endpoints"]["ui"] == "/ui"


def test_audit_persists_and_redacts_secrets(tmp_path, monkeypatch):
    db = tmp_path / "audit.sqlite3"
    monkeypatch.setattr(app, "AUDIT_DB_PATH", str(db))
    app._init_audit_db()
    app._audit("/danger", "POST", 200, "host=x token=abc password=hunter2 api_key=secret")
    app._audit_log.clear()

    with TestClient(app.app) as client:
        response = client.get("/audit", headers={"Authorization": "Bearer test-token"})

    assert response.status_code == 200
    entry = response.json()[0]
    assert entry["endpoint"] == "/danger"
    assert "abc" not in entry["detail"]
    assert "hunter2" not in entry["detail"]
    assert "secret" not in entry["detail"]
    assert entry["detail"].count("[REDACTED]") == 3


def test_audit_records_ai_and_web_ui_actor(tmp_path, monkeypatch):
    db = tmp_path / "audit.sqlite3"
    monkeypatch.setattr(app, "AUDIT_DB_PATH", str(db))

    async def empty_status():
        return {}

    monkeypatch.setattr(app, "_collect_service_status", empty_status)
    app._ui_sessions.clear()
    with TestClient(app.app) as client:
        ai = client.get(
            "/status",
            headers={"Authorization": "Bearer test-token", "X-Butler-Actor": "Trulla"},
        )
        assert ai.status_code == 200

        client.headers.pop("Authorization", None)
        client.get("/ui/session")
        web = client.get("/status")
        assert web.status_code == 200

        entries = client.get("/audit").json()

    assert [item["actor"] for item in entries[:2]] == ["Weboberfläche", "Trulla"]


def test_ui_audit_has_actor_column_and_filter():
    with TestClient(app.app) as client:
        response = client.get("/ui")
    assert '<th>Akteur</th>' in response.text
    assert 'id="audit-actor"' in response.text


def test_wireguard_status_returns_redacted_live_state(monkeypatch):
    payload = {
        "interface": "wg0",
        "addresses": ["10.11.12.1/32"],
        "listen_port": 37888,
        "service_active": True,
        "service_enabled": True,
        "routes": [{"dst": "10.11.12.3", "prefsrc": "10.11.12.1"}],
        "peers": [{
            "public_key": "peer-public-key",
            "endpoint": "203.0.113.9:51820",
            "allowed_ips": ["10.11.12.3/32"],
            "latest_handshake": 123,
            "rx_bytes": 456,
            "tx_bytes": 789,
            "persistent_keepalive": 25,
        }],
    }
    monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "debian", "ip": "141.94.237.199"})
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=30: (0, json.dumps(payload), ""))

    with TestClient(app.app) as client:
        response = client.get(
            "/network/wireguard/guck-vps",
            headers={"Authorization": "Bearer test-token"},
        )

    assert response.status_code == 200
    assert response.json()["host"] == "guck-vps"
    assert response.json()["peers"][0]["allowed_ips"] == ["10.11.12.3/32"]
    assert "private" not in response.text.lower()


def test_wireguard_status_rejects_unknown_host_without_ssh(monkeypatch):
    monkeypatch.setattr(app, "_find_inventory_host", lambda host: None)
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("SSH must not run")))

    with TestClient(app.app) as client:
        response = client.get(
            "/network/wireguard/does-not-exist",
            headers={"Authorization": "Bearer test-token"},
        )

    assert response.status_code == 404


def test_wireguard_status_command_uses_sudo_and_accepts_off_keepalive():
    import base64
    import re

    command = app._wireguard_status_command()
    encoded = re.search(r"b64decode\('([^']+)'\)", command).group(1)
    script = base64.b64decode(encoded).decode()

    assert '["sudo", "-n", "wg", "show", "wg0", "dump"]' in script
    assert '0 if fields[7] == "off" else int(fields[7])' in script


def test_wireguard_peer_remove_is_scoped_and_audited(monkeypatch):
    calls = []
    monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "debian", "ip": "141.94.237.199"})
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=30: (calls.append((host, command, timeout)) or (0, json.dumps({"status": "removed", "removed_routes": ["10.7.1.0/24"]}), "")))

    with TestClient(app.app) as client:
        response = client.request(
            "DELETE",
            "/network/wireguard/guck-vps/peer",
            headers={"Authorization": "Bearer test-token"},
            json={"public_key": "A" * 43 + "=", "expected_allowed_ip": "10.7.1.0/24", "dry_run": False},
        )

    assert response.status_code == 200
    assert response.json()["status"] == "removed"
    assert calls[0][0] == "debian@141.94.237.199"
    assert calls[0][2] == 45


def test_wireguard_peer_remove_rejects_non_allowlisted_host(monkeypatch):
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("SSH must not run")))
    with TestClient(app.app) as client:
        response = client.request(
            "DELETE",
            "/network/wireguard/node7/peer",
            headers={"Authorization": "Bearer test-token"},
            json={"public_key": "A" * 43 + "=", "expected_allowed_ip": "10.7.1.0/24", "dry_run": True},
        )
    assert response.status_code == 403


def test_tts_generate_returns_cloned_wav(monkeypatch):
    captured = {}

    class FakeResponse:
        status_code = 200
        content = b"RIFF" + b"test-wave"

    class FakeClient:
        def __init__(self, **_kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return False

        async def post(self, url, json):
            captured["url"] = url
            captured["json"] = json
            return FakeResponse()

    monkeypatch.setattr(app.httpx, "AsyncClient", FakeClient)
    with TestClient(app.app) as client:
        response = client.post(
            "/tts/generate",
            headers={"Authorization": "Bearer test-token"},
            json={"text": "Hallo Sascha", "voice": "deep_thought.mp3", "language": "de"},
        )
    assert response.status_code == 200
    assert response.headers["content-type"].startswith("audio/wav")
    assert response.content.startswith(b"RIFF")
    assert captured["json"]["voice_mode"] == "clone"
    assert captured["json"]["reference_audio_filename"] == "deep_thought.mp3"


def test_tts_generate_validates_text_and_voice_before_backend(monkeypatch):
    monkeypatch.setattr(
        app.httpx,
        "AsyncClient",
        lambda **_kwargs: (_ for _ in ()).throw(AssertionError("backend must not be called")),
    )
    with TestClient(app.app) as client:
        empty = client.post(
            "/tts/generate",
            headers={"Authorization": "Bearer test-token"},
            json={"text": ""},
        )
        traversal = client.post(
            "/tts/generate",
            headers={"Authorization": "Bearer test-token"},
            json={"text": "Hallo", "voice": "../secret.wav"},
        )
    assert empty.status_code == 422
    assert traversal.status_code == 422


def test_tts_generate_rejects_non_wav_backend_response(monkeypatch):
    class FakeResponse:
        status_code = 200
        content = b"not audio"

    class FakeClient:
        def __init__(self, **_kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return False

        async def post(self, _url, json):
            return FakeResponse()

    monkeypatch.setattr(app.httpx, "AsyncClient", FakeClient)
    with TestClient(app.app) as client:
        response = client.post(
            "/tts/generate",
            headers={"Authorization": "Bearer test-token"},
            json={"text": "Hallo"},
        )
    assert response.status_code == 502
    assert "invalid audio" in response.text


def test_tts_bridge_deploy_installs_secrets_without_logging_them(monkeypatch):
    calls = []
    monkeypatch.setattr(app, "BUTLER_TOKEN", "butler-secret")
    monkeypatch.setattr(app, "_vault_cache", {})
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=30: (calls.append((host, command, timeout)) or (0, "", "")))
    with TestClient(app.app) as client:
        response = client.post(
            "/tts/bridge/deploy",
            headers={"Authorization": "Bearer butler-secret"},
            json={"rotate_client_token": True},
        )
    assert response.status_code == 200
    assert response.json()["listen"] == "0.0.0.0:8099"
    assert len(response.json()["client_token"]) >= 32
    assert calls[0][0] == "sascha@10.5.85.5"
    assert "butler-secret" not in calls[0][1]


def test_sysctl_audit_reads_fixed_keys_from_inventory_host(monkeypatch):
    payload = {
        "live": {"net.ipv4.tcp_congestion_control": "bbr"},
        "persistent": {"net.ipv4.tcp_congestion_control": [{"file": "/etc/sysctl.d/99-net-tuning.conf", "value": "bbr"}]},
        "errors": {},
    }
    calls = []
    monkeypatch.setattr(app, "_find_inventory_host", lambda name: {"name": name, "user": "root", "ip": "10.5.85.16"})
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, __import__("json").dumps(payload), "")))
    with TestClient(app.app) as client:
        response = client.get("/system/sysctl/node6", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["live"]["net.ipv4.tcp_congestion_control"] == "bbr"
    assert calls[0][0] == "root@10.5.85.16"
    assert "base64.b64decode" in calls[0][1]


def test_sysctl_audit_rejects_unknown_host_without_ssh(monkeypatch):
    monkeypatch.setattr(app, "_find_inventory_host", lambda _name: None)
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
    with TestClient(app.app) as client:
        response = client.get("/system/sysctl/not-there", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 404


def test_host_forensics_is_read_only_and_uses_inventory(monkeypatch):
    payload = {"docker_binary": {"rc": 0, "stdout": "/usr/bin/docker", "stderr": ""}}
    calls = []
    monkeypatch.setattr(app, "_find_inventory_host", lambda name: {"name": name, "user": "root", "ip": "10.5.85.13"})
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, __import__("json").dumps(payload), "")))
    with TestClient(app.app) as client:
        response = client.get("/system/forensics/node3?since_hours=24", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["checks"] == payload
    assert calls[0][0] == "root@10.5.85.13"
    assert calls[0][2] == 60
    assert "base64.b64decode" in calls[0][1]
    command = app._host_forensics_command(24)
    for destructive in ("systemctl restart", "systemctl stop", "systemctl disable", "docker rm", "docker system prune", "iptables -F", "rm -rf"):
        assert destructive not in command


def test_host_forensics_rejects_unknown_host_and_invalid_window(monkeypatch):
    monkeypatch.setattr(app, "_find_inventory_host", lambda _name: None)
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
    with TestClient(app.app) as client:
        missing = client.get("/system/forensics/not-there", headers={"Authorization": "Bearer test-token"})
        bad_window = client.get("/system/forensics/node3?since_hours=999", headers={"Authorization": "Bearer test-token"})
    assert missing.status_code == 404
    assert bad_window.status_code == 422


def test_docker_residue_cleanup_defaults_to_dry_run(monkeypatch):
    payload = {"dry_run": True, "before_rule_count": 25, "removed_rules": [], "removed_chains": [], "removed_links": [], "removed_paths": [], "errors": []}
    calls = []
    monkeypatch.setattr(app, "_find_inventory_host", lambda name: {"name": name, "user": "root", "ip": "10.5.85.13"})
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, __import__("json").dumps(payload), "")))
    with TestClient(app.app) as client:
        response = client.post("/system/cleanup/docker-residue/node3", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["dry_run"] is True
    assert calls[0][0] == "root@10.5.85.13"
    assert calls[0][2] == 90


def test_docker_residue_cleanup_refuses_active_docker(monkeypatch):
    payload = {"dry_run": False, "error": "Docker or containerd is still installed/active; refusing residue cleanup"}
    monkeypatch.setattr(app, "_find_inventory_host", lambda name: {"name": name, "user": "root", "ip": "10.5.85.13"})
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (2, __import__("json").dumps(payload), ""))
    with TestClient(app.app) as client:
        response = client.post("/system/cleanup/docker-residue/node3?dry_run=false", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 409


def test_iso_builder_restore_defaults_to_dry_run_and_has_no_free_target(monkeypatch):
    payload = {"dry_run": True, "restored": [], "removed_outputs": [], "validation": {}}
    calls = []
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, __import__("json").dumps(payload), "")))
    with TestClient(app.app) as client:
        response = client.post("/system/restore/iso-builder", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["dry_run"] is True
    assert calls[0][0] == app.AUTOMATION1
    assert calls[0][2] == 120
    command = app._iso_builder_restore_command(True)
    encoded = command.split("base64.b64decode('", 1)[1].split("')", 1)[0]
    decoded = __import__("base64").b64decode(encoded).decode()
    assert "origin/master" in decoded
    assert "/app-config/ansible" in decoded


def test_caddy_site_remove_defaults_to_dry_run(monkeypatch):
    payload = {"hostname": "wiki.sascha-lutz.de", "dry_run": True, "found": True, "line_start": 10, "line_end": 13}
    calls = []
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, __import__("json").dumps(payload), "")))
    with TestClient(app.app) as client:
        response = client.delete("/caddy/site/wiki.sascha-lutz.de", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["dry_run"] is True
    assert calls[0][0] == app.CADDY_HOST
    assert calls[0][2] == 90


def test_caddy_site_remove_rejects_unmanaged_hostname_before_ssh(monkeypatch):
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
    with TestClient(app.app) as client:
        response = client.delete("/caddy/site/example.com", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 400


def test_dns_rrset_dry_run_returns_only_selected_records(monkeypatch):
    async def fake_lookup(zone):
        assert zone == "sascha-lutz.de"
        return {"id": 96805, "name": zone}, [
            {"name": "wiki", "type": "A", "records": [{"value": "46.225.230.72"}]},
            {"name": "wiki", "type": "AAAA", "records": [{"value": "::1"}]},
            {"name": "git", "type": "A", "records": [{"value": "46.225.230.72"}]},
        ], {"Authorization": "Bearer hidden"}
    monkeypatch.setattr(app, "_hetzner_zone_and_rrsets", fake_lookup)
    with TestClient(app.app) as client:
        response = client.delete("/dns/rrset/sascha-lutz.de/wiki", headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    body = response.json()
    assert body["dry_run"] is True
    assert [item["type"] for item in body["rrsets"]] == ["A", "AAAA"]
    assert body["deleted"] == []


def test_hetzner_dns_token_accepts_single_sanitized_vault_alias(monkeypatch):
    token = "a" * 48
    monkeypatch.setattr(app, "_vault_cache", {"hetzner-dns-api": f"Hetzner DNS API Token: {token}\nFür sascha-lutz.de", "other": "ignored"})
    monkeypatch.setattr(app, "_read", lambda _name: None)
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not sync vault")))
    assert app._get_hetzner_dns_token() == token


def test_uptime_monitor_remove_defaults_to_dry_run(monkeypatch):
    payload = {"monitor_id": 71, "expected_name": "Outline Wiki", "dry_run": True, "found": {"id": 71, "name": "Outline Wiki"}}
    monkeypatch.setattr(app, "_ssh", lambda target, command, timeout=120: (0, json.dumps(payload), ""))
    with TestClient(app.app) as client:
        response = client.delete("/uptime/monitor/71", params={"expected_name": "Outline Wiki"}, headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 200
    assert response.json()["dry_run"] is True


def test_uptime_monitor_remove_rejects_invalid_expected_name_before_ssh(monkeypatch):
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
    with TestClient(app.app) as client:
        response = client.delete("/uptime/monitor/71", params={"expected_name": "Outline; rm -rf /"}, headers={"Authorization": "Bearer test-token"})
    assert response.status_code == 400


def test_invalid_log_target_is_rejected_before_ssh():
    with TestClient(app.app) as client:
        response = client.get(
            "/logs/tdarr/fileflows;rm",
            headers={"Authorization": "Bearer test-token"},
        )
    assert response.status_code == 400


def test_inventory_rejects_invalid_ip_before_ssh():
    with TestClient(app.app) as client:
        response = client.post(
            "/inventory/host",
            headers={"Authorization": "Bearer test-token"},
            json={"name": "bad-host", "ip": "not-an-ip", "group": "auto"},
        )
    assert response.status_code == 400


def test_inventory_upsert_uses_base64_script(monkeypatch):
    calls = []

    def fake_ssh(host, command, timeout=600):
        calls.append((host, command, timeout))
        return 0, "updated", ""

    monkeypatch.setattr(app, "_ssh", fake_ssh)
    with TestClient(app.app) as client:
        response = client.post(
            "/inventory/host",
            headers={"Authorization": "Bearer test-token"},
            json={"name": "pfannkuchen", "ip": "46.225.230.72", "group": "vps", "user": "root"},
        )
    assert response.status_code == 200
    assert "base64.b64decode" in calls[0][1]
    assert "\\nname =" not in calls[0][1]


def test_ansible_run_supports_safe_tune_action_and_syncs_approved_files(monkeypatch):
    calls = []

    def fake_ssh(host, command, timeout=600):
        calls.append((host, command, timeout))
        return 0, "changed=1 failed=0", ""

    monkeypatch.setattr(app, "_ssh", fake_ssh)
    with TestClient(app.app) as client:
        response = client.post(
            "/ansible/run",
            headers={"Authorization": "Bearer test-token"},
            json={"hostname": "emby-sascha", "action": "tune"},
        )
    assert response.status_code == 200
    assert response.json()["action"] == "tune"
    assert "git fetch origin master" in calls[0][1]
    assert "git show origin/master:roles/sysctl/tasks/main.yml" in calls[0][1]
    assert "git pull --ff-only" not in calls[0][1]
    assert "bash pfannkuchen.sh tune emby-sascha" in calls[0][1]
    assert len(calls) == 1


def test_ansible_run_supports_scoped_nfs_action(monkeypatch):
    calls = []

    def fake_ssh(host, command, timeout=600):
        calls.append((host, command, timeout))
        return 0, "changed=1 failed=0", ""

    monkeypatch.setattr(app, "_ssh", fake_ssh)
    with TestClient(app.app) as client:
        response = client.post(
            "/ansible/run",
            headers={"Authorization": "Bearer test-token"},
            json={"hostname": "arrapps", "action": "nfs"},
        )
    assert response.status_code == 200
    assert response.json()["action"] == "nfs"
    command = calls[0][1]
    assert "mkdir -p roles/nfs_stability/tasks" in command
    assert "git show origin/master:roles/nfs_stability/tasks/main.yml" in command
    assert "git show origin/master:nfs-stability.yml" in command
    assert "bash pfannkuchen.sh nfs arrapps" in command
    assert "git pull --ff-only" not in command
    assert len(calls) == 1


def test_ansible_run_rejects_unknown_action_and_shell_metacharacters(monkeypatch):
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH")))
    with TestClient(app.app) as client:
        bad_action = client.post(
            "/ansible/run",
            headers={"Authorization": "Bearer test-token"},
            json={"hostname": "emby-sascha", "action": "shell"},
        )
        bad_host = client.post(
            "/ansible/run",
            headers={"Authorization": "Bearer test-token"},
            json={"hostname": "emby-sascha;id", "action": "tune"},
        )
    assert bad_action.status_code == 400
    assert bad_host.status_code == 400


def test_docker_inspect_returns_sanitized_summary(monkeypatch):
    raw = [{
        "Name": "/fileflows",
        "Config": {"Image": "revenz/fileflows:26.06", "Env": ["API_TOKEN=secret", "TZ=Europe/Berlin"]},
        "State": {"Status": "running", "Running": True, "OOMKilled": False, "ExitCode": 0},
        "RestartCount": 0,
        "HostConfig": {"Runtime": "nvidia", "Memory": 0, "MemoryReservation": 0, "NanoCpus": 0,
                       "DeviceRequests": [], "RestartPolicy": {"Name": "always"}, "LogConfig": {"Type": "json-file"}},
        "Mounts": [{"Type": "bind", "Source": "/app-config/fileflows-data", "Destination": "/app/Data", "RW": True}],
    }]
    monkeypatch.setattr(app, "_find_inventory_host", lambda _name: {"user": "sascha", "ip": "10.2.1.104"})
    monkeypatch.setattr(app, "_ssh", lambda *args, **kwargs: (0, __import__("json").dumps(raw), ""))
    with TestClient(app.app) as client:
        response = client.get(
            "/docker/inspect/tdarr/fileflows",
            headers={"Authorization": "Bearer test-token"},
        )
    assert response.status_code == 200
    result = response.json()
    assert result["runtime"] == "nvidia"
    assert result["environment_keys"] == ["API_TOKEN", "TZ"]
    assert "secret" not in response.text


def test_classify_http_status_distinguishes_auth_and_route_errors():
    assert app._classify_http_status(200, {200}) == "healthy"
    assert app._classify_http_status(401, {200}) == "auth_failed"
    assert app._classify_http_status(403, {200}) == "auth_failed"
    assert app._classify_http_status(404, {200}) == "misconfigured"
    assert app._classify_http_status(429, {200}) == "degraded"
    assert app._classify_http_status(503, {200}) == "degraded"


def test_service_auth_headers_support_lightweight_health_probes(monkeypatch):
    monkeypatch.setattr(app, "_get_key", lambda _cfg: "test-secret")
    assert app._service_auth({"auth": "bearer"})["headers"] == {
        "Authorization": "Bearer test-secret"
    }
    assert app._service_auth({"auth": "apikey"})["headers"] == {
        "X-Api-Key": "test-secret"
    }
    assert app._service_auth({"auth": "n8n"})["headers"] == {
        "X-N8N-API-KEY": "test-secret"
    }


def _archive_json(start, name="archive"):
    return __import__("json").dumps([{"archives": [{"start": start, "name": name}]}])


def test_backup_item_reports_age_and_severity():
    now = datetime.now(timezone.utc)
    recent = (now - timedelta(hours=4)).isoformat()
    warning = (now - timedelta(hours=36)).isoformat()
    critical = (now - timedelta(hours=60)).isoformat()

    assert app._backup_item(0, _archive_json(recent), "")["state"] == "healthy"
    assert app._backup_item(0, _archive_json(warning), "")["state"] == "warning"
    assert app._backup_item(0, _archive_json(critical), "")["state"] == "critical"
    assert app._backup_item(1, "", "timeout")["state"] == "unknown"


def test_ssh_uses_writable_runtime_known_hosts(monkeypatch):
    captured = {}

    class Result:
        returncode = 0
        stdout = "ok"
        stderr = ""

    def fake_run(args, **_kwargs):
        captured["args"] = args
        return Result()

    monkeypatch.setattr(app._sp, "run", fake_run)
    assert app._ssh("sascha@example", "true", timeout=1)[0] == 0
    joined = " ".join(captured["args"])
    assert "UserKnownHostsFile=/tmp/butler_known_hosts" in joined


def test_ssh_timeout_is_normalized_instead_of_crashing_collection(monkeypatch):
    def timeout(*_args, **_kwargs):
        raise app._sp.TimeoutExpired(cmd=["ssh"], timeout=1)

    monkeypatch.setattr(app._sp, "run", timeout)
    rc, out, err = app._ssh("sascha@example", "true", timeout=1)
    assert (rc, out) == (124, "")
    assert "timed out" in err.lower()


def test_service_login_failure_is_isolated(monkeypatch):
    monkeypatch.setattr(app, "SERVICES", {
        "dockhand": {"url": "http://dockhand.invalid", "auth": "session", "health_path": "/api/health"}
    })

    async def failed_login(_client):
        raise RuntimeError("login failed")

    monkeypatch.setattr(app, "_dockhand_login", failed_login)
    result = asyncio.run(app._collect_service_status())
    assert result["dockhand"]["status"] == "offline"
    assert result["dockhand"]["reachable"] is False


def test_backup_collection_runs_hosts_concurrently(monkeypatch):
    active = 0
    max_active = 0

    monkeypatch.setattr(app, "_get_inventory_hosts", lambda: [
        {"name": f"vm-{index}", "user": "sascha", "ip": f"10.1.1.{index}"}
        for index in range(1, 5)
    ])
    monkeypatch.setattr(app, "_config", {})

    def fake_ssh(*_args, **_kwargs):
        nonlocal active, max_active
        active += 1
        max_active = max(max_active, active)
        time.sleep(0.05)
        active -= 1
        return 0, _archive_json(datetime.now(timezone.utc).isoformat()), ""

    monkeypatch.setattr(app, "_ssh", fake_ssh)
    result = asyncio.run(app._collect_backup_status(concurrency=4))

    assert max_active > 1
    assert result["summary"] == {
        "total": 4, "healthy": 4, "warning": 0, "critical": 0, "unknown": 0, "exempt": 0
    }


def test_backup_policy_exempts_host_without_borg_call(monkeypatch):
    monkeypatch.setattr(app, "_get_inventory_hosts", lambda: [
        {"name": "guck-vps", "user": "debian", "ip": "141.94.237.199"}
    ])
    monkeypatch.setattr(app, "_config", {"backup": {"exempt_hosts": ["guck-vps"]}})
    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not call borg")))

    result = asyncio.run(app._collect_backup_status())

    assert result["summary"] == {
        "total": 1, "healthy": 0, "warning": 0, "critical": 0, "unknown": 0, "exempt": 1
    }
    assert result["hosts"]["guck-vps"] == {
        "state": "exempt", "ok": True, "reason": "backup policy exemption"
    }


def test_overview_openapi_has_stable_enums_and_schema():
    schema = app.app.openapi()
    response_schema = schema["paths"]["/overview"]["get"]["responses"]["200"]["content"]["application/json"]["schema"]
    assert response_schema["$ref"].endswith("/OverviewResponse")
    overview_schema = schema["components"]["schemas"]["OverviewResponse"]
    assert overview_schema["properties"]["overall_state"]["enum"] == ["healthy", "warning", "critical"]
    finding_schema = schema["components"]["schemas"]["OverviewFinding"]
    assert finding_schema["properties"]["severity"]["enum"] == ["healthy", "warning", "critical"]


def test_doctor_correlates_host_layers(monkeypatch):
    async def snapshot():
        return {
            "services": {},
            "hosts": {"guck-vps": {"reachable": True, "containers": ["caddy: Up 3 days"]}},
            "backups": {"summary": {}, "hosts": {"guck-vps": {"state": "exempt", "ok": True}}},
            "disks": {"guck-vps": {"pct": "8%"}},
        }

    monkeypatch.setattr(app, "_collect_operational_snapshot", snapshot)
    with TestClient(app.app) as client:
        response = client.get("/doctor/guck-vps", headers={"Authorization": "Bearer test-token"})

    assert response.status_code == 200
    result = response.json()
    assert result["state"] == "healthy"
    assert result["layers"]["host"]["reachable"] is True
    assert result["layers"]["backup"]["state"] == "exempt"
    assert result["layers"]["disk"]["pct"] == "8%"
    assert result["findings"] == []


def test_drift_reports_inventory_coverage_gaps(monkeypatch):
    async def snapshot():
        return {
            "services": {},
            "hosts": {"vm-a": {"reachable": True}, "vm-b": {"reachable": True}, "node1": {"reachable": True}},
            "backups": {"summary": {}, "hosts": {"vm-a": {"state": "healthy"}, "orphan": {"state": "healthy"}}},
            "disks": {"vm-a": {"pct": "10%"}, "node1": {"pct": "20%"}},
        }

    monkeypatch.setattr(app, "_collect_operational_snapshot", snapshot)
    with TestClient(app.app) as client:
        response = client.get("/drift", headers={"Authorization": "Bearer test-token"})

    assert response.status_code == 200
    result = response.json()
    assert result["state"] == "warning"
    assert {item["code"] for item in result["findings"]} == {
        "inventory_missing_backup", "inventory_missing_disk", "backup_without_inventory"
    }


def test_maintenance_preflight_blocks_active_target_backup(monkeypatch):
    async def snapshot():
        return {
            "services": {},
            "hosts": {"emby-chris": {"reachable": True, "containers": ["emby: Up 2 days"]}},
            "backups": {"summary": {}, "hosts": {"emby-chris": {"state": "healthy"}}},
            "disks": {"emby-chris": {"pct": "30%"}},
        }

    async def active_backups():
        return {"emby-chris": "active"}

    monkeypatch.setattr(app, "_collect_operational_snapshot", snapshot)
    monkeypatch.setattr(app, "_collect_active_backups", active_backups)
    with TestClient(app.app) as client:
        response = client.get(
            "/maintenance/preflight?action=docker&target=emby-chris",
            headers={"Authorization": "Bearer test-token"},
        )

    assert response.status_code == 200
    result = response.json()
    assert result["safe"] is False
    assert result["blockers"] == [{"code": "backup_active", "target": "emby-chris"}]


def test_overview_is_compact_deterministic_and_light_model_friendly(monkeypatch):
    async def service_data():
        return {
            "ok": {"status": "healthy", "reachable": True, "http": 200},
            "bad-auth": {"status": "auth_failed", "reachable": True, "http": 401},
        }

    async def host_data():
        return {
            "vm1": {"reachable": True, "containers": ["app: Up 1 hour (healthy)"]},
            "vm2": {"reachable": False, "containers": [], "error": "timeout"},
        }

    async def backup_data():
        return {
            "summary": {"total": 1, "healthy": 0, "warning": 1, "critical": 0, "unknown": 0},
            "hosts": {"vm1": {"state": "warning", "age_hours": 36}},
        }

    async def disk_data():
        return {"vm1": {"pct": "85%"}}

    monkeypatch.setattr(app, "_collect_service_status", service_data)
    monkeypatch.setattr(app, "_collect_health_all", host_data)
    monkeypatch.setattr(app, "_collect_backup_status", backup_data)
    monkeypatch.setattr(app, "_collect_disk_usage", disk_data)

    with TestClient(app.app) as client:
        response = client.get("/overview", headers={"Authorization": "Bearer test-token"})

    assert response.status_code == 200
    result = response.json()
    assert result["schema_version"] == 1
    assert result["overall_state"] == "critical"
    assert result["summary"] == {"critical": 2, "warning": 2, "healthy": 2}
    assert [item["code"] for item in result["findings"]] == [
        "host_unreachable", "service_auth_failed", "backup_warning", "disk_high"
    ]


def test_proxy_route_validation_restricts_domain_and_upstream():
    assert app._validate_proxy_route("speed.guck.tv", "127.0.0.1:8080") == (
        "speed.guck.tv", "127.0.0.1:8080", "guck.tv", "speed"
    )

    for domain, upstream in [
        ("guck.tv", "127.0.0.1:8080"),
        ("speed.evil.example", "127.0.0.1:8080"),
        ("speed.guck.tv", "10.0.0.1:8080"),
        ("speed.guck.tv", "127.0.0.1:70000"),
        ("speed.guck.tv;rm", "127.0.0.1:8080"),
    ]:
        try:
            app._validate_proxy_route(domain, upstream)
        except ValueError:
            pass
        else:
            raise AssertionError(f"unsafe route accepted: {domain} -> {upstream}")


def test_proxy_route_endpoint_configures_caddy_and_dns(monkeypatch):
    calls = []

    def fake_caddy(domain, upstream):
        calls.append(("caddy", domain, upstream))
        return {"status": "reloaded", "backup": "/app-config/caddy/Caddyfile.bak-test"}

    async def fake_dns(zone, name, token_override=None):
        calls.append(("dns", zone, name, token_override))
        return {"zone_id": 123, "records": ["A", "AAAA"]}

    monkeypatch.setattr(app, "_configure_caddy_route", fake_caddy)
    monkeypatch.setattr(app, "_upsert_dns_records", fake_dns)
    with TestClient(app.app) as client:
        response = client.post(
            "/vps/proxy-route",
            headers={"Authorization": "Bearer test-token"},
            json={"domain": "speed.guck.tv", "upstream": "127.0.0.1:8080", "dns_token": "test-dns-token"},
        )

    assert response.status_code == 200
    assert response.json()["status"] == "configured"
    assert calls == [
        ("caddy", "speed.guck.tv", "127.0.0.1:8080"),
        ("dns", "guck.tv", "speed", "test-dns-token"),
    ]


def test_hetzner_token_refreshes_vault_cache_when_missing(monkeypatch):
    reads = iter([None, "refreshed-token"])
    calls = []
    monkeypatch.setattr(app, "_read", lambda _name: next(reads))
    monkeypatch.setattr(app, "_load_vault_cache", lambda: calls.append("reload"))
    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, "vault-sync: ok", "")))

    assert app._get_hetzner_dns_token() == "refreshed-token"
    assert calls[0][0] == "sascha@10.4.1.116"
    assert calls[0][1] == "sudo bash /data/stacks/homelab-butler/vault-sync.sh"
    assert calls[1] == "reload"


def test_speedtest_deploy_requires_strong_secrets_and_uses_full_git_app(monkeypatch):
    calls = []

    async def fake_fetch(repo, path):
        calls.append(("fetch", repo, path))
        return f"content:{path}"

    def fake_deploy(files, password, session_secret):
        calls.append(("deploy", files, password, session_secret))
        return {"status": "deployed", "health": "ok", "application": "streamscope"}

    monkeypatch.setattr(app, "_fetch_forgejo_text", fake_fetch)
    monkeypatch.setattr(app, "_deploy_speedtest_compose", fake_deploy)
    with TestClient(app.app) as client:
        weak = client.post(
            "/vps/speedtest/deploy",
            headers={"Authorization": "Bearer test-token"},
            json={"stats_password": "short", "session_secret": "long-session-secret-with-entropy"},
        )
        response = client.post(
            "/vps/speedtest/deploy",
            headers={"Authorization": "Bearer test-token"},
            json={
                "stats_password": "correct-horse-battery-staple",
                "session_secret": "streamscope-session-secret-with-entropy",
            },
        )

    assert weak.status_code == 400
    assert response.status_code == 200
    assert response.json()["application"] == "streamscope"
    assert ("fetch", "sascha/speedtest", "compose.yaml") in calls
    assert ("fetch", "sascha/speedtest", "streamscope/static/assets/app.js") in calls
    deploy = calls[-1]
    assert deploy[0] == "deploy"
    assert deploy[1]["compose.yaml"] == "content:compose.yaml"
    assert deploy[2] == "correct-horse-battery-staple"
    assert deploy[3] == "streamscope-session-secret-with-entropy"