tests/test_whatsapp_read.py — Desktop Chat Reader

# By Zero Width Lab (zerowidthlab.com)

from __future__ import annotations

import json
import io
import importlib.util
import os
import re
import sqlite3
import signal
import shutil
import time
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
from pathlib import Path

__author__ = "Zero Width Lab"


class WhatsAppReadCliTests(unittest.TestCase):
    def setUp(self) -> None:
        self.package_root = Path(__file__).resolve().parent.parent
        self.script_path = self.package_root / "scripts" / "whatsapp_read.py"
        self.temp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp_dir.cleanup)
        self.temp_path = Path(self.temp_dir.name)
        self.db_path = self.temp_path / "ChatStorage.sqlite"
        self._create_fixture_database(self.db_path)

    def _create_fixture_database(
        self, db_path: Path, omitted_tables: set[str] | None = None
    ) -> None:
        omitted = omitted_tables or set()
        conn = sqlite3.connect(db_path)
        try:
            schemas = {
                "ZWACHATSESSION": """
                CREATE TABLE ZWACHATSESSION (
                    Z_PK INTEGER PRIMARY KEY,
                    ZCONTACTJID TEXT,
                    ZPARTNERNAME TEXT,
                    ZSESSIONTYPE INTEGER
                )
                """,
                "ZWAMESSAGE": """
                CREATE TABLE ZWAMESSAGE (
                    Z_PK INTEGER PRIMARY KEY,
                    ZCHATSESSION INTEGER,
                    ZMESSAGEDATE REAL,
                    ZISFROMME INTEGER,
                    ZMESSAGETYPE INTEGER,
                    ZTEXT TEXT,
                    ZGROUPMEMBER INTEGER
                )
                """,
                "ZWAGROUPMEMBER": """
                CREATE TABLE ZWAGROUPMEMBER (
                    Z_PK INTEGER PRIMARY KEY,
                    ZMEMBERJID TEXT
                )
                """,
                "ZWAPROFILEPUSHNAME": """
                CREATE TABLE ZWAPROFILEPUSHNAME (
                    ZJID TEXT,
                    ZPUSHNAME TEXT
                )
                """,
                "ZWAMEDIAITEM": """
                CREATE TABLE ZWAMEDIAITEM (
                    ZMESSAGE INTEGER,
                    ZTITLE TEXT,
                    ZMEDIALOCALPATH TEXT
                )
                """,
            }
            for table_name, schema in schemas.items():
                if table_name not in omitted:
                    conn.execute(schema)

            if "ZWACHATSESSION" not in omitted:
                conn.executemany(
                    """
                    INSERT INTO ZWACHATSESSION
                        (Z_PK, ZCONTACTJID, ZPARTNERNAME, ZSESSIONTYPE)
                    VALUES (?, ?, ?, ?)
                    """,
                    [
                        (1, "15550000001@s.whatsapp.net", "Alice", 0),
                        (2, "120000000000000001@g.us", "Test Group", 0),
                    ],
                )
            if "ZWAGROUPMEMBER" not in omitted:
                conn.executemany(
                    "INSERT INTO ZWAGROUPMEMBER (Z_PK, ZMEMBERJID) VALUES (?, ?)",
                    [
                        (10, "15550000002@s.whatsapp.net"),
                        (11, "15550000003@s.whatsapp.net"),
                    ],
                )
            if "ZWAPROFILEPUSHNAME" not in omitted:
                conn.execute(
                    "INSERT INTO ZWAPROFILEPUSHNAME (ZJID, ZPUSHNAME) VALUES (?, ?)",
                    ("15550000002@s.whatsapp.net", "Bob"),
                )
            if "ZWAMESSAGE" not in omitted:
                conn.executemany(
                    """
                    INSERT INTO ZWAMESSAGE
                        (Z_PK, ZCHATSESSION, ZMESSAGEDATE, ZISFROMME,
                         ZMESSAGETYPE, ZTEXT, ZGROUPMEMBER)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    [
                        (100, 1, 800000001, 1, 0, "My outbound note", None),
                        (101, 1, 800000002, 0, 0, "Alice inbound reply", None),
                        (102, 2, 800000003, 0, 0, "Lantern status update", 10),
                        (103, 2, 800000004, 0, 0, "Fallback sender update", 11),
                    ],
                )
            conn.commit()
        finally:
            conn.close()

    def _database_snapshots(self) -> dict[Path, bytes]:
        return {
            path: path.read_bytes()
            for path in self.temp_path.rglob("*")
            if path.is_file()
        }

    def run_raw_cli(
        self,
        *arguments: str,
        env_overrides: dict[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        before = self._database_snapshots()
        env = os.environ.copy()
        env["HOME"] = str(self.temp_path)
        env.pop("WHATSAPP_DB", None)
        if env_overrides:
            env.update(env_overrides)

        result = subprocess.run(
            [sys.executable, str(self.script_path), *arguments],
            cwd=self.package_root,
            capture_output=True,
            text=True,
            check=False,
            env=env,
        )

        after = self._database_snapshots()
        self.assertEqual(set(after), set(before), "CLI created or removed a fixture file")
        for path, contents in before.items():
            self.assertEqual(after[path], contents, f"CLI modified {path}")
        return result

    def run_cli(
        self, *arguments: str, db_path: Path | None = None
    ) -> subprocess.CompletedProcess[str]:
        selected_db = db_path if db_path is not None else self.db_path
        return self.run_raw_cli(
            "--db",
            str(selected_db),
            "--json",
            *arguments,
        )

    def parse_successful_json(self, result: subprocess.CompletedProcess[str]):
        self.assertEqual(result.returncode, 0, result.stderr)
        return json.loads(result.stdout)

    def load_cli_module(self):
        spec = importlib.util.spec_from_file_location("whatsapp_read", self.script_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module

    def test_unsupported_platform_refused_before_database_io(self):
        module = self.load_cli_module()
        with patch.object(module.sys, "platform", "win32"), patch.object(module.Path, "stat") as stat_call:
            with self.assertRaises(SystemExit):
                module.connect(self.db_path)
            stat_call.assert_not_called()

    def test_input_limits_reject_large_file_value_and_result_limit(self):
        module = self.load_cli_module()
        with patch.object(module, "MAX_DATABASE_BYTES", 1), patch.object(module.sqlite3, "connect") as connect_call:
            with self.assertRaises(SystemExit):
                module.connect(self.db_path)
            connect_call.assert_not_called()
        result = self.run_cli("chats", "--limit", "1001")
        self.assertEqual(result.returncode, 2)
        conn = sqlite3.connect(self.db_path)
        conn.execute("UPDATE ZWAMESSAGE SET ZTEXT = ? WHERE Z_PK = 100", ("a" * (1024 * 1024 + 1),))
        conn.commit()
        conn.close()
        result = self.run_cli("read", "Alice")
        self.assertEqual(result.returncode, 1)
        self.assertNotIn("Traceback", result.stderr)

    def test_malformed_file_and_returned_values_report_clear_errors(self):
        bad = self.temp_path / "malformed.sqlite"
        bad.write_bytes(b"not a SQLite database")
        result = self.run_cli("chats", db_path=bad)
        self.assertEqual(result.returncode, 1)
        self.assertIn("file is not a database", result.stderr)
        for column, value in [("ZMESSAGEDATE", "bad-date"), ("ZTEXT", b"bad-blob"), ("ZISFROMME", "wrong")]:
            with self.subTest(column=column):
                conn = sqlite3.connect(self.db_path)
                original = conn.execute(f"SELECT {column} FROM ZWAMESSAGE WHERE Z_PK = 100").fetchone()[0]
                conn.execute(f"UPDATE ZWAMESSAGE SET {column} = ? WHERE Z_PK = 100", (value,))
                conn.commit()
                conn.close()
                result = self.run_cli("read", "Alice")
                self.assertEqual(result.returncode, 1)
                self.assertIn("Malformed database", result.stderr)
                self.assertNotIn("Traceback", result.stderr)
                conn = sqlite3.connect(self.db_path)
                conn.execute(f"UPDATE ZWAMESSAGE SET {column} = ? WHERE Z_PK = 100", (original,))
                conn.commit()
                conn.close()
        module = self.load_cli_module()
        for value in ("wrong", float("inf"), 1e300):
            with self.assertRaises(SystemExit):
                module.to_iso(value)

    def test_query_row_and_work_budgets(self):
        module = self.load_cli_module()
        conn = sqlite3.connect(":memory:")
        self.addCleanup(conn.close)
        with self.assertRaises(SystemExit):
            module.query(conn, "WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n WHERE x<1001) SELECT x FROM n")
        stderr = io.StringIO()
        with patch.object(module.time, "monotonic", side_effect=[0, 31]), patch.object(module.sys, "stderr", stderr):
            with self.assertRaises(SystemExit):
                module.query(conn, "WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n WHERE x<10000) SELECT sum(x) FROM n")
        self.assertIn("30-second wall-clock limit", stderr.getvalue())
        self.assertEqual(module.query(conn, "SELECT 1"), [(1,)])

    def test_large_history_chats_and_read_with_and_without_index(self):
        conn = sqlite3.connect(self.db_path)
        conn.executemany("INSERT INTO ZWAMESSAGE VALUES (?,1,?,0,0,'synthetic',NULL)",
                         ((i, 800000000 + i) for i in range(1000, 101000)))
        conn.commit()
        for indexed in (False, True):
            if indexed:
                conn.execute("CREATE INDEX message_chat ON ZWAMESSAGE(ZCHATSESSION)")
                conn.commit()
            self.assertEqual(self.parse_successful_json(self.run_cli("chats"))[0]["messages"], 100002)
            self.assertEqual(len(self.parse_successful_json(self.run_cli("read", "Alice"))), 50)
        conn.close()

    def test_live_wal_visible_and_snapshot_private_and_cleaned(self):
        writer = sqlite3.connect(self.db_path)
        self.addCleanup(writer.close)
        writer.execute("PRAGMA journal_mode=WAL")
        writer.execute("INSERT INTO ZWAPROFILEPUSHNAME VALUES ('synthetic', 'WAL entry')")
        writer.commit()
        before = self._database_snapshots()
        module = self.load_cli_module()
        connection = module.connect(self.db_path)
        directory = connection.snapshot_directory
        self.assertEqual(directory.stat().st_mode & 0o777, 0o700)
        self.assertTrue(all(p.stat().st_mode & 0o777 == 0o600 for p in directory.iterdir()))
        self.assertIn(('synthetic', 'WAL entry'), module.query(connection, "SELECT ZJID,ZPUSHNAME FROM ZWAPROFILEPUSHNAME"))
        connection.close()
        self.assertFalse(directory.exists())
        self.assertEqual(before, self._database_snapshots())
        self.assertTrue(any(x["name"] == "WAL entry" for x in self.parse_successful_json(self.run_cli("contacts"))))

    def test_changing_sources_retry_three_times_then_refuse_and_cleanup(self):
        module = self.load_cli_module()
        original_read = module.os.read
        original_temp = module.tempfile.mkdtemp
        directories = []
        def temp(**kwargs):
            directory = original_temp(**kwargs)
            directories.append(Path(directory))
            return directory
        def changed_read(fd, count):
            chunk = original_read(fd, count)
            if chunk:
                info = self.db_path.stat()
                os.utime(self.db_path, ns=(info.st_atime_ns, info.st_mtime_ns + 1))
            return chunk
        with patch.object(module.tempfile, "mkdtemp", side_effect=temp), patch.object(module.os, "read", side_effect=changed_read):
            with self.assertRaises(SystemExit):
                module.connect(self.db_path)
        self.assertEqual(len(directories), 3)
        self.assertTrue(all(not d.exists() for d in directories))

    def test_changed_source_retries_then_succeeds(self):
        module = self.load_cli_module()
        original_read = module.os.read
        changed = False
        def read(fd, count):
            nonlocal changed
            chunk = original_read(fd, count)
            if chunk and not changed:
                changed = True
                info = self.db_path.stat()
                os.utime(self.db_path, ns=(info.st_atime_ns, info.st_mtime_ns + 1))
            return chunk
        with patch.object(module.os, "read", side_effect=read), patch.object(module.tempfile, "mkdtemp", wraps=module.tempfile.mkdtemp) as create:
            connection = module.connect(self.db_path)
            self.assertEqual(create.call_count, 2)
            directory = connection.snapshot_directory
            connection.close()
            self.assertFalse(directory.exists())

    def test_combined_sidecar_cap_refused(self):
        module = self.load_cli_module()
        sidecar = Path(str(self.db_path) + "-wal")
        sidecar.write_bytes(b"synthetic")
        with patch.object(module, "MAX_DATABASE_BYTES", self.db_path.stat().st_size + 1):
            with self.assertRaises(SystemExit):
                module.connect(self.db_path)

    def test_snapshot_cleanup_on_error_and_keyboard_interrupt(self):
        module = self.load_cli_module()
        original_temp = module.tempfile.mkdtemp
        for error in (OSError("synthetic failure"), KeyboardInterrupt()):
            directories = []
            def temp(**kwargs):
                directory = original_temp(**kwargs)
                directories.append(Path(directory))
                return directory
            with patch.object(module.tempfile, "mkdtemp", side_effect=temp), patch.object(module.os, "read", side_effect=error):
                with self.assertRaises((SystemExit, KeyboardInterrupt)):
                    module.connect(self.db_path)
            self.assertTrue(directories)
            self.assertTrue(all(not d.exists() for d in directories))
        with patch.object(module, "cmd_chats", side_effect=KeyboardInterrupt()), patch.object(module.sys, "argv", ["reader", "--db", str(self.db_path), "chats"]), patch.object(module.tempfile, "mkdtemp", side_effect=temp):
            with self.assertRaises(KeyboardInterrupt):
                module.main()
        self.assertTrue(all(not d.exists() for d in directories))

    def test_missing_parent_has_missing_path_message(self):
        result = self.run_cli("chats", db_path=self.temp_path / "absent" / "db.sqlite")
        self.assertEqual(result.returncode, 1)
        self.assertIn("parent folder not found", result.stderr)
        self.assertNotIn("symlinked", result.stderr)

    def test_cleanup_when_connection_or_return_is_interrupted(self):
        module = self.load_cli_module()
        original_connect = module.sqlite3.connect
        original_cli_connect = module.connect
        previous = {n: signal.getsignal(n) for n in (signal.SIGTERM, signal.SIGHUP)}
        for stage in ("sqlite", "attribute", "return"):
            with self.subTest(stage=stage), tempfile.TemporaryDirectory() as temp:
                root = Path(temp)
                def interrupted_sqlite(*args, **kwargs):
                    connection = original_connect(*args, **kwargs)
                    connection.close()
                    raise KeyboardInterrupt()
                held_connections = []
                def interrupted_return(*args, **kwargs):
                    held_connections.append(original_cli_connect(*args, **kwargs))
                    raise KeyboardInterrupt()
                with patch.object(module.tempfile, "tempdir", str(root)):
                    with self.assertRaises(KeyboardInterrupt):
                        if stage == "sqlite":
                            with patch.object(module.sqlite3, "connect", side_effect=interrupted_sqlite):
                                module.connect(self.db_path)
                        elif stage == "attribute":
                            class InterruptedConnection(module.SnapshotConnection):
                                def __setattr__(self, name, value):
                                    if name == "snapshot_directory" and value is not None:
                                        raise KeyboardInterrupt()
                                    super().__setattr__(name, value)
                            with patch.object(module, "SnapshotConnection", InterruptedConnection):
                                module.connect(self.db_path)
                        else:
                            with patch.object(module, "connect", side_effect=interrupted_return), patch.object(module.sys, "argv", ["reader", "--db", str(self.db_path), "chats"]):
                                module.main()
                self.assertEqual(list(root.iterdir()), [])
                self.assertEqual(previous, {n: signal.getsignal(n) for n in previous})
                for connection in held_connections:
                    connection.close()

    def test_termination_signals_cleanup_real_subprocess(self):
        for number in (signal.SIGTERM, signal.SIGHUP):
            for stage in ("copy", "query"):
                with self.subTest(number=number, stage=stage), tempfile.TemporaryDirectory() as temp:
                    root = Path(temp)
                    ready = root / "ready"
                    runner = root / "runner.py"
                    runner.write_text(
                        "import importlib.util, pathlib, sys, tempfile, time\n"
                        "spec = importlib.util.spec_from_file_location('reader', sys.argv[1])\n"
                        "reader = importlib.util.module_from_spec(spec)\n"
                        "spec.loader.exec_module(reader)\n"
                        "tempfile.tempdir = sys.argv[3]\n"
                        "ready = pathlib.Path(sys.argv[4])\n"
                        "def wait(*args):\n"
                        "    ready.write_text('ready')\n"
                        "    while True: time.sleep(0.05)\n"
                        "if sys.argv[5] == 'copy': reader.os.read = wait\n"
                        "else: reader.cmd_chats = wait\n"
                        "sys.argv = ['reader', '--db', sys.argv[2], 'chats']\n"
                        "reader.main()\n"
                    )
                    process = subprocess.Popen([sys.executable, str(runner), str(self.script_path), str(self.db_path), str(root), str(ready), stage], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                    try:
                        deadline = time.monotonic() + 10
                        while not ready.exists() and process.poll() is None and time.monotonic() < deadline:
                            time.sleep(0.01)
                        self.assertTrue(ready.exists(), "subprocess did not reach snapshot stage")
                        self.assertTrue(list(root.glob("desktop-chat-reader-*")))
                        process.send_signal(number)
                        output, error = process.communicate(timeout=10)
                        self.assertEqual(process.returncode, 128 + number, error.decode())
                        self.assertEqual(list(root.glob("desktop-chat-reader-*")), [])
                    finally:
                        if process.poll() is None:
                            process.kill()
                        process.communicate(timeout=10)

    def test_symlinked_database_and_sidecars_are_refused(self):
        for suffix in ("", "-wal", "-shm"):
            with self.subTest(suffix=suffix):
                if suffix:
                    alias = Path(str(self.db_path) + suffix)
                    alias.symlink_to(self.db_path)
                    selected = self.db_path
                else:
                    alias = self.temp_path / "alias.sqlite"
                    alias.symlink_to(self.db_path)
                    selected = alias
                result = self.run_cli("contacts", db_path=selected)
                self.assertEqual(result.returncode, 1)
                self.assertIn("symlink", result.stderr)
                alias.unlink()

    def test_chats_lists_direct_and_group_chat(self) -> None:
        chats = self.parse_successful_json(self.run_cli("chats"))
        chats_by_jid = {chat["jid"]: chat for chat in chats}

        self.assertEqual(
            set(chats_by_jid),
            {"15550000001@s.whatsapp.net", "120000000000000001@g.us"},
        )
        self.assertFalse(chats_by_jid["15550000001@s.whatsapp.net"]["is_group"])
        self.assertTrue(chats_by_jid["120000000000000001@g.us"]["is_group"])

    def test_direct_chat_resolves_me_and_partner_name(self) -> None:
        messages = self.parse_successful_json(self.run_cli("read", "Alice"))

        self.assertEqual([message["sender"] for message in messages], ["ME", "Alice"])
        self.assertEqual(
            [message["text"] for message in messages],
            ["My outbound note", "Alice inbound reply"],
        )

    def test_group_chat_resolves_pushname_and_raw_jid_fallback(self) -> None:
        messages = self.parse_successful_json(
            self.run_cli("read", "120000000000000001@g.us")
        )
        senders_by_text = {message["text"]: message["sender"] for message in messages}

        self.assertEqual(senders_by_text["Lantern status update"], "Bob")
        fallback_sender = senders_by_text["Fallback sender update"]
        self.assertEqual(fallback_sender, "15550000003@s.whatsapp.net")
        self.assertNotEqual(fallback_sender, "Bob")
        self.assertTrue(fallback_sender)

    def test_search_finds_keyword_in_message_body(self) -> None:
        matches = self.parse_successful_json(self.run_cli("search", "Lantern"))

        self.assertEqual(len(matches), 1)
        self.assertEqual(matches[0]["text"], "Lantern status update")
        self.assertEqual(matches[0]["chat_jid"], "120000000000000001@g.us")
        self.assertEqual(matches[0]["sender"], "Bob")

    def test_nonexistent_database_reports_not_found(self) -> None:
        missing_path = Path(self.temp_dir.name) / "missing.sqlite"
        result = self.run_cli("chats", db_path=missing_path)

        self.assertNotEqual(result.returncode, 0)
        self.assertIn("not found", result.stderr.lower())

    def test_database_connection_is_explicitly_read_only(self) -> None:
        source = self.script_path.read_text(encoding="utf-8")

        self.assertIn('f"{copied.as_uri()}?mode=ro"', source)
        self.assertIn("uri=True", source)

    def test_global_options_work_before_and_after_subcommand_and_db_wins(self) -> None:
        missing_env_db = self.temp_path / "environment.sqlite"
        env = {"WHATSAPP_DB": str(missing_env_db)}
        before = self.parse_successful_json(
            self.run_raw_cli(
                "--json",
                "--db",
                str(self.db_path),
                "chats",
                env_overrides=env,
            )
        )
        after = self.parse_successful_json(
            self.run_raw_cli(
                "chats",
                "--json",
                "--db",
                str(self.db_path),
                env_overrides=env,
            )
        )
        self.assertEqual(before, after)

        result = self.run_raw_cli("chats", env_overrides={"WHATSAPP_DB": ""})
        expected_default = (
            self.temp_path
            / "Library"
            / "Group Containers"
            / "group.net.whatsapp.WhatsApp.shared"
            / "ChatStorage.sqlite"
        )
        self.assertNotEqual(result.returncode, 0)
        self.assertIn(str(expected_default), result.stderr)

    def test_percent_and_hash_in_database_filename_are_uri_safe(self) -> None:
        special_db = self.temp_path / "Chat # 100%.sqlite"
        special_db.write_bytes(self.db_path.read_bytes())
        before_names = sorted(path.name for path in self.temp_path.iterdir())

        chats = self.parse_successful_json(
            self.run_raw_cli("chats", "--db", str(special_db), "--json")
        )

        self.assertEqual(len(chats), 2)
        self.assertEqual(
            sorted(path.name for path in self.temp_path.iterdir()), before_names
        )

    def test_text_only_strips_whitespace_before_limit(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.executemany("INSERT INTO ZWAMESSAGE VALUES (?, 1, ?, 0, 0, ?, NULL)",
                             [(300, 800000030, "kept"), (301, 800000031, "\n"),
                              (302, 800000032, " \t\r\n\u2003"), (303, 800000033, None)])
        rows = self.parse_successful_json(self.run_cli("read", "Alice", "--text-only", "--limit", "1"))
        self.assertEqual([row["text"] for row in rows], ["kept"])

    def test_json_control_character_encoding_and_search_truncation(self):
        body = "needle" + "x" * 200 + "\x01\x7f\x80\x9f\u2028end"
        name = "N" * 40
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("UPDATE ZWACHATSESSION SET ZPARTNERNAME=? WHERE Z_PK=1", (name,))
            conn.execute("INSERT INTO ZWAMESSAGE VALUES (300, 1, 800000030, 0, 0, ?, NULL)", (body,))
        result = self.run_cli("search", "needle")
        rows = self.parse_successful_json(result)
        self.assertEqual(rows[0]["text"], body)
        self.assertEqual(rows[0]["chat"], name)
        self.assertIn("\\u0001", result.stdout)
        for character in ("\x7f", "\x80", "\x9f", "\u2028"):
            self.assertIn(character, result.stdout)
        result = self.run_raw_cli("--db", str(self.db_path), "search", "needle")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn(name[:28] + " / ", result.stdout)
        self.assertNotIn(name[:29] + " / ", result.stdout)
        self.assertIn(body[:180], result.stdout)
        self.assertNotIn(body[:181], result.stdout)

    def test_search_treats_percent_and_underscore_as_literals(self) -> None:
        conn = sqlite3.connect(self.db_path)
        try:
            conn.executemany(
                """
                INSERT INTO ZWAMESSAGE
                    (Z_PK, ZCHATSESSION, ZMESSAGEDATE, ZISFROMME,
                     ZMESSAGETYPE, ZTEXT, ZGROUPMEMBER)
                VALUES (?, 1, ?, 0, 0, ?, NULL)
                """,
                [
                    (200, 800000010, "Literal 100% complete"),
                    (201, 800000011, "Literal under_score"),
                    (202, 800000012, "Wildcard distractor"),
                ],
            )
            conn.commit()
        finally:
            conn.close()

        percent = self.parse_successful_json(self.run_cli("search", "%"))
        underscore = self.parse_successful_json(self.run_cli("search", "_"))

        self.assertEqual([item["text"] for item in percent], ["Literal 100% complete"])
        self.assertEqual(
            [item["text"] for item in underscore], ["Literal under_score"]
        )

    def test_invalid_numeric_and_mutually_exclusive_arguments_fail(self) -> None:
        invalid_cases = [
            ("chats", "--limit", "0"),
            ("chats", "--limit", "-1"),
            ("chats", "--days", "-1"),
            ("chats", "--days", "nan"),
            ("chats", "--days", "inf"),
        ]
        for arguments in invalid_cases:
            with self.subTest(arguments=arguments):
                result = self.run_raw_cli(
                    "--db", str(self.db_path), *arguments
                )
                self.assertNotEqual(result.returncode, 0)
                self.assertIn("error:", result.stderr.lower())

        result = self.run_raw_cli(
            "--db",
            str(self.db_path),
            "chats",
            "--groups-only",
            "--dms-only",
        )
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("not allowed with argument", result.stderr.lower())

    def test_equal_message_dates_have_stable_primary_key_order(self) -> None:
        conn = sqlite3.connect(self.db_path)
        try:
            conn.executemany(
                """
                INSERT INTO ZWAMESSAGE
                    (Z_PK, ZCHATSESSION, ZMESSAGEDATE, ZISFROMME,
                     ZMESSAGETYPE, ZTEXT, ZGROUPMEMBER)
                VALUES (?, 1, 800000020, 0, 0, ?, NULL)
                """,
                [(210, "tie lower"), (211, "tie higher")],
            )
            conn.commit()
        finally:
            conn.close()

        messages = self.parse_successful_json(
            self.run_cli("read", "Alice", "--limit", "2")
        )
        self.assertEqual(
            [message["text"] for message in messages], ["tie lower", "tie higher"]
        )

    def test_duplicate_media_rows_do_not_duplicate_messages_or_consume_limit(self) -> None:
        conn = sqlite3.connect(self.db_path)
        try:
            conn.executemany(
                """
                INSERT INTO ZWAMESSAGE
                    (Z_PK, ZCHATSESSION, ZMESSAGEDATE, ZISFROMME,
                     ZMESSAGETYPE, ZTEXT, ZGROUPMEMBER)
                VALUES (?, 1, ?, 0, ?, ?, NULL)
                """,
                [
                    (220, 800000020, 0, "ordinary before media"),
                    (221, 800000021, 1, None),
                ],
            )
            conn.executemany(
                """
                INSERT INTO ZWAMEDIAITEM (ZMESSAGE, ZTITLE, ZMEDIALOCALPATH)
                VALUES (221, ?, ?)
                """,
                [
                    ("first title", "/tmp/first.jpg"),
                    ("second title", "/tmp/second.jpg"),
                ],
            )
            conn.commit()
        finally:
            conn.close()

        messages = self.parse_successful_json(
            self.run_cli("read", "Alice", "--limit", "2")
        )
        self.assertEqual(len(messages), 2)
        self.assertEqual(messages[0]["text"], "ordinary before media")
        self.assertEqual(messages[1]["text"], "[image: first title]")

    def test_duplicate_chat_sessions_are_aggregated_by_jid(self) -> None:
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                """
                INSERT INTO ZWACHATSESSION
                    (Z_PK, ZCONTACTJID, ZPARTNERNAME, ZSESSIONTYPE)
                VALUES (3, '15550000001@s.whatsapp.net', 'Zed Alice', 0)
                """
            )
            conn.execute(
                """
                INSERT INTO ZWAMESSAGE
                    (Z_PK, ZCHATSESSION, ZMESSAGEDATE, ZISFROMME,
                     ZMESSAGETYPE, ZTEXT, ZGROUPMEMBER)
                VALUES (230, 3, 800000030, 0, 0, 'new session', NULL)
                """
            )
            conn.commit()
        finally:
            conn.close()

        chats = self.parse_successful_json(self.run_cli("chats"))
        alice_rows = [
            chat
            for chat in chats
            if chat["jid"] == "15550000001@s.whatsapp.net"
        ]
        self.assertEqual(len(alice_rows), 1)
        self.assertEqual(alice_rows[0]["name"], "Zed Alice")
        self.assertEqual(alice_rows[0]["messages"], 3)

    def test_contacts_json_tolerates_null_push_name(self) -> None:
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "INSERT INTO ZWAPROFILEPUSHNAME (ZJID, ZPUSHNAME) VALUES (?, NULL)",
                ("15550000004@s.whatsapp.net",),
            )
            conn.commit()
        finally:
            conn.close()

        contacts = self.parse_successful_json(self.run_cli("contacts"))
        null_contact = next(
            item
            for item in contacts
            if item["jid"] == "15550000004@s.whatsapp.net"
        )
        self.assertEqual(null_contact["name"], "")

    def test_python_310_refuses_before_snapshot_or_database_access(self) -> None:
        interpreter = shutil.which("python3.10")
        if interpreter is None:
            self.skipTest("Python 3.10 on PATH is required for older-interpreter rejection coverage")
        env = os.environ.copy()
        env["TMPDIR"] = str(self.temp_path)
        before = set(self.temp_path.iterdir())
        result = subprocess.run(
            [interpreter, str(self.script_path), "--db", str(self.temp_path / "missing.sqlite"), "chats"],
            capture_output=True, text=True, check=False, env=env,
        )
        self.assertEqual(result.returncode, 1)
        self.assertEqual(result.stdout, "")
        self.assertEqual(result.stderr.strip(), "Desktop Chat Reader needs Python 3.11 or newer; this is 3.10")
        self.assertEqual(set(self.temp_path.iterdir()), before)
        self.assertFalse(list(self.temp_path.glob("desktop-chat-reader-*")))

    def test_json_text_trimming_captions_and_documented_caveat(self) -> None:
        body = "  Body\x1b\ninside  "
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("UPDATE ZWAMESSAGE SET ZTEXT=? WHERE Z_PK=101", (body,))
            conn.execute("INSERT INTO ZWAMEDIAITEM VALUES (101, '  Caption  ', NULL)")
        messages = self.parse_successful_json(self.run_cli("read", "Alice"))
        self.assertEqual(messages[-1]["text"], "Body\x1b\ninside — Caption")
        matches = self.parse_successful_json(self.run_cli("search", "Body"))
        self.assertEqual(matches[0]["text"], "Body\x1b\ninside")
        for name in ("README.md", "SKILL.md"):
            text = (self.package_root / name).read_text()
            self.assertIn("JSON message text is trimmed at both ends.", text)
            self.assertIn("distinct trimmed caption", text)
            self.assertIn("`search` returns only the trimmed body.", text)
            self.assertNotIn("retains unsanitized text", text)

    def test_human_output_is_printable_but_json_keeps_raw_text(self) -> None:
        raw_text = "\x1b[31mAlert\r\nSecond line"
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                """
                INSERT INTO ZWAMESSAGE
                    (Z_PK, ZCHATSESSION, ZMESSAGEDATE, ZISFROMME,
                     ZMESSAGETYPE, ZTEXT, ZGROUPMEMBER)
                VALUES (240, 1, 800000040, 0, 0, ?, NULL)
                """,
                (raw_text,),
            )
            conn.commit()
        finally:
            conn.close()

        text_result = self.run_raw_cli(
            "--db", str(self.db_path), "read", "Alice", "--limit", "1"
        )
        self.assertEqual(text_result.returncode, 0, text_result.stderr)
        self.assertNotIn("\x1b", text_result.stdout)
        self.assertNotIn("\r", text_result.stdout)
        self.assertIn("Alert Second line", text_result.stdout)

        messages = self.parse_successful_json(
            self.run_cli("read", "Alice", "--limit", "1")
        )
        self.assertEqual(messages[0]["text"], raw_text)

    def test_missing_profile_push_name_table_reports_schema_drift(self) -> None:
        missing_table_db = self.temp_path / "missing-profile.sqlite"
        self._create_fixture_database(
            missing_table_db, omitted_tables={"ZWAPROFILEPUSHNAME"}
        )

        result = self.run_raw_cli(
            "--db", str(missing_table_db), "--json", "chats"
        )

        self.assertNotEqual(result.returncode, 0)
        self.assertIn("schema drift", result.stderr.lower())
        self.assertIn("ZWAPROFILEPUSHNAME", result.stderr)

    def test_every_json_timestamp_has_seconds_and_utc_offset(self) -> None:
        timestamp_pattern = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2}$"
        payloads = [
            self.parse_successful_json(self.run_cli("chats")),
            self.parse_successful_json(self.run_cli("read", "Alice")),
            self.parse_successful_json(self.run_cli("search", "Lantern")),
        ]
        timestamps = [
            value
            for payload in payloads
            for item in payload
            for key, value in item.items()
            if key == "at" or key.endswith("_at")
        ]
        self.assertTrue(timestamps)
        for value in timestamps:
            with self.subTest(value=value):
                self.assertRegex(value, timestamp_pattern)


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

License: MIT (read it).

Back to the listing