tests/test_chat_export_to_transcript.py — Chat Export to Transcript

# By Zero Width Lab (zerowidthlab.com)
from __future__ import annotations

import json
import importlib.util
import os
from unittest.mock import patch
import subprocess
import sys
import tempfile
import unittest
import zipfile
from pathlib import Path

__author__ = "Zero Width Lab"
tempfile.tempdir = str(Path(tempfile.gettempdir()).resolve())
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "chat_export_to_transcript.py"
FIXTURES = ROOT / "tests" / "fixtures"


class ConverterTests(unittest.TestCase):
    def cli(self, *args: str) -> subprocess.CompletedProcess[str]:
        return subprocess.run([sys.executable, str(SCRIPT), *map(str, args)],
                              capture_output=True, text=True, check=False)

    def dry_run_and_live(self, source: Path, expected_platform: str, extra: list[str] | None = None) -> tuple[list[dict], str]:
        with tempfile.TemporaryDirectory() as temp:
            out = Path(temp) / "out"
            args = ["convert", str(source), "--out", str(out), *(extra or [])]
            preview = self.cli(*args, "--dry-run")
            self.assertEqual(preview.returncode, 0, preview.stderr)
            self.assertFalse(out.exists())
            digest = preview.stdout.split("approval digest: ")[1].splitlines()[0]
            denied = self.cli(*args)
            self.assertEqual(denied.returncode, 2)
            self.assertFalse(out.exists())
            wrong = self.cli(*args, "--approve", "0" * 64)
            self.assertEqual(wrong.returncode, 2)
            self.assertFalse(out.exists())
            live = self.cli(*args, "--approve", digest)
            self.assertEqual(live.returncode, 0, live.stderr)
            records = [json.loads(line) for line in (out / "messages.jsonl").read_text().splitlines()]
            self.assertTrue(records)
            self.assertEqual({record["platform"] for record in records}, {expected_platform})
            transcript = next((out / "transcripts").glob("*.txt")).read_text()
            self.assertEqual(len(list((out / "transcripts").glob("*.txt"))), 1)
            self.assertEqual(self.cli(*args, "--approve", digest).returncode, 2)
            return records, transcript

    def test_discord_header_and_duplicate_ids(self):
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            for folder in ("first", "second"):
                channel = root / "messages" / folder
                channel.mkdir(parents=True)
                (channel / "channel.json").write_text(json.dumps({"id": "123", "name": "Readable room"}))
                (channel / "messages.json").write_text(json.dumps([{"ID": "1", "Timestamp": "20240101T120000.1Z", "Contents": "hello"}]))
                if folder == "first":
                    rows, skipped = module.parse_discord(root)
                    self.assertEqual(skipped, [])
                    files = module.render_files(rows)
                    self.assertIn("transcripts/123-a665a459.txt", files)
                    self.assertTrue(files["transcripts/123-a665a459.txt"].decode().startswith("Discord channel: Readable room (ID: 123)\n"))
                    self.assertEqual(rows[0]["channel_name"], "Readable room")
            with self.assertRaisesRegex(module.CliError, "duplicate Discord channel ID 123: folders first and second"):
                module.parse_discord(root)

    def test_timestamp_forms_are_interpreter_independent(self):
        module = self.module()
        for fraction in ("1", "12", "123", "1234", "12345", "123456", "1234567"):
            for stamp in ("2024-01-02T03:04:05." + fraction + "Z", "20240102T030405," + fraction + "+0000"):
                self.assertEqual(module.valid_timestamp(stamp), "2024-01-02T03:04:05+00:00")
        self.assertEqual(module.valid_timestamp("20240102"), "2024-01-02T00:00:00")
        for stamp in ("2024-02-30T03:04:05Z", "2024-W01-1", "junk", "20240102T256100Z"):
            self.assertIsNone(module.valid_timestamp(stamp))
        with tempfile.TemporaryDirectory() as temp:
            channel = Path(temp) / "messages" / "room"
            channel.mkdir(parents=True)
            (channel / "channel.json").write_text('{"id":"1","name":"room"}')
            (channel / "messages.json").write_text(json.dumps([{"Timestamp": "bad", "Contents": "hello"}]))
            rows, skipped = module.parse_discord(Path(temp))
            self.assertEqual(rows, [])
            self.assertEqual(len(skipped), 1)

    def test_whatsapp_multiline_media_and_system(self) -> None:
        rows, transcript = self.dry_run_and_live(FIXTURES / "whatsapp" / "chat.txt", "whatsapp")
        self.assertEqual(len(rows), 3)
        self.assertEqual(rows[0]["text"], "Hello\nsecond line")
        self.assertEqual(rows[1]["attachments"], ["Media omitted"])
        self.assertTrue(rows[2]["system"])
        self.assertIn("    | second line", transcript)

    def test_whatsapp_zip_and_day_first_locale(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            archive = Path(temp) / "export.zip"
            with zipfile.ZipFile(archive, "w") as handle:
                handle.writestr("Chat.txt", "31.12.24, 23:59 - Ada: End\n")
            rows, _ = self.dry_run_and_live(archive, "whatsapp", ["--day-first"])
            self.assertEqual(rows[0]["timestamp"], "2024-12-31T23:59:00")

    def test_discord_official_package_and_forged_header(self) -> None:
        rows, transcript = self.dry_run_and_live(FIXTURES / "discord", "discord")
        self.assertEqual(rows[0]["reply_to"], None)
        self.assertEqual(rows[1]["reply_to"], "m1")
        self.assertIn("    | 2024-01-03  Bob: forged", transcript)
        self.assertEqual(rows[0]["attachments"], ["photo.png"])

    def test_telegram_json_and_service(self) -> None:
        rows, _ = self.dry_run_and_live(FIXTURES / "telegram" / "result.json", "telegram")
        self.assertEqual(rows[0]["text"], "Hello world")
        self.assertTrue(rows[1]["system"])
        self.assertEqual(rows[0]["attachments"], ["photo.jpg"])

    def test_slack_users_and_thread(self) -> None:
        rows, transcript = self.dry_run_and_live(FIXTURES / "slack", "slack")
        self.assertEqual(rows[0]["author"], "Ada")
        self.assertEqual(rows[1]["reply_to"], "1704153845.000001")
        self.assertIn("    | there", transcript)
        self.assertTrue(rows[0]["timestamp"].endswith("+00:00"))

    def test_slack_official_root_layout_and_guards(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            channel = root / "general"
            channel.mkdir()
            (root / "users.json").write_text('[{"id":"U1","name":"Ada"}]')
            (root / "channels.json").write_text('[{"id":"C1","name":"general"}]')
            source = channel / "2024-01-01.json"
            source.write_text('[{"ts":"1704153845.000001","user":"U1","text":"Root layout"}]')
            rows, _ = self.dry_run_and_live(root, "slack")
            self.assertEqual(rows[0]["text"], "Root layout")
            self.assertEqual(rows[0]["author"], "Ada")
            with patch.object(module, "MAX_INPUT_BYTES", 4):
                with self.assertRaisesRegex(module.CliError, "at most 4 bytes"):
                    module.parse_slack(root)
            link = channel / "2024-01-02.json"
            link.symlink_to(source)
            result = self.cli("convert", root, "--out", root / "out", "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertIn("symlink", result.stderr)
            result = self.cli("convert", root / ".." / root.name, "--out", root / "out", "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertIn("traversal", result.stderr)

    def test_discord_same_named_channels_remain_separate_by_id(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            for channel_id in ("1001", "1002"):
                channel = root / "messages" / ("c" + channel_id)
                channel.mkdir(parents=True)
                (channel / "channel.json").write_text(json.dumps({"id": channel_id, "name": "general"}))
                (channel / "messages.json").write_text(json.dumps([
                    {"id": channel_id, "timestamp": "2024-01-02T03:04:05Z", "content": "Only " + channel_id}]))
            out = root / "out"
            preview = self.cli("convert", root, "--out", out, "--dry-run")
            self.assertEqual(preview.returncode, 0, preview.stderr)
            digest = preview.stdout.split("approval digest: ")[1].splitlines()[0]
            live = self.cli("convert", root, "--out", out, "--approve", digest)
            self.assertEqual(live.returncode, 0, live.stderr)
            rows = [json.loads(line) for line in (out / "messages.jsonl").read_text().splitlines()]
            self.assertEqual({row["conversation"] for row in rows}, {"1001", "1002"})
            files = list((out / "transcripts").glob("*.txt"))
            self.assertEqual(len(files), 2)
            for channel_id in ("1001", "1002"):
                text = next(file for file in files if file.name.startswith(channel_id + "-")).read_text()
                self.assertIn("Only " + channel_id, text)
                self.assertNotIn("Only " + ("1002" if channel_id == "1001" else "1001"), text)

    def test_invalid_whatsapp_header_is_counted_not_merged(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "chat.txt"
            source.write_text("1/2/24, 9:04 AM - Ada: Good\n99/99/24, 9:05 AM - Bea: Invalid\ninvalid continuation\n1/2/24, 9:06 AM - Ada: Last\n")
            rows, skipped = self.module().parse_whatsapp(source.read_bytes(), "chat", False)
            self.assertEqual([row["text"] for row in rows], ["Good", "Last"])
            self.assertEqual(len(skipped), 2)
            self.assertIn("line 2: invalid date or time", skipped[0])
            result = self.cli("convert", source, "--out", Path(temp) / "out", "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertIn("status: incomplete", result.stdout)
            self.assertIn("skipped: 2", result.stdout)

    def test_test_filesystem_disclosure_is_exact(self) -> None:
        readme = (ROOT / "README.md").read_text()
        evidence = " ".join(json.loads((ROOT / "source-package.json").read_text())["tests"]["evidence"])
        for text in (readme, evidence):
            self.assertIn("synthetic", text)
            self.assertIn("temporary folder", text)
            self.assertIn("output files", text)
            self.assertIn("remove that folder", text)
        listing = ROOT / "packaging" / "listing-values.json"
        if listing.exists():
            permissions = json.loads(listing.read_text())["permissions"]
            self.assertIn("synthetic fixtures and output files in a temporary folder", permissions)
            self.assertNotIn("tests read package files only", permissions)

    def test_partial_records_fail_closed_then_allow(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "chat.txt"
            source.write_text("orphan line\n1/2/24, 9:04 AM - Ada: Good\n")
            out = Path(temp) / "out"
            args = ["convert", str(source), "--out", str(out)]
            preview = self.cli(*args, "--dry-run")
            self.assertEqual(preview.returncode, 2)
            self.assertIn("status: incomplete", preview.stdout)
            self.assertIn("skipped: 1", preview.stdout)
            digest = preview.stdout.split("approval digest: ")[1].splitlines()[0]
            blocked = self.cli(*args, "--approve", digest)
            self.assertEqual(blocked.returncode, 2)
            self.assertFalse(out.exists())
            allowed = self.cli(*args, "--approve", digest, "--allow-partial")
            self.assertEqual(allowed.returncode, 0, allowed.stderr)
            self.assertTrue((out / "messages.jsonl").exists())

    def test_untrusted_controls_are_escaped(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "chat.txt"
            source.write_text("1/2/24, 9:04 AM - A\tB: hi\u2028bad\x1b[31m\n")
            rows, transcript = self.dry_run_and_live(source, "whatsapp")
            self.assertEqual(rows[0]["author"], "A\tB")
            self.assertIn("A\\u0009B", transcript)
            self.assertIn("\\u001b", transcript)
            self.assertNotIn("\x1b", transcript)

    def test_html_and_symlink_refused(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "result.html"
            source.write_text("<html></html>")
            out = Path(temp) / "out"
            failed = self.cli("convert", source, "--out", out, "--format", "telegram", "--dry-run")
            self.assertEqual(failed.returncode, 2)
            self.assertIn("HTML is unsupported", failed.stderr)
            link = Path(temp) / "link.txt"
            link.symlink_to(FIXTURES / "whatsapp" / "chat.txt")
            self.assertEqual(self.cli("convert", link, "--out", out, "--dry-run").returncode, 2)

    def test_help_has_signature(self) -> None:
        result = self.cli("--help")
        self.assertEqual(result.returncode, 0)
        self.assertTrue(result.stdout.rstrip().endswith("By Zero Width Lab (zerowidthlab.com)"))

    def test_discord_csv_and_missing_metadata_are_partial(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            channel = Path(temp) / "messages" / "test-channel"
            channel.mkdir(parents=True)
            (channel / "messages.csv").write_text('ID,Timestamp,Author,Contents\nm1,2024-01-02T03:04:05Z,Ada,Hello\n')
            out = Path(temp) / "out"
            result = self.cli("convert", temp, "--out", out, "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertIn("channel.json missing", result.stdout)

    def test_header_shaped_whatsapp_body_starts_record(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "chat.txt"
            source.write_text("1/2/24, 9:04 AM - Ada: First\n1/2/24, 9:05 AM - Bea: Second\n")
            rows, _ = self.dry_run_and_live(source, "whatsapp")
            self.assertEqual([row["text"] for row in rows], ["First", "Second"])

    def test_discord_missing_author_is_unknown(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            channel = Path(temp) / "messages" / "sample"
            channel.mkdir(parents=True)
            (channel / "channel.json").write_text('{"name":"Sample"}')
            (channel / "messages.json").write_text('[{"id":"m1","timestamp":"2024-01-02T03:04:05Z","content":"Hello"}]')
            rows, transcript = self.dry_run_and_live(Path(temp), "discord")
            self.assertEqual(rows[0]["author"], "Unknown")
            self.assertIn("Unknown: Hello", transcript)

    def test_malformed_telegram_record_is_partial(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "result.json"
            source.write_text('{"name":"Fake","messages":[{"id":1,"date":"2024-01-02T03:04:05","text":"Good"},{"id":2,"date":"bad","text":"Bad"}]}')
            result = self.cli("convert", source, "--out", Path(temp) / "out", "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertIn("record 2", result.stdout)

    def test_changed_source_invalidates_digest(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "chat.txt"
            source.write_text("1/2/24, 9:04 AM - Ada: Original\n")
            out = Path(temp) / "out"
            args = ["convert", str(source), "--out", str(out)]
            preview = self.cli(*args, "--dry-run")
            digest = preview.stdout.split("approval digest: ")[1].splitlines()[0]
            source.write_text("1/2/24, 9:04 AM - Ada: Changed\n")
            result = self.cli(*args, "--approve", digest)
            self.assertEqual(result.returncode, 2)
            self.assertFalse(out.exists())

    def test_all_malformed_records_still_report_incomplete(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "result.json"
            source.write_text('{"messages":[{"date":"invalid","text":"Bad"}]}')
            out = Path(temp) / "out"
            result = self.cli("convert", source, "--out", out, "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertIn("status: incomplete", result.stdout)
            self.assertIn("record 1", result.stdout)
            self.assertFalse(out.exists())

    def module(self):
        spec = importlib.util.spec_from_file_location("converter", SCRIPT)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module

    def test_swapped_output_parent_refused_before_write(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            parent = root / "parent"
            parent.mkdir()
            outside = root / "outside"
            outside.mkdir()
            out = parent / "out"
            files = {"messages.jsonl": b"test"}
            approved = module.output_plan(out)
            digest = module.plan_digest(files, out)
            parent.rename(root / "old-parent")
            parent.symlink_to(outside, target_is_directory=True)
            with self.assertRaises((module.CliError, OSError)):
                module.write_files(out, files, approved)
            with self.assertRaisesRegex(module.CliError, "physical path"):
                module.plan_digest(files, out)
            self.assertFalse((outside / "out").exists())
            self.assertEqual(len(digest), 64)

    def test_replaced_existing_output_invalidates_digest(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            out = Path(temp) / "out"
            out.mkdir()
            files = {"messages.jsonl": b"test"}
            approved = module.output_plan(out)
            digest = module.plan_digest(files, out)
            out.rename(Path(temp) / "old")
            out.mkdir()
            self.assertNotEqual(digest, module.plan_digest(files, out))
            with self.assertRaises(module.CliError):
                module.write_files(out, files, approved)
            self.assertFalse(list(out.iterdir()))

    def test_missing_output_ancestor_is_not_created(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            missing = Path(temp) / "missing"
            result = self.cli("convert", FIXTURES / "whatsapp/chat.txt", "--out", missing / "out", "--dry-run")
            self.assertEqual(result.returncode, 2)
            self.assertFalse(missing.exists())

    def test_zip_traversal_rejected_before_member_read(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "export.zip"
            for name in ("../secret.txt", "/absolute.txt", "%2e%2e/secret.txt", "..\\secret.txt"):
                with zipfile.ZipFile(source, "w") as archive:
                    archive.writestr("chat.txt", "1/2/24, 9:04 AM - Ada: Hello")
                    archive.writestr(name, "secret")
                with patch.object(zipfile.ZipFile, "open", side_effect=AssertionError("member read")):
                    with self.assertRaises(module.CliError):
                        module.parse_whatsapp_path(source, False)

    def test_export_ancestor_symlink_refuses_read_and_listing(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            outside = root / "outside"
            outside.mkdir()
            (outside / "secret.json").write_text("secret")
            selected = root / "selected"
            selected.mkdir()
            (selected / "channel").symlink_to(outside, target_is_directory=True)
            with self.assertRaises(module.CliError):
                module.read_file(selected / "channel/secret.json")
            with self.assertRaisesRegex(module.CliError, "physical path"):
                module.directory_children(selected / "channel")
            with self.assertRaises(module.CliError):
                module.directory_children(selected)

    def test_local_percent_and_backslash_paths_are_literal(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            source = root / "Photos 100%\\literal.txt"
            source.write_text("1/2/24, 9:04 AM - Ada: Hello\n")
            destination = root / "out%41\\literal"
            preview = self.cli("convert", source, "--out", destination, "--dry-run")
            self.assertEqual(preview.returncode, 0, preview.stderr)
            digest = preview.stdout.split("approval digest: ")[1].splitlines()[0]
            live = self.cli("convert", source, "--out", destination, "--approve", digest)
            self.assertEqual(live.returncode, 0, live.stderr)
            self.assertTrue((destination / "messages.jsonl").is_file())
            module.validate_path(Path("%2e%2e/literal"))
            with self.assertRaises(module.CliError):
                module.validate_path(Path("bad\0name"))

    def test_symlink_component_explains_physical_path(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            real = root / "physical"
            real.mkdir()
            alias = root / "alias"
            alias.symlink_to(real, target_is_directory=True)
            with self.assertRaisesRegex(module.CliError, "a path component is a symlink; use the physical path"):
                module.open_directory(alias)

    def test_symlink_anywhere_inside_export_is_refused(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            for relative in ("unrelated", "messages/unrelated", "messages/channel/unrelated", "misc/deep/unrelated"):
                export = root / "export"
                export.mkdir(exist_ok=True)
                channel = export / "messages/channel"
                channel.mkdir(parents=True, exist_ok=True)
                (channel / "channel.json").write_text('{"name":"Synthetic"}')
                (channel / "messages.json").write_text('[{"ID":"1","Timestamp":"2024-01-01T00:00:00Z","Contents":"Hello","Author":"Ada"}]')
                link = export / relative
                link.parent.mkdir(parents=True, exist_ok=True)
                link.symlink_to(root / "absent")
                result = self.cli("convert", export, "--format", "discord", "--out", root / "out", "--dry-run")
                self.assertEqual(result.returncode, 2, result.stderr)
                self.assertIn("symlink in export", result.stderr)
                link.unlink()
                self.assertFalse((root / "out").exists())

    def test_platform_guard_precedes_io(self) -> None:
        module = self.module()
        with patch.object(module.sys, "platform", "win32"), patch.object(module.os, "open", side_effect=AssertionError("I/O")):
            with self.assertRaisesRegex(module.CliError, "Windows is unsupported"):
                module.run(["convert", "absent.txt", "--out", "out", "--dry-run"])

    def test_oversize_and_fifo_inputs_rejected(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            source = Path(temp) / "chat.txt"
            source.write_bytes(b"12345")
            with patch.object(module, "MAX_INPUT_BYTES", 4):
                with self.assertRaisesRegex(module.CliError, "at most 4 bytes"):
                    module.read_file(source)
            fifo = Path(temp) / "pipe"
            os.mkfifo(fifo)
            with self.assertRaisesRegex(module.CliError, "regular file"):
                module.read_file(fifo)

    def test_malformed_nested_fields_and_json_are_explicit_errors(self) -> None:
        module = self.module()
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            (root / "users.json").write_text('[{"id":"U1","profile":[]}]')
            with self.assertRaisesRegex(module.CliError, "profile must be an object"):
                module.parse_slack(root)
            source = root / "result.json"
            source.write_text('{"messages":[{"date":"2024-01-02","text":"hello","photo":{}}]}')
            with self.assertRaisesRegex(module.CliError, "photo must be text"):
                module.parse_telegram(source)
            source.write_text("{")
            with self.assertRaisesRegex(module.CliError, "invalid JSON"):
                module.parse_telegram(source)


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

License: MIT (read it).

Back to the listing