diff --git a/README.md b/README.md index 85e2d391..649f9cd4 100644 --- a/README.md +++ b/README.md @@ -787,6 +787,7 @@ For AI-agent driven interactive testing and current test command tiers, see: - `docs/ai_agent_tui_testing.md` - `docs/agent_testing_roadmap.md` +- `docs/agent_test_format.md` ### Exploring Nostr Index Size Limits diff --git a/docs/README.md b/docs/README.md index 98a6dc44..46ba888a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -36,6 +36,7 @@ This directory contains SeedPass project documentation. - `agent_testing_roadmap.md` - `ai_agent_tui_testing.md` +- `agent_test_format.md` - `installer_testing.md` - `ci-health-2026-02-19.md` - `agent-handoffs/README.md` diff --git a/docs/agent_test_format.md b/docs/agent_test_format.md new file mode 100644 index 00000000..9857760b --- /dev/null +++ b/docs/agent_test_format.md @@ -0,0 +1,102 @@ +# Agent Test Format + +This document defines the repeatable "agent-style" validation format for +SeedPass TUI v2, CLI/CUI, and API surfaces. + +## Goals + +- Reproduce user and agent workflows deterministically. +- Generate machine-readable artifacts for regressions. +- Keep a stable command format for local runs and CI runs. + +## Output Contract + +Each harness run should emit: + +- `status`: `passed` or `failed` +- `failure`: short error detail when failed +- `duration_sec`: run duration +- `steps`: per-step status/timing +- `coverage_points`: workflow checkpoints as booleans +- optional transcripts/activity logs + +Reports are written to timestamped artifact folders under `artifacts/`. + +## TUI v2 Harness + +Script: + +```bash +python3 scripts/ai_tui2_agent_test.py --scenario extended --verbose +``` + +Artifacts: + +- `artifacts/agent_tui2_test//report.json` + +Scenarios: + +- `core`: startup/search/document edit/graph open/archive-restore +- `extended`: `core` + palette unknown command + help overlay/navigation checks + +## Legacy Interactive Harness + +Script: + +```bash +python3 scripts/ai_tui_agent_test.py --scenario extended --python-bin .venv/bin/python --verbose +``` + +Artifacts: + +- `artifacts/agent_tui_test//report.json` +- transcript files (`transcript.clean.txt`, `transcript.raw.txt`) + +Use this for legacy interactive flow regression checks. + +## CLI/CUI Agent-Style Validation + +Primary command suites: + +```bash +pytest -q src/tests/test_cli*.py src/tests/test_typer_cli.py src/tests/test_cli_agent_mode.py +``` + +This covers user CLI commands and agent command flows. + +## API Agent-Style Validation + +Primary command suites: + +```bash +pytest -q src/tests/test_api*.py +``` + +This covers auth, policy enforcement, entry/document operations, and vault +endpoints from an API client perspective. + +## TUI v2 Deep Coverage Command + +To track TUI v2 function/path coverage: + +```bash +pytest -q \ + src/tests/test_tui_v2_helpers.py \ + src/tests/test_tui_v2_textual_interactions.py \ + src/tests/test_tui_v2_parity_scenarios.py \ + src/tests/test_tui_v2_large_vault_validation.py \ + src/tests/test_tui_v2_action_matrix.py \ + --cov=seedpass.tui_v2.app \ + --cov-report=term-missing +``` + +## Determinism Gate Integration + +Run deterministic suite: + +```bash +./scripts/run_determinism_tests.sh +``` + +This includes `scripts/check_determinism_suite.py` to enforce suite shape and +minimum determinism test count before executing tests. diff --git a/memory-update.md b/memory-update.md index 2d700c86..022d2cd3 100644 --- a/memory-update.md +++ b/memory-update.md @@ -206,3 +206,37 @@ - UI copy now emphasizes current selection/filter context while keeping deterministic action affordances unchanged; full TUI-focused test slice remains green. - 2026-03-01: Reduced TUI v2 status noise by deduplicating repeated status messages before appending to status bar/activity log. - Verified critical coverage gate against existing full coverage artifact (`artifacts/coverage/coverage.json`): all default threshold modules pass (main/core manager/core encryption/core migrations). + +## 2026-03-01 Determinism test infrastructure hardening +- Added a dedicated determinism lane with `scripts/run_determinism_tests.sh` and wired `scripts/run_ci_tests.sh` to run it by default (`DETERMINISM_GATE=1`, disable with `DETERMINISM_GATE=0`). +- Added pytest marker support for determinism (`pyproject.toml`, `src/tests/conftest.py`) and a `--determinism-only` selector that skips non-determinism tests. +- Expanded deterministic regression vectors in `src/tests/test_deterministic_artifact_regression.py`: + - pinned entropy-stream SHA-256 vectors for indexes `0,1,352,1024` + - pinned password outputs for policy variants (default, safe specials, no specials, exclude ambiguous, restricted special charset) +- Marked existing deterministic suites so they run in the determinism lane: password helpers/policy plus seed, SSH, PGP, and Nostr determinism tests. +- Verified determinism lane now executes 23 focused tests and fails fast on derivation drift. + +## 2026-03-01 Determinism gates + cross-process + size-tier sync tests +- Added `scripts/check_determinism_suite.py` and wired it into `scripts/run_determinism_tests.sh`. +- Determinism gate now enforces both a minimum test count (`DETERMINISM_MIN_TESTS`, default 25) and required determinism files, preventing accidental suite erosion. +- Added unit tests for determinism gate parser logic in `src/tests/test_check_determinism_suite.py`. +- Added cross-process determinism regression in `src/tests/test_cross_process_determinism.py`; derives artifacts in separate Python processes and asserts exact JSON identity. +- Expanded size-tier sync integrity coverage: + - `src/tests/test_document_file_io.py` includes large document import/export and portable backup restore invariance checks (hash/equality). + - `src/tests/test_full_sync_roundtrip.py` includes parameterized large document Nostr sync roundtrip with content hash checks. + - `src/tests/test_nostr_snapshot.py` includes chunk hash and gzip roundtrip integrity across multiple payload size tiers. +- Current determinism lane status after updates: 43 passed, 916 skipped. + +## 2026-03-01 TUI v2 agent harness + deep coverage expansion +- Added `scripts/ai_tui2_agent_test.py` as a deterministic agent-style TUI v2 harness using Textual pilot mode; emits timestamped JSON reports in `artifacts/agent_tui2_test/`. +- Added test/process documentation: `docs/agent_test_format.md` with repeatable commands for TUI v2, legacy interactive harness, CLI/CUI, API, and determinism gate runs. +- Fixed legacy harness invalid-input regression after Document type addition: in `scripts/ai_tui_agent_test.py`, Add Entry invalid choice now uses `99` instead of `9`. +- Added `src/tests/test_tui_v2_action_matrix.py` to exercise broad TUI v2 action and palette paths (including guards/failure/retry/event handlers). +- Fixed a real TUI v2 help-overlay bug in `src/seedpass/tui_v2/app.py` where `[/]` text triggered Textual markup parsing errors (`MarkupError`). +- Latest TUI v2 focused coverage run: + - Command: test helper/interactions/parity/large-vault/action-matrix suites with `--cov=seedpass.tui_v2.app` + - Result: 25 passed, 2 skipped + - Coverage: 86% for `src/seedpass/tui_v2/app.py` (up from ~57%). +- Harness validation results: + - `scripts/ai_tui2_agent_test.py --scenario extended`: passed + - `scripts/ai_tui_agent_test.py --scenario core`: passed diff --git a/pyproject.toml b/pyproject.toml index 5b618bab..a7f3c5cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,6 +75,7 @@ mypy_path = "src" [tool.pytest.ini_options] markers = [ "network: tests that require network access", + "determinism: tests that lock deterministic artifact derivation behavior", ] filterwarnings = [ "ignore:'imghdr' is deprecated and slated for removal in Python 3.13:DeprecationWarning:pgpy.constants", diff --git a/scripts/ai_tui2_agent_test.py b/scripts/ai_tui2_agent_test.py new file mode 100755 index 00000000..85f16b50 --- /dev/null +++ b/scripts/ai_tui2_agent_test.py @@ -0,0 +1,368 @@ +#!/usr/bin/env python3 +""" +Deterministic AI-agent harness for SeedPass TUI v2. + +This harness uses Textual's test pilot API to simulate user/agent behavior and +produces a machine-readable report for CI and local regressions. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import time +from dataclasses import dataclass, asdict +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +from seedpass.tui_v2.app import check_tui2_runtime, launch_tui2 + + +@dataclass +class StepResult: + step_id: str + description: str + status: str + duration_sec: float + detail: str | None = None + + +class HarnessError(RuntimeError): + pass + + +class HarnessEntryService: + def __init__(self) -> None: + self.entries = { + 1: { + "id": 1, + "kind": "document", + "label": "Runbook", + "content": "alpha", + "file_type": "md", + "tags": ["ops"], + "archived": False, + }, + 2: { + "id": 2, + "kind": "password", + "label": "Console Login", + "archived": False, + }, + } + self.links = { + 1: [{"target": 2, "relation": "references", "note": "credential"}], + 2: [], + } + + def search_entries(self, query: str, kinds: list[str] | None = None): + q = (query or "").strip().lower() + out = [] + for idx in sorted(self.entries.keys()): + entry = self.entries[idx] + kind = str(entry.get("kind", "password")) + if kinds and kind not in kinds: + continue + label = str(entry.get("label", "")) + if q and q not in label.lower(): + continue + out.append( + ( + idx, + label, + None, + None, + bool(entry.get("archived", False)), + SimpleNamespace(value=kind), + ) + ) + return out + + def retrieve_entry(self, entry_id: int): + return dict(self.entries.get(int(entry_id), {})) + + def modify_entry(self, entry_id: int, **kwargs): + entry = self.entries[int(entry_id)] + for key, value in kwargs.items(): + if value is not None: + entry[key] = value + + def archive_entry(self, entry_id: int): + self.entries[int(entry_id)]["archived"] = True + + def restore_entry(self, entry_id: int): + self.entries[int(entry_id)]["archived"] = False + + def get_links(self, entry_id: int): + return [dict(link) for link in self.links.get(int(entry_id), [])] + + def add_link( + self, + entry_id: int, + target_id: int, + *, + relation: str = "related_to", + note: str = "", + ): + links = self.links.setdefault(int(entry_id), []) + links.append({"target": int(target_id), "relation": relation, "note": note}) + return [dict(link) for link in links] + + def remove_link( + self, entry_id: int, target_id: int, *, relation: str | None = None + ): + src = self.links.setdefault(int(entry_id), []) + kept = [] + for link in src: + if int(link.get("target", -1)) != int(target_id): + kept.append(link) + continue + if relation is not None and str(link.get("relation")) != relation: + kept.append(link) + self.links[int(entry_id)] = kept + return [dict(link) for link in kept] + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run deterministic TUI v2 agent validation harness." + ) + parser.add_argument( + "--scenario", + choices=("core", "extended"), + default="extended", + help="Validation scenario profile (default: extended).", + ) + parser.add_argument( + "--output-dir", + type=Path, + default=Path("artifacts/agent_tui2_test"), + help="Directory for report artifacts.", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Print step progress.", + ) + return parser.parse_args() + + +def _status(app) -> str: + return str(app.query_one("#status").render()) + + +def _assert_contains(haystack: str, needle: str, context: str) -> None: + if needle not in haystack: + raise HarnessError(f"{context}: expected {needle!r} in {haystack!r}") + + +async def _scenario_core( + app, service: HarnessEntryService, coverage: dict[str, bool], pilot +) -> None: + from textual.widgets import Input + + await pilot.pause() + coverage["startup"] = True + + app.query_one("#search", Input).value = "runbook" + app._load_entries(query="runbook", reset_page=True) + await pilot.pause() + coverage["search"] = True + + app._run_palette_command("open 1") + await pilot.pause() + app.action_edit_document() + await pilot.pause() + app.query_one("#doc-edit-label", Input).value = "Runbook Updated" + app.query_one("#doc-edit-file-type", Input).value = "txt" + app.query_one("#doc-edit-tags", Input).value = "ops,core" + if len(app.query("#doc-edit-content")) > 0: + area = app.query_one("#doc-edit-content") + if hasattr(area, "load_text"): + area.load_text("alpha\nbeta") + else: + area.text = "alpha\nbeta" + else: + app.query_one("#doc-edit-content-single", Input).value = "alpha\nbeta" + app.action_save_document() + await pilot.pause() + _assert_contains(_status(app), "Saved document", "save-doc") + coverage["document_edit"] = True + + app._run_palette_command("link-add 2 references auth") + await pilot.pause() + app.action_open_link_target() + await pilot.pause() + _assert_contains(_status(app), "Opened linked entry 2", "link-open") + coverage["graph_navigation"] = True + + app._run_palette_command("open 1") + await pilot.pause() + app.action_toggle_archive() + await pilot.pause() + if "archived" not in _status(app): + raise HarnessError(f"archive: expected archived status, got {_status(app)!r}") + app._run_palette_command("open 1") + await pilot.pause() + app.action_toggle_archive() + await pilot.pause() + if service.entries[1].get("archived", False): + raise HarnessError( + "restore: expected entry 1 to be unarchived after second toggle" + ) + coverage["archive_restore"] = True + + +async def _scenario_extended( + app, service: HarnessEntryService, coverage: dict[str, bool], pilot +) -> None: + await _scenario_core(app, service, coverage, pilot) + app.action_open_palette() + await pilot.pause() + app._run_palette_command("unknown-cmd") + await pilot.pause() + _assert_contains(_status(app), "Unknown command", "palette-unknown") + + app._run_palette_command("page 1") + await pilot.pause() + coverage["palette_navigation"] = True + + app.action_toggle_help() + await pilot.pause() + _assert_contains(_status(app), "Help opened", "help-open") + app.action_cancel_document_edit() + await pilot.pause() + _assert_contains(_status(app), "Help closed", "help-close") + coverage["help_overlay"] = True + + +async def _run_scenario( + app, service: HarnessEntryService, coverage: dict[str, bool], scenario: str +) -> None: + async with app.run_test() as pilot: + if scenario == "core": + await _scenario_core(app, service, coverage, pilot) + else: + await _scenario_extended(app, service, coverage, pilot) + + +def _run_step( + step_id: str, + description: str, + fn, + results: list[StepResult], + *, + verbose: bool = False, +): + start = time.monotonic() + try: + asyncio.run(fn()) + except Exception as exc: + duration = time.monotonic() - start + results.append( + StepResult( + step_id=step_id, + description=description, + status="failed", + duration_sec=duration, + detail=str(exc), + ) + ) + raise + duration = time.monotonic() - start + results.append( + StepResult( + step_id=step_id, + description=description, + status="passed", + duration_sec=duration, + ) + ) + if verbose: + print(f"[PASS] {step_id} ({duration:.2f}s)") + + +def main() -> int: + args = _parse_args() + runtime = check_tui2_runtime() + if not runtime.get("textual_available", False): + print("Error: TUI v2 runtime unavailable. Install textual.") + return 2 + + artifact_dir = args.output_dir / datetime.now(timezone.utc).strftime( + "%Y%m%dT%H%M%SZ" + ) + artifact_dir.mkdir(parents=True, exist_ok=True) + + service = HarnessEntryService() + holder: dict[str, object] = {} + + def _hook(app): + holder["app"] = app + + launched = launch_tui2(entry_service_factory=lambda: service, app_hook=_hook) + if not launched: + print("Error: failed to launch TUI v2 app") + return 2 + app = holder.get("app") + if app is None: + print("Error: app hook did not receive app") + return 2 + + coverage = { + "startup": False, + "search": False, + "document_edit": False, + "graph_navigation": False, + "archive_restore": False, + "palette_navigation": False, + "help_overlay": False, + } + results: list[StepResult] = [] + start = time.monotonic() + status = "passed" + failure = "" + try: + if args.scenario == "core": + _run_step( + "core", + "Drive core TUI v2 user/agent workflow.", + lambda: _run_scenario(app, service, coverage, "core"), + results, + verbose=args.verbose, + ) + else: + _run_step( + "extended", + "Drive extended TUI v2 user/agent workflow.", + lambda: _run_scenario(app, service, coverage, "extended"), + results, + verbose=args.verbose, + ) + except Exception as exc: + status = "failed" + failure = str(exc) + + report = { + "timestamp_utc": datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"), + "scenario": args.scenario, + "status": status, + "failure": failure, + "duration_sec": round(time.monotonic() - start, 3), + "coverage_points": coverage, + "steps": [asdict(item) for item in results], + "activity_log": list(getattr(app, "_activity_log", [])), + } + report_file = artifact_dir / "report.json" + report_file.write_text(json.dumps(report, indent=2), encoding="utf-8") + print(f"AI agent TUI v2 report: {report_file}") + print(f"Status: {status}") + if failure: + print(f"Failure: {failure}") + return 0 if status == "passed" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ai_tui_agent_test.py b/scripts/ai_tui_agent_test.py index b7880347..9f67fd4c 100755 --- a/scripts/ai_tui_agent_test.py +++ b/scripts/ai_tui_agent_test.py @@ -358,7 +358,7 @@ def _invalid_input_checks( timeout: float, *, main_menu_choice: str = "99", - add_entry_choice: str = "9", + add_entry_choice: str = "99", retrieve_index: str = "abc", ) -> None: runner.sendline(main_menu_choice) diff --git a/scripts/check_determinism_suite.py b/scripts/check_determinism_suite.py new file mode 100644 index 00000000..6309a13e --- /dev/null +++ b/scripts/check_determinism_suite.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +"""Validate determinism test-suite shape before running CI-heavy jobs.""" + +from __future__ import annotations + +import argparse +import shutil +import subprocess +import sys +from pathlib import Path + +DEFAULT_REQUIRED_FILES = [ + "src/tests/test_cross_process_determinism.py", + "src/tests/test_deterministic_artifact_regression.py", + "src/tests/test_document_file_io.py", + "src/tests/test_full_sync_roundtrip.py", + "src/tests/test_nostr_snapshot.py", + "src/tests/test_password_generation_policy.py", + "src/tests/test_password_helpers.py", + "src/tests/test_seed_entry.py", + "src/tests/test_ssh_entry.py", + "src/tests/test_pgp_entry.py", + "src/tests/test_nostr_entry.py", +] + + +def _extract_nodeids(collected_output: str) -> list[str]: + nodeids: list[str] = [] + for line in collected_output.splitlines(): + raw = line.strip() + if "::" not in raw: + continue + if raw.startswith("<") or raw.startswith("="): + continue + nodeids.append(raw) + return nodeids + + +def _extract_file_set(nodeids: list[str]) -> set[str]: + files: set[str] = set() + for nodeid in nodeids: + files.add(nodeid.split("::", 1)[0].replace("\\", "/")) + return files + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Ensure determinism tests remain comprehensive." + ) + parser.add_argument( + "--pytest-bin", + default="pytest", + help="pytest executable to use (default: pytest).", + ) + parser.add_argument( + "--min-tests", + type=int, + default=25, + help="Minimum required collected tests with marker 'determinism'.", + ) + parser.add_argument( + "--tests-root", + default="src/tests", + help="Root test path for pytest collection (default: src/tests).", + ) + parser.add_argument( + "--require-file", + action="append", + default=[], + help="Required file that must contribute at least one determinism test.", + ) + parser.add_argument( + "--no-default-required-files", + action="store_true", + help="Disable built-in required file checks.", + ) + return parser.parse_args() + + +def main() -> int: + args = _parse_args() + pytest_bin = shutil.which(args.pytest_bin) if args.pytest_bin else None + if pytest_bin is None: + print(f"Error: pytest binary not found: {args.pytest_bin}", file=sys.stderr) + return 2 + + cmd = [ + pytest_bin, + "--collect-only", + "-q", + "-m", + "determinism", + args.tests_root, + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + if result.returncode != 0: + print("Error: failed to collect determinism tests.", file=sys.stderr) + if result.stdout: + print(result.stdout, file=sys.stderr) + if result.stderr: + print(result.stderr, file=sys.stderr) + return 2 + + nodeids = _extract_nodeids(result.stdout) + files = _extract_file_set(nodeids) + required_files = ( + [] if args.no_default_required_files else list(DEFAULT_REQUIRED_FILES) + ) + required_files.extend(args.require_file) + required_files = [f.replace("\\", "/") for f in required_files] + missing_files = sorted([f for f in required_files if f not in files]) + + print(f"Determinism tests collected: {len(nodeids)}") + if files: + print(f"Determinism files present: {len(files)}") + + if len(nodeids) < args.min_tests: + print( + f"FAIL: determinism suite below minimum test count ({len(nodeids)} < {args.min_tests})", + file=sys.stderr, + ) + return 1 + + if missing_files: + print( + "FAIL: determinism suite missing required files:\n- " + + "\n- ".join(missing_files), + file=sys.stderr, + ) + return 1 + + print("Determinism suite gate passed.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_ci_tests.sh b/scripts/run_ci_tests.sh index 6441cbc9..e6f292d9 100755 --- a/scripts/run_ci_tests.sh +++ b/scripts/run_ci_tests.sh @@ -3,6 +3,10 @@ set -eo pipefail mkdir -p artifacts/coverage +if [[ "${DETERMINISM_GATE:-1}" == "1" ]]; then + ./scripts/run_determinism_tests.sh +fi + pytest_args=(-vv) if [[ -n "${STRESS_ARGS:-}" ]]; then pytest_args+=(${STRESS_ARGS}) diff --git a/scripts/run_determinism_tests.sh b/scripts/run_determinism_tests.sh new file mode 100755 index 00000000..d62157e4 --- /dev/null +++ b/scripts/run_determinism_tests.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +set -eo pipefail + +timeout_bin="timeout" +if ! command -v "$timeout_bin" >/dev/null 2>&1; then + if command -v gtimeout >/dev/null 2>&1; then + timeout_bin="gtimeout" + else + timeout_bin="" + fi +fi + +pytest_args=( + -q + --maxfail=1 + --determinism-only + src/tests +) + +pytest_bin="pytest" +if ! command -v "$pytest_bin" >/dev/null 2>&1; then + if [[ -x ".venv/bin/pytest" ]]; then + pytest_bin=".venv/bin/pytest" + else + echo "pytest not found on PATH and .venv/bin/pytest is missing" >&2 + exit 2 + fi +fi + +if [[ -n "$timeout_bin" ]]; then + $timeout_bin 2m python3 scripts/check_determinism_suite.py \ + --pytest-bin "$pytest_bin" \ + --min-tests "${DETERMINISM_MIN_TESTS:-25}" + $timeout_bin 10m "$pytest_bin" "${pytest_args[@]}" +else + echo "timeout command not found; running determinism tests without timeout" >&2 + python3 scripts/check_determinism_suite.py \ + --pytest-bin "$pytest_bin" \ + --min-tests "${DETERMINISM_MIN_TESTS:-25}" + "$pytest_bin" "${pytest_args[@]}" +fi diff --git a/src/seedpass/tui_v2/app.py b/src/seedpass/tui_v2/app.py index c7fe823d..0fb6fc97 100644 --- a/src/seedpass/tui_v2/app.py +++ b/src/seedpass/tui_v2/app.py @@ -325,7 +325,7 @@ def launch_tui2( "", "Core : / search j jump-id p/n page r refresh", "Modes : Ctrl+P palette e edit-doc Ctrl+S save Esc cancel/close", - "Graph : l relation filter [/] link select o open link target", + "Graph : l relation filter brackets link select o open link target", "Resilience: x retry last error", "Pane Focus: 1 left 2 center 3 right", "", diff --git a/src/tests/conftest.py b/src/tests/conftest.py index 9035d926..0241af68 100644 --- a/src/tests/conftest.py +++ b/src/tests/conftest.py @@ -76,23 +76,37 @@ def pytest_addoption(parser: pytest.Parser) -> None: default=None, help="maximum entries for nostr index size test", ) + parser.addoption( + "--determinism-only", + action="store_true", + default=False, + help="run only determinism regression tests", + ) def pytest_configure(config: pytest.Config) -> None: config.addinivalue_line("markers", "stress: long running stress tests") config.addinivalue_line("markers", "desktop: desktop only tests") + config.addinivalue_line( + "markers", "determinism: locks deterministic derivation behavior" + ) def pytest_collection_modifyitems( config: pytest.Config, items: list[pytest.Item] ) -> None: - if config.getoption("--stress"): + if config.getoption("--determinism-only"): + skip_non_determinism = pytest.mark.skip(reason="running determinism-only suite") + for item in items: + if "determinism" not in item.keywords: + item.add_marker(skip_non_determinism) return - skip_stress = pytest.mark.skip(reason="need --stress option to run") - for item in items: - if "stress" in item.keywords: - item.add_marker(skip_stress) + if not config.getoption("--stress"): + skip_stress = pytest.mark.skip(reason="need --stress option to run") + for item in items: + if "stress" in item.keywords: + item.add_marker(skip_stress) if not config.getoption("--desktop"): skip_desktop = pytest.mark.skip(reason="need --desktop option to run") diff --git a/src/tests/test_check_determinism_suite.py b/src/tests/test_check_determinism_suite.py new file mode 100644 index 00000000..4dadaf13 --- /dev/null +++ b/src/tests/test_check_determinism_suite.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + + +def _load_module(): + repo_root = Path(__file__).resolve().parents[2] + script_path = repo_root / "scripts" / "check_determinism_suite.py" + spec = importlib.util.spec_from_file_location( + "check_determinism_suite", script_path + ) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +gate = _load_module() + + +def test_extract_nodeids_filters_noise(): + output = """ + +src/tests/test_seed_entry.py::test_seed_phrase_determinism +============================ test session starts ============================ +src/tests/test_pgp_entry.py::test_pgp_key_determinism +""" + nodeids = gate._extract_nodeids(output) + assert nodeids == [ + "src/tests/test_seed_entry.py::test_seed_phrase_determinism", + "src/tests/test_pgp_entry.py::test_pgp_key_determinism", + ] + + +def test_extract_file_set_normalizes_windows_paths(): + nodeids = [ + r"src\tests\test_seed_entry.py::test_seed_phrase_determinism", + "src/tests/test_pgp_entry.py::test_pgp_key_determinism", + ] + files = gate._extract_file_set(nodeids) + assert "src/tests/test_seed_entry.py" in files + assert "src/tests/test_pgp_entry.py" in files diff --git a/src/tests/test_cross_process_determinism.py b/src/tests/test_cross_process_determinism.py new file mode 100644 index 00000000..842e6482 --- /dev/null +++ b/src/tests/test_cross_process_determinism.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.determinism + +TEST_SEED = ( + "abandon abandon abandon abandon abandon abandon abandon abandon abandon " + "abandon abandon about" +) + + +def _run_derivation_subprocess(index: int) -> dict[str, str]: + repo_root = Path(__file__).resolve().parents[2] + env = dict(os.environ) + pythonpath = env.get("PYTHONPATH", "") + src_path = str(repo_root / "src") + env["PYTHONPATH"] = f"{src_path}:{pythonpath}" if pythonpath else src_path + + code = f""" +import hashlib +import json +from bip_utils import Bip39SeedGenerator +from local_bip85.bip85 import BIP85 +from nostr.coincurve_keys import Keys +from seedpass.core.password_generation import PasswordGenerator, derive_seed_phrase, derive_ssh_key_pair +from seedpass.core.totp import TotpManager + +seed = "{TEST_SEED}" +index = {index} + +class _MnemonicSeedEncoder: + @staticmethod + def derive_seed_from_mnemonic(mnemonic: str) -> bytes: + return Bip39SeedGenerator(mnemonic).Generate() + +seed_bytes = Bip39SeedGenerator(seed).Generate() +bip85 = BIP85(seed_bytes) +pg = PasswordGenerator(_MnemonicSeedEncoder(), seed, bip85) +pw = pg.generate_password(length=24, index=index) +ssh_priv, ssh_pub = derive_ssh_key_pair(seed, index) +seed24 = derive_seed_phrase(bip85, index, 24) +totp_secret = TotpManager.derive_secret(seed, index) +totp_code = TotpManager.current_code(seed, index, timestamp=1700000000) +entropy = bip85.derive_entropy(index=index, entropy_bytes=32) +nostr_keys = Keys(priv_k=entropy.hex()) +npub = Keys.hex_to_bech32(nostr_keys.public_key_hex(), "npub") +nsec = Keys.hex_to_bech32(nostr_keys.private_key_hex(), "nsec") + +print(json.dumps({{ + "password": pw, + "ssh_priv_sha256": hashlib.sha256(ssh_priv.encode("utf-8")).hexdigest(), + "ssh_pub_sha256": hashlib.sha256(ssh_pub.encode("utf-8")).hexdigest(), + "seed24": seed24, + "totp_secret": totp_secret, + "totp_code_at_1700000000": totp_code, + "npub": npub, + "nsec": nsec, +}}, sort_keys=True)) +""" + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + cwd=str(repo_root), + env=env, + check=True, + ) + return json.loads(result.stdout.strip()) + + +@pytest.mark.parametrize("index", [0, 352, 1024]) +def test_artifacts_are_identical_across_processes(index: int): + first = _run_derivation_subprocess(index) + second = _run_derivation_subprocess(index) + assert first == second diff --git a/src/tests/test_deterministic_artifact_regression.py b/src/tests/test_deterministic_artifact_regression.py index 2615ac9d..3cb2fc6c 100644 --- a/src/tests/test_deterministic_artifact_regression.py +++ b/src/tests/test_deterministic_artifact_regression.py @@ -1,11 +1,13 @@ import hashlib from dataclasses import dataclass +import pytest from bip_utils import Bip39SeedGenerator from local_bip85.bip85 import BIP85 from nostr.coincurve_keys import Keys from seedpass.core.password_generation import ( + PasswordPolicy, PasswordGenerator, derive_pgp_key, derive_seed_phrase, @@ -18,6 +20,8 @@ TEST_SEED = ( "abandon abandon about" ) +pytestmark = pytest.mark.determinism + @dataclass(frozen=True) class DeterministicVector: @@ -104,10 +108,12 @@ class _MnemonicSeedEncoder: return Bip39SeedGenerator(mnemonic).Generate() -def _build_password_generator(seed_phrase: str) -> PasswordGenerator: +def _build_password_generator( + seed_phrase: str, policy: PasswordPolicy | None = None +) -> PasswordGenerator: seed_bytes = Bip39SeedGenerator(seed_phrase).Generate() bip85 = BIP85(seed_bytes) - return PasswordGenerator(_MnemonicSeedEncoder(), seed_phrase, bip85) + return PasswordGenerator(_MnemonicSeedEncoder(), seed_phrase, bip85, policy=policy) def test_deterministic_vectors_index_352_regression(): @@ -177,3 +183,82 @@ def test_repeatability_for_same_seed_and_index(): totp_secret_1 = TotpManager.derive_secret(TEST_SEED, vector.index) totp_secret_2 = TotpManager.derive_secret(TEST_SEED, vector.index) assert totp_secret_1 == totp_secret_2 + + +@pytest.mark.parametrize( + "index,expected_sha256", + [ + (0, "4e4df93e6a35f03eb5108f7bfca41f4469a13c287876cfb1b0c6c4fa48d4026f"), + (1, "c28902c1ba33cacab86d2fdc4524cee594809277337f9e77a4124e6946c3ae80"), + (352, "85ff7ed5b2be3277f822fb0c141b35ba8636badbd9adec089531172d689fa815"), + ( + 1024, + "834a72f2fbea70964239bbc085c6ba4262032b961542333e2eef5ccdc05214b4", + ), + ], +) +def test_password_entropy_stream_hash_regression(index: int, expected_sha256: str): + """Pin entropy stream outputs for known indexes to catch derivation drift.""" + generator = _build_password_generator(TEST_SEED) + stream = generator._derive_password_entropy(index=index) + assert hashlib.sha256(stream).hexdigest() == expected_sha256 + + +@pytest.mark.parametrize( + "policy,index,length,expected_password", + [ + ( + PasswordPolicy(), + 352, + 24, + "$oEEz0$606DE2g9o(%$wH;Uu", + ), + ( + PasswordPolicy(special_mode="safe"), + 352, + 24, + "o66U^o!g%D0w!6E2E=ZH6uw!", + ), + ( + PasswordPolicy(include_special_chars=False, min_special=0), + 352, + 24, + "0E6jwUo92EZ96GDoeuoHiU56", + ), + ( + PasswordPolicy(exclude_ambiguous=True), + 17, + 20, + "59ik9Z3k=UJW_%)tZm_6", + ), + ( + PasswordPolicy( + allowed_special_chars="@#", + min_uppercase=2, + min_lowercase=2, + min_digits=2, + min_special=2, + ), + 99, + 22, + "x@N#M82#Jwf#7b8u99NM5@", + ), + ], + ids=[ + "default", + "safe_special_mode", + "no_special_characters", + "exclude_ambiguous", + "restricted_special_charset", + ], +) +def test_password_policy_vector_regression( + policy: PasswordPolicy, + index: int, + length: int, + expected_password: str, +): + """Lock policy-specific password outputs to prevent accidental behavior changes.""" + generator = _build_password_generator(TEST_SEED, policy=policy) + password = generator.generate_password(length=length, index=index) + assert password == expected_password diff --git a/src/tests/test_document_file_io.py b/src/tests/test_document_file_io.py index 50f96432..33b735e9 100644 --- a/src/tests/test_document_file_io.py +++ b/src/tests/test_document_file_io.py @@ -1,10 +1,15 @@ from pathlib import Path from tempfile import TemporaryDirectory +import hashlib +import pytest from helpers import TEST_PASSWORD, TEST_SEED, create_vault from seedpass.core.backup import BackupManager from seedpass.core.config_manager import ConfigManager from seedpass.core.entry_management import EntryManager +from seedpass.core.portable_backup import export_backup, import_backup + +pytestmark = pytest.mark.determinism def _make_entry_manager(tmp_path: Path) -> EntryManager: @@ -14,6 +19,24 @@ def _make_entry_manager(tmp_path: Path) -> EntryManager: return EntryManager(vault, backup_mgr) +def _make_managers(tmp_path: Path) -> tuple[EntryManager, BackupManager]: + vault, _ = create_vault(tmp_path, TEST_SEED, TEST_PASSWORD) + cfg_mgr = ConfigManager(vault, tmp_path) + backup_mgr = BackupManager(tmp_path, cfg_mgr) + return EntryManager(vault, backup_mgr), backup_mgr + + +def _deterministic_text_bytes(size: int) -> str: + """Return deterministic ASCII text of at least *size* bytes.""" + block = ( + "SeedPass deterministic document payload block. " + "0123456789 abcdefghijklmnopqrstuvwxyz\n" + ) + repeats = (size // len(block)) + 1 + text = block * repeats + return text[:size] + + def test_import_document_file_roundtrip(): with TemporaryDirectory() as tmpdir: base = Path(tmpdir) @@ -43,3 +66,90 @@ def test_export_document_file_writes_expected_content(): assert dest.exists() assert dest.suffix == ".txt" assert dest.read_text(encoding="utf-8") == "line1\nline2" + + +@pytest.mark.parametrize("size_bytes", [1024, 512 * 1024, 1024 * 1024]) +def test_import_export_large_document_roundtrip_is_byte_stable(size_bytes: int): + with TemporaryDirectory() as tmpdir: + base = Path(tmpdir) + entry_mgr = _make_entry_manager(base) + content = _deterministic_text_bytes(size_bytes) + source = base / "large_doc.md" + source.write_text(content, encoding="utf-8") + + idx = entry_mgr.import_document_file(source, tags=["large", "determinism"]) + entry = entry_mgr.retrieve_entry(idx) + assert entry is not None + assert entry["file_type"] == "md" + assert entry["content"] == content + + out_dir = base / "out" + exported = entry_mgr.export_document_file(idx, output_path=out_dir) + exported_content = exported.read_text(encoding="utf-8") + assert exported_content == content + assert ( + hashlib.sha256(exported_content.encode("utf-8")).hexdigest() + == hashlib.sha256(content.encode("utf-8")).hexdigest() + ) + + +@pytest.mark.parametrize("encrypt", [True, False], ids=["encrypted", "plaintext"]) +def test_portable_backup_restore_preserves_document_content_and_extension( + encrypt: bool, +): + with TemporaryDirectory() as tmpdir: + base = Path(tmpdir) + entry_mgr, backup_mgr = _make_managers(base) + content = _deterministic_text_bytes(256 * 1024) + + idx = entry_mgr.add_document( + "Ops Runbook", + content, + file_type="py", + notes="critical", + tags=["ops", "runbook"], + ) + before = entry_mgr.retrieve_entry(idx) + assert before is not None + + export_path = base / "portable_export.json" + export_backup( + entry_mgr.vault, + backup_mgr, + export_path, + parent_seed=TEST_SEED, + encrypt=encrypt, + ) + + entry_mgr.modify_entry( + idx, + label="mutated", + content="mutated content", + file_type="txt", + notes="changed", + tags=["changed"], + ) + + import_backup( + entry_mgr.vault, + backup_mgr, + export_path, + parent_seed=TEST_SEED, + ) + entry_mgr.clear_cache() + after = entry_mgr.retrieve_entry(idx) + assert after == before + + +def test_backup_manager_restore_latest_preserves_large_document_content(): + with TemporaryDirectory() as tmpdir: + base = Path(tmpdir) + entry_mgr, backup_mgr = _make_managers(base) + content = _deterministic_text_bytes(300 * 1024) + idx = entry_mgr.add_document("Design Doc", content, file_type="txt") + before = entry_mgr.retrieve_entry(idx) + assert before is not None + + backup_mgr.restore_latest_backup() + restored = entry_mgr.retrieve_entry(idx) + assert restored == before diff --git a/src/tests/test_full_sync_roundtrip.py b/src/tests/test_full_sync_roundtrip.py index c3a46312..143bfbaf 100644 --- a/src/tests/test_full_sync_roundtrip.py +++ b/src/tests/test_full_sync_roundtrip.py @@ -1,7 +1,9 @@ import asyncio +import hashlib from pathlib import Path from tempfile import TemporaryDirectory +import pytest from helpers import create_vault from seedpass.core.entry_management import EntryManager @@ -9,6 +11,8 @@ from seedpass.core.backup import BackupManager from seedpass.core.config_manager import ConfigManager from seedpass.core.manager import PasswordManager, EncryptionMode +pytestmark = pytest.mark.determinism + def _init_pm(dir_path: Path, client) -> PasswordManager: vault, enc_mgr = create_vault(dir_path) @@ -29,6 +33,15 @@ def _init_pm(dir_path: Path, client) -> PasswordManager: return pm +def _deterministic_text_bytes(size: int) -> str: + block = ( + "SeedPass sync payload block for deterministic Nostr/backup tests. " + "0123456789 abcdefghijklmnopqrstuvwxyz\n" + ) + repeats = (size // len(block)) + 1 + return (block * repeats)[:size] + + def test_full_sync_roundtrip(dummy_nostr_client): client, relay = dummy_nostr_client with TemporaryDirectory() as tmpdir: @@ -66,7 +79,8 @@ def test_full_sync_roundtrip(dummy_nostr_client): assert sorted(labels) == ["site1", "site2"] -def test_large_document_sync_roundtrip(dummy_nostr_client): +@pytest.mark.parametrize("size_bytes", [64 * 1024, 512 * 1024, 2 * 1024 * 1024]) +def test_large_document_sync_roundtrip(dummy_nostr_client, size_bytes: int): client, relay = dummy_nostr_client with TemporaryDirectory() as tmpdir: base = Path(tmpdir) @@ -78,7 +92,8 @@ def test_large_document_sync_roundtrip(dummy_nostr_client): pm_a = _init_pm(dir_a, client) pm_b = _init_pm(dir_b, client) - large_content = "line\n" * 120_000 + large_content = _deterministic_text_bytes(size_bytes) + expected_hash = hashlib.sha256(large_content.encode("utf-8")).hexdigest() doc_id = pm_a.entry_manager.add_document( "LargeDoc", large_content, @@ -95,4 +110,7 @@ def test_large_document_sync_roundtrip(dummy_nostr_client): assert restored is not None assert restored["kind"] == "document" assert restored["file_type"] == "txt" - assert restored["content"] == large_content + assert ( + hashlib.sha256(restored["content"].encode("utf-8")).hexdigest() + == expected_hash + ) diff --git a/src/tests/test_nostr_entry.py b/src/tests/test_nostr_entry.py index 53509d1c..e37093db 100644 --- a/src/tests/test_nostr_entry.py +++ b/src/tests/test_nostr_entry.py @@ -1,6 +1,7 @@ import sys from pathlib import Path from tempfile import TemporaryDirectory +import pytest from helpers import create_vault, TEST_SEED, TEST_PASSWORD @@ -13,6 +14,8 @@ from seedpass.core.backup import BackupManager from seedpass.core.vault import Vault from seedpass.core.config_manager import ConfigManager +pytestmark = pytest.mark.determinism + def test_nostr_key_determinism(): with TemporaryDirectory() as tmpdir: diff --git a/src/tests/test_nostr_snapshot.py b/src/tests/test_nostr_snapshot.py index 684f27d0..58262307 100644 --- a/src/tests/test_nostr_snapshot.py +++ b/src/tests/test_nostr_snapshot.py @@ -8,12 +8,21 @@ import os import asyncio import sys from unittest.mock import patch +import pytest sys.path.append(str(Path(__file__).resolve().parents[1])) from nostr.client import prepare_snapshot, NostrClient from seedpass.core.encryption import EncryptionManager +pytestmark = pytest.mark.determinism + + +def _deterministic_bytes(size: int) -> bytes: + block = b"SeedPass snapshot chunk payload 0123456789 abcdefghijklmnopqrstuvwxyz\n" + repeats = (size // len(block)) + 1 + return (block * repeats)[:size] + def test_prepare_snapshot_roundtrip(): data = b"a" * 70000 @@ -26,6 +35,30 @@ def test_prepare_snapshot_roundtrip(): assert data == gzip.decompress(joined) +@pytest.mark.parametrize( + "size_bytes,limit", + [ + (64 * 1024, 16 * 1024), + (512 * 1024, 64 * 1024), + (2 * 1024 * 1024, 128 * 1024), + ], +) +def test_prepare_snapshot_chunk_hash_integrity_by_size_tier( + size_bytes: int, limit: int +): + payload = _deterministic_bytes(size_bytes) + manifest, chunks = prepare_snapshot(payload, limit) + assert len(chunks) == len(manifest.chunks) + assert len(chunks) >= 1 + + for meta, chunk in zip(manifest.chunks, chunks): + assert hashlib.sha256(chunk).hexdigest() == meta.hash + assert meta.size == len(chunk) + + joined = b"".join(chunks) + assert gzip.decompress(joined) == payload + + class DummyEvent: def __init__(self, content): self._content = content diff --git a/src/tests/test_password_generation_policy.py b/src/tests/test_password_generation_policy.py index 62a20953..b905163a 100644 --- a/src/tests/test_password_generation_policy.py +++ b/src/tests/test_password_generation_policy.py @@ -1,11 +1,14 @@ import string from pathlib import Path import sys +import pytest sys.path.append(str(Path(__file__).resolve().parents[1])) from seedpass.core.password_generation import PasswordGenerator, PasswordPolicy +pytestmark = pytest.mark.determinism + class DummyEnc: def derive_seed_from_mnemonic(self, mnemonic): diff --git a/src/tests/test_password_helpers.py b/src/tests/test_password_helpers.py index fb80683f..8fa13d88 100644 --- a/src/tests/test_password_helpers.py +++ b/src/tests/test_password_helpers.py @@ -1,6 +1,9 @@ import string +import pytest from seedpass.core.password_generation import PasswordGenerator, PasswordPolicy +pytestmark = pytest.mark.determinism + class DummyEnc: def derive_seed_from_mnemonic(self, mnemonic): diff --git a/src/tests/test_pgp_entry.py b/src/tests/test_pgp_entry.py index 9cb7c28a..69bce9bb 100644 --- a/src/tests/test_pgp_entry.py +++ b/src/tests/test_pgp_entry.py @@ -1,6 +1,7 @@ import sys from pathlib import Path from tempfile import TemporaryDirectory +import pytest from helpers import create_vault, TEST_SEED, TEST_PASSWORD @@ -10,6 +11,8 @@ from seedpass.core.entry_management import EntryManager from seedpass.core.backup import BackupManager from seedpass.core.config_manager import ConfigManager +pytestmark = pytest.mark.determinism + def test_pgp_key_determinism(): with TemporaryDirectory() as tmpdir: diff --git a/src/tests/test_seed_entry.py b/src/tests/test_seed_entry.py index 305083b4..7b1d1f86 100644 --- a/src/tests/test_seed_entry.py +++ b/src/tests/test_seed_entry.py @@ -1,6 +1,7 @@ import sys from pathlib import Path from tempfile import TemporaryDirectory +import pytest from helpers import create_vault, TEST_SEED, TEST_PASSWORD from mnemonic import Mnemonic @@ -14,6 +15,8 @@ from seedpass.core.password_generation import derive_seed_phrase from local_bip85.bip85 import BIP85 from bip_utils import Bip39SeedGenerator +pytestmark = pytest.mark.determinism + def test_seed_phrase_determinism(): with TemporaryDirectory() as tmpdir: diff --git a/src/tests/test_ssh_entry.py b/src/tests/test_ssh_entry.py index f2fee6c1..23902c8a 100644 --- a/src/tests/test_ssh_entry.py +++ b/src/tests/test_ssh_entry.py @@ -1,6 +1,7 @@ import sys from pathlib import Path from tempfile import TemporaryDirectory +import pytest from helpers import create_vault, TEST_SEED, TEST_PASSWORD @@ -11,6 +12,8 @@ from seedpass.core.backup import BackupManager from seedpass.core.vault import Vault from seedpass.core.config_manager import ConfigManager +pytestmark = pytest.mark.determinism + def test_add_and_retrieve_ssh_key_pair(): with TemporaryDirectory() as tmpdir: diff --git a/src/tests/test_tui_v2_action_matrix.py b/src/tests/test_tui_v2_action_matrix.py new file mode 100644 index 00000000..5be5c0b2 --- /dev/null +++ b/src/tests/test_tui_v2_action_matrix.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +pytest.importorskip("textual") +from textual.widgets import Input, ListView + +from seedpass.tui_v2.app import launch_tui2 + + +class MatrixEntryService: + def __init__( + self, + entries: list[dict], + *, + fail_search: bool = False, + fail_links: bool = False, + fail_archive: bool = False, + fail_restore: bool = False, + fail_modify: bool = False, + fail_add_link: bool = False, + fail_remove_link: bool = False, + ) -> None: + self._entries = {int(entry["id"]): dict(entry) for entry in entries} + self._links = { + int(entry["id"]): [dict(link) for link in entry.get("links", [])] + for entry in entries + } + self.fail_search = fail_search + self.fail_links = fail_links + self.fail_archive = fail_archive + self.fail_restore = fail_restore + self.fail_modify = fail_modify + self.fail_add_link = fail_add_link + self.fail_remove_link = fail_remove_link + + def search_entries(self, query: str, kinds: list[str] | None = None): + if self.fail_search: + raise RuntimeError("search failed") + q = (query or "").strip().lower() + rows = [] + for entry_id in sorted(self._entries.keys()): + entry = self._entries[entry_id] + kind = str(entry.get("kind", "password")) + if kinds and kind not in kinds: + continue + label = str(entry.get("label", "")) + if q and q not in label.lower(): + continue + rows.append( + ( + entry_id, + label, + None, + None, + bool(entry.get("archived", False)), + SimpleNamespace(value=kind), + ) + ) + return rows + + def retrieve_entry(self, entry_id: int): + if int(entry_id) not in self._entries: + return {} + return dict(self._entries[int(entry_id)]) + + def modify_entry(self, entry_id: int, **kwargs) -> None: + if self.fail_modify: + raise RuntimeError("modify failed") + entry = self._entries[int(entry_id)] + for key, value in kwargs.items(): + if value is not None: + entry[key] = value + + def archive_entry(self, entry_id: int) -> None: + if self.fail_archive: + raise RuntimeError("archive failed") + self._entries[int(entry_id)]["archived"] = True + + def restore_entry(self, entry_id: int) -> None: + if self.fail_restore: + raise RuntimeError("restore failed") + self._entries[int(entry_id)]["archived"] = False + + def get_links(self, entry_id: int): + if self.fail_links: + raise RuntimeError("links failed") + return [dict(link) for link in self._links.get(int(entry_id), [])] + + def add_link( + self, + entry_id: int, + target_id: int, + *, + relation: str = "related_to", + note: str = "", + ): + if self.fail_add_link: + raise RuntimeError("link add failed") + links = self._links.setdefault(int(entry_id), []) + links.append({"target": int(target_id), "relation": relation, "note": note}) + return [dict(link) for link in links] + + def remove_link( + self, entry_id: int, target_id: int, *, relation: str | None = None + ): + if self.fail_remove_link: + raise RuntimeError("link remove failed") + src = self._links.setdefault(int(entry_id), []) + keep = [] + for link in src: + if int(link.get("target", -1)) != int(target_id): + keep.append(link) + continue + if relation is not None and str(link.get("relation")) != relation: + keep.append(link) + self._links[int(entry_id)] = keep + return [dict(link) for link in keep] + + +def _build_app(*, service: MatrixEntryService | None = None, factory=None): + holder: dict[str, object] = {} + + def _hook(app): + holder["app"] = app + + launched = launch_tui2( + entry_service_factory=factory if factory is not None else (lambda: service), + app_hook=_hook, + ) + assert launched is True + app = holder.get("app") + assert app is not None + return app + + +def _status(app) -> str: + return str(app.query_one("#status").render()) + + +@pytest.mark.anyio +async def test_tui2_matrix_init_retry_failure_and_success() -> None: + calls = {"n": 0} + svc = MatrixEntryService([{"id": 1, "kind": "document", "label": "Doc"}]) + + def _factory(): + calls["n"] += 1 + if calls["n"] < 2: + raise RuntimeError("boom") + return svc + + app = _build_app(factory=_factory) + async with app.run_test() as pilot: + await pilot.pause() + assert "Unable to initialize entry service" in _status(app) + app.action_retry_last_error() + await pilot.pause() + assert any( + "Entry service initialized" in msg + for msg in getattr(app, "_activity_log", []) + ) + list_view = app.query_one("#entry-list", ListView) + assert len(list_view.children) == 1 + + +@pytest.mark.anyio +async def test_tui2_matrix_service_unavailable_and_no_retry() -> None: + app = _build_app(service=None) + async with app.run_test() as pilot: + await pilot.pause() + assert "Entry service unavailable" in str( + app.query_one("#entry-detail").render() + ) + app.action_retry_last_error() + await pilot.pause() + assert "No retry action available" in _status(app) + + +@pytest.mark.anyio +async def test_tui2_matrix_actions_palette_events_and_guards() -> None: + entries = [ + { + "id": 1, + "kind": "document", + "label": "Doc One", + "content": "line", + "file_type": "md", + "tags": ["a"], + "links": [ + {"target": 2, "relation": "references", "note": "a-note"}, + {"target": 3, "relation": "contains", "note": ""}, + ], + }, + {"id": 2, "kind": "password", "label": "Pw Two"}, + {"id": 3, "kind": "key_value", "label": "KV Three"}, + ] + [ + {"id": i, "kind": "document", "label": f"Doc {i}", "content": "x"} + for i in range(4, 235) + ] + svc = MatrixEntryService(entries) + app = _build_app(service=svc) + + async with app.run_test() as pilot: + await pilot.pause() + + app.action_toggle_help() + await pilot.pause() + assert "Help opened" in _status(app) + app.action_open_palette() + await pilot.pause() + assert "Palette opened" in _status(app) + app.action_cancel_document_edit() + await pilot.pause() + assert "Palette closed" in _status(app) + app.action_cancel_document_edit() + await pilot.pause() + assert app.help_open is False + + app._run_palette_command("") + app._run_palette_command("open") + app._run_palette_command("open abc") + app._run_palette_command("jump") + app._run_palette_command("jump abc") + app._run_palette_command("page") + app._run_palette_command("page abc") + app._run_palette_command("page 0") + app._run_palette_command("link-filter") + app._run_palette_command("link-add") + app._run_palette_command("link-add no") + app._run_palette_command("link-rm") + app._run_palette_command("link-rm no") + app._run_palette_command("filter") + app._run_palette_command("unknown-cmd") + await pilot.pause() + + app._run_palette_command("help") + app._run_palette_command("search Doc") + app._run_palette_command("filter document") + app._run_palette_command("page-next") + app._run_palette_command("page-prev") + app._run_palette_command("page 2") + app._run_palette_command("open 1") + app._run_palette_command("jump 2") + await pilot.pause() + + app.action_focus_left() + app.action_focus_center() + app.action_focus_right() + app.action_focus_search() + app.action_focus_jump() + await pilot.pause() + + app.action_next_page() + app.action_prev_page() + await pilot.pause() + + app._run_palette_command("open 1") + await pilot.pause() + app.action_cycle_link_filter() + app.link_relation_filter = "all" + app._update_links_panel() + app.action_next_link() + app.action_prev_link() + app.action_open_link_target() + await pilot.pause() + assert "Opened linked entry" in _status(app) + + app._current_links = [{"target": "bad", "relation": "references", "note": ""}] + app._current_link_cursor = 0 + app.action_open_link_target() + await pilot.pause() + assert "Invalid link target" in _status(app) + + app.action_toggle_archive() + await pilot.pause() + assert "archived" in _status(app) or "restored" in _status(app) + app._run_palette_command("archive") + app._run_palette_command("restore") + await pilot.pause() + + app._run_palette_command("open 1") + app.action_edit_document() + await pilot.pause() + app.action_focus_search() + app.action_focus_jump() + app.action_cycle_link_filter() + app.action_open_palette() + app.action_toggle_archive() + await pilot.pause() + + app.query_one("#doc-edit-label", Input).value = "Doc Updated" + app.query_one("#doc-edit-file-type", Input).value = "txt" + app.query_one("#doc-edit-tags", Input).value = "a,b" + if len(app.query("#doc-edit-content")) > 0: + area = app.query_one("#doc-edit-content") + if hasattr(area, "load_text"): + area.load_text("new-content") + else: + area.text = "new-content" + else: + app.query_one("#doc-edit-content-single", Input).value = "new-content" + + app.on_input_changed( + SimpleNamespace(input=SimpleNamespace(id="doc-edit-label")) + ) + app.on_text_area_changed(None) + app.action_save_document() + await pilot.pause() + assert "Saved document" in _status(app) + + app._run_palette_command("open 1") + app.action_edit_document() + await pilot.pause() + app.action_save_document() + await pilot.pause() + assert "No document changes to save" in _status(app) + app.action_cancel_document_edit() + await pilot.pause() + + app.on_input_submitted( + SimpleNamespace(input=SimpleNamespace(id="quick-jump"), value="") + ) + app.on_input_submitted( + SimpleNamespace(input=SimpleNamespace(id="quick-jump"), value="x") + ) + app.on_input_submitted( + SimpleNamespace(input=SimpleNamespace(id="quick-jump"), value="2") + ) + app.on_input_submitted( + SimpleNamespace(input=SimpleNamespace(id="search"), value="Doc One") + ) + app.on_input_submitted( + SimpleNamespace(input=SimpleNamespace(id="command-palette"), value="help") + ) + await pilot.pause() + + app._run_palette_command("filter all") + app._run_palette_command("search Doc") + await pilot.pause() + first = app.query_one("#entry-list", ListView).children[0] + app.on_list_view_selected(SimpleNamespace(item=first)) + await pilot.pause() + + app._run_palette_command("retry") + await pilot.pause() + + +@pytest.mark.anyio +async def test_tui2_matrix_failure_paths_and_retry() -> None: + svc = MatrixEntryService( + [ + {"id": 1, "kind": "document", "label": "Doc", "content": "x"}, + {"id": 2, "kind": "password", "label": "Pw"}, + ], + fail_links=True, + fail_archive=True, + fail_modify=True, + fail_add_link=True, + fail_remove_link=True, + ) + app = _build_app(service=svc) + async with app.run_test() as pilot: + await pilot.pause() + app._show_entry(1) + await pilot.pause() + assert "Links unavailable:" in str(app.query_one("#link-detail").render()) + + app.action_retry_last_error() + await pilot.pause() + assert "retry" in _status(app).lower() or "failed" in _status(app).lower() + + app._run_palette_command("link-add 2 references nope") + await pilot.pause() + assert "link-add failed" in _status(app) + + app._run_palette_command("link-rm 2 references") + await pilot.pause() + assert "link-rm failed" in _status(app) + + app.action_toggle_archive() + await pilot.pause() + assert "Archive/restore failed" in _status(app) + + app.action_edit_document() + await pilot.pause() + app.query_one("#doc-edit-label", Input).value = "Doc2" + app.action_save_document() + await pilot.pause() + assert "Failed to save document" in _status(app)