feat(P3): submission-readiness input contract — schema + validator + docs + tests (REQ-217..220)
REQ-217: schemas/submission-readiness.schema.json (JSON Schema draft 2020-12) defines acceptable-to-start as a superset gate above contract.schema.json: contractId, environment, tags (5 Nova tags D-054), policyPreconditions, profile (developer|agentic), appSource (repo+ref), per-env mandatory (W3.E: qa→e2eSuite+loadTest, prod→runbook+dashboard+oncall, dr→drDrillRef), agentic markers (naturalLanguageIntent+confidenceAtSubmission+agentTrace). REQ-218: core/submission_readiness.py validator with check_readiness() + ReadinessResult (structured pass/fail + reason codes). Wired as contract_ingestor.py --check-readiness (D-133). Reason codes: MISSING_TAGS, ENV_MISSING_MANDATORY, AGENTIC_MISSING_INTENT, MISSING_APP_SOURCE, POLICY_PRECONDITION_MISSING. Never raises — all failures are reason codes. REQ-219: docs/submission-readiness.md (good + rejected examples + reason-code catalog + compliance-standard equivalence). REQ-220: tests/test_submission_readiness.py — 16 tests, all pass. Covers: good-pass, good-agentic-pass, missing-tags, empty-tag, qa-missing-e2e, prod-missing-runbook, dr-missing-drdrill, prod-all-pass, agentic-missing-all, agentic-missing-one, missing-appsource, appsource-missing-ref, empty-policy, result-structure. ---ci--- project: acdl phase: 3 milestone: v1.18 status: execute requirements: covered: [REQ-217, REQ-218, REQ-219, REQ-220] partial: [] ---/ci---
This commit is contained in:
Jon Chery
@@ -499,4 +499,23 @@ def lambda_handler(event, context):
|
||||
return {"statusCode": 401, "body": json.dumps({"error": str(e)})}
|
||||
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
|
||||
except Exception as e: # pragma: no cover - defensive top-level guard
|
||||
return {"statusCode": 500, "body": json.dumps({"error": str(e)})}
|
||||
return {"statusCode": 500, "body": json.dumps({"error": str(e)})}
|
||||
|
||||
|
||||
# --- CLI: --check-readiness (D-133, REQ-218) ---------------------------
|
||||
# Invoked as: python3 -m core.lambda.contract_ingestor --check-readiness <submission.json>
|
||||
# Delegates to core.submission_readiness.check_readiness() and prints the
|
||||
# structured ReadinessResult. Exits 0 if ready, 1 if not.
|
||||
if __name__ == "__main__": # pragma: no cover - CLI entry
|
||||
import sys
|
||||
if "--check-readiness" in sys.argv:
|
||||
sys.path.insert(
|
||||
0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
)
|
||||
from core.submission_readiness import cli_main
|
||||
|
||||
# Strip the --check-readiness flag; pass the file path.
|
||||
rest = [a for a in sys.argv[1:] if a != "--check-readiness"]
|
||||
sys.exit(cli_main(["check-readiness"] + rest))
|
||||
else:
|
||||
print("Usage: python3 -m core.lambda.contract_ingestor --check-readiness <submission.json>")
|
||||
@@ -0,0 +1,193 @@
|
||||
"""core/submission_readiness.py — Nova submission-readiness validator (REQ-218).
|
||||
|
||||
Defines what is acceptable to start — a superset gate ABOVE
|
||||
contract.schema.json validity. Invoked as
|
||||
``contract_ingestor.py --check-readiness`` (D-133). Returns a structured
|
||||
ReadinessResult (pass/fail per check, with reason codes). On fail → the
|
||||
ingestor rejects with a citizen-developer-facing error (not a stack
|
||||
trace). On pass → proceeds to existing contract ingestion.
|
||||
|
||||
The validator calls contract.schema.json validation first (the shape),
|
||||
then the readiness checks (the gate): tags, env mandatory, policy
|
||||
preconditions, profile:agentic markers, appSource.
|
||||
|
||||
Reason codes:
|
||||
MISSING_TAGS — one or more required Nova tags are absent
|
||||
ENV_MISSING_MANDATORY:<env>:<field> — a per-env mandatory field is missing
|
||||
AGENTIC_MISSING_INTENT — profile=agentic but naturalLanguageIntent absent
|
||||
MISSING_APP_SOURCE — appSource (repo + ref) is missing
|
||||
POLICY_PRECONDITION_MISSING — a declared policy precondition is absent
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
_SCHEMA_DIR = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "schemas"
|
||||
)
|
||||
|
||||
REQUIRED_TAGS = [
|
||||
"nova:owner",
|
||||
"nova:contract",
|
||||
"nova:environment",
|
||||
"nova:cost-center",
|
||||
"nova:ref",
|
||||
]
|
||||
|
||||
ENV_MANDATORY: dict[str, list[str]] = {
|
||||
"dev": [], # dev requires only the base contract shape (id+environment+infrastructure)
|
||||
"qa": ["validation.e2eSuite", "validation.loadTest"],
|
||||
"prod": ["runbook", "dashboard", "oncall"],
|
||||
"dr": ["drDrillRef"],
|
||||
}
|
||||
|
||||
AGENTIC_REQUIRED = ["naturalLanguageIntent", "confidenceAtSubmission", "agentTrace"]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ReadinessResult:
|
||||
"""Structured result of the submission-readiness gate."""
|
||||
|
||||
ready: bool
|
||||
reason_codes: list[str] = field(default_factory=list)
|
||||
contract_id: str | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"ready": self.ready,
|
||||
"reason_codes": self.reason_codes,
|
||||
"contractId": self.contract_id,
|
||||
}
|
||||
|
||||
def __str__(self) -> str:
|
||||
if self.ready:
|
||||
return f"READY — contract {self.contract_id} passes submission-readiness gate"
|
||||
codes = "; ".join(self.reason_codes) if self.reason_codes else "unknown"
|
||||
return f"NOT READY — contract {self.contract_id}: {codes}"
|
||||
|
||||
|
||||
def _validate_contract_schema(contract: dict[str, Any]) -> list[str]:
|
||||
"""Validate the contract against contract.schema.json (the shape).
|
||||
Returns a list of reason codes (empty if valid). Falls back to no-op
|
||||
if jsonschema or the schema file is unavailable (the contract is
|
||||
validated upstream by run_platform.sh in the normal path).
|
||||
"""
|
||||
codes: list[str] = []
|
||||
try:
|
||||
import jsonschema
|
||||
|
||||
schema_path = os.path.join(_SCHEMA_DIR, "contract.schema.json")
|
||||
with open(schema_path) as f:
|
||||
schema = json.load(f)
|
||||
jsonschema.validate(instance=contract, schema=schema)
|
||||
except (OSError, ImportError):
|
||||
pass
|
||||
except jsonschema.ValidationError as e:
|
||||
codes.append(f"CONTRACT_SCHEMA_INVALID:{e.message}")
|
||||
return codes
|
||||
|
||||
|
||||
def _get_nested(data: dict[str, Any], dotted_key: str) -> Any:
|
||||
parts = dotted_key.split(".")
|
||||
val: Any = data
|
||||
for p in parts:
|
||||
if not isinstance(val, dict) or p not in val:
|
||||
return None
|
||||
val = val[p]
|
||||
return val
|
||||
|
||||
|
||||
def check_readiness(submission: dict[str, Any]) -> ReadinessResult:
|
||||
"""Run the full submission-readiness gate.
|
||||
|
||||
1. Validate the contract shape (contract.schema.json).
|
||||
2. Validate the readiness schema (submission-readiness.schema.json).
|
||||
3. Run the semantic readiness checks (tags, env mandatory, agentic, appSource, policy).
|
||||
|
||||
Returns a ReadinessResult. Never raises — all failures are reason codes.
|
||||
"""
|
||||
contract_id = submission.get("contractId") or submission.get("id", "unknown")
|
||||
codes: list[str] = []
|
||||
|
||||
# Step 1: contract shape validation
|
||||
contract_shape = {k: v for k, v in submission.items() if k in ("id", "name", "environment", "infrastructure")}
|
||||
if contract_shape:
|
||||
codes.extend(_validate_contract_schema(contract_shape))
|
||||
|
||||
# Step 2: readiness schema validation
|
||||
try:
|
||||
import jsonschema
|
||||
|
||||
schema_path = os.path.join(_SCHEMA_DIR, "submission-readiness.schema.json")
|
||||
with open(schema_path) as f:
|
||||
readiness_schema = json.load(f)
|
||||
jsonschema.validate(instance=submission, schema=readiness_schema)
|
||||
except (OSError, ImportError):
|
||||
pass
|
||||
except jsonschema.ValidationError as e:
|
||||
codes.append(f"READINESS_SCHEMA_INVALID:{e.message}")
|
||||
|
||||
# Step 3: semantic checks (reason codes for citizen-developer-facing errors)
|
||||
|
||||
# 3a: tags
|
||||
tags = submission.get("tags", {})
|
||||
missing_tags = [t for t in REQUIRED_TAGS if t not in tags or not tags[t]]
|
||||
if missing_tags:
|
||||
codes.append(f"MISSING_TAGS:{','.join(missing_tags)}")
|
||||
|
||||
# 3b: env mandatory (W3.E per-env table)
|
||||
env = submission.get("environment")
|
||||
if env and env in ENV_MANDATORY:
|
||||
for field_key in ENV_MANDATORY[env]:
|
||||
val = _get_nested(submission, field_key)
|
||||
if val is None:
|
||||
codes.append(f"ENV_MISSING_MANDATORY:{env}:{field_key}")
|
||||
|
||||
# 3c: agentic profile markers
|
||||
if submission.get("profile") == "agentic":
|
||||
for marker in AGENTIC_REQUIRED:
|
||||
if not submission.get(marker):
|
||||
codes.append(f"AGENTIC_MISSING_INTENT:{marker}")
|
||||
|
||||
# 3d: appSource
|
||||
app_source = submission.get("appSource")
|
||||
if not app_source or not app_source.get("repo") or not app_source.get("ref"):
|
||||
codes.append("MISSING_APP_SOURCE")
|
||||
|
||||
# 3e: policy preconditions (warn if declared but not enforced this milestone)
|
||||
policy = submission.get("policyPreconditions", {})
|
||||
if not policy:
|
||||
codes.append("POLICY_PRECONDITION_MISSING")
|
||||
|
||||
ready = len(codes) == 0
|
||||
return ReadinessResult(ready=ready, reason_codes=codes, contract_id=contract_id)
|
||||
|
||||
|
||||
def cli_main(argv: list[str]) -> int:
|
||||
"""CLI entry: python3 -m core.submission_readiness <contract.json>
|
||||
|
||||
Also invoked via contract_ingestor.py --check-readiness (D-133).
|
||||
Prints the ReadinessResult to stdout; exits 0 if ready, 1 if not.
|
||||
"""
|
||||
if len(argv) < 2:
|
||||
print("Usage: submission_readiness <contract.json>", file=sys.stderr)
|
||||
return 2
|
||||
path = argv[1]
|
||||
try:
|
||||
with open(path) as f:
|
||||
submission = json.load(f)
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f"ERROR: cannot read {path}: {e}", file=sys.stderr)
|
||||
return 2
|
||||
result = check_readiness(submission)
|
||||
print(result)
|
||||
print(json.dumps(result.to_dict(), indent=2))
|
||||
return 0 if result.ready else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(cli_main(sys.argv))
|
||||
Reference in New Issue
Block a user