650 lines
22 KiB
Python
650 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
try:
|
|
import psycopg
|
|
except ImportError:
|
|
psycopg = None
|
|
|
|
|
|
def parse_dotnet_connection_string(s: str) -> dict[str, str | None]:
|
|
"""Parse .NET connection string format to psycopg-compatible dict.
|
|
|
|
Input: "Host=x;Port=n;Database=d;Username=u;Password=p;Search Path=s;"
|
|
Output: {host, port, dbname, user, password, options (if Search Path present)}
|
|
Keys are case-insensitive; unknown keys ignored.
|
|
"""
|
|
result: dict[str, str | None] = {}
|
|
parts = s.split(";")
|
|
search_path_value = None
|
|
|
|
for part in parts:
|
|
part = part.strip()
|
|
if not part:
|
|
continue
|
|
if "=" not in part:
|
|
continue
|
|
key, value = part.split("=", 1)
|
|
key_lower = key.strip().lower()
|
|
value = value.strip()
|
|
|
|
if key_lower == "host":
|
|
result["host"] = value
|
|
elif key_lower == "port":
|
|
result["port"] = value
|
|
elif key_lower == "database":
|
|
result["dbname"] = value
|
|
elif key_lower == "username":
|
|
result["user"] = value
|
|
elif key_lower == "password":
|
|
result["password"] = value
|
|
elif key_lower == "search path":
|
|
search_path_value = value
|
|
|
|
if search_path_value:
|
|
result["options"] = f"-c search_path={search_path_value}"
|
|
|
|
return result
|
|
|
|
|
|
def load_spec(spec_path: Path) -> dict[str, Any]:
|
|
"""Load and parse the WBS spec YAML."""
|
|
if not spec_path.exists():
|
|
raise FileNotFoundError(f"Spec file not found: {spec_path}")
|
|
return yaml.safe_load(spec_path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def resolve_db_connection(root: Path, spec: dict[str, Any]) -> str | None:
|
|
"""Resolve PostgreSQL connection string.
|
|
|
|
Resolution order:
|
|
1. env QE_WBS_PG_DSN (psycopg DSN format)
|
|
2. env ConnectionStrings__DefaultConnection (.NET format, convert to psycopg)
|
|
3. appsettings.Development.json ConnectionStrings.DefaultConnection (.NET format, convert)
|
|
"""
|
|
import os
|
|
|
|
# Try env QE_WBS_PG_DSN
|
|
dsn = os.environ.get("QE_WBS_PG_DSN")
|
|
if dsn:
|
|
return dsn
|
|
|
|
# Try env ConnectionStrings__DefaultConnection (.NET format)
|
|
dotnet_str = os.environ.get("ConnectionStrings__DefaultConnection")
|
|
if dotnet_str:
|
|
parsed = parse_dotnet_connection_string(dotnet_str)
|
|
return _build_psycopg_dsn(parsed)
|
|
|
|
# Try appsettings.Development.json
|
|
appsettings_path = root / spec.get("meta", {}).get("db_connection", {}).get("dotnet_appsettings", "")
|
|
if appsettings_path and appsettings_path.is_relative_to(root):
|
|
full_path = root / appsettings_path
|
|
if full_path.exists():
|
|
try:
|
|
appsettings = json.loads(full_path.read_text(encoding="utf-8"))
|
|
conn_str = appsettings.get("ConnectionStrings", {}).get("DefaultConnection", "")
|
|
if conn_str:
|
|
parsed = parse_dotnet_connection_string(conn_str)
|
|
return _build_psycopg_dsn(parsed)
|
|
except Exception:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def _build_psycopg_dsn(parsed: dict[str, str | None]) -> str:
|
|
"""Build psycopg DSN from parsed dict."""
|
|
parts = []
|
|
for key in ["host", "port", "dbname", "user", "password"]:
|
|
val = parsed.get(key)
|
|
if val:
|
|
parts.append(f"{key}={val}")
|
|
# The evidence queries use fully-qualified schema names. Do not emit the
|
|
# parsed Search Path as a libpq DSN option: unquoted values are interpreted
|
|
# as connection keywords by psycopg (for example, "search_path"), which
|
|
# makes an otherwise valid .NET connection string fail to connect.
|
|
return " ".join(parts)
|
|
|
|
|
|
def check_pg_query(root: Path, check: dict[str, Any], dsn: str | None) -> tuple[bool, dict[str, Any]]:
|
|
"""Verify pg_query check type."""
|
|
import os
|
|
|
|
if not dsn:
|
|
return False, {"error": "No PostgreSQL connection available"}
|
|
|
|
sql = check.get("sql", "")
|
|
expect = check.get("expect", {})
|
|
observed = None
|
|
error_msg = None
|
|
connected_as_host = "unknown"
|
|
|
|
if psycopg is not None:
|
|
try:
|
|
conn = psycopg.connect(dsn)
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(sql)
|
|
row = cursor.fetchone()
|
|
observed = row[0] if row else None
|
|
cursor.close()
|
|
conn.close()
|
|
connected_as_host = dsn.split("host=")[-1].split()[0] if "host=" in dsn else "unknown"
|
|
except Exception as e:
|
|
conn.close()
|
|
error_msg = f"psycopg execution failed: {e}"
|
|
except Exception as e:
|
|
error_msg = f"psycopg connection failed: {e}"
|
|
else:
|
|
error_msg = "psycopg not installed"
|
|
|
|
# Fallback to psql CLI if psycopg failed or is not installed
|
|
if observed is None:
|
|
import shutil
|
|
psql_path = shutil.which("psql")
|
|
if psql_path:
|
|
import re
|
|
dsn_parts = {}
|
|
matches = re.findall(r"(\w+)\s*=\s*(?:'([^']*)'|(\S+))", dsn)
|
|
for key, val1, val2 in matches:
|
|
dsn_parts[key] = val1 or val2
|
|
|
|
cmd = [psql_path]
|
|
if "host" in dsn_parts:
|
|
cmd.extend(["-h", dsn_parts["host"]])
|
|
connected_as_host = dsn_parts["host"]
|
|
if "port" in dsn_parts:
|
|
cmd.extend(["-p", dsn_parts["port"]])
|
|
if "user" in dsn_parts:
|
|
cmd.extend(["-U", dsn_parts["user"]])
|
|
if "dbname" in dsn_parts:
|
|
cmd.extend(["-d", dsn_parts["dbname"]])
|
|
cmd.extend(["-t", "-A", "-c", sql])
|
|
|
|
env = dict(os.environ)
|
|
if "password" in dsn_parts:
|
|
env["PGPASSWORD"] = dsn_parts["password"]
|
|
|
|
try:
|
|
res = subprocess.run(cmd, env=env, capture_output=True, text=True, check=True)
|
|
stdout_val = res.stdout.strip()
|
|
if stdout_val:
|
|
observed = stdout_val
|
|
error_msg = None
|
|
except Exception as e:
|
|
stderr_msg = e.stderr if hasattr(e, "stderr") else ""
|
|
error_msg = f"psql fallback failed: {e}. stderr: {stderr_msg}. (Previous: {error_msg})"
|
|
else:
|
|
error_msg = f"psql CLI not found. (Previous: {error_msg})"
|
|
|
|
if error_msg is not None:
|
|
return False, {"error": error_msg, "sql": sql}
|
|
|
|
# Try to coerce to numeric for comparison
|
|
if observed is not None:
|
|
try:
|
|
observed_float = float(observed)
|
|
if observed_float.is_integer():
|
|
observed = int(observed_float)
|
|
else:
|
|
observed = observed_float
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Check expectations
|
|
passed = True
|
|
if "min" in expect:
|
|
min_val = expect["min"]
|
|
try:
|
|
if observed is None or float(observed) < float(min_val):
|
|
passed = False
|
|
except (ValueError, TypeError):
|
|
passed = False
|
|
if "max" in expect and passed:
|
|
max_val = expect["max"]
|
|
try:
|
|
if observed is None or float(observed) > float(max_val):
|
|
passed = False
|
|
except (ValueError, TypeError):
|
|
passed = False
|
|
if "equals" in expect and passed:
|
|
eq_val = expect["equals"]
|
|
try:
|
|
if isinstance(eq_val, (int, float)) or (isinstance(eq_val, str) and eq_val.replace(".","",1).isdigit()):
|
|
if float(observed) != float(eq_val):
|
|
passed = False
|
|
else:
|
|
if str(observed) != str(eq_val):
|
|
passed = False
|
|
except (ValueError, TypeError):
|
|
if str(observed) != str(eq_val):
|
|
passed = False
|
|
|
|
return passed, {
|
|
"sql": sql,
|
|
"observed": observed,
|
|
"expected": expect,
|
|
"connected_as_host": connected_as_host
|
|
}
|
|
|
|
|
|
def check_log_pattern(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
|
"""Verify log_pattern check type."""
|
|
from pathlib import Path as PathlibPath
|
|
|
|
file_glob = check.get("file_glob", "")
|
|
pattern_str = check.get("pattern", "")
|
|
expect = check.get("expect", {})
|
|
min_matches = expect.get("min_matches")
|
|
max_matches = expect.get("max_matches")
|
|
max_age_hours = expect.get("max_age_hours")
|
|
|
|
if not file_glob or not pattern_str:
|
|
return False, {"error": "Missing file_glob or pattern"}
|
|
|
|
try:
|
|
pattern = re.compile(pattern_str)
|
|
except re.error as e:
|
|
return False, {"error": f"Invalid regex: {e}"}
|
|
|
|
# Glob for files
|
|
glob_path = root / file_glob
|
|
matched_files = list(glob_path.parent.glob(glob_path.name)) if "*" in file_glob else (
|
|
[glob_path] if glob_path.exists() else []
|
|
)
|
|
|
|
# Handle ** in glob
|
|
if "**" in file_glob:
|
|
parts = file_glob.split("**")
|
|
base = root / parts[0] if parts[0] else root
|
|
suffix = parts[-1] if len(parts) > 1 else "*"
|
|
matched_files = list(base.glob(f"**/{suffix}"))
|
|
|
|
# Filter by age if needed
|
|
import time as time_module
|
|
now = time_module.time()
|
|
if max_age_hours:
|
|
max_age_seconds = max_age_hours * 3600
|
|
matched_files = [f for f in matched_files if f.is_file() and (now - f.stat().st_mtime) <= max_age_seconds]
|
|
|
|
# Count matches
|
|
match_count = 0
|
|
matched_lines = []
|
|
for file_path in matched_files:
|
|
try:
|
|
content = file_path.read_text(encoding="utf-8", errors="replace")
|
|
for line in content.splitlines():
|
|
if pattern.search(line):
|
|
match_count += 1
|
|
if len(matched_lines) < 200:
|
|
matched_lines.append(f"{file_path.name}: {line}")
|
|
except Exception:
|
|
pass
|
|
|
|
# Check expectations
|
|
passed = True
|
|
if min_matches is not None and match_count < min_matches:
|
|
passed = False
|
|
if max_matches is not None and match_count > max_matches:
|
|
passed = False
|
|
|
|
# Special case: if neither bound given, default min_matches=1
|
|
if min_matches is None and max_matches is None:
|
|
if match_count < 1:
|
|
passed = False
|
|
|
|
return passed, {
|
|
"file_glob": file_glob,
|
|
"pattern": pattern_str,
|
|
"observed": match_count,
|
|
"expected": expect,
|
|
"matched_lines": matched_lines[:200]
|
|
}
|
|
|
|
|
|
def check_json_gate(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
|
"""Verify json_gate check type."""
|
|
path_str = check.get("path", "")
|
|
expect = check.get("expect", {})
|
|
|
|
if not path_str:
|
|
return False, {"error": "Missing path"}
|
|
|
|
json_path = root / path_str
|
|
if not json_path.exists():
|
|
return False, {"error": f"JSON file not found: {json_path}", "path": path_str}
|
|
|
|
try:
|
|
payload = json.loads(json_path.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
return False, {"error": f"Failed to parse JSON: {e}", "path": path_str}
|
|
|
|
# Check each expectation key
|
|
passed = True
|
|
details = {}
|
|
for key, expected_val in expect.items():
|
|
# Dot-notation path support
|
|
observed_val = payload
|
|
for part in key.split("."):
|
|
if isinstance(observed_val, dict):
|
|
observed_val = observed_val.get(part)
|
|
else:
|
|
observed_val = None
|
|
break
|
|
|
|
# Check if expected_val is a comparison operator string
|
|
if isinstance(expected_val, str):
|
|
# Check two-character operators first
|
|
for op, op_str in [(">=", ">="), ("<=", "<="), (">", ">"), ("<", "<")]:
|
|
if expected_val.startswith(op_str):
|
|
try:
|
|
expected_num = float(expected_val[len(op_str):])
|
|
observed_num = float(observed_val) if observed_val is not None else None
|
|
if observed_num is None:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_val}"
|
|
elif op == ">=" and observed_num < expected_num:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_num}"
|
|
elif op == "<=" and observed_num > expected_num:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_num}"
|
|
elif op == ">" and observed_num <= expected_num:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_num}"
|
|
elif op == "<" and observed_num >= expected_num:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_num}"
|
|
except (ValueError, TypeError):
|
|
passed = False
|
|
details[key] = f"Failed to parse comparison: {expected_val}"
|
|
break
|
|
else:
|
|
# Plain equality
|
|
if observed_val != expected_val:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_val}"
|
|
else:
|
|
# Plain equality
|
|
if observed_val != expected_val:
|
|
passed = False
|
|
details[key] = f"Expected {expected_val}, got {observed_val}"
|
|
|
|
return passed, {
|
|
"path": path_str,
|
|
"expected": expect,
|
|
"details": details if details else "All checks passed"
|
|
}
|
|
|
|
|
|
def check_file_exists(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
|
"""Verify file_exists check type."""
|
|
paths = check.get("paths", [])
|
|
expect = check.get("expect", {})
|
|
min_bytes = expect.get("min_bytes")
|
|
|
|
if not paths:
|
|
return False, {"error": "Missing paths"}
|
|
|
|
details = {}
|
|
passed = True
|
|
for path_str in paths:
|
|
file_path = root / path_str
|
|
exists = file_path.exists()
|
|
details[path_str] = {"exists": exists}
|
|
|
|
if not exists:
|
|
passed = False
|
|
continue
|
|
|
|
if file_path.is_file() and min_bytes is not None:
|
|
size = file_path.stat().st_size
|
|
details[path_str]["size"] = size
|
|
if size < min_bytes:
|
|
passed = False
|
|
|
|
return passed, {
|
|
"paths": paths,
|
|
"expected": expect,
|
|
"details": details
|
|
}
|
|
|
|
|
|
def check_playwright_report(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
|
"""Verify playwright_report check type."""
|
|
report_path_str = check.get("report", "")
|
|
spec_file = check.get("spec_file", "")
|
|
expect = check.get("expect", {})
|
|
passed_min = expect.get("passed_min", 1)
|
|
failed_expected = expect.get("failed", 0)
|
|
|
|
if not report_path_str or not spec_file:
|
|
return False, {"error": "Missing report or spec_file"}
|
|
|
|
report_path = root / report_path_str
|
|
if not report_path.exists():
|
|
return False, {"error": f"Playwright report not found: {report_path}"}
|
|
|
|
try:
|
|
report = json.loads(report_path.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
return False, {"error": f"Failed to parse report: {e}"}
|
|
|
|
# Walk suites to find specs matching spec_file
|
|
passed_count = 0
|
|
failed_count = 0
|
|
|
|
def walk_suites(suites_list):
|
|
nonlocal passed_count, failed_count
|
|
if not suites_list:
|
|
return
|
|
for suite in suites_list:
|
|
# Recurse into nested suites
|
|
if "suites" in suite:
|
|
walk_suites(suite["suites"])
|
|
# Check specs
|
|
if "specs" in suite:
|
|
for spec in suite["specs"]:
|
|
if spec_file in spec.get("file", ""):
|
|
if spec.get("ok"):
|
|
passed_count += 1
|
|
else:
|
|
failed_count += 1
|
|
|
|
suites = report.get("suites", [])
|
|
walk_suites(suites)
|
|
|
|
passed = (passed_count >= passed_min) and (failed_count == failed_expected)
|
|
|
|
return passed, {
|
|
"report": report_path_str,
|
|
"spec_file": spec_file,
|
|
"passed_count": passed_count,
|
|
"failed_count": failed_count,
|
|
"expected": expect
|
|
}
|
|
|
|
|
|
def run_verification_commands(root: Path, task_id: str, commands: list[str]) -> list[dict[str, Any]]:
|
|
"""Execute verification commands (excluding self-references)."""
|
|
results = []
|
|
evidence_dir = root / "Temp" / "evidence" / task_id
|
|
evidence_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
for idx, cmd in enumerate(commands):
|
|
# Skip commands that reference this script to avoid recursion
|
|
if "verify_wbs_task_v1.py" in cmd:
|
|
continue
|
|
|
|
try:
|
|
# Write to command log file
|
|
log_file = evidence_dir / f"command_{idx}.log"
|
|
result = subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
cwd=root,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace"
|
|
)
|
|
output = result.stdout + result.stderr
|
|
log_file.write_text(output, encoding="utf-8")
|
|
results.append({
|
|
"index": idx,
|
|
"command": cmd,
|
|
"returncode": result.returncode,
|
|
"log_file": str(log_file.relative_to(root))
|
|
})
|
|
except Exception as e:
|
|
results.append({
|
|
"index": idx,
|
|
"command": cmd,
|
|
"error": str(e)
|
|
})
|
|
|
|
return results
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
"""Main entry point."""
|
|
parser = argparse.ArgumentParser(description="Verify a single WBS task evidence")
|
|
parser.add_argument("--task", required=True, help="Task ID (e.g., QE-M0-01)")
|
|
parser.add_argument("--repo-root", default=None, help="Repository root path")
|
|
parser.add_argument("--spec", default=None, help="Spec YAML file path")
|
|
parser.add_argument("--run-commands", action="store_true", help="Execute verification commands")
|
|
|
|
args = parser.parse_args(argv)
|
|
task_id = args.task
|
|
|
|
# Resolve root
|
|
if args.repo_root:
|
|
root = Path(args.repo_root).resolve()
|
|
else:
|
|
root = Path(__file__).resolve().parents[1]
|
|
|
|
# Resolve spec path
|
|
if args.spec:
|
|
spec_path = Path(args.spec).resolve()
|
|
else:
|
|
spec_path = root / "spec" / "60_quant_engine_wbs.yaml"
|
|
|
|
# Load spec
|
|
try:
|
|
spec = load_spec(spec_path)
|
|
except FileNotFoundError as e:
|
|
print(f"ERROR: {e}")
|
|
return 2
|
|
|
|
# Find task
|
|
tasks = spec.get("tasks", {})
|
|
if task_id not in tasks:
|
|
print(f"ERROR: Task {task_id} not found in spec")
|
|
return 2
|
|
|
|
task = tasks[task_id]
|
|
evidence_checks = task.get("evidence_checks", [])
|
|
verification_commands = task.get("success_criteria", {}).get("verification_commands", [])
|
|
|
|
# Create evidence directory
|
|
evidence_dir = root / "Temp" / "evidence" / task_id
|
|
evidence_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Resolve DB connection
|
|
dsn = resolve_db_connection(root, spec)
|
|
|
|
# Run checks
|
|
check_results = []
|
|
gate = "PASS"
|
|
|
|
for check_idx, check in enumerate(evidence_checks):
|
|
check_type = check.get("type")
|
|
passed = False
|
|
detail = {}
|
|
|
|
try:
|
|
if check_type == "pg_query":
|
|
passed, detail = check_pg_query(root, check, dsn)
|
|
elif check_type == "log_pattern":
|
|
passed, detail = check_log_pattern(root, check)
|
|
elif check_type == "json_gate":
|
|
passed, detail = check_json_gate(root, check)
|
|
elif check_type == "file_exists":
|
|
passed, detail = check_file_exists(root, check)
|
|
elif check_type == "playwright_report":
|
|
passed, detail = check_playwright_report(root, check)
|
|
else:
|
|
passed = False
|
|
detail = {"error": f"Unknown check type: {check_type}"}
|
|
except Exception as e:
|
|
passed = False
|
|
detail = {"error": str(e)}
|
|
|
|
if not passed:
|
|
gate = "FAIL"
|
|
|
|
check_results.append({
|
|
"index": check_idx,
|
|
"type": check_type,
|
|
"gate": "PASS" if passed else "FAIL",
|
|
"detail": detail
|
|
})
|
|
|
|
# Run commands if requested
|
|
if args.run_commands:
|
|
run_verification_commands(root, task_id, verification_commands)
|
|
|
|
# Prepare verdict
|
|
verdict = {
|
|
"task_id": task_id,
|
|
"formula_id": "QUANT_ENGINE_WBS_TASK_V1",
|
|
"gate": gate,
|
|
"generated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
|
"checks": check_results,
|
|
"input_hashes": {}
|
|
}
|
|
|
|
# Save verdict
|
|
verdict_path = evidence_dir / "verdict.json"
|
|
verdict_path.write_text(json.dumps(verdict, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
# Append lineage event
|
|
lineage_log = root / "runtime" / "lineage_events.jsonl"
|
|
lineage_log.parent.mkdir(parents=True, exist_ok=True)
|
|
event = {
|
|
"node_id": f"wbs_{task_id}",
|
|
"command": f"python tools/verify_wbs_task_v1.py --task {task_id}",
|
|
"returncode": 0 if gate == "PASS" else 1,
|
|
"elapsed_sec": 0,
|
|
"gate": gate,
|
|
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
}
|
|
lineage_log.write_text(json.dumps(event, ensure_ascii=False) + "\n", encoding="utf-8", errors="append")
|
|
|
|
# Print summary (ASCII only)
|
|
print(f"Task: {task_id}")
|
|
for check_result in check_results:
|
|
check_gate = check_result["gate"]
|
|
check_type = check_result["type"]
|
|
status = "PASS" if check_gate == "PASS" else "FAIL"
|
|
print(f" [{status}] {check_type}")
|
|
print(f"Gate: {gate}")
|
|
|
|
return 0 if gate == "PASS" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|