112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = ["jsonschema>=4.23,<5", "PyYAML>=6,<7"]
|
|
# ///
|
|
"""Validate tenancy.yaml files against the canonical syntax and semantics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
from typing import Any
|
|
|
|
import jsonschema
|
|
import yaml
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
|
SCHEMA = ROOT / "canon/schemas/tenancy-posture_v0.1.schema.json"
|
|
AXES = ("I", "A", "E", "P", "R", "V")
|
|
FLOORS = {"I": 1, "A": 1, "E": 0, "P": 0, "R": 1, "V": 0}
|
|
|
|
|
|
def services(document: dict[str, Any]) -> list[dict[str, Any]]:
|
|
if "services" in document:
|
|
return document["services"]
|
|
return [document]
|
|
|
|
|
|
def validate_semantics(document: dict[str, Any], path: pathlib.Path) -> list[str]:
|
|
errors: list[str] = []
|
|
entries = services(document)
|
|
names = [entry["service"] for entry in entries]
|
|
if len(names) != len(set(names)):
|
|
errors.append("service names must be unique")
|
|
|
|
for entry in entries:
|
|
name = entry["service"]
|
|
posture = entry["tenancy"]
|
|
current = posture["current"]
|
|
reason = posture.get("reason", {})
|
|
gap = posture.get("gap", {})
|
|
evidence = entry.get("evidence", {})
|
|
|
|
reviewed = dt.date.fromisoformat(posture["reviewed"])
|
|
review_due = dt.date.fromisoformat(posture["review_due"])
|
|
if review_due < reviewed:
|
|
errors.append(f"{name}: review_due precedes reviewed")
|
|
|
|
for axis in AXES:
|
|
level = current[axis]
|
|
explained = bool(reason.get(axis) or gap.get(axis))
|
|
if level == "n/a" or (isinstance(level, int) and level <= FLOORS[axis]):
|
|
if not explained:
|
|
errors.append(f"{name}: {axis}{level} needs reason or gap text")
|
|
continue
|
|
key = f"{axis}{level}"
|
|
if key not in evidence:
|
|
errors.append(f"{name}: current {key} has no evidence entry")
|
|
|
|
implemented = posture.get("implemented", {})
|
|
for axis, level in implemented.items():
|
|
current_level = current[axis]
|
|
if current_level != "n/a" and level != "n/a" and level <= current_level:
|
|
errors.append(
|
|
f"{name}: implemented {axis}{level} must be above current {axis}{current_level}"
|
|
)
|
|
|
|
provider = entry.get("provider", {})
|
|
for axis, reach in provider.get("axes", {}).items():
|
|
available = reach["available"]
|
|
maximum = reach["maximum"]
|
|
if available != "n/a" and maximum != "n/a" and available > maximum:
|
|
errors.append(
|
|
f"{name}: provider {axis} available {available} exceeds maximum {maximum}"
|
|
)
|
|
|
|
return [f"{path}: {error}" for error in errors]
|
|
|
|
|
|
def validate(path: pathlib.Path, schema: dict[str, Any]) -> list[str]:
|
|
document = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
validator = jsonschema.Draft202012Validator(
|
|
schema, format_checker=jsonschema.FormatChecker()
|
|
)
|
|
errors = [
|
|
f"{path}: {'/'.join(str(part) for part in error.absolute_path) or '<root>'}: {error.message}"
|
|
for error in sorted(validator.iter_errors(document), key=lambda item: list(item.path))
|
|
]
|
|
if not errors:
|
|
errors.extend(validate_semantics(document, path))
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("declarations", nargs="+", type=pathlib.Path)
|
|
args = parser.parse_args()
|
|
schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
|
|
errors = [error for path in args.declarations for error in validate(path, schema)]
|
|
if errors:
|
|
print("\n".join(errors), file=sys.stderr)
|
|
return 1
|
|
for path in args.declarations:
|
|
print(f"{path}: valid")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|