path: root/.ai/scripts/tests/test_inbox_send.py

"""Tests for inbox-send.py — universal cross-project inbox messaging tool.

The script:
- discovers .ai projects with an inbox/ subdirectory under known roots,
- writes a text message as a dated .org file in the target's inbox/, or
- copies a file into the target's inbox/ with a dated, source-tagged name.

All discovery is roots-driven (env var INBOX_SEND_ROOTS overrides the
defaults) so tests can sandbox everything inside tmp_path.
"""

import subprocess
from pathlib import Path

import pytest

SCRIPT = Path(__file__).parent.parent / "inbox-send.py"


@pytest.fixture
def project_root(tmp_path):
    """Build a fake project under tmp_path/projects/<name>/ with .ai/ + top-level inbox/."""
    def _make(name: str, has_inbox: bool = True) -> Path:
        proj = tmp_path / "projects" / name
        proj.mkdir(parents=True, exist_ok=True)
        (proj / ".ai").mkdir(exist_ok=True)
        if has_inbox:
            (proj / "inbox").mkdir(exist_ok=True)
        return proj
    return _make


@pytest.fixture
def run_script(tmp_path):
    """Invoke inbox-send with sandboxed roots via INBOX_SEND_ROOTS env var."""
    def _run(args, cwd=None, roots=None, expect_failure=False):
        env = {}
        # Preserve PATH and a few essentials for python3 to launch.
        import os as _os
        env["PATH"] = _os.environ.get("PATH", "")
        env["HOME"] = _os.environ.get("HOME", "/tmp")
        if roots:
            env["INBOX_SEND_ROOTS"] = ":".join(str(r) for r in roots)
        cmd = ["python3", str(SCRIPT)] + args
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=cwd or tmp_path,
            env=env,
            check=not expect_failure,
        )
        return result
    return _run


# ----------------------------------------------------------------------
# Discovery (--list)
# ----------------------------------------------------------------------

class TestInboxSendDiscovery:
    """Discovering available .ai projects under the configured roots."""

    def test_inbox_send_list_detects_projects_with_ai_inbox(self, project_root, run_script, tmp_path):
        """Normal: --list shows projects that have .ai/inbox/."""
        project_root("foo")
        project_root("bar")
        result = run_script(["--list"], roots=[tmp_path / "projects"])
        assert "foo" in result.stdout
        assert "bar" in result.stdout

    def test_inbox_send_list_skips_projects_without_inbox(self, project_root, run_script, tmp_path):
        """Boundary: project with .ai/ but no inbox/ is not surfaced."""
        project_root("withinbox", has_inbox=True)
        project_root("noinbox", has_inbox=False)
        result = run_script(["--list"], roots=[tmp_path / "projects"])
        assert "withinbox" in result.stdout
        assert "noinbox" not in result.stdout

    def test_inbox_send_list_skips_current_project(self, project_root, run_script, tmp_path):
        """Normal: --list excludes the project the user is currently in."""
        cwd_project = project_root("current")
        project_root("other")
        result = run_script(["--list"], cwd=cwd_project, roots=[tmp_path / "projects"])
        assert "other" in result.stdout
        assert "current" not in result.stdout

    def test_inbox_send_list_empty_when_no_projects(self, run_script, tmp_path):
        """Boundary: no projects under roots → friendly informational message."""
        (tmp_path / "projects").mkdir()
        result = run_script(["--list"], roots=[tmp_path / "projects"])
        assert result.returncode == 0
        assert "No projects" in result.stdout

    def test_inbox_send_list_handles_missing_root(self, run_script, tmp_path):
        """Boundary: configured root doesn't exist → skip silently."""
        result = run_script(["--list"], roots=[tmp_path / "does-not-exist"])
        assert result.returncode == 0

    def test_inbox_send_list_displays_dot_stripped_name(self, project_root, run_script, tmp_path):
        """Dotted project basenames display dot-stripped (.emacs.d → emacsd)."""
        project_root(".emacs.d")
        result = run_script(["--list"], roots=[tmp_path / "projects"])
        assert "emacsd" in result.stdout


class TestInboxSendDotAlias:
    """A dotted project basename resolves both verbatim and dot-stripped."""

    def test_resolves_by_dot_stripped_alias(self, project_root, run_script, tmp_path):
        """'emacsd' delivers to the .emacs.d project."""
        project_root(".emacs.d")
        cwd = project_root("source")
        run_script(
            ["emacsd", "--text", "hi"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / ".emacs.d" / "inbox").iterdir())
        assert len(files) == 1

    def test_resolves_by_exact_dotted_name_still(self, project_root, run_script, tmp_path):
        """Backward-compat: the verbatim '.emacs.d' target still resolves."""
        project_root(".emacs.d")
        cwd = project_root("source")
        run_script(
            [".emacs.d", "--text", "hi"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / ".emacs.d" / "inbox").iterdir())
        assert len(files) == 1

    def test_exact_match_wins_over_alias(self, project_root, run_script, tmp_path):
        """An exact basename match is preferred over a dot-stripped collision."""
        project_root("emacsd")        # exact
        project_root(".emacs.d")      # would also normalize to 'emacsd'
        cwd = project_root("source")
        run_script(
            ["emacsd", "--text", "hi"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        exact = list((tmp_path / "projects" / "emacsd" / "inbox").iterdir())
        dotted = list((tmp_path / "projects" / ".emacs.d" / "inbox").iterdir())
        assert len(exact) == 1
        assert dotted == []


# ----------------------------------------------------------------------
# Slug derivation from text and from filenames
# ----------------------------------------------------------------------

def _slug_from(inbox_files, source_name):
    """Helper: extract the slug from a deposited file's basename."""
    assert len(inbox_files) == 1
    name = inbox_files[0].stem
    marker = f"from-{source_name}-"
    return name.split(marker, 1)[1]


class TestInboxSendNaming:
    """Slug derivation from --text (and override via --name)."""

    def test_inbox_send_text_slug_hyphenated_lowercase(self, project_root, run_script, tmp_path):
        """Normal: 'ATM cash reminder' → slug 'atm-cash-reminder'."""
        project_root("target")
        cwd = project_root("source")
        run_script(
            ["target", "--text", "ATM cash reminder"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert _slug_from(files, "source") == "atm-cash-reminder"

    def test_inbox_send_text_slug_truncated_at_word_boundary(self, project_root, run_script, tmp_path):
        """Normal: long text truncated under 40 chars at the nearest word boundary."""
        project_root("target")
        cwd = project_root("source")
        long_text = (
            "Please review the SOFWeek prep doc and confirm the AirBnB kitchen details"
        )
        run_script(
            ["target", "--text", long_text],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        slug = _slug_from(files, "source")
        assert slug.startswith("please-review-the-sofweek")
        assert len(slug) <= 40
        # Truncation should land on a word boundary (last char is a letter/digit, not mid-word).
        assert "-" not in slug[-1]

    def test_inbox_send_text_slug_strips_punctuation(self, project_root, run_script, tmp_path):
        """Normal: punctuation stripped, lowercased."""
        project_root("target")
        cwd = project_root("source")
        run_script(
            ["target", "--text", "Hey! What's the plan? See you @ 5PM."],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        slug = _slug_from(files, "source")
        for ch in "!?'@.":
            assert ch not in slug
        assert slug == slug.lower()

    def test_inbox_send_name_override_overrides_slug(self, project_root, run_script, tmp_path):
        """Normal: --name wins over derived slug."""
        project_root("target")
        cwd = project_root("source")
        run_script(
            ["target", "--text", "ok", "--name", "pre-call-ack"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert _slug_from(files, "source") == "pre-call-ack"


# ----------------------------------------------------------------------
# --text mode end-to-end
# ----------------------------------------------------------------------

class TestInboxSendText:
    """--text mode writes a .org file with the message body."""

    def test_inbox_send_text_writes_org_file_with_message(self, project_root, run_script, tmp_path):
        """Normal: produces a .org file whose body contains the message."""
        project_root("target")
        cwd = project_root("source")
        run_script(
            ["target", "--text", "Remember the ATM run"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert len(files) == 1
        assert files[0].suffix == ".org"
        body = files[0].read_text()
        assert "Remember the ATM run" in body

    def test_inbox_send_text_filename_includes_source_project_name(self, project_root, run_script, tmp_path):
        """Normal: filename includes 'from-<source>-' so the target knows where it came from."""
        project_root("target")
        cwd = project_root("emacs")
        run_script(
            ["target", "--text", "hello"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert "from-emacs-" in files[0].name


# ----------------------------------------------------------------------
# --file mode end-to-end
# ----------------------------------------------------------------------

class TestInboxSendFile:
    """--file mode copies the source file into the target inbox."""

    def test_inbox_send_file_copies_text_file(self, project_root, run_script, tmp_path):
        """Normal: copies a text file to the target inbox, preserving content."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "doc.org"
        src.write_text("file content")
        run_script(
            ["target", "--file", str(src)],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert len(files) == 1
        assert files[0].read_text() == "file content"

    def test_inbox_send_file_preserves_extension(self, project_root, run_script, tmp_path):
        """Normal: extension carried from source file."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "image.png"
        src.write_bytes(b"\x89PNG\r\n...")
        run_script(
            ["target", "--file", str(src)],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert files[0].suffix == ".png"

    def test_inbox_send_file_slug_from_source_basename(self, project_root, run_script, tmp_path):
        """Normal: filename slug derived from the source file's basename when --name omitted."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "branching-strategy-notes.md"
        src.write_text("notes")
        run_script(
            ["target", "--file", str(src)],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert "branching-strategy-notes" in files[0].name

    def test_inbox_send_file_preserves_dots_in_basename(self, project_root, run_script, tmp_path):
        """Boundary: a dotted stem keeps its dots — the engine.plugin.org plugin namespace must survive transit."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "triage-intake.personal-gmail.org"
        src.write_text("plugin")
        run_script(
            ["target", "--file", str(src)],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert len(files) == 1
        assert _slug_from(files, "source") == "triage-intake.personal-gmail"
        assert files[0].suffix == ".org"

    def test_inbox_send_file_spaces_become_hyphens(self, project_root, run_script, tmp_path):
        """Normal: whitespace in a filename stem still collapses to hyphens; dots are what's preserved, not spaces."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "meeting notes draft.org"
        src.write_text("x")
        run_script(
            ["target", "--file", str(src)],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert _slug_from(files, "source") == "meeting-notes-draft"

    def test_inbox_send_file_name_override(self, project_root, run_script, tmp_path):
        """Normal: --name overrides the basename-derived slug; extension preserved."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "random.pdf"
        src.write_bytes(b"%PDF-1.4...")
        run_script(
            ["target", "--file", str(src), "--name", "branching-strategy"],
            cwd=cwd, roots=[tmp_path / "projects"],
        )
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert "branching-strategy" in files[0].name
        assert files[0].suffix == ".pdf"


# ----------------------------------------------------------------------
# Errors and refusal cases
# ----------------------------------------------------------------------

class TestInboxSendErrors:
    """Refusal cases — surface clearly, exit non-zero, leave filesystem untouched."""

    def test_inbox_send_refuses_unknown_target(self, project_root, run_script, tmp_path):
        """Error: target project not found in discovery → refuse."""
        cwd = project_root("source")
        result = run_script(
            ["nonexistent", "--text", "hi"],
            cwd=cwd, roots=[tmp_path / "projects"],
            expect_failure=True,
        )
        assert result.returncode != 0

    def test_inbox_send_refuses_no_text_and_no_file(self, project_root, run_script, tmp_path):
        """Error: must provide one of --text / --file."""
        project_root("target")
        cwd = project_root("source")
        result = run_script(
            ["target"],
            cwd=cwd, roots=[tmp_path / "projects"],
            expect_failure=True,
        )
        assert result.returncode != 0

    def test_inbox_send_refuses_both_text_and_file(self, project_root, run_script, tmp_path):
        """Error: --text and --file are mutually exclusive."""
        project_root("target")
        cwd = project_root("source")
        src = tmp_path / "doc.org"
        src.write_text("x")
        result = run_script(
            ["target", "--text", "hi", "--file", str(src)],
            cwd=cwd, roots=[tmp_path / "projects"],
            expect_failure=True,
        )
        assert result.returncode != 0

    def test_inbox_send_refuses_missing_source_file(self, project_root, run_script, tmp_path):
        """Error: --file path doesn't exist → refuse."""
        project_root("target")
        cwd = project_root("source")
        result = run_script(
            ["target", "--file", str(tmp_path / "definitely-missing.org")],
            cwd=cwd, roots=[tmp_path / "projects"],
            expect_failure=True,
        )
        assert result.returncode != 0

    def test_inbox_send_refuses_empty_text(self, project_root, run_script, tmp_path):
        """Error: empty --text refused; nothing written to target inbox."""
        project_root("target")
        cwd = project_root("source")
        result = run_script(
            ["target", "--text", "   "],
            cwd=cwd, roots=[tmp_path / "projects"],
            expect_failure=True,
        )
        assert result.returncode != 0
        files = list((tmp_path / "projects" / "target" / "inbox").iterdir())
        assert files == []


# ----------------------------------------------------------------------
# Filename collisions (two sends deriving the same name must not overwrite)
# ----------------------------------------------------------------------

def _load_module():
    import importlib.util
    spec = importlib.util.spec_from_file_location("inbox_send", SCRIPT)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


class TestFilenameCollisions:
    """Two sends in the same minute with the same leading phrase derived
    identical filenames and the second silently overwrote the first
    (a message was lost this way, 2026-07-02)."""

    def test_send_text_same_minute_same_phrase_keeps_both(self, tmp_path):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        now = datetime(2026, 7, 2, 5, 42, 0)
        prefix = "identical leading phrase long enough to fill the whole slug budget entirely"
        first = mod.send_text(inbox, prefix + " tail one", "archsetup", None, now)
        second = mod.send_text(inbox, prefix + " tail two", "archsetup", None, now)
        assert first != second
        assert first.exists() and second.exists()
        assert first.name != second.name
        assert "tail one" in first.read_text()
        assert "tail two" in second.read_text()

    def test_send_text_collision_suffix_increments(self, tmp_path):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        now = datetime(2026, 7, 2, 5, 42, 0)
        paths = [mod.send_text(inbox, "same lead phrase differs later A", "src", "fixed-slug", now)
                 for _ in range(3)]
        names = [p.name for p in paths]
        assert names[0].endswith("fixed-slug.org")
        assert names[1].endswith("fixed-slug-2.org")
        assert names[2].endswith("fixed-slug-3.org")

    def test_send_file_collision_preserves_extension(self, tmp_path):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        src = tmp_path / "note.org"
        src.write_text("body one")
        now = datetime(2026, 7, 2, 5, 42, 0)
        first = mod.send_file(inbox, src, "src", None, now)
        src.write_text("body two")
        second = mod.send_file(inbox, src, "src", None, now)
        assert second.name.endswith("note-2.org")
        assert first.read_text() == "body one"
        assert second.read_text() == "body two"

    def test_cli_two_rapid_sends_lose_nothing(self, project_root, run_script, tmp_path):
        project_root("sender")
        target = project_root("receiver")
        roots = [tmp_path / "projects"]
        prefix = "identical leading phrase long enough to fill the whole slug budget entirely"
        run_script(["receiver", "--text", prefix + " message one"],
                   cwd=tmp_path / "projects" / "sender", roots=roots)
        run_script(["receiver", "--text", prefix + " message two"],
                   cwd=tmp_path / "projects" / "sender", roots=roots)
        files = list((target / "inbox").iterdir())
        assert len(files) == 2
        bodies = "".join(f.read_text() for f in files)
        assert "message one" in bodies and "message two" in bodies


class TestAtomicWrite:
    """A send wrote straight to the destination path in another project's
    inbox/, and write_text truncates on open, so any mid-write failure left a
    zero-byte .org there. inbox-status counts that phantom as a pending
    handoff, blocking a turn in the receiving project over a file with no
    content (2026-07-23). The write must be atomic: the inbox sees a complete
    file or nothing."""

    def test_send_text_writes_utf8(self, tmp_path):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        now = datetime(2026, 7, 23, 4, 36, 0)
        # An em dash and an accented char — both non-ASCII.
        dest = mod.send_text(inbox, "accent café and dash — here", "src", None, now)
        # Reading as utf-8 must round-trip; a locale-encoded write would raise
        # under a C locale, and reading back proves the bytes are utf-8.
        assert "—" in dest.read_text(encoding="utf-8")

    def test_send_text_no_partial_on_write_failure(self, tmp_path, monkeypatch):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        now = datetime(2026, 7, 23, 4, 36, 0)
        # Force the atomic finalize to fail after the temp file is written.
        def boom(*a, **k):
            raise OSError("disk full")
        monkeypatch.setattr(mod.os, "replace", boom)
        with pytest.raises(OSError):
            mod.send_text(inbox, "a message that should never half-land", "src", None, now)
        # No phantom, no leftover temp: the inbox is empty.
        assert list(inbox.iterdir()) == []

    def test_send_text_leaves_no_temp_on_success(self, tmp_path):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        now = datetime(2026, 7, 23, 4, 36, 0)
        dest = mod.send_text(inbox, "clean send", "src", None, now)
        assert list(inbox.iterdir()) == [dest]

    def test_send_file_no_partial_on_write_failure(self, tmp_path, monkeypatch):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        src = tmp_path / "note.org"
        src.write_text("body")
        now = datetime(2026, 7, 23, 4, 36, 0)
        def boom(*a, **k):
            raise OSError("disk full")
        monkeypatch.setattr(mod.os, "replace", boom)
        with pytest.raises(OSError):
            mod.send_file(inbox, src, "src", None, now)
        assert list(inbox.iterdir()) == []

    def test_send_file_leaves_no_temp_on_success(self, tmp_path):
        from datetime import datetime
        mod = _load_module()
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        src = tmp_path / "note.org"
        src.write_text("payload")
        now = datetime(2026, 7, 23, 4, 36, 0)
        dest = mod.send_file(inbox, src, "src", None, now)
        assert list(inbox.iterdir()) == [dest]
        assert dest.read_text() == "payload"


class TestSmallerDefects:
    """Two low-severity defects found reading inbox-send during the 2026-07-23
    sweep: an unreadable source raised an uncaught traceback instead of the
    clean error every other failure path produces, and a roots config naming
    both a parent and one of its children listed the same project twice."""

    def test_unreadable_source_gives_clean_error_not_traceback(
        self, project_root, run_script, tmp_path
    ):
        project_root("sender")
        project_root("receiver")
        roots = [tmp_path / "projects"]
        src = tmp_path / "secret.bin"
        src.write_text("x")
        src.chmod(0o000)
        try:
            result = run_script(
                ["receiver", "--file", str(src)],
                cwd=tmp_path / "projects" / "sender",
                roots=roots,
                expect_failure=True,
            )
        finally:
            src.chmod(0o644)
        assert result.returncode == 1
        # The clean "inbox-send: <message>" shape, not a Python traceback.
        assert result.stderr.startswith("inbox-send:")
        assert "Traceback" not in result.stderr

    def test_discover_projects_dedupes_parent_and_child_root(self, tmp_path):
        mod = _load_module()
        # A project directory, reachable both as a child of its parent root and
        # as a root in its own right.
        parent = tmp_path / "projects"
        proj = parent / "app"
        (proj / ".ai").mkdir(parents=True)
        (proj / "inbox").mkdir()
        found = mod.discover_projects([parent, proj])
        resolved = [p.resolve() for p in found]
        assert resolved.count(proj.resolve()) == 1