Architecture-as-Code/tools/validate.py

#!/usr/bin/env python3
"""Pārbauda repozitoriju pret shēmu un datu līguma noteikumiem (tās pašas pārbaudes, ko veic VDZP savienotājs).

  python3 tools/validate.py            (no repozitorija saknes; pip install pyyaml jsonschema)

Pārbaudes:
  catalogue_valid      domenas.yaml atbilst #/$defs/Catalogue; slug unikāli; katrai pārveidotajai jomai mape eksistē
  schema_valid         katrs jomas reģistrs atbilst savas lomas tipam (schemas/arhitektura-0.1.schema.json, x-roles)
  ids_unique           lokālie identifikatori jomā ir unikāli
  edges_resolve        katras saites from un to ir šīs jomas elements
"""
import json
import sys
from pathlib import Path

import yaml
from jsonschema import Draft202012Validator

ROOT = Path(__file__).resolve().parents[1]
SCHEMA = json.loads((ROOT / "schemas/arhitektura-0.1.schema.json").read_text(encoding="utf-8"))

ENTITY = {"organizations": "organisation", "functions": "function", "services": "service",
          "information_resources": "information_resource", "systems": "system", "goals": "goal",
          "roadmap": "roadmap_item", "interactions": "interaction", "risks": "risk", "as_is_catalog": "as_is_component"}


def validator(defname):
    return Draft202012Validator({"$schema": SCHEMA["$schema"], "$defs": SCHEMA["$defs"], "$ref": f"#/$defs/{defname}"})


def errors(defname, doc, where):
    return [f"{where}: {'/'.join(map(str, e.absolute_path)) or '(sakne)'}: {e.message[:200]}"
            for e in sorted(validator(defname).iter_errors(doc), key=lambda e: list(e.absolute_path))][:20]


def files_of(domain_dir, rel):
    p = domain_dir / rel
    if p.is_dir():
        return sorted(p.rglob("*.yaml"))
    return [p] if p.exists() else []


def domain(domain_dir):
    """-> (entities, edges, problems) for one domain folder."""
    slug = domain_dir.name
    problems = {"schema_valid": [], "ids_unique": [], "edges_resolve": []}
    man_path = domain_dir / "manifest.yaml"
    man = yaml.safe_load(man_path.read_text(encoding="utf-8"))
    problems["schema_valid"] += errors("Manifest", man, f"{slug}/manifest.yaml")
    if man.get("id") != f"domain.{slug}":
        problems["schema_valid"].append(f"{slug}/manifest.yaml: id {man.get('id')} ≠ domain.{slug}")
    entities, edges = [], []
    for role, rel in (man.get("registers") or {}).items():
        defname = SCHEMA["x-roles"].get(role, "AnyObject")
        for f in files_of(domain_dir, rel):
            doc = yaml.safe_load(f.read_text(encoding="utf-8")) or {}
            where = f"{slug}/{f.relative_to(domain_dir)}"
            problems["schema_valid"] += errors(defname, doc, where)
            if role == "relations":
                edges += [dict(e, file=where) for e in doc.get("edges", []) if isinstance(e, dict)]
            elif role in ENTITY:
                items = [doc] if defname == "Goal" else doc.get("items", []) if isinstance(doc, dict) else []
                entities += [dict(it, _type=ENTITY[role], _file=where) for it in items if isinstance(it, dict) and it.get("id")]
    # the domain and its subdomains are nodes too (edges start from domain.<slug>)
    meta = domain_dir / "registers/00-meta/domain.yaml"
    nodes = {f"domain.{slug}"}
    if meta.exists():
        m = yaml.safe_load(meta.read_text(encoding="utf-8")) or {}
        nodes |= {s["id"] for s in m.get("subdomains", []) if isinstance(s, dict) and s.get("id")}
    seen = {}
    for e in entities:
        if e["id"] in seen:
            problems["ids_unique"].append(f"{slug}: {e['id']} ({seen[e['id']]} un {e['_file']})")
        seen[e["id"]] = e["_file"]
    known = set(seen) | nodes
    for e in edges:
        for k in ("from", "to"):
            if e.get(k) not in known:
                problems["edges_resolve"].append(f"{e['file']}: {e.get('from')} -{e.get('type')}-> {e.get('to')} ({k} nav jomā)")
    return man, entities, edges, problems


def main():
    cat = yaml.safe_load((ROOT / "domenas.yaml").read_text(encoding="utf-8"))
    bad = errors("Catalogue", cat, "domenas.yaml")
    slugs = [d.get("slug") for d in cat.get("domains", [])]
    bad += [f"domenas.yaml: slug {s} atkārtojas" for s in set(slugs) if slugs.count(s) > 1]
    for d in cat.get("domains", []):
        if d.get("as_code", {}).get("status") == "transformed" and not (ROOT / d["as_code"].get("path", "-") / "manifest.yaml").exists():
            bad.append(f"domenas.yaml: {d['slug']}: {d['as_code'].get('path')}/manifest.yaml nav")
    for p in sorted((ROOT / "domains").glob("*/manifest.yaml")):
        if p.parent.name not in slugs:
            bad.append(f"domains/{p.parent.name}: jomas nav domenas.yaml")
    total = {"catalogue_valid": bad}
    for p in sorted((ROOT / "domains").glob("*/manifest.yaml")):
        man, ents, edges, prob = domain(p.parent)
        print(f"{p.parent.name}: {len(ents)} elementi, {len(edges)} saites")
        for k, v in prob.items():
            total.setdefault(k, []).extend(v)
    ok = True
    for k, v in total.items():
        print(f"{k}: {'OK' if not v else f'{len(v)} kļūdas'}")
        for x in v[:10]:
            print("   ", x)
        ok &= not v
    n = sum(1 for d in cat["domains"] if d["as_code"]["status"] == "transformed")
    print(f"Jomas: {len(cat['domains'])}, pārveidotas kodā: {n}")
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()