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