tests/test_discord_cache_export.py — Chat Cache Exporter

#!/usr/bin/env python3
"""Test Chat Cache Exporter with synthetic local cache entries.

By Zero Width Lab (zerowidthlab.com)
"""

from __future__ import annotations

import contextlib
import errno
import gzip
import io
import importlib.util
import json
import os
import re
import runpy
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
CHANNEL_ONE = "111111111111111111"
CHANNEL_TWO = "222222222222222222"
ZSTD_MAGIC = b"\x28\xb5\x2f\xfd"
MAX_CACHE_FILE_BYTES = 128 * 1024 * 1024
HAS_STANDARD_ZSTD = (
    importlib.util.find_spec("compression") is not None
    and importlib.util.find_spec("compression.zstd") is not None
)
WITHOUT_STANDARD_ZSTD = (
    "import runpy, sys; "
    "sys.modules['compression'] = None; "
    "sys.argv = sys.argv[1:]; "
    "runpy.run_path(sys.argv[0], run_name='__main__')"
)
HEADER_PATTERN = r"(?:[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}|[?]{4}-[?]{2}-[?]{2}T[?]{2}:[?]{2}:[?]{2})  .*"


class DiscordCacheExportCliTests(unittest.TestCase):
    def setUp(self) -> None:
        self.package_root = Path(__file__).resolve().parent.parent
        self.script_path = self.package_root / "scripts" / "discord_cache_export.py"
        self.fixture_path = (
            self.package_root / "tests" / "fixtures" / "synthetic-messages.json"
        )
        self.messages = json.loads(self.fixture_path.read_text(encoding="utf-8"))
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.temp_path = Path(self.temp.name).resolve()

    def write_entry(
        self,
        cache_dir: Path,
        name: str,
        payload: object,
        *,
        channel_id: str = CHANNEL_ONE,
    ) -> Path:
        url = (
            f"https://discord.com/api/v9/channels/{channel_id}/messages?limit=50"
        ).encode("ascii")
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        path = cache_dir / name
        path.write_bytes(os.urandom(12) + url + b"\x00" + gzip.compress(body))
        return path

    def make_cache(
        self,
        *,
        include_duplicate: bool = True,
        include_non_list: bool = False,
        include_zstd_junk: bool = False,
    ) -> Path:
        cache_dir = self.temp_path / "cache"
        cache_dir.mkdir()
        self.write_entry(cache_dir, "entry-one", self.messages)
        if include_duplicate:
            self.write_entry(cache_dir, "entry-two", [self.messages[1]])
        (cache_dir / "not-messages").write_bytes(
            b"https://discord.com/api/v9/users/@me" + gzip.compress(b"[]")
        )
        if include_non_list:
            self.write_entry(cache_dir, "object-response", {"message": "not a list"})
        if include_zstd_junk:
            url = (
                f"https://discord.com/api/v10/channels/{CHANNEL_ONE}/messages?limit=50"
            ).encode("ascii")
            (cache_dir / "zstd-junk").write_bytes(url + b"\x00" + ZSTD_MAGIC + b"junk")
        return cache_dir

    def run_cli(
        self,
        *arguments: str,
        env_updates: dict[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        env = os.environ.copy()
        if env_updates:
            env.update(env_updates)
        return subprocess.run(
            [sys.executable, str(self.script_path), *arguments],
            cwd=self.package_root,
            capture_output=True,
            text=True,
            check=False,
            env=env,
        )

    def run_cli_without_standard_zstd(
        self, *arguments: str
    ) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            [sys.executable, "-c", WITHOUT_STANDARD_ZSTD, str(self.script_path), *arguments],
            cwd=self.package_root,
            capture_output=True,
            text=True,
            check=False,
        )

    def dry_run_then_export(
        self, cache_dir: Path, out_dir: Path, *extra: str
    ) -> subprocess.CompletedProcess[str]:
        preview = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir),
            *extra, "--dry-run",
        )
        self.assertEqual(preview.returncode, 0, preview.stderr)
        return self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), *extra
        )

    def export_transcript(self, messages: list[dict[str, object]]) -> list[str]:
        cache_dir = self.temp_path / "transcript-cache"
        cache_dir.mkdir()
        self.write_entry(cache_dir, "entry", messages)
        out_dir = self.temp_path / "transcript-export"
        result = self.dry_run_then_export(cache_dir, out_dir)
        self.assertEqual(result.returncode, 0, result.stderr)
        path = out_dir / "transcripts" / f"{CHANNEL_ONE}.txt"
        return path.read_text(encoding="utf-8").split("\n")[:-1]

    def assert_transcript_lines_are_unambiguous(self, lines: list[str]) -> list[str]:
        headers = []
        for line in lines:
            if line.startswith("    | "):
                continue
            self.assertIsNotNone(re.fullmatch(HEADER_PATTERN, line), line)
            headers.append(line)
        return headers

    def test_scan_prints_summary_table_and_resolved_cache(self) -> None:
        cache_dir = self.make_cache()

        result = self.run_cli("scan", "--cache-dir", str(cache_dir))

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("channel", result.stdout)
        self.assertIn("msgs", result.stdout)
        self.assertIn("authors", result.stdout)
        self.assertIn(CHANNEL_ONE, result.stdout)
        self.assertIn(CHANNEL_TWO, result.stdout)
        self.assertIn("2024-01-01", result.stdout)
        self.assertIn("2024-03-03", result.stdout)
        self.assertIn(str(cache_dir.absolute()), result.stderr)

    def test_scan_json_reports_counts_and_channel_summaries(self) -> None:
        cache_dir = self.make_cache()

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["scanned"], 3)
        self.assertEqual(summary["matched"], 2)
        self.assertEqual(summary["messages"], 3)
        self.assertEqual(summary["channels"], 2)
        self.assertFalse(summary["incomplete"])
        self.assertEqual(summary["skipped"], {"unreadable": 0, "zstd_skipped": 0})
        by_id = {item["channel_id"]: item for item in summary["channel_summaries"]}
        self.assertEqual(by_id[CHANNEL_ONE]["messages"], 2)
        self.assertEqual(by_id[CHANNEL_ONE]["top_display_names"], ["user_a", "User B"])

    def test_export_writes_expected_json_transcripts_and_private_modes(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "chosen-export"

        result = self.dry_run_then_export(cache_dir, out_dir)

        self.assertEqual(result.returncode, 0, result.stderr)
        messages_path = out_dir / "messages.json"
        first_transcript = out_dir / "transcripts" / f"{CHANNEL_ONE}.txt"
        second_transcript = out_dir / "transcripts" / f"{CHANNEL_TWO}.txt"
        self.assertTrue(messages_path.is_file())
        self.assertTrue(first_transcript.is_file())
        self.assertTrue(second_transcript.is_file())
        exported = json.loads(messages_path.read_text(encoding="utf-8"))
        self.assertEqual(exported["tool"], "discord-cache-export")
        self.assertEqual(exported["version"], "0.1.2")
        self.assertEqual(exported["messages"], 3)
        self.assertEqual(list(exported["channels"]), [CHANNEL_ONE, CHANNEL_TWO])
        ordered_ids = [message["id"] for message in exported["channels"][CHANNEL_ONE]]
        self.assertEqual(
            ordered_ids, ["117000000000000001", "118000000000000002"]
        )
        transcript = first_transcript.read_text(encoding="utf-8")
        self.assertIn(
            "2024-01-01T01:02:03  user_a: Earlier synthetic message  "
            "[attachment: synthetic.txt]",
            transcript,
        )
        self.assertIn(
            "2024-02-02T03:04:05  User B: Later synthetic message  "
            "[embed: https://example.invalid/synthetic-embed]",
            transcript,
        )
        if os.name != "nt":
            self.assertEqual(stat.S_IMODE(out_dir.stat().st_mode), 0o700)
            self.assertEqual(stat.S_IMODE(messages_path.stat().st_mode), 0o600)
            self.assertEqual(stat.S_IMODE(first_transcript.stat().st_mode), 0o600)

    def test_export_dry_run_lists_files_and_writes_only_approval_state(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "dry-run-export"

        result = self.run_cli(
            "export",
            "--cache-dir",
            str(cache_dir),
            "--out",
            str(out_dir),
            "--dry-run",
            "--json",
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(
            summary["would_write"],
            [
                str(out_dir / "messages.json"),
                str(out_dir / "transcripts" / f"{CHANNEL_ONE}.txt"),
                str(out_dir / "transcripts" / f"{CHANNEL_TWO}.txt"),
            ],
        )
        state_path = out_dir / ".discord-cache-export-approved.json"
        self.assertEqual(summary["approval_file"], str(state_path))
        self.assertEqual(len(summary["approval_digest"]), 64)
        self.assertTrue(
            all(
                character in "0123456789abcdef"
                for character in summary["approval_digest"]
            )
        )
        self.assertTrue(out_dir.is_dir())
        self.assertEqual([path.name for path in out_dir.iterdir()], [state_path.name])
        state = json.loads(state_path.read_text(encoding="utf-8"))
        self.assertEqual(state["approval_digest"], summary["approval_digest"])
        self.assertNotIn(str(state_path), summary["would_write"])
        if os.name != "nt":
            self.assertEqual(stat.S_IMODE(out_dir.stat().st_mode), 0o700)
            self.assertEqual(stat.S_IMODE(state_path.stat().st_mode), 0o600)

    def test_live_export_without_dry_run_is_refused_and_writes_nothing(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "unapproved-export"

        result = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir)
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("no dry-run approval state", result.stderr)
        self.assertIn("export --dry-run", result.stderr)
        self.assertIn("confirmation", result.stderr)
        self.assertEqual(result.stdout, "")
        self.assertFalse(out_dir.exists())

    def test_live_export_into_existing_empty_directory_needs_dry_run(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "empty-unapproved-export"
        out_dir.mkdir()

        result = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir)
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("no dry-run approval state", result.stderr)
        self.assertEqual(list(out_dir.iterdir()), [])

    def test_live_export_after_cache_change_is_refused(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        out_dir = self.temp_path / "cache-changed-export"
        preview = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--dry-run"
        )
        self.assertEqual(preview.returncode, 0, preview.stderr)
        added = {**self.messages[0], "id": "121000000000000005"}
        self.write_entry(cache_dir, "added-after-preview", [added])

        result = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir)
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("changed since --dry-run", result.stderr)
        self.assertEqual(
            [path.name for path in out_dir.iterdir()],
            [".discord-cache-export-approved.json"],
        )

    def test_live_export_accepts_unchanged_approved_plan(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "approved-export"

        preview = self.run_cli(
            "export",
            "--cache-dir",
            str(cache_dir),
            "--out",
            str(out_dir),
            "--dry-run",
            "--json",
        )
        live = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--json"
        )

        self.assertEqual(preview.returncode, 0, preview.stderr)
        self.assertEqual(live.returncode, 0, live.stderr)
        self.assertTrue((out_dir / "messages.json").is_file())
        self.assertTrue(
            (out_dir / ".discord-cache-export-approved.json").is_file()
        )

    def test_failed_export_removes_only_its_own_files_and_allows_retry(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "failing-export"
        preview = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--dry-run"
        )
        self.assertEqual(preview.returncode, 0, preview.stderr)
        state_path = out_dir / ".discord-cache-export-approved.json"
        approved_state = state_path.read_bytes()

        namespace = runpy.run_path(str(self.script_path))
        module_globals = namespace["write_export"].__globals__
        real_os = module_globals["os"]
        opened: list[int] = []

        class PartiallyWritingHandle:
            def __init__(self, handle):
                self.handle = handle

            def __enter__(self):
                return self

            def __exit__(self, exception_type, exception, traceback):
                self.handle.close()
                return False

            def write(self, text):
                self.handle.write(text[:5])
                self.handle.flush()
                raise OSError(errno.ENOSPC, "synthetic disk full")

        class SecondWriteFailsOs:
            def __getattr__(self, name):
                return getattr(real_os, name)

            @staticmethod
            def fdopen(descriptor, *arguments, **keywords):
                handle = real_os.fdopen(descriptor, *arguments, **keywords)
                mode = arguments[0] if arguments else keywords.get("mode", "r")
                if "w" not in mode:
                    return handle
                opened.append(descriptor)
                if len(opened) == 2:
                    return PartiallyWritingHandle(handle)
                return handle

        module_globals["os"] = SecondWriteFailsOs()
        stderr = io.StringIO()
        with contextlib.redirect_stderr(stderr), contextlib.redirect_stdout(io.StringIO()):
            code = namespace["main"](
                ["export", "--cache-dir", str(cache_dir), "--out", str(out_dir)]
            )

        self.assertEqual(code, 1)
        self.assertIn("synthetic disk full", stderr.getvalue())
        self.assertEqual(len(opened), 2)
        self.assertEqual(sorted(path.name for path in out_dir.iterdir()), [state_path.name])
        self.assertEqual(state_path.read_bytes(), approved_state)

        retry = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir)
        )
        self.assertEqual(retry.returncode, 0, retry.stderr)
        self.assertTrue((out_dir / "messages.json").is_file())
        self.assertTrue((out_dir / "transcripts" / f"{CHANNEL_ONE}.txt").is_file())

    def test_live_export_rejects_changed_approved_plan(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "changed-export"
        preview = self.run_cli(
            "export",
            "--cache-dir",
            str(cache_dir),
            "--out",
            str(out_dir),
            "--dry-run",
            "--json",
        )
        self.assertEqual(preview.returncode, 0, preview.stderr)
        original_digest = json.loads(preview.stdout)["approval_digest"]
        changed = {
            **self.messages[0],
            "id": "120000000000000004",
            "content": "New synthetic message after preview",
        }
        self.write_entry(cache_dir, "new-after-preview", [changed])

        rejected = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--json"
        )

        self.assertEqual(rejected.returncode, 1)
        self.assertIn("changed since --dry-run", rejected.stderr)
        self.assertIn("re-confirm", rejected.stderr)
        self.assertFalse((out_dir / "messages.json").exists())
        self.assertFalse((out_dir / "transcripts").exists())

        refreshed = self.run_cli(
            "export",
            "--cache-dir",
            str(cache_dir),
            "--out",
            str(out_dir),
            "--dry-run",
            "--json",
        )
        self.assertEqual(refreshed.returncode, 0, refreshed.stderr)
        self.assertNotEqual(
            json.loads(refreshed.stdout)["approval_digest"], original_digest
        )

        accepted = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--json"
        )
        self.assertEqual(accepted.returncode, 0, accepted.stderr)
        self.assertTrue((out_dir / "messages.json").is_file())

    def test_live_export_rejects_changed_content_with_same_message_ids(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        out_dir = self.temp_path / "changed-content-export"
        preview = self.run_cli(
            "export",
            "--cache-dir",
            str(cache_dir),
            "--out",
            str(out_dir),
            "--dry-run",
        )
        self.assertEqual(preview.returncode, 0, preview.stderr)
        changed_messages = [
            {
                **message,
                "content": (
                    "Changed synthetic content"
                    if message["id"] == self.messages[0]["id"]
                    else message["content"]
                ),
            }
            for message in self.messages
        ]
        self.write_entry(cache_dir, "entry-one", changed_messages)

        result = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir)
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("changed since --dry-run", result.stderr)
        self.assertFalse((out_dir / "messages.json").exists())

    def test_repeatable_channel_filter_limits_scan_and_export(self) -> None:
        cache_dir = self.make_cache()
        scan = self.run_cli(
            "scan",
            "--cache-dir",
            str(cache_dir),
            "--channel",
            CHANNEL_TWO,
            "--channel",
            CHANNEL_TWO,
            "--json",
        )
        self.assertEqual(scan.returncode, 0, scan.stderr)
        summary = json.loads(scan.stdout)
        self.assertEqual(summary["messages"], 1)
        self.assertEqual(summary["channels"], 1)

        out_dir = self.temp_path / "filtered-export"
        export = self.dry_run_then_export(
            cache_dir, out_dir, "--channel", CHANNEL_TWO
        )
        self.assertEqual(export.returncode, 0, export.stderr)
        transcript_names = sorted(path.name for path in (out_dir / "transcripts").iterdir())
        self.assertEqual(transcript_names, [f"{CHANNEL_TWO}.txt"])
        payload = json.loads((out_dir / "messages.json").read_text(encoding="utf-8"))
        self.assertEqual(list(payload["channels"]), [CHANNEL_TWO])

    def test_unknown_channel_is_exit_one_and_lists_available_ids(self) -> None:
        cache_dir = self.make_cache()

        result = self.run_cli(
            "scan", "--cache-dir", str(cache_dir), "--channel", "999999999999999999"
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("Available", result.stderr)
        self.assertIn(CHANNEL_ONE, result.stderr)
        self.assertIn(CHANNEL_TWO, result.stderr)
        self.assertNotIn("traceback", result.stderr.lower())

    def test_missing_cache_directory_is_exit_one(self) -> None:
        missing = self.temp_path / "missing"

        result = self.run_cli("scan", "--cache-dir", str(missing))

        self.assertEqual(result.returncode, 1)
        self.assertIn(str(missing.absolute()), result.stderr)
        self.assertIn("not found", result.stderr.lower())

    def test_zero_matched_entries_is_exit_one_with_hint(self) -> None:
        cache_dir = self.temp_path / "empty-cache"
        cache_dir.mkdir()
        (cache_dir / "unrelated").write_bytes(b"synthetic cache data")

        result = self.run_cli("scan", "--cache-dir", str(cache_dir))

        self.assertEqual(result.returncode, 1)
        self.assertIn("none matched", result.stderr.lower())
        self.assertIn("cleared", result.stderr.lower())
        self.assertIn("api version", result.stderr.lower())

    def test_partial_result_returns_two_unless_caller_allows_it(self) -> None:
        cache_dir = self.make_cache(include_zstd_junk=True)

        failed = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")
        allowed = self.run_cli(
            "scan", "--cache-dir", str(cache_dir), "--json", "--allow-partial"
        )

        self.assertEqual(failed.returncode, 2, failed.stderr)
        self.assertEqual(allowed.returncode, 0, allowed.stderr)
        for result in (failed, allowed):
            summary = json.loads(result.stdout)
            self.assertTrue(summary["incomplete"])
            self.assertGreater(
                summary["skipped"]["unreadable"]
                + summary["skipped"]["zstd_skipped"],
                0,
            )
            self.assertIn("incomplete", result.stderr.lower())

    def test_zstd_stream_reader_uses_a_bounded_read(self) -> None:
        requested_sizes: list[int] = []

        class BoundedReader:
            def __enter__(self):
                return self

            def __exit__(self, exception_type, exception, traceback):
                return False

            def read(self, size):
                requested_sizes.append(size)
                return b"x" * size

        class FakeStandardZstd:
            @staticmethod
            def open(source, mode):
                self.assertEqual(mode, "rb")
                self.assertTrue(source.read(4).startswith(ZSTD_MAGIC))
                return BoundedReader()

        namespace = runpy.run_path(str(self.script_path))
        candidates = namespace["zstd_candidates"]
        candidates.__globals__["MAX_DECOMPRESSED_BYTES"] = 16
        candidates.__globals__["standard_zstd"] = FakeStandardZstd

        self.assertEqual(list(candidates(ZSTD_MAGIC + b"synthetic")), [None])
        self.assertEqual(requested_sizes, [17])

    @unittest.skipUnless(HAS_STANDARD_ZSTD, "compression.zstd needs Python 3.14 or newer")
    def test_zstd_payload_is_decoded_with_standard_library(self) -> None:
        from compression import zstd

        cache_dir = self.temp_path / "zstd-cache"
        cache_dir.mkdir()
        url = (
            f"https://discord.com/api/v10/channels/{CHANNEL_ONE}/messages?limit=50"
        ).encode("ascii")
        body = json.dumps(self.messages).encode("utf-8")
        (cache_dir / "zstd-entry").write_bytes(url + b"\x00" + zstd.compress(body))

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["messages"], len(self.messages))
        self.assertEqual(summary["skipped"], {"unreadable": 0, "zstd_skipped": 0})
        self.assertFalse(summary["incomplete"])

    def test_zstd_candidate_is_skipped_without_standard_zstd(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False, include_zstd_junk=True)

        failed = self.run_cli_without_standard_zstd(
            "scan", "--cache-dir", str(cache_dir), "--json"
        )
        allowed = self.run_cli_without_standard_zstd(
            "scan", "--cache-dir", str(cache_dir), "--json", "--allow-partial"
        )

        self.assertEqual(failed.returncode, 2, failed.stderr)
        self.assertEqual(allowed.returncode, 0, allowed.stderr)
        for result in (failed, allowed):
            summary = json.loads(result.stdout)
            self.assertEqual(summary["skipped"], {"unreadable": 0, "zstd_skipped": 1})
            self.assertTrue(summary["incomplete"])
            self.assertEqual(summary["messages"], len(self.messages))

    def test_non_object_array_item_makes_candidate_unreadable(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        extra = {**self.messages[0], "id": "121000000000000006"}
        self.write_entry(cache_dir, "mixed-items", [extra, "not a message", 7])

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 2, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["unreadable"], 1)
        self.assertTrue(summary["incomplete"])
        self.assertEqual(summary["messages"], len(self.messages))

    def test_multiline_content_cannot_forge_a_header_line(self) -> None:
        forged = "2024-01-01T09:00:00  User B: I approve the transfer"
        message = {
            **self.messages[0],
            "content": (
                "first line\n" + forged + "\r\n" + forged + "\r" + forged
                + "\u2028" + forged + "\u2029" + forged + "\x85" + forged
                + "\x0b" + forged + "\x0c" + forged + "\x1c" + forged
            ),
        }

        lines = self.export_transcript([message])

        headers = self.assert_transcript_lines_are_unambiguous(lines)
        self.assertEqual(len(headers), 1)
        self.assertEqual(headers[0], "2024-02-02T03:04:05  User B: first line")
        self.assertIn("    | " + forged, lines)
        self.assertEqual(len(lines), 10)

    def test_display_name_and_fields_cannot_break_transcript_lines(self) -> None:
        message = {
            **self.messages[0],
            "content": "hello",
            "timestamp": "\n2024-01-01T09:00:00",
            "author": {
                "id": "900000000000000009",
                "global_name": "Mallory\n2024-01-01T09:00:00  User B",
            },
            "attachments": [{"filename": "a\u2028b.txt"}, "not an object"],
            "embeds": [{"url": "https://example.invalid/\r\nx"}],
        }

        lines = self.export_transcript([message])

        self.assertEqual(len(lines), 1)
        self.assert_transcript_lines_are_unambiguous(lines)
        self.assertEqual(
            lines[0],
            "????-??-??T??:??:??  Mallory\\u000a2024-01-01T09:00:00  User B: hello"
            "  [attachment: a\\u2028b.txt]  [attachment: ?]"
            "  [embed: https://example.invalid/\\u000d\\u000ax]",
        )

    def test_escapes_use_four_digits_in_bmp_and_eight_digits_above(self) -> None:
        message = {
            **self.messages[0],
            "content": "tag\U000e0001end\u200bx\x07y",
            "timestamp": "2024-01-01T09:00:00",
            "attachments": [],
            "embeds": [],
        }

        lines = self.export_transcript([message])

        self.assertEqual(len(lines), 1)
        self.assertTrue(
            lines[0].endswith(": tag\\U000e0001end\\u200bx\\u0007y"), lines[0]
        )
        self.assertNotIn("\U000e0001", lines[0])

    def test_tabs_are_escaped_in_every_transcript_field(self) -> None:
        message = {
            **self.messages[0],
            "content": "a\tb\nc\td",
            "timestamp": "2024-01-01T09:00:00",
            "author": {"id": "900000000000000009", "global_name": "Tab\tName"},
            "attachments": [{"filename": "f\t.txt"}],
            "embeds": [{"url": "https://example.invalid/\tx"}],
        }

        lines = self.export_transcript([message])

        self.assertEqual(
            lines,
            [
                "2024-01-01T09:00:00  Tab\\u0009Name: a\\u0009b",
                "    | c\\u0009d  [attachment: f\\u0009.txt]"
                "  [embed: https://example.invalid/\\u0009x]",
            ],
        )
        self.assertFalse(any("\t" in line for line in lines))

    def test_summary_table_escapes_display_names_and_dates(self) -> None:
        message = {
            **self.messages[0],
            "timestamp": "\n2024\x1b[2J-01-01",
            "author": {
                "id": "900000000000000009",
                "global_name": "Eve\n\x1b[31mforged\trow",
            },
        }
        cache_dir = self.temp_path / "summary-cache"
        cache_dir.mkdir()
        self.write_entry(cache_dir, "entry", [message])
        out_dir = self.temp_path / "summary-export"

        scan = self.run_cli("scan", "--cache-dir", str(cache_dir))
        preview = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--dry-run"
        )
        live = self.run_cli("export", "--cache-dir", str(cache_dir), "--out", str(out_dir))

        for result in (scan, preview, live):
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertNotIn("\x1b", result.stdout)
            self.assertNotIn("\t", result.stdout)
            rows = [
                line for line in result.stdout.splitlines()
                if line.startswith(CHANNEL_ONE)
            ]
            self.assertEqual(len(rows), 1, result.stdout)
            self.assertIn(
                "????-??-??  ????-??-??  "
                "Eve\\u000a\\u001b[31mforged\\u0009row",
                rows[0],
            )
        json_scan = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")
        summary = json.loads(json_scan.stdout)["channel_summaries"][0]
        self.assertEqual(summary["first_date"], "????-??-??")
        self.assertEqual(summary["last_date"], "????-??-??")

    def test_zero_channel_export_creates_no_transcripts_directory(self) -> None:
        cache_dir = self.temp_path / "empty-array-cache"
        cache_dir.mkdir()
        self.write_entry(cache_dir, "entry", [])
        out_dir = self.temp_path / "empty-array-export"

        preview = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir),
            "--dry-run", "--json",
        )
        self.assertEqual(preview.returncode, 0, preview.stderr)
        self.assertEqual(
            json.loads(preview.stdout)["would_write"], [str(out_dir / "messages.json")]
        )
        live = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--json"
        )

        self.assertEqual(live.returncode, 0, live.stderr)
        self.assertEqual(
            json.loads(live.stdout)["written"], [str(out_dir / "messages.json")]
        )
        self.assertEqual(
            sorted(path.name for path in out_dir.iterdir()),
            [".discord-cache-export-approved.json", "messages.json"],
        )
        self.assertFalse(os.path.lexists(out_dir / "transcripts"))

    def test_non_list_gzip_payload_is_unreadable_and_partial(self) -> None:
        cache_dir = self.make_cache(include_non_list=True)

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 2, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["unreadable"], 1)
        self.assertTrue(summary["incomplete"])

    def test_non_ascii_digit_message_id_is_unreadable_and_partial(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        invalid = {
            **self.messages[0],
            "id": "\u00b2\u00b2\u00b2\u00b2",
        }
        self.write_entry(cache_dir, "non-ascii-id", [invalid])

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 2, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["unreadable"], 1)
        self.assertTrue(summary["incomplete"])
        self.assertEqual(summary["messages"], len(self.messages))
        self.assertIn(
            CHANNEL_ONE,
            {item["channel_id"] for item in summary["channel_summaries"]},
        )

    def test_is_file_error_counts_as_scanned_and_unreadable(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        real_entries = list(os.scandir(cache_dir))
        message_entry = next(
            entry for entry in real_entries if entry.name == "entry-one"
        )
        test_case = self

        class FailingEntry:
            name = "entry-failing-stat"
            path = str(cache_dir / name)

            def is_file(self, *, follow_symlinks):
                test_case.assertFalse(follow_symlinks)
                raise OSError("synthetic stat failure")

        class SyntheticScandir:
            def __enter__(self):
                return iter([message_entry, FailingEntry()])

            def __exit__(self, exception_type, exception, traceback):
                return False

        namespace = runpy.run_path(str(self.script_path))
        recover = namespace["recover"]
        with mock.patch.object(
            recover.__globals__["os"],
            "scandir",
            return_value=SyntheticScandir(),
        ):
            result = recover(cache_dir)

        self.assertEqual(result.scanned, 2)
        self.assertEqual(result.matched, 1)
        self.assertEqual(result.unreadable, 1)
        self.assertTrue(result.incomplete)

    def test_truncated_gzip_stream_is_unreadable_and_partial(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        url = (
            f"https://discord.com/api/v9/channels/{CHANNEL_ONE}/messages?limit=50"
        ).encode("ascii")
        compressed = gzip.compress(json.dumps(self.messages).encode("utf-8"))
        (cache_dir / "truncated-gzip").write_bytes(url + compressed[:-8])

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 2, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["unreadable"], 1)
        self.assertTrue(summary["incomplete"])

    def test_oversized_cache_file_is_skipped_as_unreadable(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        oversized = cache_dir / "oversized-entry"
        with oversized.open("wb") as handle:
            handle.truncate(MAX_CACHE_FILE_BYTES + 1)

        result = self.run_cli(
            "scan", "--cache-dir", str(cache_dir), "--json", "--allow-partial"
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["unreadable"], 1)
        self.assertTrue(summary["incomplete"])

    def test_duplicate_message_ids_are_deduplicated_and_sorted(self) -> None:
        cache_dir = self.make_cache(include_duplicate=True)

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["messages"], 3)
        first = next(
            item for item in summary["channel_summaries"] if item["channel_id"] == CHANNEL_ONE
        )
        self.assertEqual(first["first_date"], "2024-01-01")
        self.assertEqual(first["last_date"], "2024-02-02")

    def test_distinct_authors_use_author_ids_not_shared_display_names(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        same_name = {
            **self.messages[0],
            "id": "120000000000000004",
            "timestamp": "2024-04-04T05:06:07.000000+00:00",
            "author": {
                "id": "900000000000000004",
                "username": "user_d",
                "global_name": "User B",
            },
        }
        self.write_entry(cache_dir, "same-name-author", [same_name])

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        first = next(
            item for item in summary["channel_summaries"] if item["channel_id"] == CHANNEL_ONE
        )
        self.assertEqual(first["authors"], 3)
        self.assertEqual(first["top_display_names"][0], "User B")

    def test_direct_regular_files_only_with_no_symlink_following(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        external = self.temp_path / "external-entry"
        external_payload = [
            {
                **self.messages[0],
                "id": "120000000000000004",
                "channel_id": "333333333333333333",
            }
        ]
        external_url = (
            "https://discord.com/api/v9/channels/333333333333333333/messages?limit=50"
        ).encode("ascii")
        external.write_bytes(external_url + gzip.compress(json.dumps(external_payload).encode()))
        nested = cache_dir / "nested"
        nested.mkdir()
        self.write_entry(nested, "nested-entry", external_payload, channel_id="333333333333333333")
        try:
            (cache_dir / "linked-entry").symlink_to(external)
        except OSError:
            self.skipTest("symlinks are unavailable")

        result = self.run_cli("scan", "--cache-dir", str(cache_dir), "--json")

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        self.assertEqual(summary["scanned"], 2)
        self.assertEqual(summary["channels"], 2)
        ids = {item["channel_id"] for item in summary["channel_summaries"]}
        self.assertNotIn("333333333333333333", ids)

    def test_unreadable_regular_file_is_counted_when_permissions_apply(self) -> None:
        cache_dir = self.make_cache(include_duplicate=False)
        blocked = self.write_entry(cache_dir, "blocked-entry", [self.messages[0]])
        blocked.chmod(0o000)
        self.addCleanup(blocked.chmod, 0o600)

        result = self.run_cli(
            "scan", "--cache-dir", str(cache_dir), "--json", "--allow-partial"
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        summary = json.loads(result.stdout)
        if summary["unreadable"] == 0:
            self.skipTest("this process can read mode 000 files")
        self.assertEqual(summary["unreadable"], 1)
        self.assertTrue(summary["incomplete"])

    def test_default_client_cache_location_is_used(self) -> None:
        home = self.temp_path / "home"
        if sys.platform == "darwin":
            cache_dir = home / "Library" / "Application Support" / "discordptb" / "Cache" / "Cache_Data"
        elif os.name == "nt":
            cache_dir = self.temp_path / "appdata" / "discordptb" / "Cache" / "Cache_Data"
        else:
            cache_dir = home / ".config" / "discordptb" / "Cache" / "Cache_Data"
        cache_dir.mkdir(parents=True)
        self.write_entry(cache_dir, "entry", self.messages)
        env = {"HOME": str(home)}
        if os.name == "nt":
            env["APPDATA"] = str(self.temp_path / "appdata")

        result = self.run_cli("scan", "--client", "discordptb", env_updates=env)

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn(str(cache_dir.absolute()), result.stderr)

    def test_export_rejects_output_inside_skill_directory(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.package_root / "export"

        result = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir), "--dry-run"
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("outside the skill directory", result.stderr)
        self.assertFalse(out_dir.exists())

    def test_export_rejects_symlink_alias_into_skill_directory(self) -> None:
        cache_dir = self.make_cache()
        alias = self.temp_path / "skill-alias"
        try:
            alias.symlink_to(self.package_root, target_is_directory=True)
        except OSError:
            self.skipTest("symlinks are unavailable")

        result = self.run_cli(
            "export",
            "--cache-dir",
            str(cache_dir),
            "--out",
            str(alias / "export"),
            "--dry-run",
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("outside the skill directory", result.stderr)
        self.assertFalse((self.package_root / "export").exists())

    def test_output_guard_rejects_case_varied_skill_directory_alias(self) -> None:
        namespace = runpy.run_path(str(self.script_path))
        validate_output_directory = namespace["validate_output_directory"]
        cli_error = namespace["CliError"]
        case_varied_root = self.package_root.with_name(
            self.package_root.name.swapcase()
        )

        with self.assertRaisesRegex(cli_error, "outside the skill directory"):
            validate_output_directory(case_varied_root / "export")

    def test_export_rejects_nonempty_directory_and_preserves_stale_files(self) -> None:
        cache_dir = self.make_cache()
        out_dir = self.temp_path / "existing-export"
        transcripts = out_dir / "transcripts"
        transcripts.mkdir(parents=True)
        stale = transcripts / "333333333333333333.txt"
        stale.write_text("stale synthetic transcript\n", encoding="utf-8")

        result = self.run_cli(
            "export", "--cache-dir", str(cache_dir), "--out", str(out_dir)
        )

        self.assertEqual(result.returncode, 1)
        self.assertIn("must be empty", result.stderr)
        self.assertEqual(stale.read_text(encoding="utf-8"), "stale synthetic transcript\n")

    def test_help_ends_with_publication_signature(self) -> None:
        result = self.run_cli("--help")

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout.rstrip().splitlines()[-1], SIGNATURE)

    def test_sources_contain_no_private_identifiers_or_unapproved_guids(self) -> None:
        source_paths = [
            path
            for path in self.package_root.rglob("*")
            if path.is_file()
            and ".git" not in path.parts
            and "__pycache__" not in path.parts
            and path.suffix.lower() in {".py", ".md", ".json", ".txt"}
        ]
        guid_pattern = (
            r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
            r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
        )
        email_pattern = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
        user_home_marker = "/" + "Users/"
        for path in source_paths:
            text = path.read_text(encoding="utf-8")
            self.assertNotIn(user_home_marker, text, str(path))
            self.assertEqual(re.findall(guid_pattern, text), [], str(path))
            self.assertEqual(re.findall(email_pattern, text), [], str(path))

    def test_shipped_sources_exclude_banned_tokens_unicode_and_vendor_names(self) -> None:
        banned_tokens = (
            "com" + "pile(",
            "ev" + "al(",
            "ex" + "ec(",
            "os." + "system",
            "shell" + "=True",
            "pick" + "le",
            "cty" + "pes",
        )
        vendor_terms = (
            "open" + "ai",
            "anth" + "ropic",
            "clau" + "de",
            "gem" + "ini",
            "copi" + "lot",
            "g" + "pt-",
        )
        forbidden_unicode = ("\u2013", "\u2014", "\u200b", "\u200c", "\u200d", "\ufeff")
        for path in self.package_root.rglob("*"):
            if not path.is_file() or ".git" in path.parts or "__pycache__" in path.parts:
                continue
            if path.suffix.lower() not in {".py", ".md", ".json", ".txt"}:
                continue
            text = path.read_text(encoding="utf-8")
            lowered = text.lower()
            for token in banned_tokens:
                self.assertNotIn(token, text, str(path))
            for term in vendor_terms:
                self.assertNotIn(term, lowered, str(path))
            for character in forbidden_unicode:
                self.assertNotIn(character, text, str(path))

    def test_output_approval_binds_resolved_path_and_both_identities(self) -> None:
        namespace = runpy.run_path(str(self.script_path))
        digest = namespace["approval_digest"]
        out = self.temp_path / "approved"
        out.mkdir()
        original = digest(out, {}, {})
        out.rename(self.temp_path / "old-output")
        out.mkdir()
        self.assertNotEqual(original, digest(out, {}, {}))
        parent = self.temp_path / "parent"
        parent.mkdir()
        out.rename(parent / "approved")
        self.assertNotEqual(original, digest(parent / "approved", {}, {}))
        parent.rename(self.temp_path / "old-parent")
        parent.mkdir()
        (self.temp_path / "old-parent" / "approved").rename(parent / "approved")
        self.assertNotEqual(digest(self.temp_path / "old-parent", {}, {}),
                            digest(parent, {}, {}))

    def test_parent_identity_changes_digest_with_same_output_inode(self) -> None:
        digest = runpy.run_path(str(self.script_path))["approval_digest"]
        parent = self.temp_path / "parent"
        parent.mkdir()
        out = parent / "export"
        out.mkdir()
        before = digest(out, {}, {})
        inode = out.stat().st_ino
        old = self.temp_path / "old-parent"
        parent.rename(old)
        parent.mkdir()
        (old / "export").rename(out)
        self.assertEqual(out.stat().st_ino, inode)
        self.assertNotEqual(before, digest(out, {}, {}))

    def test_parent_swap_between_verification_and_open_is_refused(self) -> None:
        cache = self.make_cache()
        parent = self.temp_path / "parent"
        parent.mkdir()
        out = parent / "export"
        self.assertEqual(self.run_cli("export", "--cache-dir", str(cache), "--out", str(out),
                                     "--dry-run").returncode, 0)
        namespace = runpy.run_path(str(self.script_path))
        scope = namespace["main"].__globals__
        original = scope["open_directory_components"]
        def swapped(path):
            parent.rename(self.temp_path / "preserved")
            parent.symlink_to(self.temp_path / "preserved", target_is_directory=True)
            return original(path)
        scope["open_directory_components"] = swapped
        with contextlib.redirect_stderr(io.StringIO()):
            code = namespace["main"](["export", "--cache-dir", str(cache), "--out", str(out)])
        self.assertEqual(code, 1)
        self.assertFalse((out / "messages.json").exists())
        self.assertFalse((out / "transcripts").exists())

    def test_parent_symlink_swap_after_approval_writes_nothing(self) -> None:
        cache = self.make_cache()
        parent = self.temp_path / "parent"
        parent.mkdir()
        out = parent / "export"
        preview = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out), "--dry-run")
        self.assertEqual(preview.returncode, 0, preview.stderr)
        preserved = self.temp_path / "preserved"
        parent.rename(preserved)
        parent.symlink_to(preserved, target_is_directory=True)
        live = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out))
        self.assertEqual(live.returncode, 1)
        self.assertEqual([p.name for p in (preserved / "export").iterdir()],
                         [".discord-cache-export-approved.json"])

    def test_parent_and_output_replacements_reject_copied_approval(self) -> None:
        cache = self.make_cache()
        for replace_parent in (False, True):
            parent = self.temp_path / str(replace_parent)
            parent.mkdir()
            out = parent / "export"
            preview = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out), "--dry-run")
            self.assertEqual(preview.returncode, 0, preview.stderr)
            state = (out / ".discord-cache-export-approved.json").read_bytes()
            target = parent if replace_parent else out
            target.rename(target.with_name(target.name + "-old"))
            target.mkdir()
            out.mkdir(exist_ok=True)
            (out / ".discord-cache-export-approved.json").write_bytes(state)
            live = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out))
            self.assertEqual(live.returncode, 1)
            self.assertIn("changed since --dry-run", live.stderr)
            self.assertFalse((out / "messages.json").exists())

    def test_swap_in_and_back_cannot_redirect_approved_export(self) -> None:
        cache = self.make_cache()
        parent = self.temp_path / "parent"
        parent.mkdir()
        out = parent / "export"
        self.assertEqual(self.run_cli("export", "--cache-dir", str(cache), "--out", str(out),
                                     "--dry-run").returncode, 0)
        evil = self.temp_path / "evil"
        (evil / "export").mkdir(parents=True)
        state_name = ".discord-cache-export-approved.json"
        (evil / "export" / state_name).write_bytes((out / state_name).read_bytes())
        held = self.temp_path / "held"
        swapped = self.temp_path / "swapped"
        namespace = runpy.run_path(str(self.script_path))
        scope = namespace["main"].__globals__
        original = scope["approval_digest"]
        calls = []
        def hooked(*args, **kwargs):
            calls.append(kwargs)
            if len(calls) == 2:
                # Restore the approved path before rechecking; fds still refer to evil.
                parent.rename(swapped)
                held.rename(parent)
            digest = original(*args, **kwargs)
            if len(calls) == 1:
                parent.rename(held)
                evil.rename(parent)
            return digest
        scope["approval_digest"] = hooked
        stderr = io.StringIO()
        with contextlib.redirect_stderr(stderr), contextlib.redirect_stdout(io.StringIO()):
            code = namespace["main"](["export", "--cache-dir", str(cache), "--out", str(out)])
        self.assertEqual(code, 1, stderr.getvalue())
        self.assertEqual(len(calls), 2)
        for destination in (parent / "export", swapped / "export"):
            self.assertEqual([p.name for p in destination.iterdir()], [state_name])

    def test_reopened_approval_state_rejects_raced_nonregular_or_invalid_payload(self) -> None:
        cache = self.make_cache()
        for replacement in ("json-list", "fifo", "oversized"):
            with self.subTest(replacement=replacement):
                out = self.temp_path / replacement
                preview = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out),
                                       "--dry-run")
                self.assertEqual(preview.returncode, 0, preview.stderr)
                namespace = runpy.run_path(str(self.script_path))
                scope = namespace["main"].__globals__
                original = scope["approval_digest"]
                state = out / ".discord-cache-export-approved.json"
                changed = []

                def hooked(*args, **kwargs):
                    digest = original(*args, **kwargs)
                    if "output_metadata" in kwargs:
                        state.rename(out / "preserved-state.json")
                        if replacement == "fifo":
                            os.mkfifo(state)
                        elif replacement == "json-list":
                            state.write_text("[]", encoding="utf-8")
                        else:
                            state.write_text(" " * (scope["MAX_APPROVAL_STATE_BYTES"] + 1),
                                             encoding="utf-8")
                        changed.append(True)
                    return digest

                scope["approval_digest"] = hooked
                stderr = io.StringIO()
                with contextlib.redirect_stderr(stderr), contextlib.redirect_stdout(io.StringIO()):
                    code = namespace["main"](["export", "--cache-dir", str(cache), "--out", str(out)])
                self.assertEqual(changed, [True])
                self.assertEqual(code, 1)
                self.assertIn({"fifo": "regular", "oversized": "large", "json-list": "changed since"}[replacement],
                              stderr.getvalue())
                self.assertFalse((out / "messages.json").exists())
                self.assertFalse((out / "transcripts").exists())

    def test_symlinked_temp_prefix_resolves_once_and_later_symlink_is_refused(self) -> None:
        cache = self.make_cache()
        real = self.temp_path / "real"
        real.mkdir()
        alias = self.temp_path / "alias"
        alias.symlink_to(real, target_is_directory=True)
        out = alias / "export"
        preview = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out),
                               "--dry-run", "--json")
        self.assertEqual(preview.returncode, 0, preview.stderr)
        self.assertEqual(json.loads(preview.stdout)["resolved_destination"], str(real / "export"))
        live = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out))
        self.assertEqual(live.returncode, 0, live.stderr)
        second = real / "second"
        second.mkdir(mode=0o755)
        self.assertEqual(self.run_cli("export", "--cache-dir", str(cache), "--out", str(second),
                                     "--dry-run").returncode, 0)
        self.assertEqual(stat.S_IMODE(second.stat().st_mode), 0o700)
        preserved = real / "preserved"
        second.rename(preserved)
        second.symlink_to(preserved, target_is_directory=True)
        refused = self.run_cli("export", "--cache-dir", str(cache), "--out", str(second))
        self.assertEqual(refused.returncode, 1)
        self.assertEqual([p.name for p in preserved.iterdir()], [".discord-cache-export-approved.json"])

    def test_directory_walk_reports_symlinks_and_non_directories_clearly(self) -> None:
        namespace = runpy.run_path(str(self.script_path))
        walk = namespace["open_directory_components"]
        real_open = os.open
        for error_number in (errno.ENOTDIR, errno.ELOOP):
            def blocked(path, flags, **kwargs):
                if "dir_fd" in kwargs:
                    raise OSError(error_number, "synthetic path failure")
                return real_open(path, flags, **kwargs)
            with mock.patch.object(os, "open", side_effect=blocked):
                with self.assertRaisesRegex(namespace["CliError"],
                                            "component is a symlink or not a directory"):
                    walk(self.temp_path)

    def test_missing_output_parent_is_never_created(self) -> None:
        cache = self.make_cache()
        for dry in ([], ["--dry-run"]):
            out = self.temp_path / "missing" / "export"
            result = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out), *dry)
            self.assertEqual(result.returncode, 1)
            self.assertIn("parent must already exist", result.stderr)
            self.assertFalse(out.parent.exists())

    def test_live_export_refuses_missing_secure_platform_features(self) -> None:
        cache = self.make_cache()
        out = self.temp_path / "unsupported"
        preview = self.run_cli("export", "--cache-dir", str(cache), "--out", str(out), "--dry-run")
        self.assertEqual(preview.returncode, 0, preview.stderr)
        namespace = runpy.run_path(str(self.script_path))
        real_os = namespace["main"].__globals__["os"]
        for feature in ("O_NOFOLLOW", "O_DIRECTORY", "supports_dir_fd"):
            class UnsupportedOs:
                def __getattr__(self, name):
                    if name == feature:
                        if name == "supports_dir_fd":
                            return set()
                        raise AttributeError(name)
                    return getattr(real_os, name)
            namespace["main"].__globals__["os"] = UnsupportedOs()
            stderr = io.StringIO()
            with contextlib.redirect_stderr(stderr):
                code = namespace["main"](["export", "--cache-dir", str(cache), "--out", str(out)])
            self.assertEqual(code, 1)
            self.assertIn("live export requires", stderr.getvalue())
            self.assertEqual([p.name for p in out.iterdir()], [".discord-cache-export-approved.json"])

    def test_markdown_signatures_are_exact_last_lines(self) -> None:
        for path in self.package_root.rglob("*.md"):
            if ".git" not in path.parts:
                self.assertEqual(path.read_text(encoding="utf-8").splitlines()[-1], SIGNATURE)

    def test_expected_shipped_files_equal_package_spec(self) -> None:
        expected = {
            "SKILL.md",
            "README.md",
            "SETUP.md",
            "LICENSE.txt",
            "references/cache-format.md",
            "agents/discord-cache-agent.md",
            "scripts/discord_cache_export.py",
            "tests/test_discord_cache_export.py",
            "tests/fixtures/synthetic-messages.json",
            "source-package.json",
        }
        specification_path = self.package_root / "packaging" / "package-spec.json"
        if specification_path.is_file():
            specification = json.loads(specification_path.read_text(encoding="utf-8"))
            declared = {item["path"] for item in specification["files"]}
            self.assertEqual(declared, expected)
            for item in specification["files"]:
                if item["path"].endswith(".py"):
                    self.assertTrue(item["executable"], item["path"])
        else:
            declared = {
                str(path.relative_to(self.package_root))
                for path in self.package_root.rglob("*")
                if path.is_file()
                and path.name != "manifest.json"
                and "__pycache__" not in path.parts
            }
            self.assertEqual(declared, expected)
        for relative_path in expected:
            self.assertTrue((self.package_root / relative_path).is_file(), relative_path)


if __name__ == "__main__":
    unittest.main()

License: MIT (read it).

Back to the listing