40-validation/validate-repository-landscape.py

#!/usr/bin/env python3
from __future__ import annotations
import argparse, json, os, sys
from pathlib import Path
from evidence_lifecycle import seal_evidence

REPOSITORIES = [
    'enterprise-architecture','solution-architecture','engineering-platform','engineering-tools','runtime'
]
EXPECTED_CLASSES = {
    'enterprise-architecture':'Knowledge Repository',
    'solution-architecture':'Knowledge Repository',
    'engineering-platform':'Engineering Repository',
    'engineering-tools':'Engineering Repository',
    'runtime':'Runtime Repository',
}
MANDATORY_CHAINS = [f'RC-{n:04d}' for n in range(1,7)]
REQUIRED_ROLES = {
    'RC-0001': {'canonical-source','specialization','realization','validation','materialization','evidence'},
    'RC-0002': {'canonical-source','specialization','realization','validation','materialization','evidence'},
    'RC-0003': {'canonical-source','specialization','realization','validation','materialization','evidence'},
    'RC-0004': {'canonical-source','specialization','realization','validation','materialization','evidence'},
    'RC-0005': {'canonical-source','specialization','realization','validation','materialization','evidence'},
    'RC-0006': {'canonical-source','specialization','realization','validation','materialization','evidence'},
}
REQUIRED_EVIDENCE = [
    'runtime:test/reports/validation/artifact-manifests.json',
    'runtime:test/reports/startup/engineering-platform-startup.json',
    'runtime:test/reports/validation/build-contract.json',
    'runtime:test/reports/validation/deployment-contract.json',
    'runtime:test/reports/validation/workspace-contract.json',
    'runtime:test/reports/validation/validation-contract.json',
    'runtime:test/reports/validation/contract-coverage-gap-review.json',
    'runtime:test/reports/validation/mandatory-discovery-closure.json',
]
VALIDATOR_REF='engineering-tools:40-validation/validate-repository-landscape.py'
CONTRACT_REF='engineering-tools:40-validation/FULL_REPOSITORY_LANDSCAPE_VALIDATION.md'


def atomic_json(path:Path,data:dict)->None:
    path.parent.mkdir(parents=True,exist_ok=True)
    tmp=path.with_suffix(path.suffix+'.tmp')
    tmp.write_text(json.dumps(data,indent=2,ensure_ascii=False)+'\n',encoding='utf-8')
    os.replace(tmp,path)


def load_roots(container:Path)->tuple[dict[str,Path],dict[str,dict]]:
    roots={}; manifests={}
    for rid in REPOSITORIES:
        root=container/rid; mp=root/'manifest.json'
        if not root.is_dir() or not mp.is_file():
            raise RuntimeError(f'missing repository root or manifest: {rid}')
        manifest=json.loads(mp.read_text(encoding='utf-8'))
        ident=manifest.get('artifactName') or manifest.get('repository',{}).get('id')
        if ident != rid: raise RuntimeError(f'repository identity mismatch: expected {rid}, got {ident}')
        roots[rid]=root; manifests[rid]=manifest
    return roots,manifests


def resolve(roots:dict[str,Path],reference:str)->Path:
    if ':' not in reference: raise RuntimeError(f'unqualified reference: {reference}')
    rid,rel=reference.split(':',1)
    if rid not in roots: raise RuntimeError(f'unknown repository reference: {reference}')
    path=roots[rid]/rel
    if not path.is_file(): raise RuntimeError(f'missing required artifact: {reference}')
    return path


def passed(doc:dict)->bool:
    for key in ('status','result','overallStatus','overall'):
        value=doc.get(key)
        if isinstance(value,str) and value.upper() in {'PASSED','PASS','SUCCESS','SUCCEEDED','VALIDATED'}:
            return True
    return False


def root_contract(rid:str,root:Path,manifest:dict)->dict:
    cls=manifest.get('repositoryClass')
    errors=[]
    if cls != EXPECTED_CLASSES[rid]: errors.append(f'expected class {EXPECTED_CLASSES[rid]}, got {cls}')
    for key in ('artifactName','version','repositoryClass','rootEntry','humanEntry','validationEntryPoints'):
        if key not in manifest: errors.append(f'missing manifest field {key}')
    entries=[]
    for key in ('rootEntry','humanEntry'):
        rel=manifest.get(key)
        if rel:
            entries.append(rel)
            if not (root/rel).is_file(): errors.append(f'missing {key}: {rel}')
    for rel in manifest.get('validationEntryPoints',[]):
        entries.append(rel)
        if not (root/rel).is_file(): errors.append(f'missing validation entry: {rel}')
    return {'repository':rid,'version':manifest.get('version'),'repositoryClass':cls,'status':'PASSED' if not errors else 'FAILED','entries':entries,'errors':errors}


def main()->None:
    ap=argparse.ArgumentParser()
    ap.add_argument('repository_roots',type=Path)
    ap.add_argument('runtime_root',type=Path)
    ap.add_argument('--work-item',default='KT-0023')
    args=ap.parse_args()
    roots,manifests=load_roots(args.repository_roots.resolve())
    runtime=args.runtime_root.resolve()
    if runtime != roots['runtime'].resolve(): raise RuntimeError('runtime root does not match repository container')

    root_results=[root_contract(rid,roots[rid],manifests[rid]) for rid in REPOSITORIES]
    references=[]; reference_errors=[]
    for rid,manifest in manifests.items():
        for item in manifest.get('crossRepositoryReferences',[]):
            if not item.get('required',False): continue
            ref=item.get('reference')
            if not ref: reference_errors.append(f'{rid}: required reference without path'); continue
            try: resolve(roots,ref); references.append(ref)
            except Exception as exc: reference_errors.append(str(exc))

    chain_results=[]; chain_refs=[]
    for cid in MANDATORY_CHAINS:
        roles={}; errors=[]
        for rid,manifest in manifests.items():
            chain=next((c for c in manifest.get('realizationChains',[]) if c.get('id')==cid),None)
            if not chain: continue
            registry=chain.get('registry')
            if registry:
                try: resolve(roots,registry); chain_refs.append(registry)
                except Exception as exc: errors.append(str(exc))
            for art in chain.get('artifacts',[]):
                if not art.get('required',False): continue
                role=art.get('role'); ref=art.get('reference')
                if not role or not ref: errors.append(f'{rid}:{cid}: incomplete artifact declaration'); continue
                qref=ref if ':' in ref else f'{rid}:{ref}'
                try: resolve(roots,qref); chain_refs.append(qref); roles.setdefault(role,[]).append(qref)
                except Exception as exc: errors.append(str(exc))
        missing=sorted(REQUIRED_ROLES[cid]-set(roles))
        if missing: errors.append('missing roles: '+', '.join(missing))
        chain_results.append({'chain':cid,'status':'PASSED' if not errors else 'FAILED','roles':sorted(roles),'missingRoles':missing,'errors':errors})

    evidence_results=[]; evidence_refs=[]
    for ref in REQUIRED_EVIDENCE:
        errors=[]; doc={}
        try:
            path=resolve(roots,ref); evidence_refs.append(ref); doc=json.loads(path.read_text(encoding='utf-8'))
            if not passed(doc): errors.append('evidence status is not successful')
            if doc.get('freshness')!='fresh': errors.append('evidence is not marked fresh')
        except Exception as exc: errors.append(str(exc))
        evidence_results.append({'reference':ref,'status':'PASSED' if not errors else 'FAILED','freshness':doc.get('freshness'),'errors':errors})

    discovery=next(x for x in evidence_results if x['reference'].endswith('mandatory-discovery-closure.json'))
    blockers=[]
    blockers += [x for x in root_results if x['status']!='PASSED']
    if reference_errors: blockers.append({'area':'crossRepositoryReferences','errors':reference_errors})
    blockers += [x for x in chain_results if x['status']!='PASSED']
    blockers += [x for x in evidence_results if x['status']!='PASSED']
    manifest={
      'schemaVersion':'1.0','validationId':args.work_item,'workItem':args.work_item,
      'status':'PASSED' if not blockers else 'FAILED',
      'repositories':root_results,
      'crossRepositoryReferences':{'status':'PASSED' if not reference_errors else 'FAILED','validated':len(set(references)),'errors':reference_errors},
      'mandatoryRealizationChains':chain_results,
      'runtimeEvidence':evidence_results,
      'completionBlockers':blockers,
      'allMandatoryRepositoriesValid':all(x['status']=='PASSED' for x in root_results),
      'allMandatoryChainsValid':all(x['status']=='PASSED' for x in chain_results),
      'allRequiredEvidenceFresh':all(x['status']=='PASSED' for x in evidence_results),
      'mandatoryDiscoveryClosureValid':discovery['status']=='PASSED',
      'phase5LandscapeCompletionCriterionMet':not blockers,
      'deploymentPerformed':False,'promotionPerformed':False,
    }
    material=f'runtime:test/validations/{args.work_item}/repository-landscape-validation-manifest.json'
    mp=runtime/'test/validations'/args.work_item/'repository-landscape-validation-manifest.json'
    atomic_json(mp,manifest)
    refs=sorted(set([CONTRACT_REF,material]+references+chain_refs+evidence_refs))
    evidence=seal_evidence(manifest,args.repository_roots.resolve(),VALIDATOR_REF,refs,work_item=args.work_item,chain='FULL-REPOSITORY-LANDSCAPE')
    atomic_json(runtime/'test/reports/validation/full-repository-landscape.json',evidence)
    print(json.dumps({'status':manifest['status'],'repositoriesPassed':sum(x['status']=='PASSED' for x in root_results),'mandatoryChainsPassed':sum(x['status']=='PASSED' for x in chain_results),'requiredEvidenceFresh':sum(x['status']=='PASSED' for x in evidence_results),'completionBlockers':len(blockers)},indent=2))
    if blockers: raise SystemExit(1)

if __name__=='__main__':
    try: main()
    except Exception as exc:
        print('ERROR: '+str(exc),file=sys.stderr); raise SystemExit(1)