shd.build/tools/game-data-capture/tests/unit/ingest/test_clipboard.py

from __future__ import annotations

import io
import json
import math
import os
import subprocess
import sys
from pathlib import Path

import pytest

import game_data_capture.ingest.clipboard as clipboard
from game_data_capture.ingest.clipboard import (
    ClipboardIntake,
    FocusedWindowMetadata,
    WatchConfig,
    focused_window,
    restricted_environment,
)
from game_data_capture.ingest.spool_helper import PNG_SIGNATURE, spool_png


def receipt(spool: Path, token: str = "receipt") -> dict[str, object]:
    return spool_png(
        __import__("io").BytesIO(PNG_SIGNATURE + b"png"),
        spool=spool,
        max_bytes=100,
        receipt_token=token,
        policy_version="v1",
    )


def unique_receipt(spool: Path, token: str, value: bytes) -> dict[str, object]:
    return spool_png(
        io.BytesIO(PNG_SIGNATURE + value),
        spool=spool,
        max_bytes=100,
        receipt_token=token,
        policy_version="v1",
    )


def config(tmp_path: Path, *, queue: int = 2, retries: int = 2) -> WatchConfig:
    return WatchConfig((tmp_path / "spool").resolve(), 100, queue, "v1", retries, 0.01, 0.1)


def test_environment_is_allowlisted_and_secret_free() -> None:
    child = restricted_environment(
        {
            "PATH": "/bin",
            "XDG_RUNTIME_DIR": "/run/user/1",
            "OPENAI_API_KEY": "sk-secret",
            "HTTP_AUTHORIZATION": "Bearer secret",
            "PYTHONPATH": "/unsafe",
            "UNRELATED": "no",
        }
    )
    assert child == {"PATH": "/bin", "XDG_RUNTIME_DIR": "/run/user/1"}


def test_text_clear_duplicate_and_backpressure_are_silent(tmp_path: Path) -> None:
    intake = ClipboardIntake(config(tmp_path, queue=1))
    item = receipt(intake.config.spool)
    intake.armed = True
    metadata = FocusedWindowMetadata(None, None, None)
    assert not intake.accept_helper_receipt(
        {"media_type": "text/plain"}, occurrence_id="a", processing_version="p"
    )
    assert not intake.accept_helper_receipt({}, occurrence_id="a", processing_version="p")
    assert intake.accept_helper_receipt(
        item, occurrence_id="a", processing_version="p", metadata=metadata
    )
    assert not intake.accept_helper_receipt(
        item, occurrence_id="a", processing_version="p", metadata=metadata
    )
    other = receipt(intake.config.spool, "other")
    assert not intake.accept_helper_receipt(
        other, occurrence_id="b", processing_version="p", metadata=metadata
    )
    assert intake.queue_depth == 1


def test_watcher_argv_places_options_before_watch_subcommand(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    launched: list[list[str]] = []

    class Fake:
        pid = os.getpid()
        stdout = None

        def poll(self) -> int | None:
            return None

        def wait(self, timeout: float | None = None) -> int:
            return 0

    def popen(args: list[str], **kwargs: object) -> Fake:
        assert kwargs["shell"] is False
        launched.append(args)
        return Fake()

    monkeypatch.setattr(subprocess, "Popen", popen)
    monkeypatch.setattr(os, "killpg", lambda _pid, _signal: None)
    intake = ClipboardIntake(config(tmp_path))
    intake.arm()
    command = launched[0]
    watch = command.index("--watch")
    assert command[:watch] == ["wl-paste", "--type", "image/png"]
    assert command[watch + 1] == sys.executable
    assert "--" not in command
    assert "game_data_capture.ingest.spool_helper" in command
    intake.pause()
    assert intake.paused
    intake.resume()
    assert len(launched) == 2
    intake.disarm()
    assert not intake.armed


def test_cancellation_disarms_and_clears_queue(tmp_path: Path) -> None:
    intake = ClipboardIntake(config(tmp_path))
    intake.armed = True
    assert intake.accept_helper_receipt(
        receipt(intake.config.spool),
        occurrence_id="occurrence",
        processing_version="processing-v1",
        metadata=FocusedWindowMetadata(None, None, None),
    )
    intake.cancel()
    assert not intake.armed and intake.take() is None


def test_receipt_integrity_versions_and_bounded_duplicate_cache(tmp_path: Path) -> None:
    intake = ClipboardIntake(config(tmp_path, queue=2))
    intake.armed = True
    metadata = FocusedWindowMetadata(None, None, None)
    first = receipt(intake.config.spool, "one")
    future = dict(first, receipt_version=2)
    assert not intake.accept_helper_receipt(
        future, occurrence_id="one", processing_version="p", metadata=metadata
    )
    assert intake.accept_helper_receipt(
        first, occurrence_id="one", processing_version="p", metadata=metadata
    )
    assert not intake.accept_helper_receipt(
        first, occurrence_id="one", processing_version="p", metadata=metadata
    )
    assert len(intake._seen_tokens) == 1
    intake.shutdown()
    assert not intake._seen_tokens and not intake._helper_payloads


def test_durable_receipt_recovery_is_null_metadata_duplicate_safe_and_acknowledged(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    intake = ClipboardIntake(config(tmp_path))
    item = receipt(intake.config.spool, "crash-token")
    intake.armed = True
    monkeypatch.setattr(
        clipboard,
        "focused_window",
        lambda: pytest.fail("recovered receipt must not query the current window"),
    )
    intake.reconcile()
    recovered = intake.take_helper_receipt()
    assert recovered == item
    assert intake.accept_helper_receipt(
        recovered, occurrence_id="recovered", processing_version="p"
    )
    accepted = intake.take()
    assert accepted is not None
    assert accepted.focused_window == FocusedWindowMetadata(None, None, None)
    assert not intake.accept_helper_receipt(item, occurrence_id="live-copy", processing_version="p")
    assert (intake.config.spool / "crash-token.receipt.json").exists()
    assert intake.acknowledge(accepted)
    assert not intake.acknowledge(accepted)
    assert not (intake.config.spool / "crash-token.receipt.json").exists()
    assert accepted.spool_path.exists()


def test_reconcile_respects_helper_backpressure_and_retains_durable_receipts(
    tmp_path: Path,
) -> None:
    intake = ClipboardIntake(config(tmp_path, queue=1))
    first = unique_receipt(intake.config.spool, "first", b"first")
    second = unique_receipt(intake.config.spool, "second", b"second")
    intake.armed = True
    intake.reconcile()
    queued = intake.take_helper_receipt()
    assert queued in (first, second)
    assert intake.accept_helper_receipt(
        queued,
        occurrence_id="first",
        processing_version="p",
        metadata=FocusedWindowMetadata(None, None, None),
    )
    remaining = second if queued == first else first
    assert (intake.config.spool / f"{remaining['receipt_token']}.receipt.json").exists()
    intake.reconcile()
    assert intake.take_helper_receipt() == remaining


def test_unacknowledged_receipt_survives_a_fresh_intake(tmp_path: Path) -> None:
    initial = ClipboardIntake(config(tmp_path))
    item = receipt(initial.config.spool, "retry-token")
    restarted = ClipboardIntake(config(tmp_path))
    restarted.armed = True
    restarted.reconcile()
    replay = restarted.take_helper_receipt()
    assert replay == item
    assert restarted.accept_helper_receipt(replay, occurrence_id="retry", processing_version="p")
    accepted = restarted.take()
    assert accepted is not None
    assert (restarted.config.spool / "retry-token.receipt.json").exists()


@pytest.mark.parametrize("field", ["max_image_bytes", "max_queue_items", "max_restarts"])
def test_bool_limits_are_rejected(tmp_path: Path, field: str) -> None:
    values: dict[str, object] = dict(
        spool=(tmp_path / "spool").resolve(),
        max_image_bytes=100,
        max_queue_items=1,
        policy_version="v1",
        max_restarts=1,
    )
    values[field] = True
    with pytest.raises(ValueError):
        WatchConfig(**values)  # type: ignore[arg-type]


def test_watcher_crash_uses_backoff_and_restart_cap(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    starts = 0

    class Dead:
        pid = os.getpid()
        stdout = None

        def poll(self) -> int:
            return 1

        def wait(self, timeout: float | None = None) -> int:
            return 1

    def popen(*_args: object, **_kwargs: object) -> Dead:
        nonlocal starts
        starts += 1
        return Dead()

    monkeypatch.setattr(subprocess, "Popen", popen)
    intake = ClipboardIntake(config(tmp_path, retries=1))
    intake.arm()
    intake.tick(0)
    assert starts == 1
    intake.tick(0.02)
    assert starts == 2
    intake.tick(0.02)
    intake.tick(1)
    assert starts == 2


def test_niri_timeout_and_malformed_json_are_unknown(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(clipboard, "_bounded_command_output", lambda *_args: None)
    assert focused_window() == FocusedWindowMetadata(None, None, None)
    monkeypatch.setattr(clipboard, "_bounded_command_output", lambda *_args: b"{broken")
    assert focused_window() == FocusedWindowMetadata(None, None, None)


def test_receipt_mutation_matrix_and_file_integrity(tmp_path: Path) -> None:
    intake = ClipboardIntake(config(tmp_path))
    intake.armed = True
    metadata = FocusedWindowMetadata(None, None, None)
    base = unique_receipt(intake.config.spool, "token", b"base")
    for key, value in (
        ("receipt_version", 2),
        ("receipt_token", ""),
        ("sha256", "0" * 64),
        ("byte_size", 9),
        ("media_type", "text/plain"),
        ("policy_version", "other"),
        ("spool_name", "wrong.png"),
    ):
        changed = dict(base)
        changed[key] = value
        assert not intake.accept_helper_receipt(
            changed, occurrence_id="ok", processing_version="p", metadata=metadata
        )
    for value in ("", "bad value"):
        assert not intake.accept_helper_receipt(
            base, occurrence_id=value, processing_version="p", metadata=metadata
        )
        assert not intake.accept_helper_receipt(
            base, occurrence_id="ok", processing_version=value, metadata=metadata
        )
    for index, (mode, content) in enumerate(
        ((0o600, b"changed"), (0o600, PNG_SIGNATURE), (0o644, None))
    ):
        item = unique_receipt(
            intake.config.spool, f"token{mode}{index}", bytes(f"{mode}{index}", "ascii")
        )
        path = intake.config.spool / str(item["spool_name"])
        if content is not None:
            path.write_bytes(content)
        path.chmod(mode)
        assert not intake.accept_helper_receipt(
            item, occurrence_id=f"o{mode}", processing_version="p", metadata=metadata
        )
    item = unique_receipt(intake.config.spool, "symlink", b"link")
    path = intake.config.spool / str(item["spool_name"])
    path.unlink()
    path.symlink_to(tmp_path / "outside")
    assert not intake.accept_helper_receipt(
        item, occurrence_id="symlink", processing_version="p", metadata=metadata
    )


def test_open_verified_and_acceptance_close_descriptors(tmp_path: Path) -> None:
    intake = ClipboardIntake(config(tmp_path))
    intake.armed = True
    item = unique_receipt(intake.config.spool, "fd", b"reader")
    before = len(list(Path("/proc/self/fd").iterdir()))
    assert intake.accept_helper_receipt(
        item,
        occurrence_id="fd",
        processing_version="p",
        metadata=FocusedWindowMetadata(None, None, None),
    )
    accepted = intake.take()
    assert accepted is not None and len(list(Path("/proc/self/fd").iterdir())) == before
    with intake.open_verified(accepted) as reader:
        assert b"".join(reader) == PNG_SIGNATURE + b"reader"
    path = accepted.spool_path
    path.write_bytes(b"changed")
    with pytest.raises(ValueError), intake.open_verified(accepted):
        pass


def test_duplicate_cache_eviction_shutdown_and_unreferenced_candidates(tmp_path: Path) -> None:
    intake = ClipboardIntake(config(tmp_path, queue=2))
    intake.armed = True
    metadata = FocusedWindowMetadata(None, None, None)
    items = [unique_receipt(intake.config.spool, f"t{i}", bytes([i])) for i in range(3)]
    for index, item in enumerate(items):
        assert intake.accept_helper_receipt(
            item, occurrence_id=f"o{index}", processing_version="p", metadata=metadata
        )
        intake.take()
    assert len(intake._seen_tokens) == 2 and not intake.accept_helper_receipt(
        items[2], occurrence_id="latest", processing_version="p", metadata=metadata
    )
    assert intake.accept_helper_receipt(
        items[0], occurrence_id="evicted", processing_version="p", metadata=metadata
    )
    intake = ClipboardIntake(config(tmp_path / "other", queue=1))
    intake.armed = True
    queued = unique_receipt(intake.config.spool, "queued", b"queued")
    extra = unique_receipt(intake.config.spool, "extra", b"extra")
    assert intake.accept_helper_receipt(
        queued, occurrence_id="queued", processing_version="p", metadata=metadata
    )
    assert [path.name for path in intake.unreferenced_spool_candidates()] == [
        str(extra["spool_name"])
    ]
    intake._helper_payloads.append({"receipt": "real"})
    intake.shutdown()
    assert (
        not intake._queue
        and not intake._helper_payloads
        and not intake._seen_tokens
        and not intake._seen_order
    )


def test_receipt_reader_bounds_spawn_failures_and_config_types(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    class Output:
        def __init__(self) -> None:
            self.parts = [
                b"x" * 1025 + b"\n",
                json.dumps({"ok": 1}).encode() + b"\n",
                json.dumps({"two": 2}).encode() + b"\n",
            ]

        def readline(self, _size: int) -> bytes:
            return self.parts.pop(0) if self.parts else b""

    class Process:
        stdout = Output()

    intake = ClipboardIntake(config(tmp_path, queue=1))
    intake._read_receipts(Process())  # type: ignore[arg-type]
    assert intake.helper_queue_depth == 1 and intake.take_helper_receipt() == {"ok": 1}
    attempts = 0

    def failing(*_args: object, **_kwargs: object) -> object:
        nonlocal attempts
        attempts += 1
        raise OSError("missing")

    monkeypatch.setattr(subprocess, "Popen", failing)
    failed = ClipboardIntake(config(tmp_path, retries=2))
    failed.arm()
    failed.tick(failed._restart_after + 0.01)
    failed.tick(failed._restart_after + 0.01)
    failed.tick(failed._restart_after + 1)
    assert attempts == 3
    for field, value in (
        ("max_image_bytes", 1.5),
        ("max_queue_items", "1"),
        ("max_restarts", 1.5),
        ("shutdown_timeout_seconds", 0),
        ("restart_backoff_seconds", math.inf),
    ):
        values: dict[str, object] = dict(
            spool=(tmp_path / "x").resolve(),
            max_image_bytes=100,
            max_queue_items=1,
            policy_version="v1",
            max_restarts=1,
        )
        values[field] = value
        with pytest.raises(ValueError):
            WatchConfig(**values)  # type: ignore[arg-type]


def test_bounded_niri_runner_and_valid_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
    assert (
        clipboard._bounded_command_output([sys.executable, "-c", "print('x'*20000)"], {}, 1) is None
    )
    assert (
        clipboard._bounded_command_output(
            [sys.executable, "-c", "import time; time.sleep(2)"], {}, 0.01
        )
        is None
    )
    monkeypatch.setattr(
        clipboard,
        "_bounded_command_output",
        lambda *_args: b'{"app_id":"game","title":"title","id":1}',
    )
    assert focused_window() == FocusedWindowMetadata("game", "title", 1)


def test_exited_watcher_is_waited_stdout_closed_and_reader_retired(tmp_path: Path) -> None:
    class Output:
        closed = False

        def close(self) -> None:
            self.closed = True

    class Child:
        pid = os.getpid()
        stdout = Output()
        waits = 0

        def poll(self) -> int:
            return 1

        def wait(self, timeout: float | None = None) -> int:
            self.waits += 1
            return 1

    class Reader:
        joined = False

        def join(self, timeout: float | None = None) -> None:
            self.joined = True

    intake = ClipboardIntake(config(tmp_path))
    child = Child()
    reader = Reader()
    intake._watcher = child  # type: ignore[assignment]
    intake._reader = reader  # type: ignore[assignment]
    intake.pause()
    assert child.waits == 1 and child.stdout.closed and reader.joined and intake._reader is None