"""Tests for ``hermes debug`` CLI command and debug utilities."""

import os
from unittest.mock import MagicMock, patch

import pytest

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
    """Set up an isolated HERMES_HOME with minimal logs."""
    home = tmp_path / ".hermes"
    home.mkdir()
    monkeypatch.setenv("HERMES_HOME", str(home))

    # Create log files
    logs_dir = home / "logs"
    logs_dir.mkdir()
    (logs_dir / "agent.log").write_text(
        "2026-04-12 17:00:00 INFO agent: session started\n"
        "2026-04-12 17:00:01 INFO tools.terminal: running ls\n"
        "2026-04-12 17:00:02 WARNING agent: high token usage\n"
    )
    (logs_dir / "errors.log").write_text(
        "2026-04-12 17:00:05 ERROR gateway.run: connection lost\n"
    )
    (logs_dir / "gateway.log").write_text(
        "2026-04-12 17:00:10 INFO gateway.run: started\n"
    )
    (logs_dir / "gui.log").write_text(
        "2026-04-12 17:00:12 INFO hermes_cli.web_server: dashboard request\n"
    )
    (logs_dir / "desktop.log").write_text(
        "2026-04-12 17:00:15 INFO desktop: backend spawned\n"
    )

    return home


# ---------------------------------------------------------------------------
# Unit tests for upload helpers
# ---------------------------------------------------------------------------





class TestUploadToPastebin:
    """Test the combined upload with fallback."""


    def test_falls_back_to_dpaste_com(self):
        from hermes_cli.debug import upload_to_pastebin

        with patch("hermes_cli.debug._upload_paste_rs",
                    side_effect=Exception("down")), \
             patch("hermes_cli.debug._upload_dpaste_com",
                    return_value="https://dpaste.com/TEST") as dp:
            url = upload_to_pastebin("content")

        assert url == "https://dpaste.com/TEST"
        dp.assert_called_once()

    def test_raises_when_both_fail(self):
        from hermes_cli.debug import upload_to_pastebin

        with patch("hermes_cli.debug._upload_paste_rs",
                    side_effect=Exception("err1")), \
             patch("hermes_cli.debug._upload_dpaste_com",
                    side_effect=Exception("err2")):
            with pytest.raises(RuntimeError, match="Failed to upload"):
                upload_to_pastebin("content")


# ---------------------------------------------------------------------------
# Log reading
# ---------------------------------------------------------------------------

class TestCaptureLogSnapshot:
    """Test _capture_log_snapshot for log reading and truncation."""




    def test_race_truncate_after_resolve_reports_empty(self, hermes_home, monkeypatch):
        """If the log is truncated between resolve and stat, say 'empty', not 'missing'."""
        log_path = hermes_home / "logs" / "agent.log"
        from hermes_cli import debug

        monkeypatch.setattr(debug, "_resolve_log_path", lambda _name: log_path)
        log_path.write_text("")

        snap = debug._capture_log_snapshot("agent", tail_lines=10)
        assert snap.path == log_path
        assert snap.full_text is None
        assert snap.tail_text == "(file empty)"


    def test_keeps_first_line_when_truncation_on_boundary(self, hermes_home):
        """When truncation lands on a line boundary, keep the first full line."""
        from hermes_cli.debug import _capture_log_snapshot

        # File must exceed the initial chunk_size (8192) used by the
        # backward-reading loop so the truncation path actually fires.
        line = "A" * 99 + "\n"  # 100 bytes per line
        num_lines = 200  # 20000 bytes
        (hermes_home / "logs" / "agent.log").write_bytes((line * num_lines).encode("utf-8"))

        # max_bytes = 1000 = 100 * 10 → cut at byte 20000 - 1000 = 19000,
        # and byte 19000 - 1 is '\n'.  Boundary hit → keep all 10 lines.
        snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=1000)
        assert snap.full_text is not None
        assert "truncated" in snap.full_text
        raw = snap.full_text.split("\n", 1)[1]
        kept = [l for l in raw.strip().splitlines() if l.startswith("A")]
        assert len(kept) == 10


class TestMissingLogNote:
    """A missing log explains itself when the writer isn't this backend.

    `hermes debug share` runs on the backend, so a desktop connected to a
    remote/docker/SSH backend can never contribute desktop.log. Reporting a
    bare absence sends triage after a client-side bug it cannot see.
    """

    def test_backend_written_log_reports_plain_absence(self, hermes_home):
        from hermes_cli.debug import _capture_log_snapshot

        (hermes_home / "logs" / "agent.log").unlink()

        snap = _capture_log_snapshot("agent", tail_lines=10)
        assert snap.full_text is None
        assert snap.tail_text == "(file not found)"

    def test_client_written_log_names_its_writer_and_path(self, hermes_home):
        from hermes_cli.debug import _capture_log_snapshot

        (hermes_home / "logs" / "desktop.log").unlink()

        snap = _capture_log_snapshot("desktop", tail_lines=10)
        assert snap.full_text is None
        assert "not on this host" in snap.tail_text
        # The reader needs the path to collect by hand on the client machine.
        assert str(hermes_home / "logs" / "desktop.log") in snap.tail_text

    def test_present_client_log_is_captured_normally(self, hermes_home):
        """A local backend still reads desktop.log — the note is only for a miss."""
        from hermes_cli.debug import _capture_log_snapshot

        snap = _capture_log_snapshot("desktop", tail_lines=10)
        assert "backend spawned" in snap.tail_text
        assert "not on this host" not in snap.tail_text

    def test_empty_client_log_is_empty_not_absent(self, hermes_home):
        """An empty file means the app ran and logged nothing — a different fact."""
        from hermes_cli.debug import _capture_log_snapshot

        (hermes_home / "logs" / "desktop.log").write_text("")

        snap = _capture_log_snapshot("desktop", tail_lines=10)
        assert snap.tail_text == "(file empty)"

    def test_report_carries_the_note_for_a_remote_backend(self, hermes_home):
        """The uploaded report — what people paste into support — must explain it."""
        from hermes_cli.debug import collect_debug_report

        (hermes_home / "logs" / "desktop.log").unlink()

        report = collect_debug_report(log_lines=10, dump_text="dump\n")
        assert "--- desktop.log" in report
        assert "not on this host" in report




# ---------------------------------------------------------------------------
# Capture log redaction (force=True applies regardless of HERMES_REDACT_SECRETS)
# ---------------------------------------------------------------------------

# A vendor-prefixed token used across redaction tests. Long enough to clear
# the redactor's `floor` parameter so it actually masks rather than fully blanks.
_REDACT_FIXTURE_TOKEN = "sk-proj-A1B2C3D4E5F6G7H8I9J0aA"


class TestCaptureLogSnapshotRedaction:
    """Pin upload-time redaction at the _capture_log_snapshot boundary."""

    @pytest.fixture
    def hermes_home_with_secret(self, tmp_path, monkeypatch):
        """Isolated HERMES_HOME whose agent.log contains a vendor-prefixed token."""
        home = tmp_path / ".hermes"
        home.mkdir()
        monkeypatch.setenv("HERMES_HOME", str(home))
        # Baseline fixture: no explicit env-var opinion. With the post-#17691
        # default of ON, the default-path tests below exercise the
        # secure-default behaviour. The `force=True` regression test
        # setenvs to "false" inline to prove force=True works even when
        # the runtime flag is disabled.
        monkeypatch.delenv("HERMES_REDACT_SECRETS", raising=False)

        logs_dir = home / "logs"
        logs_dir.mkdir()
        (logs_dir / "agent.log").write_text(
            f"2026-04-12 17:00:00 INFO config: api_key={_REDACT_FIXTURE_TOKEN} loaded\n"
        )
        (logs_dir / "errors.log").write_text("")
        (logs_dir / "gateway.log").write_text("")
        return home

    def test_default_redacts_tail_and_full_text(self, hermes_home_with_secret):
        from hermes_cli.debug import _capture_log_snapshot

        snap = _capture_log_snapshot("agent", tail_lines=10)

        # Both views the upload uses must be sanitized.
        assert _REDACT_FIXTURE_TOKEN not in snap.tail_text
        assert snap.full_text is not None
        assert _REDACT_FIXTURE_TOKEN not in snap.full_text

    def test_redact_false_passes_through(self, hermes_home_with_secret):
        from hermes_cli.debug import _capture_log_snapshot

        snap = _capture_log_snapshot("agent", tail_lines=10, redact=False)

        # Original token survives when the caller opts out.
        assert _REDACT_FIXTURE_TOKEN in snap.tail_text
        assert _REDACT_FIXTURE_TOKEN in (snap.full_text or "")

    def test_force_true_works_when_redaction_disabled(
        self, hermes_home_with_secret, monkeypatch
    ):
        """Regression test: redact_sensitive_text short-circuits without force=True.

        If a future refactor drops `force=True` from `_redact_log_text`, this
        test fails immediately. Without `force=True`, the redactor returns the
        input unchanged when HERMES_REDACT_SECRETS=false, and the share-time
        redaction feature ships silently broken for users who opted out of
        runtime redaction (e.g. developers working on the redactor itself).
        """

        # Force the runtime flag off so we're exercising the force=True path,
        # not the default-on path.
        monkeypatch.setenv("HERMES_REDACT_SECRETS", "false")

        from hermes_cli.debug import _capture_log_snapshot

        assert os.environ.get("HERMES_REDACT_SECRETS", "") == "false"

        snap = _capture_log_snapshot("agent", tail_lines=10)

        assert _REDACT_FIXTURE_TOKEN not in snap.tail_text
        assert snap.full_text is not None
        assert _REDACT_FIXTURE_TOKEN not in snap.full_text

    def test_default_redacts_email_addresses_for_public_share(
        self, hermes_home_with_secret
    ):
        from hermes_cli.debug import _capture_log_snapshot

        log_path = hermes_home_with_secret / "logs" / "agent.log"
        log_path.write_text(
            "2026-04-12 17:00:00 INFO gateway.run: "
            "inbound message: platform=bluebubbles "
            "user=person@example.com chat=iMessage;-;person@example.com msg='hello'\n"
        )

        snap = _capture_log_snapshot("agent", tail_lines=10)

        assert "person@example.com" not in snap.tail_text
        assert "[REDACTED_EMAIL]" in snap.tail_text
        assert snap.full_text is not None
        assert "person@example.com" not in snap.full_text


    def test_capture_default_log_snapshots_threads_redact(
        self, hermes_home_with_secret
    ):
        from hermes_cli.debug import _capture_default_log_snapshots

        snaps = _capture_default_log_snapshots(50)

        # Default threads redact=True to all three captured logs.
        assert _REDACT_FIXTURE_TOKEN not in snaps["agent"].tail_text
        assert _REDACT_FIXTURE_TOKEN not in (snaps["agent"].full_text or "")



# ---------------------------------------------------------------------------
# Debug report collection
# ---------------------------------------------------------------------------



# ---------------------------------------------------------------------------
# CLI entry point — run_debug_share
# ---------------------------------------------------------------------------

class TestRunDebugShare:
    """Test the run_debug_share CLI handler."""




    def test_share_uploads_five_pastes(self, hermes_home, capsys):
        """Successful share uploads report + agent.log + gateway.log + gui.log + desktop.log."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False
        args.nous = False

        call_count = [0]
        uploaded_content = []
        def _mock_upload(content, expiry_days=7):
            call_count[0] += 1
            uploaded_content.append(content)
            return f"https://paste.rs/paste{call_count[0]}"

        with patch("hermes_cli.dump.run_dump") as mock_dump, \
             patch("hermes_cli.debug.upload_to_pastebin",
                    side_effect=_mock_upload):
            mock_dump.side_effect = lambda a: print("--- hermes dump ---\nversion: test\n--- end dump ---")
            run_debug_share(args)

        out = capsys.readouterr().out
        # Should have 5 uploads: report, agent.log, gateway.log, gui.log, desktop.log
        assert call_count[0] == 5
        assert "paste.rs/paste1" in out  # Report
        assert "paste.rs/paste2" in out  # agent.log
        assert "paste.rs/paste3" in out  # gateway.log
        assert "paste.rs/paste4" in out  # gui.log
        assert "paste.rs/paste5" in out  # desktop.log
        assert "Report" in out
        assert "agent.log" in out
        assert "gateway.log" in out
        assert "gui.log" in out
        assert "desktop.log" in out

        # Each log paste should start with the dump header
        agent_paste = uploaded_content[1]
        assert "--- hermes dump ---" in agent_paste
        assert "--- full agent.log ---" in agent_paste
        gateway_paste = uploaded_content[2]
        assert "--- hermes dump ---" in gateway_paste
        assert "--- full gateway.log ---" in gateway_paste
        gui_paste = uploaded_content[3]
        assert "--- hermes dump ---" in gui_paste
        assert "--- full gui.log ---" in gui_paste
        desktop_paste = uploaded_content[4]
        assert "--- hermes dump ---" in desktop_paste
        assert "--- full desktop.log ---" in desktop_paste




# ---------------------------------------------------------------------------
# Share-time redaction wiring + visible banner
# ---------------------------------------------------------------------------

class TestRunDebugShareRedaction:
    """End-to-end: --no-redact flag, banner injection, default behavior."""

    @pytest.fixture
    def hermes_home_with_secret(self, tmp_path, monkeypatch):
        """Isolated HERMES_HOME whose agent.log contains a vendor-prefixed token."""
        home = tmp_path / ".hermes"
        home.mkdir()
        monkeypatch.setenv("HERMES_HOME", str(home))
        monkeypatch.delenv("HERMES_REDACT_SECRETS", raising=False)

        logs_dir = home / "logs"
        logs_dir.mkdir()
        (logs_dir / "agent.log").write_text(
            f"2026-04-12 17:00:00 INFO config: api_key={_REDACT_FIXTURE_TOKEN} loaded\n"
        )
        (logs_dir / "errors.log").write_text("")
        (logs_dir / "gateway.log").write_text(
            f"2026-04-12 17:00:01 INFO gateway.run: token {_REDACT_FIXTURE_TOKEN}\n"
        )
        return home

    def test_default_share_redacts_uploaded_content(
        self, hermes_home_with_secret, capsys
    ):
        """The uploaded report and full-log pastes do not contain the raw token."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False
        args.nous = False
        args.no_redact = False

        captured: list[str] = []

        def fake_upload(content, expiry_days=7):
            captured.append(content)
            return f"https://paste.rs/{len(captured)}"

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \
             patch("hermes_cli.debug.upload_to_pastebin", side_effect=fake_upload):
            run_debug_share(args)

        # At least the report plus one full log paste reached the upload path.
        assert len(captured) >= 2
        for content in captured:
            assert _REDACT_FIXTURE_TOKEN not in content, (
                "raw token leaked into upload-bound content"
            )

    def test_fallback_provider_key_never_reaches_upload_bound_content(self, hermes_home_with_secret):
        """``hermes dump`` quotes ``fallback_providers`` from config.yaml; neither the CLI share
        bundle nor the gateway /debug report may carry a fallback entry's api_key."""
        from hermes_cli.debug import _capture_dump, collect_debug_report, collect_share_bundle

        fallback_key = "fbk-opaque-0123456789abcdefghij"
        (hermes_home_with_secret / "config.yaml").write_text(
            "fallback_providers:\n"
            "  - provider: custom\n"
            "    model: backup-model\n"
            "    base_url: https://backup.example/v1\n"
            f"    api_key: {fallback_key}\n", encoding="utf-8")

        uploads = [*collect_share_bundle(log_lines=20).values(),
                   collect_debug_report(log_lines=20, dump_text=_capture_dump())]

        assert any("backup-model" in text for text in uploads)  # the dump really ran
        assert not [text for text in uploads if fallback_key in text]

    @pytest.mark.parametrize(("yaml_value", "secret"), [
        ("87419362508741936250", "87419362508741936250"),  # unquoted: YAML int
        ("'opaque***Fallback0123456789'", "opaque***Fallback0123456789"),  # looks pre-masked
        ("{value: nestedOpaque0123456789abcdef}", "nestedOpaque0123456789abcdef"),
    ])
    def test_fallback_api_key_masked_whatever_its_yaml_shape(self, hermes_home_with_secret, yaml_value, secret):
        """The runtime ``str()``s any ``api_key`` value (fallback_config.resolve_entry_api_key), so the
        dump masks the field itself rather than relying on text redaction to recognise the value."""
        from hermes_cli.debug import _capture_dump, collect_debug_report, collect_share_bundle

        (hermes_home_with_secret / "config.yaml").write_text(
            "fallback_providers:\n"
            "  - provider: custom\n"
            "    model: backup-model\n"
            f"    api_key: {yaml_value}\n", encoding="utf-8")

        texts = [_capture_dump(redact=False),  # `hermes dump` stdout, as printed
                 *collect_share_bundle(log_lines=20).values(),
                 collect_debug_report(log_lines=20, dump_text=_capture_dump())]

        assert "backup-model" in texts[0]
        assert not [text for text in texts if secret in text]

    @pytest.mark.parametrize(("field", "yaml_value", "secret"), [
        ("token", "87419362508741936250", "87419362508741936250"),
        ("auth_token", "'opaque***AuthToken0123456789'", "opaque***AuthToken0123456789"),
        ("client_secret", "{value: clientSecretOpaque0123456789}", "clientSecretOpaque0123456789"),
        ("password", "98127364509812736450", "98127364509812736450"),
        ("clientApiKey", "'opaque***CamelKey0123456789'", "opaque***CamelKey0123456789"),
    ])
    def test_fallback_secret_fields_masked_by_repo_policy(self, hermes_home_with_secret, field, yaml_value, secret):
        """Every field ``agent.redact`` treats as a credential is masked by field, not only ``api_key``;
        env-var names and token budgets stay readable."""
        from hermes_cli.debug import _capture_dump, collect_debug_report, run_debug_share

        (hermes_home_with_secret / "config.yaml").write_text(
            "fallback_providers:\n"
            "  - provider: custom\n"
            "    model: backup-model\n"
            "    key_env: BACKUP_KEY_ENV_NAME\n"
            "    api_key_env: BACKUP_API_KEY_ENV_NAME\n"
            "    max_tokens: 4321\n"
            f"    {field}: {yaml_value}\n", encoding="utf-8")

        uploaded: list[str] = []
        args = MagicMock(lines=20, expire=1, local=False, nous=False, no_redact=False)
        with patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \
             patch("hermes_cli.debug._schedule_auto_delete"), \
             patch("hermes_cli.debug.upload_to_pastebin",
                   side_effect=lambda content, expiry_days=1: uploaded.append(content) or "https://paste.rs/x"):
            run_debug_share(args)  # what actually reaches the paste service

        assert any("backup-model" in text for text in uploaded)  # the dump really reached the sink
        assert not [text for text in uploaded if secret in text]  # checked first: the sink alone catches a leak

        texts = [_capture_dump(redact=False),  # `hermes dump` stdout, as printed
                 collect_debug_report(log_lines=20, dump_text=_capture_dump())]

        assert all(s in texts[0] for s in ("backup-model", "BACKUP_KEY_ENV_NAME", "BACKUP_API_KEY_ENV_NAME", "4321"))
        assert not [text for text in texts if secret in text]

    @pytest.mark.parametrize("base_url", [
        "https://user:{s}@backup.example/v1",
        "https://backup.example/v1?key={s}",
        "https://backup.example/v1?X-Amz-Signature={s}",
        "https://backup.example/v1?X-Goog-Signature={s}",
        "https://backup.example/v1?sv=2024-11-04&sig={s}",
        "https://backup.example/v1#access_token={s}&view=public",
    ])
    def test_fallback_base_url_credential_never_reaches_upload_bound_content(self, hermes_home_with_secret, base_url):
        """A fallback ``base_url`` can carry its credential in the URL (userinfo, query, fragment). The
        dump is config made to be pasted, so ``hermes dump`` itself and every upload of it get strict
        URL-credential redaction, not the log policy."""
        from hermes_cli.debug import _capture_dump, collect_debug_report, collect_share_bundle

        secret = "urlCredOpaque0123456789abcdef"
        (hermes_home_with_secret / "config.yaml").write_text(
            "fallback_providers:\n"
            "  - provider: custom\n"
            "    model: backup-model\n"
            f"    base_url: '{base_url.format(s=secret)}'\n", encoding="utf-8")

        texts = [_capture_dump(redact=False),  # `hermes dump` stdout, as printed
                 *collect_share_bundle(log_lines=20).values(),
                 collect_debug_report(log_lines=20, dump_text=_capture_dump())]

        assert all("backup-model" in text and "backup.example" in text for text in texts[:2])
        assert not [text for text in texts if secret in text]

    def test_capture_dump_is_independent_strict_redaction_boundary(self):
        """A future dump formatting mistake cannot leak through debug upload paths."""
        from hermes_cli.debug import _capture_dump

        raw_key = "opaqueBoundaryKeyABC123456789"
        raw_password = "boundaryPasswordABC123456789"
        raw_signature = "boundarySignatureABC123456789"

        def leaky_dump(_args):
            print(
                "{'api_key': '" + raw_key + "', "
                "'base_url': 'https://user:" + raw_password
                + "@example.test/v1?X-Amz-Signature=" + raw_signature + "'}"
            )

        with patch("hermes_cli.dump.run_dump", side_effect=leaky_dump):
            safe = _capture_dump()
            raw = _capture_dump(redact=False)

        for secret in (raw_key, raw_password, raw_signature):
            assert secret not in safe
            assert secret in raw

    def test_default_share_includes_redaction_banner(
        self, hermes_home_with_secret, capsys
    ):
        """Each upload-bound paste carries the visible redaction banner."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False
        args.nous = False
        args.no_redact = False

        captured: list[str] = []

        def fake_upload(content, expiry_days=7):
            captured.append(content)
            return f"https://paste.rs/{len(captured)}"

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \
             patch("hermes_cli.debug.upload_to_pastebin", side_effect=fake_upload):
            run_debug_share(args)

        for content in captured:
            assert "redacted at upload time" in content, (
                "redaction banner missing from upload-bound content"
            )

    def test_no_redact_flag_disables_redaction_and_banner(
        self, hermes_home_with_secret, capsys
    ):
        """--no-redact preserves original log content and omits the banner."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False
        args.nous = False
        args.no_redact = True

        captured: list[str] = []

        def fake_upload(content, expiry_days=7):
            captured.append(content)
            return f"https://paste.rs/{len(captured)}"

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \
             patch("hermes_cli.debug.upload_to_pastebin", side_effect=fake_upload):
            run_debug_share(args)

        # The agent.log paste should now contain the raw token.
        assert any(_REDACT_FIXTURE_TOKEN in c for c in captured), (
            "expected raw token in --no-redact upload"
        )
        # No banner anywhere when redaction is disabled.
        for content in captured:
            assert "redacted at upload time" not in content, (
                "banner present with --no-redact"
            )


# ---------------------------------------------------------------------------
# run_debug router
# ---------------------------------------------------------------------------



# ---------------------------------------------------------------------------
# Argparse integration
# ---------------------------------------------------------------------------

# ---------------------------------------------------------------------------
# Delete / auto-delete
# ---------------------------------------------------------------------------



class TestDeletePaste:
    def test_delete_sends_delete_request(self):
        from hermes_cli.debug import delete_paste

        mock_resp = MagicMock()
        mock_resp.status = 200
        mock_resp.__enter__ = lambda s: s
        mock_resp.__exit__ = MagicMock(return_value=False)

        with patch("hermes_cli.debug.urllib.request.urlopen",
                    return_value=mock_resp) as mock_open:
            result = delete_paste("https://paste.rs/abc123")

        assert result is True
        req = mock_open.call_args[0][0]
        assert req.method == "DELETE"
        assert "paste.rs/abc123" in req.full_url

    def test_dpaste_url_error_explains_no_delete(self):
        """dpaste.com pastes have no owner token, so the user must be told the
        paste cannot be deleted and will expire on its own (#106164)."""
        from hermes_cli.debug import delete_paste

        with pytest.raises(ValueError):
            delete_paste("https://dpaste.com/ABC123")


class TestScheduleAutoDelete:
    """``_schedule_auto_delete`` used to spawn a detached Python subprocess
    per call (one per paste URL batch).  Those subprocesses slept 6 hours
    and accumulated forever under repeated use — 15+ orphaned interpreters
    were observed in production.

    The new implementation is stateless: it records pending deletions to
    ``~/.hermes/pastes/pending.json`` and lets ``_sweep_expired_pastes``
    handle the DELETE requests synchronously on the next ``hermes debug``
    invocation.
    """


    def test_records_pending_to_json(self, hermes_home):
        """Scheduled URLs are persisted to pending.json with expiration."""
        from hermes_cli.debug import _schedule_auto_delete, _pending_file
        import json

        _schedule_auto_delete(
            ["https://paste.rs/abc", "https://paste.rs/def"],
            delay_seconds=10,
        )

        pending_path = _pending_file()
        assert pending_path.exists()

        entries = json.loads(pending_path.read_text())
        assert len(entries) == 2
        urls = {e["url"] for e in entries}
        assert urls == {"https://paste.rs/abc", "https://paste.rs/def"}

        # expire_at is ~now + delay_seconds
        import time
        for e in entries:
            assert e["expire_at"] > time.time()
            assert e["expire_at"] <= time.time() + 15



    def test_dedupes_same_url(self, hermes_home):
        """Same URL recorded twice → one entry with the later expire_at."""
        from hermes_cli.debug import _schedule_auto_delete, _load_pending

        _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=10)
        _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=100)

        entries = _load_pending()
        assert len(entries) == 1
        assert entries[0]["url"] == "https://paste.rs/dup"


class TestSweepExpiredPastes:
    """Test the opportunistic sweep that replaces the sleeping subprocess."""


    def test_sweep_deletes_expired_entries(self, hermes_home):
        from hermes_cli.debug import (
            _sweep_expired_pastes,
            _save_pending,
            _load_pending,
        )
        import time

        # Seed pending.json with one expired + one future entry
        _save_pending([
            {"url": "https://paste.rs/expired", "expire_at": time.time() - 100},
            {"url": "https://paste.rs/future", "expire_at": time.time() + 3600},
        ])

        delete_calls = []

        def fake_delete(url):
            delete_calls.append(url)
            return True

        with patch("hermes_cli.debug.delete_paste", side_effect=fake_delete):
            deleted, remaining = _sweep_expired_pastes()

        assert delete_calls == ["https://paste.rs/expired"]
        assert deleted == 1
        assert remaining == 1

        entries = _load_pending()
        urls = {e["url"] for e in entries}
        assert urls == {"https://paste.rs/future"}

    def test_sweep_leaves_future_entries_alone(self, hermes_home):
        from hermes_cli.debug import _sweep_expired_pastes, _save_pending
        import time

        _save_pending([
            {"url": "https://paste.rs/future1", "expire_at": time.time() + 3600},
            {"url": "https://paste.rs/future2", "expire_at": time.time() + 7200},
        ])

        with patch("hermes_cli.debug.delete_paste") as mock_delete:
            deleted, remaining = _sweep_expired_pastes()

        mock_delete.assert_not_called()
        assert deleted == 0
        assert remaining == 2

    def test_sweep_survives_network_failure(self, hermes_home):
        """Failed DELETEs stay in pending.json until the 24h grace window."""
        from hermes_cli.debug import (
            _sweep_expired_pastes,
            _save_pending,
            _load_pending,
        )
        import time

        _save_pending([
            {"url": "https://paste.rs/flaky", "expire_at": time.time() - 100},
        ])

        with patch(
            "hermes_cli.debug.delete_paste",
            side_effect=Exception("network down"),
        ):
            deleted, remaining = _sweep_expired_pastes()

        # Failure within 24h grace → kept for retry
        assert deleted == 0
        assert remaining == 1
        assert len(_load_pending()) == 1








# ---------------------------------------------------------------------------
# build_debug_share — structured core used by the dashboard endpoint
# ---------------------------------------------------------------------------


class TestBuildDebugShare:
    """The shared core that returns structured paste URLs (not printed text).

    Backs both ``hermes debug share`` (CLI) and ``POST /api/ops/debug-share``
    (dashboard). The dashboard renders ``urls`` as real, copyable links, so the
    contract here is the return value, not stdout.
    """



    def test_redaction_keeps_secrets_out_of_payload(self, hermes_home):
        from hermes_cli.debug import build_debug_share

        secret = "sk-proj-SUPERSECRETtoken1234567890"
        (hermes_home / "logs" / "agent.log").write_text(
            f"line one\nauthorization token={secret}\nline three\n"
        )

        uploaded = []

        def _upload(content, expiry_days=7):
            uploaded.append(content)
            return "https://paste.rs/x"

        with patch("hermes_cli.dump.run_dump"), patch(
            "hermes_cli.debug.upload_to_pastebin", side_effect=_upload
        ), patch("hermes_cli.debug._schedule_auto_delete"):
            result = build_debug_share(log_lines=50, redact=True)

        assert result.redacted is True
        joined = "\n".join(uploaded)
        assert secret not in joined, "secret leaked into upload payload"

    def test_optional_log_failure_is_collected_not_raised(self, hermes_home):
        from hermes_cli.debug import build_debug_share

        count = [0]

        def _upload(content, expiry_days=7):
            count[0] += 1
            # First call (the required Report) succeeds; a later one fails.
            if count[0] == 2:
                raise RuntimeError("paste service hiccup")
            return f"https://paste.rs/p{count[0]}"

        with patch("hermes_cli.dump.run_dump"), patch(
            "hermes_cli.debug.upload_to_pastebin", side_effect=_upload
        ), patch("hermes_cli.debug._schedule_auto_delete"):
            result = build_debug_share(log_lines=50, redact=True)

        assert "Report" in result.urls
        assert len(result.failures) == 1
        assert "paste service hiccup" in result.failures[0]


# ---------------------------------------------------------------------------
# Shared bundle collection + Nous-S3 path
# ---------------------------------------------------------------------------

class TestCollectShareBundle:


    def test_redaction_keeps_secrets_out(self, hermes_home):
        from hermes_cli.debug import collect_share_bundle

        secret = "sk-proj-abcdefghijklmnopqrstuvwxyz1234567890"
        (hermes_home / "logs" / "agent.log").write_text(
            f"line one\nOPENAI_API_KEY={secret}\nline three\n"
        )
        with patch("hermes_cli.dump.run_dump"):
            redacted = collect_share_bundle(log_lines=50, redact=True)
            unredacted = collect_share_bundle(log_lines=50, redact=False)

        # Sanity: without redaction the secret is present in the bundle.
        assert secret in "\n".join(unredacted.values())
        # With redaction it must be scrubbed everywhere.
        assert secret not in "\n".join(redacted.values())

    def test_redaction_masks_url_credentials(self, hermes_home):
        """Log-time redaction leaves ``?token=`` and ``user:pass@`` in URLs for tool flows;
        the upload must not carry them."""
        from hermes_cli.debug import collect_share_bundle

        query_token = "Q7fK2mZp9RtX4vLb8NcW1yHs"
        password = "Pw7Kq2Lm9Xs4Vb"
        (hermes_home / "logs" / "agent.log").write_text(
            "2026-09-29 01:00:00 INFO plugins.web.firecrawl.provider: Firecrawl scraping: "
            f"https://files.example.com/export.csv?token={query_token}&page=2\n"
            f"2026-09-29 01:00:01 INFO agent: using proxy http://alice:{password}@10.0.0.5:3128\n"
        )
        with patch("hermes_cli.dump.run_dump"):
            bundle = "\n".join(collect_share_bundle(log_lines=50, redact=True).values())

        assert query_token not in bundle
        assert password not in bundle
        # Query credentials are masked in place; strict URL redaction masks the
        # complete userinfo field while preserving the host and port.
        assert "export.csv?token=***&page=2" in bundle
        assert "***:***@10.0.0.5:3128" in bundle




class TestBuildNousBundle:
    def test_envelope_shape_and_gzip(self, hermes_home):
        import gzip
        import json as _json

        from hermes_cli.debug import build_nous_bundle

        files = {"report": "hello", "agent.log": "log line"}
        blob = build_nous_bundle(files, redact=True)

        # It's gzip — magic bytes.
        assert blob[:2] == b"\x1f\x8b"
        envelope = _json.loads(gzip.decompress(blob).decode())
        assert envelope["format"] == "hermes-debug-share/1"
        assert envelope["redacted"] is True
        assert envelope["files"] == files
        assert "created" in envelope

    def test_redacted_false_recorded(self):
        import gzip
        import json as _json

        from hermes_cli.debug import build_nous_bundle

        blob = build_nous_bundle({"report": "x"}, redact=False)
        envelope = _json.loads(gzip.decompress(blob).decode())
        assert envelope["redacted"] is False


class TestRunDebugShareNous:
    def _args(self, **over):
        class _A:
            lines = 50
            expire = 7
            local = False
            nous = True
            no_redact = False
            yes = True

        a = _A()
        for k, v in over.items():
            setattr(a, k, v)
        return a

    def test_nous_success_prints_view_url(self, hermes_home, capsys):
        from hermes_cli.debug import run_debug_share

        res = {
            "id": "id-1",
            "viewUrl": "https://support.example.com/diagnostics/id-1",
            "expiresAt": "2026-06-20T00:00:00Z",
        }
        with patch("hermes_cli.dump.run_dump"), patch(
            "hermes_cli.diagnostics_upload.share_to_nous", return_value=res
        ) as share:
            run_debug_share(self._args())

        out = capsys.readouterr().out
        assert "https://support.example.com/diagnostics/id-1" in out
        assert "2026-06-20T00:00:00Z" in out
        # The blob passed to share_to_nous must be gzip bytes.
        blob = share.call_args[0][0]
        assert isinstance(blob, (bytes, bytearray)) and blob[:2] == b"\x1f\x8b"

    def test_nous_failure_suggests_local(self, hermes_home, capsys):
        from hermes_cli.debug import run_debug_share

        with patch("hermes_cli.dump.run_dump"), patch(
            "hermes_cli.diagnostics_upload.share_to_nous",
            side_effect=RuntimeError("service down"),
        ):
            with pytest.raises(SystemExit) as exc:
                run_debug_share(self._args())
        assert exc.value.code == 1
        err = capsys.readouterr().err
        assert "--local" in err

    def test_nous_does_not_touch_pastebin(self, hermes_home):
        from hermes_cli.debug import run_debug_share

        res = {"id": "id-1", "viewUrl": "https://v"}
        with patch("hermes_cli.dump.run_dump"), patch(
            "hermes_cli.diagnostics_upload.share_to_nous", return_value=res
        ), patch("hermes_cli.debug.upload_to_pastebin") as paste:
            run_debug_share(self._args())
        paste.assert_not_called()


class TestDebugSlashCommand:
    """`/debug [nous|local]` parsing in the CLI/TUI handler.

    The classic CLI and the TUI slash worker both dispatch through
    ``HermesCLI.process_command`` → ``_handle_debug_command(cmd_original)``,
    which parses an optional destination word and builds the args namespace
    handed to ``run_debug_share``.
    """

    def _handler(self):
        from hermes_cli.cli_commands_mixin import CLICommandsMixin

        class _Stub(CLICommandsMixin):
            pass

        return _Stub()._handle_debug_command

    def _captured(self, cmd_original):
        captured = {}

        def _fake_run(args):
            captured.update(vars(args))

        with patch("hermes_cli.debug.run_debug_share", _fake_run):
            self._handler()(cmd_original)
        return captured

    def test_bare_debug_defaults_to_paste(self):
        c = self._captured("/debug")
        assert c["nous"] is False and c["local"] is False
        # The slash command IS the consent action → skip the [y/N] prompt
        # (input() would hang inside prompt_toolkit's event loop).
        assert c["yes"] is True


    def test_word_parsing_is_case_insensitive(self):
        c = self._captured("/debug NOUS")
        assert c["nous"] is True




class TestShareConsentGate:
    """`hermes debug share` requires explicit consent before uploading.

    Uses SimpleNamespace rather than MagicMock so ``args.yes`` is a real
    ``False`` — a MagicMock auto-provides a truthy ``.yes`` and would silently
    bypass the very gate under test.
    """

    def _args(self, **over):
        from types import SimpleNamespace

        base = dict(lines=50, expire=7, local=False, nous=False,
                    no_redact=False, yes=False)
        base.update(over)
        return SimpleNamespace(**base)




    def test_non_interactive_requires_yes(self, hermes_home, capsys, monkeypatch):
        """No TTY + no --yes → exit(1), never upload silently."""
        from hermes_cli.debug import run_debug_share

        monkeypatch.setattr("sys.stdin.isatty", lambda: False)

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin") as mock_upload:
            with pytest.raises(SystemExit) as exc:
                run_debug_share(self._args())

        assert exc.value.code == 1
        mock_upload.assert_not_called()
        assert "--yes" in capsys.readouterr().err


    def test_local_never_prompts(self, hermes_home, capsys, monkeypatch):
        """--local renders to stdout and must not prompt or upload."""
        from hermes_cli.debug import run_debug_share

        def _boom(_):
            raise AssertionError("input() must not be called for --local")

        monkeypatch.setattr("builtins.input", _boom)

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin") as mock_upload:
            run_debug_share(self._args(local=True))

        mock_upload.assert_not_called()
        assert "Aborted" not in capsys.readouterr().out

