feat: generate evidence-based fit assessments

This commit is contained in:
2026-08-20 12:58:53 +02:00
parent 83ccb7f198
commit 5d4535f7b5
12 changed files with 1487 additions and 10 deletions
@@ -0,0 +1,126 @@
#!/usr/bin/env python3
"""Generate the human capability-fit report from its machine-readable input."""
from __future__ import annotations
import argparse
import os
from pathlib import Path
import sys
import tempfile
META_ROOT = Path(__file__).resolve().parents[2]
ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments"
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT):
if str(tools_root) not in sys.path:
sys.path.insert(0, str(tools_root))
from govoplan_assessment.report_generator import ( # noqa: E402
AssessmentGenerationError,
load_bounded_json,
render_assessment_markdown,
validate_report_input,
)
DEFAULT_ASSESSMENT = META_ROOT / "docs" / "capability-fit-current.json"
DEFAULT_SCHEMA = META_ROOT / "docs" / "capability-fit.schema.json"
DEFAULT_OUTPUT = (
META_ROOT
/ "docs"
/ "evidence"
/ "snapshots"
/ "CAPABILITY_AND_INFRASTRUCTURE_FIT.generated.md"
)
MAX_OUTPUT_BYTES = 16 * 1024 * 1024
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Render a deterministic human report from one capability-fit JSON input."
)
parser.add_argument("--assessment", type=Path, default=DEFAULT_ASSESSMENT)
parser.add_argument("--schema", type=Path, default=DEFAULT_SCHEMA)
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
parser.add_argument(
"--check",
action="store_true",
help="Fail when the output is missing or differs instead of writing it.",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
assessment = load_bounded_json(args.assessment, label="assessment")
schema = load_bounded_json(args.schema, label="assessment schema")
validate_report_input(assessment=assessment, schema=schema)
rendered = render_assessment_markdown(assessment)
encoded = rendered.encode("utf-8")
if len(encoded) > MAX_OUTPUT_BYTES:
raise AssessmentGenerationError(
f"Generated report exceeds the {MAX_OUTPUT_BYTES}-byte output limit"
)
if args.check:
try:
current = args.output.read_bytes()
except OSError:
current = None
if current != encoded:
print(
f"Capability-fit report is stale: {args.output}",
file=sys.stderr,
)
return 2
print(f"Capability-fit report is current: {args.output}")
return 0
_atomic_write(args.output, encoded)
print(f"Generated capability-fit report: {args.output}")
return 0
except AssessmentGenerationError as exc:
print(str(exc), file=sys.stderr)
return 1
def _atomic_write(path: Path, content: bytes) -> None:
if not path.parent.is_dir():
raise AssessmentGenerationError(
f"Output parent directory does not exist: {path.parent}"
)
if path.is_symlink():
raise AssessmentGenerationError("Output path must not be a symbolic link")
descriptor = -1
temporary_name = ""
try:
descriptor, temporary_name = tempfile.mkstemp(
prefix=".govoplan-fit-report-",
suffix=".tmp",
dir=path.parent,
)
os.fchmod(descriptor, 0o644)
with os.fdopen(descriptor, "wb", closefd=True) as handle:
descriptor = -1
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary_name, path)
temporary_name = ""
except OSError as exc:
raise AssessmentGenerationError(
f"Could not write generated report atomically: {exc}"
) from exc
finally:
if descriptor >= 0:
os.close(descriptor)
if temporary_name:
try:
os.unlink(temporary_name)
except FileNotFoundError:
pass
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,551 @@
"""Deterministically render one validated capability-fit assessment as Markdown."""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence
from jsonschema import Draft202012Validator, FormatChecker
from jsonschema.exceptions import SchemaError
MAX_ASSESSMENT_BYTES = 16 * 1024 * 1024
STATUS_DEFINITIONS = (
(
"verified",
"Implemented and directly exercised by evidence appropriate to the stated scope.",
),
(
"available_unconfigured",
"Implemented with supporting evidence, but not configured and exercised in the target.",
),
(
"partial",
"A useful subset exists, but a material part of the requirement is missing or unproved.",
),
(
"scaffold",
"Contracts or structure exist, but the end-to-end capability is not usable.",
),
(
"external_system",
"The deployment or another system must supply the capability.",
),
(
"planned",
"Only a concept, backlog item, or design direction exists.",
),
(
"not_fit",
"Evidence shows that the assessed composition cannot meet the requirement.",
),
(
"not_assessed",
"The requirement or target environment is not sufficiently known.",
),
)
class AssessmentGenerationError(ValueError):
"""The assessment cannot be safely validated or rendered."""
def load_bounded_json(path: Path, *, label: str) -> dict[str, Any]:
try:
size = path.stat().st_size
except OSError as exc:
raise AssessmentGenerationError(f"Could not inspect {label}: {exc}") from exc
if size > MAX_ASSESSMENT_BYTES:
raise AssessmentGenerationError(
f"{label} exceeds the {MAX_ASSESSMENT_BYTES}-byte input limit"
)
try:
payload = json.loads(
path.read_text(encoding="utf-8"),
object_pairs_hook=_unique_object,
)
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise AssessmentGenerationError(f"Could not read {label}: {exc}") from exc
if not isinstance(payload, dict):
raise AssessmentGenerationError(f"{label} must contain one JSON object")
return payload
def validate_report_input(
*,
assessment: Mapping[str, Any],
schema: Mapping[str, Any],
) -> None:
try:
Draft202012Validator.check_schema(schema)
except SchemaError as exc:
raise AssessmentGenerationError(
f"Assessment schema is invalid: {exc.message}"
) from exc
errors = sorted(
Draft202012Validator(
schema,
format_checker=FormatChecker(),
).iter_errors(assessment),
key=lambda item: tuple(str(part) for part in item.absolute_path),
)
if errors:
details = "; ".join(
f"{_json_path(error.absolute_path)}: {error.message}"
for error in errors[:20]
)
raise AssessmentGenerationError(f"Assessment does not match schema: {details}")
_validate_references(assessment)
_reject_sensitive_keys(assessment)
def render_assessment_markdown(assessment: Mapping[str, Any]) -> str:
"""Return stable Markdown derived only from a validated assessment object."""
assessment_hash = hashlib.sha256(
json.dumps(
assessment,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
).encode("utf-8")
).hexdigest()
scope = _mapping(assessment["scope"])
release = _mapping(assessment["release"])
profile = _mapping(assessment["deployment_profile"])
lines = [
"# GovOPlaN Capability and IT-Infrastructure Fit Assessment",
"",
"> Generated from [`capability-fit-current.json`](../../capability-fit-current.json).",
"> Edit and validate the machine-readable assessment, then regenerate this file;",
"> do not maintain conclusions independently in Markdown.",
"",
"This is an evidence-based fit assessment, not a production approval or",
"security certification. Repository or manifest existence alone never counts",
"as an implemented capability. Unknown target requirements remain explicitly",
"`not_assessed`.",
"",
"## Assessment record",
"",
"| Field | Value |",
"| --- | --- |",
f"| Assessment ID | `{_cell(assessment['assessment_id'])}` |",
f"| Schema version | `govoplan.fit-assessment/{_cell(assessment['schema_version'])}` |",
f"| Assessed on | {_cell(assessment['assessed_at'])} |",
f"| Scope | {_cell(scope['title'])} |",
f"| Release | `{_cell(release['ref'])}` ({_cell(release['kind'])}) |",
f"| Meta commit | `{_cell(release['meta_commit'])}` |",
f"| Deployment profile | `{_cell(profile['id'])}` · `{_cell(profile['status'])}` |",
f"| Configuration packages | {_inline_list(release['configuration_packages'], code=True)} |",
f"| Canonical input SHA-256 | `{assessment_hash}` |",
"",
"## Controlled status vocabulary",
"",
"| Status | Meaning |",
"| --- | --- |",
]
lines.extend(
f"| `{status}` | {_cell(description)} |"
for status, description in STATUS_DEFINITIONS
)
lines.extend(
[
"",
"## Scope and reference journeys",
"",
"Reference journeys:",
"",
*_bullets(scope["reference_journeys"]),
"",
"Explicitly postponed:",
"",
*_bullets(scope["postponed"]),
"",
"## Facts",
"",
*_bullets(assessment["facts"]),
"",
"## Decisions",
"",
*_bullets(assessment["decisions"]),
"",
"## Assumptions",
"",
*_bullets(assessment["assumptions"]),
"",
"## Unresolved decisions",
"",
*_bullets(assessment["open_questions"]),
"",
"## Pinned release and composition",
"",
f"Release reproducible: **{'yes' if release['reproducible'] else 'no'}**.",
"",
]
)
lines.extend(_notes(release.get("notes", [])))
lines.extend(
[
"",
"| Module | Repository and commit | Manifest version | Enabled | Role |",
"| --- | --- | --- | --- | --- |",
]
)
for module_value in assessment["composition"]:
module = _mapping(module_value)
lines.append(
"| `{}` | `{}` @ `{}` | `{}` | {} | {} |".format(
_cell(module["module_id"]),
_cell(module["repository"]),
_cell(module["commit"]),
_cell(module["manifest_version"]),
"yes" if module["enabled"] else "no",
_cell(module["role"]),
)
)
lines.extend(
[
"",
"## Deployment profile",
"",
f"Status: `{_cell(profile['status'])}`",
"",
_text(profile["description"]),
"",
"Evidence:",
"",
*_bullets(_evidence_labels(profile["evidence"])),
"",
"## Recommended scenarios",
"",
]
)
for scenario_value in assessment["scenarios"]:
scenario = _mapping(scenario_value)
lines.extend(
[
f"### {_text(scenario['label'])}",
"",
f"Status: `{_cell(scenario['status'])}`",
"",
_text(scenario["recommendation"]),
"",
f"Composition: {_inline_list(scenario['composition'], code=True)}.",
"",
"Topology:",
"",
*_bullets(scenario["topology"]),
"",
"Conditions:",
"",
*_bullets(scenario["conditions"]),
"",
]
)
functional_context = _mapping(assessment["functional_context"])
lines.extend(
[
"## Functional matrix context",
"",
"### Required modules",
"",
*_bullets(functional_context["required_modules"]),
"",
"### Optional modules",
"",
*_bullets(functional_context["optional_modules"]),
"",
"### External systems and connectors",
"",
*_bullets(functional_context["external_systems"]),
"",
"### Missing contracts",
"",
*_bullets(functional_context["missing_contracts"]),
"",
"### Policy decisions",
"",
*_bullets(functional_context["policy_decisions"]),
"",
"### Manual workarounds",
"",
*_bullets(functional_context["manual_workarounds"]),
"",
"### Blockers",
"",
*_bullets(functional_context["blockers"]),
"",
]
)
lines.extend(
[
"## Assessment questionnaire",
"",
"Every required area remains visible even when its target answer is unknown.",
"",
"| Area | Question | State | Answer | Evidence |",
"| --- | --- | --- | --- | --- |",
]
)
questionnaire = _mapping(assessment["questionnaire"])
for area, answers in questionnaire.items():
for answer_value in _sequence(answers):
answer = _mapping(answer_value)
raw_answer = answer["answer"]
answer_text = (
_inline_list(raw_answer)
if isinstance(raw_answer, list)
else _text(raw_answer) if raw_answer is not None else ""
)
lines.append(
"| {} | {} | `{}` | {} | {} |".format(
_cell(area.replace("_", " ").title()),
_cell(answer["question"]),
_cell(answer["state"]),
_cell(answer_text),
_cell("; ".join(_evidence_labels(answer["evidence"])) or ""),
)
)
lines.extend(_assessed_matrix("Functional capability matrix", assessment["capabilities"]))
lines.extend(_assessed_matrix("Infrastructure matrix", assessment["infrastructure"]))
lines.extend(
[
"## Data flows and trust boundaries",
"",
"| Flow | From → to | Data | Trust boundary | Controls |",
"| --- | --- | --- | --- | --- |",
]
)
for flow_value in assessment["data_flows"]:
flow = _mapping(flow_value)
lines.append(
"| `{}` | {}{} | {} | {} | {} |".format(
_cell(flow["id"]),
_cell(flow["from"]),
_cell(flow["to"]),
_cell(_inline_list(flow["data"])),
_cell(flow["trust_boundary"]),
_cell(_inline_list(flow["controls"])),
)
)
lines.extend(
[
"",
"## Risks and residual risks",
"",
"| Risk | Impact | Treatment | Owner | Residual risk |",
"| --- | --- | --- | --- | --- |",
]
)
for risk_value in assessment["risks"]:
risk = _mapping(risk_value)
lines.append(
"| **{}**<br>{} | {} | {} | {} | {} |".format(
_cell(risk["id"]),
_cell(risk["statement"]),
_cell(risk["impact"]),
_cell(risk["treatment"]),
_cell(risk["owner"] or "unassigned"),
_cell(risk["residual_risk"]),
)
)
lines.extend(
[
"",
"## Recommendations",
"",
*_bullets(assessment["recommendations"]),
"",
"## Proof-of-concept and promotion checks",
"",
*_numbered(assessment["proof_checks"]),
"",
"## Generation contract",
"",
"This report is deterministic output from the schema-validated JSON companion.",
"The generator rejects duplicate JSON keys, schema drift, secret-bearing field",
"names, stale checked-in output, and oversized inputs. A new assessment or",
"release changes the canonical input hash and requires review of the affected",
"evidence and conclusions through the release-aware reassessment tool.",
"",
]
)
return "\n".join(lines)
def _assessed_matrix(title: str, values: object) -> list[str]:
lines = [
"",
f"## {title}",
"",
"| Requirement | Status | Evidence | Conditions and gaps | Recommendation and proof |",
"| --- | --- | --- | --- | --- |",
]
for item_value in _sequence(values):
item = _mapping(item_value)
conditions = [f"Condition: {value}" for value in item["conditions"]]
gaps = [f"Gap: {value}" for value in item["gaps"]]
risks = [f"Risk: {value}" for value in item["risks"]]
lines.append(
"| **{}**<br>{} | `{}` | {} | {} | {}<br>**Proof:** {} |".format(
_cell(item["id"]),
_cell(item["requirement"]),
_cell(item["status"]),
_cell("; ".join(_evidence_labels(item["evidence"])) or "Explicit absence of evidence"),
_cell("; ".join([*conditions, *gaps, *risks]) or ""),
_cell(item["recommendation"] or ""),
_cell(item["proof_check"] or ""),
)
)
return lines
def _evidence_labels(values: object) -> list[str]:
labels: list[str] = []
for value in _sequence(values):
item = _mapping(value)
label = f"{item['kind']}/{item['scope']}: {item['locator']}"
if item.get("note"):
label += f" ({item['note']})"
labels.append(label)
return labels
def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise AssessmentGenerationError(f"Duplicate JSON key: {key!r}")
result[key] = value
return result
def _reject_sensitive_keys(value: object, path: tuple[str, ...] = ()) -> None:
forbidden = {
"access_token",
"api_key",
"credential_value",
"password",
"private_key",
"refresh_token",
"secret",
}
if isinstance(value, Mapping):
for key, nested in value.items():
normalized = str(key).strip().casefold()
if normalized in forbidden:
raise AssessmentGenerationError(
f"Assessment contains forbidden sensitive field {_json_path((*path, str(key)))}"
)
_reject_sensitive_keys(nested, (*path, str(key)))
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
for index, nested in enumerate(value):
_reject_sensitive_keys(nested, (*path, str(index)))
def _validate_references(assessment: Mapping[str, Any]) -> None:
composition = [_mapping(item) for item in _sequence(assessment["composition"])]
module_ids = [str(item["module_id"]) for item in composition]
if len(module_ids) != len(set(module_ids)):
raise AssessmentGenerationError("Composition contains duplicate module IDs")
modules = {str(item["module_id"]): item for item in composition}
context = _mapping(assessment["functional_context"])
required = {str(item) for item in _sequence(context["required_modules"])}
optional = {str(item) for item in _sequence(context["optional_modules"])}
unknown_context = (required | optional) - set(modules)
if unknown_context:
raise AssessmentGenerationError(
"Functional context references unknown modules: "
+ ", ".join(sorted(unknown_context))
)
if required & optional:
raise AssessmentGenerationError(
"Functional context cannot mark a module both required and optional"
)
for scenario_value in _sequence(assessment["scenarios"]):
scenario = _mapping(scenario_value)
referenced = {str(item) for item in _sequence(scenario["composition"])}
unknown = referenced - set(modules)
if unknown:
raise AssessmentGenerationError(
f"Scenario {scenario['id']!r} references unknown modules: "
+ ", ".join(sorted(unknown))
)
disabled = sorted(
module_id
for module_id in referenced
if not bool(modules[module_id]["enabled"])
)
if disabled:
raise AssessmentGenerationError(
f"Scenario {scenario['id']!r} references disabled modules: "
+ ", ".join(disabled)
)
for collection in ("capabilities", "infrastructure", "data_flows", "risks"):
identifiers = [
str(_mapping(item)["id"])
for item in _sequence(assessment[collection])
]
if len(identifiers) != len(set(identifiers)):
raise AssessmentGenerationError(
f"Assessment contains duplicate {collection} IDs"
)
def _mapping(value: object) -> Mapping[str, Any]:
if not isinstance(value, Mapping):
raise AssessmentGenerationError("Validated assessment contains a non-object value")
return value
def _sequence(value: object) -> Sequence[Any]:
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
raise AssessmentGenerationError("Validated assessment contains a non-list value")
return value
def _text(value: object) -> str:
return str(value).strip()
def _cell(value: object) -> str:
return _text(value).replace("|", "\\|").replace("\r", " ").replace("\n", " ")
def _inline_list(values: object, *, code: bool = False) -> str:
items = [_text(item) for item in _sequence(values)]
if not items:
return "none"
if code:
return ", ".join(f"`{_cell(item)}`" for item in items)
return "; ".join(items)
def _bullets(values: object) -> list[str]:
items = [_text(item) for item in _sequence(values)]
return [f"- {item}" for item in items] or ["- None recorded."]
def _numbered(values: object) -> list[str]:
return [f"{index}. {_text(item)}" for index, item in enumerate(_sequence(values), 1)]
def _notes(values: object) -> list[str]:
items = _bullets(values)
return ["Release notes:", "", *items]
def _json_path(parts: Iterable[object]) -> str:
suffix = "".join(f"[{part}]" if str(part).isdigit() else f".{part}" for part in parts)
return f"${suffix}"
__all__ = (
"AssessmentGenerationError",
"MAX_ASSESSMENT_BYTES",
"load_bounded_json",
"render_assessment_markdown",
"validate_report_input",
)
+2
View File
@@ -46,6 +46,8 @@ cd "$META_ROOT"
"$PYTHON" -m unittest tests.test_module_package_workflows tests.test_package_registry_release
"$PYTHON" -m unittest tests.test_deployment_installer
"$PYTHON" -m unittest tests.test_capability_fit_evidence
"$PYTHON" -m unittest tests.test_capability_fit_generation tests.test_capability_fit_review
"$PYTHON" tools/assessments/generate-capability-fit-report.py --check
"$PYTHON" -m unittest tests.test_configuration_package_artifacts
"$PYTHON" -m unittest tests.test_institutional_governance_journey
"$PYTHON" -m unittest tests.test_institutional_service_journey
+2
View File
@@ -26,6 +26,8 @@ cd "$ROOT"
"$PYTHON" "$META_ROOT/tools/checks/check_dependency_boundaries.py"
cd "$META_ROOT"
"$PYTHON" -m unittest tests.test_capability_fit_generation
"$PYTHON" tools/assessments/generate-capability-fit-report.py --check
"$PYTHON" -m unittest tests.test_configuration_package_artifacts
PYTHONPATH="$META_ROOT/../govoplan-portal/src:$META_ROOT/../govoplan-forms/src:$META_ROOT/../govoplan-forms-runtime/src:$META_ROOT/../govoplan-cases/src:$ROOT/src${PYTHONPATH:+:$PYTHONPATH}" \
"$PYTHON" -m unittest tests.test_institutional_service_journey