217 lines
10 KiB
Python
217 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import py_compile
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ERRORS: list[str] = []
|
|
WARNINGS: list[str] = []
|
|
CHECKS: list[str] = []
|
|
|
|
|
|
def ok(message: str) -> None: CHECKS.append(message)
|
|
def fail(message: str) -> None: ERRORS.append(message)
|
|
def warn(message: str) -> None: WARNINGS.append(message)
|
|
|
|
|
|
def active_files(pattern: str):
|
|
for path in ROOT.rglob(pattern):
|
|
rel = path.relative_to(ROOT)
|
|
if rel.parts[:1] == ('attachments',):
|
|
continue
|
|
if rel.parts[:2] == ('research', 'original'):
|
|
continue
|
|
if any(part in {'bin', 'obj', 'node_modules', '__pycache__', '.git'} for part in rel.parts):
|
|
continue
|
|
yield path
|
|
|
|
required = [
|
|
'README.md', 'AGENTS.md', 'global.json', 'KArtSell.sln',
|
|
'docs/v12_2/00_EXECUTIVE_STRATEGIC_PROPOSAL.md',
|
|
'docs/v12_2/01_SOURCE_COVERAGE_MATRIX.csv',
|
|
'docs/v12_2/06_SLICE_CATALOGUE.csv',
|
|
'docs/v12_2/07_DETAILED_WBS_MASTER.csv',
|
|
'docs/v12_2/08_TECH_DEBT_REGISTER.csv',
|
|
'docs/v12_2/09_TRACEABILITY_MATRIX.csv',
|
|
'docs/v12_2/10_DECISION_LOG.csv',
|
|
'db/migrations/0014_signal_engine_lot_weight_and_policy_trace.sql',
|
|
'research/hardening/test_policy_contract.py',
|
|
'templates/dotnet/VerticalSlice/Endpoint.cs.template',
|
|
'templates/vue/Feature/schema.ts.template',
|
|
'templates/sql/Migration.sql.template',
|
|
'templates/hangfire/ReliableJob.cs.template',
|
|
]
|
|
for name in required:
|
|
if not (ROOT/name).exists(): fail(f'missing required file: {name}')
|
|
if not ERRORS: ok(f'required files present: {len(required)}')
|
|
|
|
# All current-session attachments must be byte-for-byte represented by source coverage.
|
|
coverage_path = ROOT/'docs/v12_2/01_SOURCE_COVERAGE_MATRIX.csv'
|
|
coverage: dict[str, str] = {}
|
|
if coverage_path.exists():
|
|
with coverage_path.open(encoding='utf-8-sig', newline='') as f:
|
|
for row in csv.DictReader(f): coverage[row['Attachment']] = row['SHA256']
|
|
attachment_dir = ROOT/'attachments/current_session'
|
|
for path in sorted(attachment_dir.glob('*')):
|
|
if not path.is_file(): continue
|
|
actual=hashlib.sha256(path.read_bytes()).hexdigest()
|
|
if coverage.get(path.name) != actual:
|
|
fail(f'attachment coverage/hash mismatch: {path.name}')
|
|
if len(coverage) != len([p for p in attachment_dir.glob('*') if p.is_file()]):
|
|
fail('source coverage row count differs from current-session attachment count')
|
|
else:
|
|
ok(f'attachment preservation verified: {len(coverage)} files')
|
|
|
|
for path in active_files('*.json'):
|
|
try: json.loads(path.read_text(encoding='utf-8'))
|
|
except Exception as exc: fail(f'invalid JSON {path.relative_to(ROOT)}: {exc}')
|
|
ok('JSON parse complete')
|
|
|
|
for path in [*active_files('*.csproj'), *active_files('*.props')]:
|
|
try: ET.parse(path)
|
|
except Exception as exc: fail(f'invalid XML {path.relative_to(ROOT)}: {exc}')
|
|
ok('MSBuild XML parse complete')
|
|
|
|
for path in active_files('*.csv'):
|
|
try:
|
|
with path.open(encoding='utf-8-sig', newline='') as f: rows=list(csv.reader(f))
|
|
if not rows: fail(f'empty CSV {path.relative_to(ROOT)}'); continue
|
|
width=len(rows[0])
|
|
bad=[i+1 for i,row in enumerate(rows) if len(row)!=width]
|
|
if bad: fail(f'ragged CSV {path.relative_to(ROOT)} rows {bad[:10]}')
|
|
except Exception as exc: fail(f'invalid CSV {path.relative_to(ROOT)}: {exc}')
|
|
ok('CSV structure complete')
|
|
|
|
# WBS structural and traceability checks.
|
|
wbs_path=ROOT/'docs/v12_2/07_DETAILED_WBS_MASTER.csv'
|
|
if wbs_path.exists():
|
|
with wbs_path.open(encoding='utf-8-sig', newline='') as f: wbs=list(csv.DictReader(f))
|
|
ids=[r['WBS_ID'] for r in wbs]
|
|
if len(ids)!=len(set(ids)): fail('duplicate WBS_ID')
|
|
if len(wbs)<225: fail(f'detailed WBS is unexpectedly small: {len(wbs)}')
|
|
slice_ids={r['Slice_ID'] for r in wbs if r['Slice_ID'].startswith('VS-')}
|
|
expected={f'VS-{i:02d}' for i in range(26)}
|
|
if slice_ids!=expected: fail(f'WBS slice coverage mismatch: missing={sorted(expected-slice_ids)} extra={sorted(slice_ids-expected)}')
|
|
required_columns={'Requirement_ID','API_ID','DB_Migration_ID','Job_Event_ID','UI_ID','Test_ID','Acceptance_Evidence','Primary_Owner','Dependency','Gate','Evidence_Class'}
|
|
if not required_columns.issubset(wbs[0]): fail('WBS required columns missing')
|
|
if any(not r['Acceptance_Evidence'].strip() or not r['Primary_Owner'].strip() for r in wbs): fail('WBS has blank acceptance or owner')
|
|
ok(f'WBS uniqueness/coverage complete: {len(wbs)} tasks, 26 slices')
|
|
|
|
for path in active_files('*.py'):
|
|
try: py_compile.compile(str(path), doraise=True)
|
|
except Exception as exc: fail(f'Python compile failed {path.relative_to(ROOT)}: {exc}')
|
|
ok('Python compile complete')
|
|
|
|
for path in active_files('*.yaml'):
|
|
text=path.read_text(encoding='utf-8')
|
|
if '\t' in text: fail(f'YAML contains tab indentation: {path.relative_to(ROOT)}')
|
|
if not text.strip(): fail(f'empty YAML: {path.relative_to(ROOT)}')
|
|
ok('YAML basic checks complete')
|
|
|
|
source_files=[p for p in [*active_files('*.cs'),*active_files('*.sql')] if 'tests' not in p.relative_to(ROOT).parts]
|
|
patterns={
|
|
'AllowAnonymous()':'anonymous module endpoint',
|
|
'IGenericRepository':'generic repository',
|
|
'DateTime.Now':'direct wall-clock use',
|
|
'DateTime.UtcNow':'direct wall-clock use',
|
|
'dynamic ':'dynamic typing in production C#',
|
|
}
|
|
for token,label in patterns.items():
|
|
hits=[str(p.relative_to(ROOT)) for p in source_files if token in p.read_text(encoding='utf-8')]
|
|
if hits: fail(f'{label} pattern {token}: {hits}')
|
|
select_star=[]
|
|
for path in source_files:
|
|
if re.search(r'\bselect\s+\*',path.read_text(encoding='utf-8'),flags=re.I): select_star.append(str(path.relative_to(ROOT)))
|
|
if select_star: fail(f'SELECT * prohibited: {select_star}')
|
|
ok('prohibited source pattern scan complete')
|
|
|
|
for path in active_files('Endpoint.cs'):
|
|
text=path.read_text(encoding='utf-8')
|
|
if 'Roles(' not in text and 'Policies(' not in text: fail(f'endpoint lacks Roles/Policies: {path.relative_to(ROOT)}')
|
|
ok('endpoint authorization declarations complete')
|
|
|
|
for path in active_files('*.cs'):
|
|
rel=path.relative_to(ROOT)
|
|
if 'Domain' in rel.parts:
|
|
text=path.read_text(encoding='utf-8')
|
|
for token in ['using Dapper','using Npgsql','using FastEndpoints','using Hangfire','HttpContext','DbConnection']:
|
|
if token in text: fail(f'domain framework dependency {token}: {rel}')
|
|
ok('domain dependency scan complete')
|
|
|
|
# Lightweight brace balance catches truncated generated source without pretending to compile C#.
|
|
for path in active_files('*.cs'):
|
|
text=path.read_text(encoding='utf-8')
|
|
# Remove raw/normal string bodies sufficiently for a structural smoke check.
|
|
scrub=re.sub(r'""".*?"""','',text,flags=re.S)
|
|
scrub=re.sub(r'"(?:\\.|[^"\\])*"','',scrub)
|
|
if scrub.count('{') != scrub.count('}'):
|
|
fail(f'C# brace imbalance: {path.relative_to(ROOT)}')
|
|
ok('C# structural smoke complete')
|
|
|
|
migration_names=sorted(p.name for p in (ROOT/'db/migrations').glob('*.sql'))
|
|
if not all(re.match(r'^\d{4}_[a-z0-9_]+\.sql$',n) for n in migration_names): fail('migration filename format must be NNNN_snake_case.sql')
|
|
if len(migration_names)!=len(set(migration_names)): fail('duplicate migration filename')
|
|
nums=[int(n[:4]) for n in migration_names]
|
|
if nums != sorted(nums): fail('migration sequence is not ordered')
|
|
ok(f'migration ordering complete: {len(migration_names)} scripts')
|
|
|
|
for path in active_files('*'):
|
|
if path.is_file() and path.name.lower() in {'testfile','temp','tmp'}: fail(f'placeholder file: {path.relative_to(ROOT)}')
|
|
ok('placeholder scan complete')
|
|
|
|
# Safety capability must be fail-closed in code and configuration.
|
|
program=(ROOT/'src/KArtSell.Host/Program.cs').read_text(encoding='utf-8')
|
|
for marker in ['!x.AutomaticOrder','!x.KisOrderAdapter','ValidateOnStart']:
|
|
if marker not in program: fail(f'missing startup capability guard: {marker}')
|
|
for cfg in [ROOT/'src/KArtSell.Host/appsettings.json',ROOT/'src/KArtSell.Host/appsettings.Development.json']:
|
|
if cfg.exists():
|
|
data=json.loads(cfg.read_text(encoding='utf-8'))
|
|
caps=data.get('Capabilities')
|
|
if caps is not None and (caps.get('AutomaticOrder') is not False or caps.get('KisOrderAdapter') is not False):
|
|
fail(f'order capability is not OFF: {cfg.relative_to(ROOT)}')
|
|
ok('automatic-order and KIS capability guards complete')
|
|
|
|
|
|
# v12.2 data-semantics and traceability checks.
|
|
input_text=(ROOT/'src/KArtSell.Modules.SignalEngine/Domain/SellDecisionInput.cs').read_text(encoding='utf-8')
|
|
service_text=(ROOT/'src/KArtSell.Modules.SignalEngine/Application/SellDecisionService.cs').read_text(encoding='utf-8')
|
|
fe_schema=(ROOT/'frontend/src/features/sell-decision/schema.ts').read_text(encoding='utf-8')
|
|
migration=(ROOT/'db/migrations/0014_signal_engine_lot_weight_and_policy_trace.sql').read_text(encoding='utf-8')
|
|
for token in ['CurrentSecurityPortfolioWeight','CurrentLotPortfolioWeight','TargetSecurityPortfolioWeightAfter']:
|
|
if token not in input_text and token not in service_text:
|
|
fail(f'missing v12.2 explicit weight token: {token}')
|
|
for token in ['policy_trace_json','SignalDecisionCreated",\n 2']:
|
|
if token not in service_text:
|
|
fail(f'missing v12.2 policy trace/event token: {token}')
|
|
for token in ['currentSecurityPortfolioWeight','currentLotPortfolioWeight','targetSecurityPortfolioWeightAfter','policyTrace']:
|
|
if token not in fe_schema:
|
|
fail(f'missing frontend runtime contract token: {token}')
|
|
for token in ['current_security_portfolio_weight','current_lot_portfolio_weight','policy_trace_json']:
|
|
if token not in migration:
|
|
fail(f'missing migration token: {token}')
|
|
if len(coverage) != 6:
|
|
fail(f'expected exactly six current-session attachments, found {len(coverage)}')
|
|
else:
|
|
ok('v12.2 explicit lot/security weight and policy-trace contracts complete')
|
|
|
|
if not (ROOT/'frontend/pnpm-lock.yaml').exists(): warn('frontend/pnpm-lock.yaml missing; G0 must generate, review and commit it')
|
|
if shutil.which('dotnet') is None: warn('dotnet SDK unavailable; restore/build/test not executed')
|
|
if shutil.which('pnpm') is None: warn('pnpm unavailable; frozen install/typecheck/test/build not executed')
|
|
if shutil.which('psql') is None: warn('PostgreSQL client unavailable; migration rehearsal not executed')
|
|
|
|
print('K-ArtSell Aegis v12.2 static validation')
|
|
print(f'root={ROOT}')
|
|
for m in CHECKS: print(f'PASS: {m}')
|
|
for m in WARNINGS: print(f'WARN: {m}')
|
|
for m in ERRORS: print(f'FAIL: {m}')
|
|
print(f'summary: pass={len(CHECKS)} warn={len(WARNINGS)} fail={len(ERRORS)}')
|
|
sys.exit(1 if ERRORS else 0)
|