mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2026-10-27 07:52:03 +00:00
Add agent mode, hardened TUI harness, coverage gates, and cross-platform installer smoke CI
This commit delivers four related improvements focused on reliability and automation while preserving existing human/TUI workflows. 1) Agent-first CLI mode (additive) - Added new `seedpass agent` command group (`src/seedpass/cli/agent.py`) and wired it into CLI root. - Implemented non-interactive profile bootstrap: `seedpass agent init` with `--seed|--seed-file|--generate-seed` and password env injection (`--password-env`). - Added policy management commands: `agent policy-show` and `agent policy-set` with allow/deny kind controls. - Added JSON-based secret retrieval: `seedpass -f <fingerprint> agent get <query> --ttl <sec>` with lease metadata and policy enforcement. - Kept regular human mode fully available; no behavioral changes to interactive TUI entrypoint. 2) TUI agent harness deepening and stabilization - Refactored harness into scenario profiles: `core`, `extended`, `stress` with `--scenario` selection. - Added deterministic seeded stress campaigns: `--stress-cycles`, `--stress-seed`. - Added scenario-specific required coverage keys and report metadata. - Added post-condition assertions in extended scenario (archive/restore consistency, lock/unlock retrieval recovery). - Hardened prompt matching for output variants and fail-fast behavior on non-zero SeedPass exit codes. - Added harness unit tests for helper logic and control flow. 3) Critical module coverage gate tooling - Added `scripts/check_critical_coverage.py` to enforce per-module thresholds from coverage JSON and report uncovered functions. - Added override controls (`--threshold`, `--no-default-thresholds`) and optional JSON result output. - Integrated coverage JSON generation into CI helper script and optional gate execution via `CRITICAL_COVERAGE_GATE=1`. - Tuned default thresholds using full-suite baseline and increased strictness: - `src/main.py >= 50%` - `src/seedpass/core/manager.py >= 60%` - `src/seedpass/core/encryption.py >= 80%` - `src/seedpass/core/migrations.py >= 90%` - Added tests for coverage gate parser/analysis logic and targeted manager worker path tests. 4) Installer validation framework for beta/main - Added cross-platform installer smoke scripts: - `scripts/installer_smoke_unix.sh` - `scripts/installer_smoke_windows.ps1` - Added GitHub Actions workflow `.github/workflows/installer-smoke.yml` with: - reduced PR matrix - full scheduled/manual matrix across OS/branch/mode - per-job log capture and artifact uploads for failure triage - Added installer testing documentation and linked from README. Additional housekeeping - Added `.venv/` to `.gitignore`. - Updated docs for new harness scenarios, seeded stress mode, post-condition metadata, coverage gate usage, and installer testing workflow.
This commit is contained in:
1 parent
dd7290e83b
commit
03958106e1
16 files changed
+1826
-54
No files matched your search
+375
-48
@@ -18,6 +18,7 @@ import argparse
|
||||
import json
|
||||
import os
|
||||
import pty
|
||||
import random
|
||||
import re
|
||||
import select
|
||||
import shutil
|
||||
@@ -142,6 +143,24 @@ def _parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Run deterministic, repeatable TUI coverage checks for SeedPass."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--scenario",
|
||||
choices=("core", "extended", "stress"),
|
||||
default="extended",
|
||||
help="Scenario profile to run (default: extended).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--stress-cycles",
|
||||
type=int,
|
||||
default=6,
|
||||
help="Deterministic menu-stress loops for --scenario stress.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--stress-seed",
|
||||
type=int,
|
||||
default=1337,
|
||||
help="RNG seed for deterministic invalid-input campaigns in stress scenario.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--repo-root",
|
||||
type=Path,
|
||||
@@ -289,7 +308,7 @@ def _add_entry_menu_open(
|
||||
)
|
||||
|
||||
|
||||
def _scenario(
|
||||
def _complete_initial_seed_setup(
|
||||
runner: TUIRunner, timeout: float, password: str, coverage: dict[str, bool]
|
||||
) -> None:
|
||||
# First-run onboarding.
|
||||
@@ -302,34 +321,26 @@ def _scenario(
|
||||
coverage["startup_onboarding"] = True
|
||||
coverage["main_menu"] = True
|
||||
|
||||
# Invalid input resilience checks.
|
||||
runner.sendline("99")
|
||||
runner.wait_for(r"Invalid choice\. Please select a valid option\.", timeout)
|
||||
_add_entry_menu_open(runner, timeout, "9")
|
||||
runner.wait_for(r"Invalid choice\.", timeout)
|
||||
runner.expect_and_send(r"Select entry type or press Enter to go back:", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
runner.sendline("2")
|
||||
runner.expect_and_send(
|
||||
r"Enter the index number of the entry to retrieve:", "abc", timeout
|
||||
)
|
||||
runner.wait_for(r"Error: Index must be a number\.", timeout)
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
coverage["invalid_input_resilience"] = True
|
||||
|
||||
# Add password in quick mode, including validation checks.
|
||||
def _add_quick_password_entry(
|
||||
runner: TUIRunner,
|
||||
timeout: float,
|
||||
*,
|
||||
label: str,
|
||||
username: str,
|
||||
url: str,
|
||||
exercise_invalid_length: bool,
|
||||
coverage: dict[str, bool],
|
||||
) -> None:
|
||||
_add_entry_menu_open(runner, timeout, "1")
|
||||
runner.expect_and_send(r"Choose mode: \[Q\]uick or \[A\]dvanced\?", "q", timeout)
|
||||
runner.expect_and_send(
|
||||
r"Enter the label or website name:", "agent.example", timeout
|
||||
)
|
||||
runner.expect_and_send(r"Enter the username \(optional\):", "agent", timeout)
|
||||
runner.expect_and_send(
|
||||
r"Enter the URL \(optional\):", "https://agent.example", timeout
|
||||
)
|
||||
runner.expect_and_send(r"Enter desired password length .*:", "3", timeout)
|
||||
runner.wait_for(r"Password length must be between 8 and 128", timeout)
|
||||
coverage["invalid_length_validation"] = True
|
||||
runner.expect_and_send(r"Enter the label or website name:", label, timeout)
|
||||
runner.expect_and_send(r"Enter the username \(optional\):", username, timeout)
|
||||
runner.expect_and_send(r"Enter the URL \(optional\):", url, timeout)
|
||||
if exercise_invalid_length:
|
||||
runner.expect_and_send(r"Enter desired password length .*:", "3", timeout)
|
||||
runner.wait_for(r"Password length must be between 8 and 128", timeout)
|
||||
coverage["invalid_length_validation"] = True
|
||||
runner.expect_and_send(r"Enter desired password length .*:", "12", timeout)
|
||||
runner.expect_and_send(r"Include special characters\? \(Y/n\):", "", timeout)
|
||||
runner.wait_for(r"Password generated and indexed with ID 0", timeout)
|
||||
@@ -337,12 +348,214 @@ def _scenario(
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
|
||||
def _invalid_input_checks(
|
||||
runner: TUIRunner,
|
||||
timeout: float,
|
||||
*,
|
||||
main_menu_choice: str = "99",
|
||||
add_entry_choice: str = "9",
|
||||
retrieve_index: str = "abc",
|
||||
) -> None:
|
||||
runner.sendline(main_menu_choice)
|
||||
runner.wait_for(r"Invalid choice\. Please select a valid option\.", timeout)
|
||||
_add_entry_menu_open(runner, timeout, add_entry_choice)
|
||||
runner.wait_for(r"Invalid choice\.", timeout)
|
||||
runner.expect_and_send(r"Select entry type or press Enter to go back:", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
runner.sendline("2")
|
||||
runner.expect_and_send(
|
||||
r"Enter the index number of the entry to retrieve:", retrieve_index, timeout
|
||||
)
|
||||
runner.wait_for(r"Error: Index must be a number\.", timeout)
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
|
||||
def _wait_for_retrieve_password_output(
|
||||
runner: TUIRunner, timeout: float, label: str
|
||||
) -> None:
|
||||
runner.wait_for(
|
||||
rf"(Retrieved Password for {re.escape(label)}|Retrieving password for '{re.escape(label)}')",
|
||||
timeout,
|
||||
)
|
||||
|
||||
|
||||
def _scenario_core(
|
||||
runner: TUIRunner,
|
||||
timeout: float,
|
||||
password: str,
|
||||
coverage: dict[str, bool],
|
||||
post_conditions: dict[str, bool],
|
||||
) -> None:
|
||||
_complete_initial_seed_setup(runner, timeout, password, coverage)
|
||||
|
||||
# Invalid input resilience checks.
|
||||
_invalid_input_checks(runner, timeout)
|
||||
coverage["invalid_input_resilience"] = True
|
||||
|
||||
# Add a password and exercise length validation.
|
||||
_add_quick_password_entry(
|
||||
runner,
|
||||
timeout,
|
||||
label="agent.example",
|
||||
username="agent",
|
||||
url="https://agent.example",
|
||||
exercise_invalid_length=True,
|
||||
coverage=coverage,
|
||||
)
|
||||
|
||||
# Retrieve and action menu.
|
||||
runner.sendline("2")
|
||||
runner.expect_and_send(
|
||||
r"Enter the index number of the entry to retrieve:", "0", timeout
|
||||
)
|
||||
runner.wait_for(r"Retrieved Password for agent\.example", timeout)
|
||||
_wait_for_retrieve_password_output(runner, timeout, "agent.example")
|
||||
if runner.try_wait_for(r"Reveal hidden fields\? \(y/N\):", timeout=0.4):
|
||||
runner.sendline("")
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
runner.wait_for(r"Entry Actions:", timeout)
|
||||
coverage["retrieve_entry"] = True
|
||||
runner.expect_and_send(r"Select an action or press Enter to return:", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
# Search.
|
||||
runner.sendline("3")
|
||||
runner.expect_and_send(r"Enter search string:", "agent", timeout)
|
||||
runner.wait_for(r"Search Results", timeout)
|
||||
runner.wait_for(r"0\. Password - agent\.example", timeout)
|
||||
coverage["search_entries"] = True
|
||||
runner.expect_and_send(
|
||||
r"Enter index to view details or press Enter to go back:", "", timeout
|
||||
)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
# List entries.
|
||||
runner.sendline("4")
|
||||
runner.expect_and_send(
|
||||
r"Select entry type or press Enter to go back:", "1", timeout
|
||||
)
|
||||
runner.wait_for(r"\[\+\] Entries:", timeout)
|
||||
runner.wait_for(r"0\. Password - agent\.example", timeout)
|
||||
coverage["list_entries"] = True
|
||||
runner.expect_and_send(
|
||||
r"Enter index to view details or press Enter to go back:", "", timeout
|
||||
)
|
||||
runner.expect_and_send(r"Select entry type or press Enter to go back:", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
# Graceful exit.
|
||||
runner.sendline("")
|
||||
runner.wait_for(r"Exiting the program\.", timeout)
|
||||
coverage["graceful_exit"] = True
|
||||
|
||||
|
||||
def _scenario_stress(
|
||||
runner: TUIRunner,
|
||||
timeout: float,
|
||||
password: str,
|
||||
coverage: dict[str, bool],
|
||||
post_conditions: dict[str, bool],
|
||||
*,
|
||||
cycles: int,
|
||||
seed: int,
|
||||
) -> None:
|
||||
_complete_initial_seed_setup(runner, timeout, password, coverage)
|
||||
|
||||
_add_quick_password_entry(
|
||||
runner,
|
||||
timeout,
|
||||
label="stress.example",
|
||||
username="stress",
|
||||
url="https://stress.example",
|
||||
exercise_invalid_length=False,
|
||||
coverage=coverage,
|
||||
)
|
||||
|
||||
rng = random.Random(seed)
|
||||
invalid_main_menu_inputs = ["99", "0", "x", "-1", "one", "1x"]
|
||||
invalid_add_entry_inputs = ["9", "0", "x", "-1", "one", "1x"]
|
||||
invalid_retrieve_inputs = ["abc", "!", "one", "1x", "index", "0x1"]
|
||||
|
||||
# Deterministic negative-input + back-navigation stress cycles.
|
||||
for _ in range(cycles):
|
||||
_invalid_input_checks(
|
||||
runner,
|
||||
timeout,
|
||||
main_menu_choice=rng.choice(invalid_main_menu_inputs),
|
||||
add_entry_choice=rng.choice(invalid_add_entry_inputs),
|
||||
retrieve_index=rng.choice(invalid_retrieve_inputs),
|
||||
)
|
||||
|
||||
runner.sendline("4")
|
||||
runner.expect_and_send(
|
||||
r"Select entry type or press Enter to go back:", "1", timeout
|
||||
)
|
||||
runner.wait_for(r"\[\+\] Entries:", timeout)
|
||||
runner.wait_for(r"0\. Password - stress\.example", timeout)
|
||||
runner.expect_and_send(
|
||||
r"Enter index to view details or press Enter to go back:", "", timeout
|
||||
)
|
||||
runner.expect_and_send(
|
||||
r"Select entry type or press Enter to go back:", "", timeout
|
||||
)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
runner.sendline("2")
|
||||
runner.expect_and_send(
|
||||
r"Enter the index number of the entry to retrieve:", "0", timeout
|
||||
)
|
||||
_wait_for_retrieve_password_output(runner, timeout, "stress.example")
|
||||
if runner.try_wait_for(r"Reveal hidden fields\? \(y/N\):", timeout=0.4):
|
||||
runner.sendline("")
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
runner.wait_for(r"Entry Actions:", timeout)
|
||||
runner.expect_and_send(
|
||||
r"Select an action or press Enter to return:", "", timeout
|
||||
)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
|
||||
coverage["invalid_input_resilience"] = True
|
||||
coverage["retrieve_entry"] = True
|
||||
coverage["list_entries"] = True
|
||||
coverage["stress_cycles"] = True
|
||||
coverage["seeded_negative_campaign"] = True
|
||||
|
||||
runner.sendline("")
|
||||
runner.wait_for(r"Exiting the program\.", timeout)
|
||||
coverage["graceful_exit"] = True
|
||||
|
||||
|
||||
def _scenario_extended(
|
||||
runner: TUIRunner,
|
||||
timeout: float,
|
||||
password: str,
|
||||
coverage: dict[str, bool],
|
||||
post_conditions: dict[str, bool],
|
||||
) -> None:
|
||||
_complete_initial_seed_setup(runner, timeout, password, coverage)
|
||||
|
||||
# Invalid input resilience checks.
|
||||
_invalid_input_checks(runner, timeout)
|
||||
coverage["invalid_input_resilience"] = True
|
||||
|
||||
# Add password in quick mode, including validation checks.
|
||||
_add_quick_password_entry(
|
||||
runner,
|
||||
timeout,
|
||||
label="agent.example",
|
||||
username="agent",
|
||||
url="https://agent.example",
|
||||
exercise_invalid_length=True,
|
||||
coverage=coverage,
|
||||
)
|
||||
|
||||
# Retrieve and action menu.
|
||||
runner.sendline("2")
|
||||
runner.expect_and_send(
|
||||
r"Enter the index number of the entry to retrieve:", "0", timeout
|
||||
)
|
||||
_wait_for_retrieve_password_output(runner, timeout, "agent.example")
|
||||
if runner.try_wait_for(r"Reveal hidden fields\? \(y/N\):", timeout=0.4):
|
||||
runner.sendline("")
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
@@ -541,6 +754,23 @@ def _scenario(
|
||||
runner.wait_for(r"Entry at index 0 modified successfully", timeout)
|
||||
coverage["archive_restore"] = True
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
runner.sendline("8")
|
||||
if runner.try_wait_for(r"Archived Entries", timeout=0.8):
|
||||
if runner.try_wait_for(r"0\. agent\.example", timeout=0.4):
|
||||
raise TUIHarnessError(
|
||||
"Archive/restore post-condition failed: restored entry is still archived."
|
||||
)
|
||||
runner.expect_and_send(
|
||||
r"Enter index to manage or press Enter to go back:", "", timeout
|
||||
)
|
||||
elif runner.try_wait_for(r"Press Enter to continue\.\.\.", timeout=0.8):
|
||||
runner.sendline("")
|
||||
else:
|
||||
raise TUIHarnessError(
|
||||
"Archive/restore post-condition failed: archived view prompt was not recognized."
|
||||
)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
post_conditions["archive_restore_consistency"] = True
|
||||
|
||||
# 2FA codes view remains accessible after mixed operations.
|
||||
runner.sendline("6")
|
||||
@@ -589,6 +819,17 @@ def _scenario(
|
||||
runner.expect_and_send(r"Select an option or press Enter to go back:", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
coverage["settings_toggles_lock_unlock"] = True
|
||||
runner.sendline("2")
|
||||
runner.expect_and_send(
|
||||
r"Enter the index number of the entry to retrieve:", "0", timeout
|
||||
)
|
||||
_wait_for_retrieve_password_output(runner, timeout, "agent.example")
|
||||
if runner.try_wait_for(r"Reveal hidden fields\? \(y/N\):", timeout=0.4):
|
||||
runner.sendline("")
|
||||
runner.expect_and_send(r"Press Enter to continue\.\.\.", "", timeout)
|
||||
runner.expect_and_send(r"Select an action or press Enter to return:", "", timeout)
|
||||
_drain_to_main_menu(runner, timeout)
|
||||
post_conditions["lock_unlock_recovers_retrieval"] = True
|
||||
|
||||
# Settings -> Stats and back.
|
||||
runner.sendline("7")
|
||||
@@ -609,6 +850,10 @@ def _scenario(
|
||||
|
||||
def main() -> int:
|
||||
args = _parse_args()
|
||||
if args.stress_cycles < 1:
|
||||
print("Error: --stress-cycles must be >= 1", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
repo_root = args.repo_root.resolve()
|
||||
src_main = repo_root / "src" / "main.py"
|
||||
if not src_main.exists():
|
||||
@@ -640,32 +885,97 @@ def main() -> int:
|
||||
runner = TUIRunner(proc, master_fd)
|
||||
|
||||
results: list[StepResult] = []
|
||||
coverage = {
|
||||
"startup_onboarding": False,
|
||||
"main_menu": False,
|
||||
"invalid_input_resilience": False,
|
||||
"add_password_quick": False,
|
||||
"invalid_length_validation": False,
|
||||
"add_totp": False,
|
||||
"add_all_entry_types": False,
|
||||
"retrieve_entry": False,
|
||||
"search_entries": False,
|
||||
"list_entries": False,
|
||||
"modify_entry": False,
|
||||
"archive_restore": False,
|
||||
"totp_codes_view": False,
|
||||
"settings_toggles_lock_unlock": False,
|
||||
"settings_stats": False,
|
||||
"graceful_exit": False,
|
||||
scenario_coverage_keys: dict[str, list[str]] = {
|
||||
"core": [
|
||||
"startup_onboarding",
|
||||
"main_menu",
|
||||
"invalid_input_resilience",
|
||||
"add_password_quick",
|
||||
"invalid_length_validation",
|
||||
"retrieve_entry",
|
||||
"search_entries",
|
||||
"list_entries",
|
||||
"graceful_exit",
|
||||
],
|
||||
"extended": [
|
||||
"startup_onboarding",
|
||||
"main_menu",
|
||||
"invalid_input_resilience",
|
||||
"add_password_quick",
|
||||
"invalid_length_validation",
|
||||
"add_totp",
|
||||
"add_all_entry_types",
|
||||
"retrieve_entry",
|
||||
"search_entries",
|
||||
"list_entries",
|
||||
"modify_entry",
|
||||
"archive_restore",
|
||||
"totp_codes_view",
|
||||
"settings_toggles_lock_unlock",
|
||||
"settings_stats",
|
||||
"graceful_exit",
|
||||
],
|
||||
"stress": [
|
||||
"startup_onboarding",
|
||||
"main_menu",
|
||||
"add_password_quick",
|
||||
"invalid_input_resilience",
|
||||
"retrieve_entry",
|
||||
"list_entries",
|
||||
"stress_cycles",
|
||||
"seeded_negative_campaign",
|
||||
"graceful_exit",
|
||||
],
|
||||
}
|
||||
scenario_post_condition_keys: dict[str, list[str]] = {
|
||||
"core": [],
|
||||
"extended": [
|
||||
"archive_restore_consistency",
|
||||
"lock_unlock_recovers_retrieval",
|
||||
],
|
||||
"stress": [],
|
||||
}
|
||||
coverage = {
|
||||
key: False
|
||||
for key in sorted({k for keys in scenario_coverage_keys.values() for k in keys})
|
||||
}
|
||||
post_conditions = {
|
||||
key: False
|
||||
for key in sorted(
|
||||
{k for keys in scenario_post_condition_keys.values() for k in keys}
|
||||
)
|
||||
}
|
||||
required_keys = scenario_coverage_keys[args.scenario]
|
||||
required_post_conditions = scenario_post_condition_keys[args.scenario]
|
||||
|
||||
failure: str | None = None
|
||||
start_time = time.monotonic()
|
||||
scenario_handlers = {
|
||||
"core": lambda: _scenario_core(
|
||||
runner, args.timeout, args.password, coverage, post_conditions
|
||||
),
|
||||
"extended": lambda: _scenario_extended(
|
||||
runner, args.timeout, args.password, coverage, post_conditions
|
||||
),
|
||||
"stress": lambda: _scenario_stress(
|
||||
runner,
|
||||
args.timeout,
|
||||
args.password,
|
||||
coverage,
|
||||
post_conditions,
|
||||
cycles=args.stress_cycles,
|
||||
seed=args.stress_seed,
|
||||
),
|
||||
}
|
||||
try:
|
||||
_run_step(
|
||||
"full_scenario",
|
||||
"Drive end-to-end deterministic TUI coverage workflow.",
|
||||
lambda: _scenario(runner, args.timeout, args.password, coverage),
|
||||
f"{args.scenario}_scenario",
|
||||
(
|
||||
"Drive end-to-end deterministic TUI coverage workflow."
|
||||
if args.scenario == "extended"
|
||||
else f"Drive deterministic {args.scenario} TUI scenario."
|
||||
),
|
||||
scenario_handlers[args.scenario],
|
||||
results,
|
||||
verbose=args.verbose,
|
||||
)
|
||||
@@ -674,6 +984,8 @@ def main() -> int:
|
||||
finally:
|
||||
exit_code = runner.close()
|
||||
total_duration = time.monotonic() - start_time
|
||||
if failure is None and exit_code != 0:
|
||||
failure = f"SeedPass exited with non-zero code: {exit_code}"
|
||||
|
||||
transcript_clean = artifact_dir / "transcript.clean.txt"
|
||||
transcript_raw = artifact_dir / "transcript.raw.txt"
|
||||
@@ -685,12 +997,27 @@ def main() -> int:
|
||||
"command": cmd,
|
||||
"repo_root": str(repo_root),
|
||||
"home_dir": str(temp_home_dir),
|
||||
"scenario": args.scenario,
|
||||
"stress_cycles": args.stress_cycles if args.scenario == "stress" else None,
|
||||
"stress_seed": args.stress_seed if args.scenario == "stress" else None,
|
||||
"required_coverage_keys": required_keys,
|
||||
"required_post_conditions": required_post_conditions,
|
||||
"duration_sec": round(total_duration, 3),
|
||||
"seedpass_exit_code": exit_code,
|
||||
"status": "passed" if failure is None and all(coverage.values()) else "failed",
|
||||
"status": (
|
||||
"passed"
|
||||
if (
|
||||
failure is None
|
||||
and all(coverage[key] for key in required_keys)
|
||||
and all(post_conditions[key] for key in required_post_conditions)
|
||||
and exit_code == 0
|
||||
)
|
||||
else "failed"
|
||||
),
|
||||
"failure": failure,
|
||||
"steps": [asdict(item) for item in results],
|
||||
"coverage_points": coverage,
|
||||
"post_conditions": post_conditions,
|
||||
"transcript_clean": str(transcript_clean),
|
||||
"transcript_raw": str(transcript_raw),
|
||||
}
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Enforce coverage thresholds for critical SeedPass modules.
|
||||
|
||||
Reads coverage.py JSON output (pytest --cov-report=json:<path>) and:
|
||||
1. Fails if critical modules are below configured coverage thresholds.
|
||||
2. Reports fully uncovered functions/methods for those modules.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
import json
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
DEFAULT_THRESHOLDS: dict[str, float] = {
|
||||
"src/main.py": 50.0,
|
||||
"src/seedpass/core/manager.py": 60.0,
|
||||
"src/seedpass/core/encryption.py": 80.0,
|
||||
"src/seedpass/core/migrations.py": 90.0,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FunctionCoverage:
|
||||
name: str
|
||||
start_line: int
|
||||
end_line: int
|
||||
covered: bool
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Validate coverage thresholds for critical modules."
|
||||
)
|
||||
parser.add_argument(
|
||||
"coverage_json",
|
||||
type=Path,
|
||||
help="Path to coverage JSON report.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--threshold",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="PATH=PERCENT",
|
||||
help="Override/add threshold mapping, e.g. src/main.py=40",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-default-thresholds",
|
||||
action="store_true",
|
||||
help="Disable built-in threshold set; use only --threshold entries.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--repo-root",
|
||||
type=Path,
|
||||
default=Path(__file__).resolve().parents[1],
|
||||
help="Repository root for resolving module paths.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--json-output",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="Optional path to write machine-readable gate results JSON.",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def _parse_threshold_overrides(values: list[str]) -> dict[str, float]:
|
||||
parsed: dict[str, float] = {}
|
||||
for item in values:
|
||||
if "=" not in item:
|
||||
raise ValueError(
|
||||
f"Invalid --threshold value {item!r}; expected PATH=PERCENT"
|
||||
)
|
||||
path, raw_percent = item.split("=", 1)
|
||||
path = path.strip()
|
||||
if not path:
|
||||
raise ValueError(f"Invalid --threshold value {item!r}; empty path")
|
||||
try:
|
||||
percent = float(raw_percent)
|
||||
except ValueError as exc:
|
||||
raise ValueError(
|
||||
f"Invalid --threshold value {item!r}; percent must be numeric"
|
||||
) from exc
|
||||
if not (0.0 <= percent <= 100.0):
|
||||
raise ValueError(
|
||||
f"Invalid --threshold value {item!r}; percent must be 0..100"
|
||||
)
|
||||
parsed[path] = percent
|
||||
return parsed
|
||||
|
||||
|
||||
def _load_coverage_json(path: Path) -> dict:
|
||||
try:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
except FileNotFoundError as exc:
|
||||
raise RuntimeError(f"Coverage JSON not found: {path}") from exc
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError(f"Invalid coverage JSON: {path}: {exc}") from exc
|
||||
|
||||
|
||||
def _normalize_path(path: str) -> str:
|
||||
return path.replace("\\", "/")
|
||||
|
||||
|
||||
def _find_file_payload(files_payload: dict, module_path: str):
|
||||
module_norm = _normalize_path(module_path)
|
||||
for candidate_key, payload in files_payload.items():
|
||||
candidate_norm = _normalize_path(candidate_key)
|
||||
if candidate_norm.endswith(module_norm):
|
||||
return payload
|
||||
return None
|
||||
|
||||
|
||||
class _FunctionCollector(ast.NodeVisitor):
|
||||
def __init__(self) -> None:
|
||||
self.stack: list[str] = []
|
||||
self.functions: list[tuple[str, int, int]] = []
|
||||
|
||||
def _collect_fn(self, node: ast.AST, fn_name: str) -> None:
|
||||
end_line = getattr(node, "end_lineno", None) or getattr(node, "lineno", 0)
|
||||
full_name = ".".join(self.stack + [fn_name])
|
||||
self.functions.append((full_name, int(node.lineno), int(end_line)))
|
||||
|
||||
def visit_ClassDef(self, node: ast.ClassDef) -> None:
|
||||
self.stack.append(node.name)
|
||||
self.generic_visit(node)
|
||||
self.stack.pop()
|
||||
|
||||
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
|
||||
self._collect_fn(node, node.name)
|
||||
self.generic_visit(node)
|
||||
|
||||
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
|
||||
self._collect_fn(node, node.name)
|
||||
self.generic_visit(node)
|
||||
|
||||
|
||||
def _compute_function_coverage(
|
||||
file_path: Path, executed_lines: set[int]
|
||||
) -> list[FunctionCoverage]:
|
||||
source = file_path.read_text(encoding="utf-8")
|
||||
tree = ast.parse(source)
|
||||
collector = _FunctionCollector()
|
||||
collector.visit(tree)
|
||||
functions: list[FunctionCoverage] = []
|
||||
for name, start_line, end_line in collector.functions:
|
||||
line_range = range(start_line, end_line + 1)
|
||||
covered = any(line in executed_lines for line in line_range)
|
||||
functions.append(
|
||||
FunctionCoverage(
|
||||
name=name,
|
||||
start_line=start_line,
|
||||
end_line=end_line,
|
||||
covered=covered,
|
||||
)
|
||||
)
|
||||
return functions
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = _parse_args()
|
||||
try:
|
||||
overrides = _parse_threshold_overrides(args.threshold)
|
||||
except ValueError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
thresholds = {} if args.no_default_thresholds else dict(DEFAULT_THRESHOLDS)
|
||||
thresholds.update(overrides)
|
||||
if not thresholds:
|
||||
print(
|
||||
"Error: no thresholds configured. Provide --threshold entries.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
||||
try:
|
||||
coverage = _load_coverage_json(args.coverage_json)
|
||||
except RuntimeError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
files_payload = coverage.get("files")
|
||||
if not isinstance(files_payload, dict):
|
||||
print("Error: coverage JSON missing 'files' map", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
repo_root = args.repo_root.resolve()
|
||||
failing_modules: list[str] = []
|
||||
result_payload: dict[str, dict] = {}
|
||||
|
||||
print("Critical module coverage gates:")
|
||||
for module_path, threshold in sorted(thresholds.items()):
|
||||
payload = _find_file_payload(files_payload, module_path)
|
||||
module_result = {
|
||||
"threshold_percent": threshold,
|
||||
"actual_percent": None,
|
||||
"meets_threshold": False,
|
||||
"uncovered_functions": [],
|
||||
"error": None,
|
||||
}
|
||||
if payload is None:
|
||||
msg = "missing from coverage report"
|
||||
print(f"- FAIL {module_path}: {msg}")
|
||||
module_result["error"] = msg
|
||||
failing_modules.append(module_path)
|
||||
result_payload[module_path] = module_result
|
||||
continue
|
||||
|
||||
summary = payload.get("summary", {})
|
||||
percent = float(summary.get("percent_covered", 0.0))
|
||||
module_result["actual_percent"] = percent
|
||||
module_result["meets_threshold"] = percent >= threshold
|
||||
|
||||
module_file = (repo_root / module_path).resolve()
|
||||
try:
|
||||
executed_lines = {int(x) for x in payload.get("executed_lines", [])}
|
||||
functions = _compute_function_coverage(module_file, executed_lines)
|
||||
except Exception as exc: # pragma: no cover - defensive report path
|
||||
module_result["error"] = f"function coverage analysis error: {exc}"
|
||||
functions = []
|
||||
|
||||
uncovered = [
|
||||
{
|
||||
"name": fn.name,
|
||||
"line": fn.start_line,
|
||||
}
|
||||
for fn in functions
|
||||
if not fn.covered
|
||||
]
|
||||
module_result["uncovered_functions"] = uncovered
|
||||
|
||||
status = "PASS" if module_result["meets_threshold"] else "FAIL"
|
||||
print(f"- {status} {module_path}: {percent:.2f}% (threshold {threshold:.2f}%)")
|
||||
if uncovered:
|
||||
preview = ", ".join(
|
||||
f"{item['name']}@L{item['line']}" for item in uncovered[:8]
|
||||
)
|
||||
suffix = " ..." if len(uncovered) > 8 else ""
|
||||
print(f" uncovered functions: {preview}{suffix}")
|
||||
|
||||
if not module_result["meets_threshold"]:
|
||||
failing_modules.append(module_path)
|
||||
result_payload[module_path] = module_result
|
||||
|
||||
output = {
|
||||
"status": "passed" if not failing_modules else "failed",
|
||||
"failing_modules": failing_modules,
|
||||
"modules": result_payload,
|
||||
"coverage_json": str(args.coverage_json.resolve()),
|
||||
}
|
||||
if args.json_output:
|
||||
args.json_output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.json_output.write_text(json.dumps(output, indent=2), encoding="utf-8")
|
||||
|
||||
if failing_modules:
|
||||
print("\nCoverage gate failed for:")
|
||||
for module in failing_modules:
|
||||
print(f"- {module}")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,58 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
IFS=$'\n\t'
|
||||
|
||||
BRANCH="${1:-beta}"
|
||||
MODE="${2:-tui}"
|
||||
|
||||
case "$MODE" in
|
||||
tui|gui|both) ;;
|
||||
*)
|
||||
echo "[ERROR] MODE must be one of: tui, gui, both" >&2
|
||||
exit 2
|
||||
;;
|
||||
esac
|
||||
|
||||
if ! command -v git >/dev/null 2>&1; then
|
||||
echo "[ERROR] git is required for installer smoke test" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
TMP_HOME="$(mktemp -d)"
|
||||
cleanup() {
|
||||
rm -rf "$TMP_HOME"
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
export HOME="$TMP_HOME"
|
||||
unset DISPLAY || true
|
||||
unset WAYLAND_DISPLAY || true
|
||||
|
||||
echo "[INFO] Smoke install start: branch=$BRANCH mode=$MODE home=$HOME"
|
||||
|
||||
bash scripts/install.sh -b "$BRANCH" -m "$MODE"
|
||||
|
||||
LAUNCHER="$HOME/.local/bin/seedpass"
|
||||
APP_DIR="$HOME/.seedpass/app"
|
||||
if [[ ! -x "$LAUNCHER" ]]; then
|
||||
echo "[ERROR] launcher missing: $LAUNCHER" >&2
|
||||
exit 1
|
||||
fi
|
||||
if [[ ! -d "$APP_DIR/.git" ]]; then
|
||||
echo "[ERROR] install dir missing git checkout: $APP_DIR" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
actual_branch="$(git -C "$APP_DIR" rev-parse --abbrev-ref HEAD)"
|
||||
if [[ "$actual_branch" != "$BRANCH" ]]; then
|
||||
echo "[ERROR] expected branch '$BRANCH', got '$actual_branch'" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
"$LAUNCHER" --help >/tmp/seedpass-installer-smoke-help.txt
|
||||
|
||||
echo "[INFO] Re-running installer to validate idempotence"
|
||||
bash scripts/install.sh -b "$BRANCH" -m "$MODE"
|
||||
"$LAUNCHER" --help >/tmp/seedpass-installer-smoke-help-2.txt
|
||||
|
||||
echo "[SUCCESS] Installer smoke test passed (branch=$BRANCH mode=$MODE)"
|
||||
@@ -0,0 +1,58 @@
|
||||
param(
|
||||
[string]$Branch = "beta",
|
||||
[ValidateSet("tui", "gui", "both")]
|
||||
[string]$Mode = "tui"
|
||||
)
|
||||
|
||||
$ErrorActionPreference = "Stop"
|
||||
|
||||
if (-not (Get-Command git -ErrorAction SilentlyContinue)) {
|
||||
throw "git is required for installer smoke test"
|
||||
}
|
||||
|
||||
$TempHome = Join-Path $env:RUNNER_TEMP ("seedpass-installer-home-" + [guid]::NewGuid().ToString("N"))
|
||||
New-Item -ItemType Directory -Path $TempHome -Force | Out-Null
|
||||
|
||||
try {
|
||||
$env:USERPROFILE = $TempHome
|
||||
$env:HOME = $TempHome
|
||||
|
||||
Write-Host "[INFO] Smoke install start: branch=$Branch mode=$Mode home=$TempHome"
|
||||
|
||||
if ($Mode -eq "tui") {
|
||||
& "$PSScriptRoot\install.ps1" -Branch $Branch
|
||||
} else {
|
||||
& "$PSScriptRoot\install.ps1" -Branch $Branch -IncludeGui
|
||||
}
|
||||
|
||||
$LauncherPath = Join-Path $env:USERPROFILE ".seedpass\app\bin\seedpass.cmd"
|
||||
$AppDir = Join-Path $env:USERPROFILE ".seedpass\app"
|
||||
if (-not (Test-Path $LauncherPath)) {
|
||||
throw "launcher missing: $LauncherPath"
|
||||
}
|
||||
if (-not (Test-Path (Join-Path $AppDir ".git"))) {
|
||||
throw "install dir missing git checkout: $AppDir"
|
||||
}
|
||||
|
||||
$ActualBranch = (git -C $AppDir rev-parse --abbrev-ref HEAD).Trim()
|
||||
if ($ActualBranch -ne $Branch) {
|
||||
throw "expected branch '$Branch', got '$ActualBranch'"
|
||||
}
|
||||
|
||||
cmd /c "`"$LauncherPath`" --help" | Out-Null
|
||||
|
||||
Write-Host "[INFO] Re-running installer to validate idempotence"
|
||||
if ($Mode -eq "tui") {
|
||||
& "$PSScriptRoot\install.ps1" -Branch $Branch
|
||||
} else {
|
||||
& "$PSScriptRoot\install.ps1" -Branch $Branch -IncludeGui
|
||||
}
|
||||
cmd /c "`"$LauncherPath`" --help" | Out-Null
|
||||
|
||||
Write-Host "[SUCCESS] Installer smoke test passed (branch=$Branch mode=$Mode)"
|
||||
}
|
||||
finally {
|
||||
if (Test-Path $TempHome) {
|
||||
Remove-Item -Recurse -Force $TempHome
|
||||
}
|
||||
}
|
||||
+13
-1
@@ -1,6 +1,8 @@
|
||||
#!/usr/bin/env bash
|
||||
set -eo pipefail
|
||||
|
||||
mkdir -p artifacts/coverage
|
||||
|
||||
pytest_args=(-vv)
|
||||
if [[ -n "${STRESS_ARGS:-}" ]]; then
|
||||
pytest_args+=(${STRESS_ARGS})
|
||||
@@ -8,7 +10,14 @@ fi
|
||||
if [[ "${RUNNER_OS:-}" == "Windows" ]]; then
|
||||
pytest_args+=(-n 1)
|
||||
fi
|
||||
pytest_args+=(--cov=src --cov-report=xml --cov-report=term-missing --cov-fail-under=20 src/tests)
|
||||
pytest_args+=(
|
||||
--cov=src
|
||||
--cov-report=xml
|
||||
--cov-report=term-missing
|
||||
--cov-report=json:artifacts/coverage/coverage.json
|
||||
--cov-fail-under=20
|
||||
src/tests
|
||||
)
|
||||
|
||||
timeout_bin="timeout"
|
||||
if ! command -v "$timeout_bin" >/dev/null 2>&1; then
|
||||
@@ -33,4 +42,7 @@ if [[ $status -eq 124 ]]; then
|
||||
tail -n 20 pytest.log
|
||||
exit 1
|
||||
fi
|
||||
if [[ $status -eq 0 && "${CRITICAL_COVERAGE_GATE:-0}" == "1" ]]; then
|
||||
python scripts/check_critical_coverage.py artifacts/coverage/coverage.json
|
||||
fi
|
||||
exit $status
|
||||
Reference in new issue
Block a user