busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

tui rework

Browse Source
This commit is contained in:
busya
2026-03-09 14:18:34 +03:00
parent 309dfdba86
commit 4f74bb8afb
8 changed files with 5497 additions and 1520 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
backend/src/scripts/clean_release_tui.py
View File

@@ -8,6 +8,7 @@
# @INVARIANT: TUI must provide a headless fallback for non-TTY environments.
import curses
import json
import os
import sys
import time
@@ -32,51 +33,14 @@ from backend.src.models.clean_release import (
ReleaseCandidate,
ResourceSourceEntry,
ResourceSourceRegistry,
RegistryStatus,
ReleaseCandidateStatus,
)
from backend.src.services.clean_release.compliance_orchestrator import CleanComplianceOrchestrator
from backend.src.services.clean_release.preparation_service import prepare_candidate
from backend.src.services.clean_release.repository import CleanReleaseRepository
from backend.src.services.clean_release.manifest_builder import build_distribution_manifest
class FakeRepository(CleanReleaseRepository):
"""
In-memory stub for the TUI to satisfy Orchestrator without a real DB.
"""
def __init__(self):
super().__init__()
# Seed with demo data for F5 demonstration
now = datetime.now(timezone.utc)
self.save_policy(CleanProfilePolicy(
policy_id="POL-ENT-CLEAN",
policy_version="1",
profile=ProfileType.ENTERPRISE_CLEAN,
active=True,
internal_source_registry_ref="REG-1",
prohibited_artifact_categories=["test-data"],
effective_from=now
))
self.save_registry(ResourceSourceRegistry(
registry_id="REG-1",
name="Default Internal Registry",
entries=[ResourceSourceEntry(
source_id="S1",
host="internal-repo.company.com",
protocol="https",
purpose="artifactory"
)],
updated_at=now,
updated_by="system"
))
self.save_candidate(ReleaseCandidate(
candidate_id="2026.03.03-rc1",
version="1.0.0",
profile=ProfileType.ENTERPRISE_CLEAN,
source_snapshot_ref="v1.0.0-rc1",
created_at=now,
created_by="system"
))
# [DEF:CleanReleaseTUI:Class]
# @PURPOSE: Curses-based application for compliance monitoring.
# @UX_STATE: READY -> Waiting for operator to start checks (F5).
@@ -87,12 +51,15 @@ class FakeRepository(CleanReleaseRepository):
class CleanReleaseTUI:
def __init__(self, stdscr: curses.window):
self.stdscr = stdscr
self.repo = FakeRepository()
self.mode = os.getenv("CLEAN_TUI_MODE", "demo").strip().lower()
self.repo = self._build_repository(self.mode)
self.orchestrator = CleanComplianceOrchestrator(self.repo)
self.candidate_id = self._resolve_candidate_id()
self.status: Any = "READY"
self.checks_progress: List[Dict[str, Any]] = []
self.violations_list: List[ComplianceViolation] = []
self.report_id: Optional[str] = None
self.last_error: Optional[str] = None
curses.start_color()
curses.use_default_colors()
@@ -102,6 +69,124 @@ class CleanReleaseTUI:
curses.init_pair(4, curses.COLOR_YELLOW, -1) # RUNNING
curses.init_pair(5, curses.COLOR_CYAN, -1) # Text
def _build_repository(self, mode: str) -> CleanReleaseRepository:
repo = CleanReleaseRepository()
if mode == "demo":
self._bootstrap_demo_repository(repo)
self._bootstrap_real_repository(repo)
return repo
def _bootstrap_demo_repository(self, repository: CleanReleaseRepository) -> None:
now = datetime.now(timezone.utc)
repository.save_policy(
CleanProfilePolicy(
policy_id="POL-ENT-CLEAN",
policy_version="1",
profile=ProfileType.ENTERPRISE_CLEAN,
active=True,
internal_source_registry_ref="REG-1",
prohibited_artifact_categories=["test-data"],
effective_from=now,
)
)
repository.save_registry(
ResourceSourceRegistry(
registry_id="REG-1",
name="Default Internal Registry",
entries=[
ResourceSourceEntry(
source_id="S1",
host="internal-repo.company.com",
protocol="https",
purpose="artifactory",
)
],
updated_at=now,
updated_by="system",
)
)
repository.save_candidate(
ReleaseCandidate(
candidate_id="2026.03.03-rc1",
version="1.0.0",
profile=ProfileType.ENTERPRISE_CLEAN,
source_snapshot_ref="v1.0.0-rc1",
created_at=now,
created_by="system",
)
)
def _bootstrap_real_repository(self, repository: CleanReleaseRepository) -> None:
bootstrap_path = os.getenv("CLEAN_TUI_BOOTSTRAP_JSON", "").strip()
if not bootstrap_path:
return
with open(bootstrap_path, "r", encoding="utf-8") as bootstrap_file:
payload = json.load(bootstrap_file)
now = datetime.now(timezone.utc)
candidate = ReleaseCandidate(
candidate_id=payload.get("candidate_id", "candidate-1"),
version=payload.get("version", "1.0.0"),
profile=ProfileType.ENTERPRISE_CLEAN,
source_snapshot_ref=payload.get("source_snapshot_ref", "snapshot-ref"),
created_at=now,
created_by=payload.get("created_by", "operator"),
status=ReleaseCandidateStatus.DRAFT,
)
repository.save_candidate(candidate)
registry_id = payload.get("registry_id", "REG-1")
entries = [
ResourceSourceEntry(
source_id=f"S-{index + 1}",
host=host,
protocol="https",
purpose="bootstrap",
enabled=True,
)
for index, host in enumerate(payload.get("allowed_hosts", []))
if str(host).strip()
]
if entries:
repository.save_registry(
ResourceSourceRegistry(
registry_id=registry_id,
name=payload.get("registry_name", "Bootstrap Internal Registry"),
entries=entries,
updated_at=now,
updated_by=payload.get("created_by", "operator"),
status=RegistryStatus.ACTIVE,
)
)
if entries:
repository.save_policy(
CleanProfilePolicy(
policy_id=payload.get("policy_id", "POL-ENT-CLEAN"),
policy_version=payload.get("policy_version", "1"),
profile=ProfileType.ENTERPRISE_CLEAN,
active=True,
internal_source_registry_ref=registry_id,
prohibited_artifact_categories=payload.get(
"prohibited_artifact_categories",
["test-data", "demo", "load-test"],
),
required_system_categories=payload.get("required_system_categories", ["core"]),
effective_from=now,
)
)
def _resolve_candidate_id(self) -> str:
env_candidate = os.getenv("CLEAN_TUI_CANDIDATE_ID", "").strip()
if env_candidate:
return env_candidate
candidate_ids = list(self.repo.candidates.keys())
if candidate_ids:
return candidate_ids[0]
return ""
def draw_header(self, max_y: int, max_x: int):
header_text = " Enterprise Clean Release Validator (TUI) "
self.stdscr.attron(curses.color_pair(1) | curses.A_BOLD)
@@ -109,8 +194,10 @@ class CleanReleaseTUI:
centered = header_text.center(max_x)
self.stdscr.addstr(0, 0, centered[:max_x])
self.stdscr.attroff(curses.color_pair(1) | curses.A_BOLD)
info_line_text = " │ Candidate: [2026.03.03-rc1] Profile: [enterprise-clean]".ljust(max_x)
candidate_text = self.candidate_id or "not-set"
profile_text = "enterprise-clean"
info_line_text = f" │ Candidate: [{candidate_text}] Profile: [{profile_text}] Mode: [{self.mode}]".ljust(max_x)
self.stdscr.addstr(2, 0, info_line_text[:max_x])
def draw_checks(self):
@@ -148,12 +235,17 @@ class CleanReleaseTUI:
def draw_sources(self):
self.stdscr.addstr(12, 3, "Allowed Internal Sources:", curses.A_BOLD)
reg = self.repo.get_registry("REG-1")
reg = None
policy = self.repo.get_active_policy()
if policy:
reg = self.repo.get_registry(policy.internal_source_registry_ref)
row = 13
if reg:
for entry in reg.entries:
self.stdscr.addstr(row, 3, f" - {entry.host}")
row += 1
else:
self.stdscr.addstr(row, 3, " - (none)")
def draw_status(self):
color = curses.color_pair(5)
@@ -173,6 +265,8 @@ class CleanReleaseTUI:
v_cat = str(v.category.value if hasattr(v.category, "value") else v.category)
msg_text = f"[{v_cat}] {v.remediation} (Loc: {v.location})"
self.stdscr.addstr(row + i, 5, msg_text[:70], curses.color_pair(3))
if self.last_error:
self.stdscr.addstr(27, 3, f"Error: {self.last_error}"[:100], curses.color_pair(3) | curses.A_BOLD)
def draw_footer(self, max_y: int, max_x: int):
footer_text = " F5 Run Check F7 Clear History F10 Exit ".center(max_x)
@@ -187,28 +281,60 @@ class CleanReleaseTUI:
self.report_id = None
self.violations_list = []
self.checks_progress = []
candidate = self.repo.get_candidate("2026.03.03-rc1")
self.last_error = None
candidate = self.repo.get_candidate(self.candidate_id) if self.candidate_id else None
policy = self.repo.get_active_policy()
if not candidate or not policy:
self.status = "FAILED"
self.last_error = "Candidate or active policy not found. Set CLEAN_TUI_CANDIDATE_ID and prepare repository data."
self.refresh_screen()
return
# Prepare a manifest with a deliberate violation for demo
artifacts = [
{"path": "src/main.py", "category": "core", "reason": "source code", "classification": "allowed"},
{"path": "test/data.csv", "category": "test-data", "reason": "test payload", "classification": "excluded-prohibited"},
]
manifest = build_distribution_manifest(
manifest_id=f"manifest-{candidate.candidate_id}",
candidate_id=candidate.candidate_id,
policy_id=policy.policy_id,
generated_by="operator",
artifacts=artifacts
)
self.repo.save_manifest(manifest)
if self.mode == "demo":
# Prepare a manifest with a deliberate violation for demonstration mode.
artifacts = [
{"path": "src/main.py", "category": "core", "reason": "source code", "classification": "allowed"},
{"path": "test/data.csv", "category": "test-data", "reason": "test payload", "classification": "excluded-prohibited"},
]
manifest = build_distribution_manifest(
manifest_id=f"manifest-{candidate.candidate_id}",
candidate_id=candidate.candidate_id,
policy_id=policy.policy_id,
generated_by="operator",
artifacts=artifacts
)
self.repo.save_manifest(manifest)
else:
manifest = self.repo.get_manifest(f"manifest-{candidate.candidate_id}")
if manifest is None:
artifacts_path = os.getenv("CLEAN_TUI_ARTIFACTS_JSON", "").strip()
if artifacts_path:
try:
with open(artifacts_path, "r", encoding="utf-8") as artifacts_file:
artifacts = json.load(artifacts_file)
if not isinstance(artifacts, list):
raise ValueError("Artifacts JSON must be a list")
prepare_candidate(
repository=self.repo,
candidate_id=candidate.candidate_id,
artifacts=artifacts,
sources=[],
operator_id="tui-operator",
)
manifest = self.repo.get_manifest(f"manifest-{candidate.candidate_id}")
except Exception as exc:
self.status = "FAILED"
self.last_error = f"Unable to prepare manifest from CLEAN_TUI_ARTIFACTS_JSON: {exc}"
self.refresh_screen()
return
if manifest is None:
self.status = "FAILED"
self.last_error = "Manifest not found. Prepare candidate first or provide CLEAN_TUI_ARTIFACTS_JSON."
self.refresh_screen()
return
# Init orchestrator sequence
check_run = self.orchestrator.start_check_run(candidate.candidate_id, policy.policy_id, "operator", "tui")
@@ -244,6 +370,7 @@ class CleanReleaseTUI:
self.report_id = None
self.violations_list = []
self.checks_progress = []
self.last_error = None
self.refresh_screen()
def refresh_screen(self):