#!/usr/bin/env python3
"""
GEARM 0.1 automated machine-conformance regression runner (GEARM-A12).

The runner executes the packaged A10/A11 testcase manifest across the machine
validation layers established by GEARM-A5..A8:

1. A7 Draft 2020-12 JSON / JSON-LD structural validation.
2. A5 offline JSON-LD expansion to RDF.
3. A6 SHACL machine constraints over the declared validation graph.
4. A8 Profile Schema structural validation.
5. Comparison of actual machine outcomes with manifest expectations.

Important semantic boundary:
A suite PASS means that the packaged machine validators behaved as expected.
It does NOT establish complete GEARM conformance where semantic assessment is
required by A3/A6 or by the human-readable GEARM specification.
"""

from __future__ import annotations

import argparse
import copy
import json
import sys
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from jsonschema import Draft202012Validator, FormatChecker
from rdflib import Dataset, Graph, Namespace, RDF, URIRef

try:
    from pyshacl import validate as pyshacl_validate
except ImportError:  # Optional cross-check engine.
    pyshacl_validate = None

SH = Namespace("http://www.w3.org/ns/shacl#")
GEARMSH = Namespace("https://specs.kivanura.org/gearm/shacl#")

EXIT_OK = 0
EXIT_DEPENDENCY_OR_RUNTIME = 2
EXIT_EXPECTATION_MISMATCH = 3

RUNNER_VERSION = "0.1.0-draft"
MACHINE_LAYERS = ("jsonSchema", "jsonLdExpansion", "shacl", "profileSchema")
ALLOWED_MACHINE_RESULTS = {"pass", "fail", "not-run"}
ALLOWED_SEMANTIC_ASSESSMENT = {"required", "not-required", "not-applicable"}
ALLOWED_COMPLETE_CONFORMANCE = {"not-claimed"}


def package_root() -> Path:
    """Return static/specs/gearm/0.1 regardless of current working directory."""
    return Path(__file__).resolve().parents[1]


def testcase_root(root: Path) -> Path:
    return root / "testcases"


def load_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def _resolve_context_value(value: Any, base_dir: Path) -> Any:
    """Inline packaged relative JSON-LD contexts for deterministic offline parsing."""
    if isinstance(value, str):
        candidate = (base_dir / value).resolve()
        if candidate.exists() and candidate.is_file():
            obj = load_json(candidate)
            return obj.get("@context", obj)
        return value
    if isinstance(value, list):
        return [_resolve_context_value(v, base_dir) for v in value]
    if isinstance(value, dict):
        return copy.deepcopy(value)
    return value


def load_jsonld_offline(path: Path) -> Dict[str, Any]:
    obj = load_json(path)
    if not isinstance(obj, dict):
        raise ValueError(f"JSON-LD testcase must be an object: {path}")
    obj = copy.deepcopy(obj)
    if "@context" in obj:
        obj["@context"] = _resolve_context_value(obj["@context"], path.parent)
    return obj


def json_schema_result(document: Any, schema: Dict[str, Any]) -> Tuple[str, List[str]]:
    validator = Draft202012Validator(schema, format_checker=FormatChecker())
    errors = sorted(validator.iter_errors(document), key=lambda e: list(e.absolute_path))
    if not errors:
        return "pass", []
    messages: List[str] = []
    for error in errors:
        loc = "$" + "".join(f"[{json.dumps(x)}]" for x in error.absolute_path)
        messages.append(f"{loc}: {error.message}")
    return "fail", messages


def jsonld_graph(path: Path) -> Graph:
    obj = load_jsonld_offline(path)
    # GEARM semantic documents may use an identified top-level @graph. JSON-LD
    # represents that as a named graph, so parse into a Dataset and return the
    # union of all quads rather than only the default graph.
    dataset = Dataset()
    dataset.parse(
        data=json.dumps(obj),
        format="json-ld",
        publicID=path.resolve().as_uri(),
    )
    graph = Graph()
    for subject, predicate, obj_value, _context in dataset.quads((None, None, None, None)):
        graph.add((subject, predicate, obj_value))
    return graph


def turtle_graph(path: Path) -> Graph:
    return Graph().parse(path, format="turtle")


def union_graphs(*graphs: Graph) -> Graph:
    result = Graph()
    for graph in graphs:
        for triple in graph:
            result.add(triple)
    return result


def invariant_id(value: Any) -> str:
    text = str(value)
    return text.rstrip("/").split("/")[-1].split("#")[-1]


def _shape_invariant(shapes_graph: Graph, shape: Any) -> Optional[str]:
    value = shapes_graph.value(shape, GEARMSH.implementsInvariant)
    return invariant_id(value) if value is not None else None


def _native_shacl_supported(shapes_graph: Graph) -> None:
    """
    Fail closed if A6 begins using SHACL features the built-in A12 engine does
    not understand. This prevents a future shape from being silently skipped.
    """
    shapes = list(shapes_graph.subjects(RDF.type, SH.NodeShape))
    if not shapes:
        raise RuntimeError("A6 contains no sh:NodeShape resources")

    allowed_shape_predicates = {
        RDF.type,
        SH.targetNode,
        SH.severity,
        SH.sparql,
        SH.property,
        GEARMSH.implementsInvariant,
        GEARMSH.constraintKind,
        GEARMSH.validationScope,
        GEARMSH.requiresSemanticAssessment,
    }
    allowed_sparql_predicates = {RDF.type, SH.message, SH.select}
    allowed_property_predicates = {SH.path, SH.hasValue, SH.message}

    for shape in shapes:
        predicates = {p for p, _ in shapes_graph.predicate_objects(shape)}
        unknown = predicates - allowed_shape_predicates
        if unknown:
            raise RuntimeError(
                f"Unsupported A6 NodeShape predicate(s) on {shape}: "
                + ", ".join(sorted(map(str, unknown)))
            )
        targets = list(shapes_graph.objects(shape, SH.targetNode))
        if len(targets) != 1:
            raise RuntimeError(f"A12 native engine requires exactly one sh:targetNode on {shape}")

        constraints = 0
        for sparql_node in shapes_graph.objects(shape, SH.sparql):
            constraints += 1
            predicates = {p for p, _ in shapes_graph.predicate_objects(sparql_node)}
            unknown = predicates - allowed_sparql_predicates
            if unknown:
                raise RuntimeError(
                    f"Unsupported SHACL-SPARQL predicate(s) on {sparql_node}: "
                    + ", ".join(sorted(map(str, unknown)))
                )
            if shapes_graph.value(sparql_node, SH.select) is None:
                raise RuntimeError(f"Missing sh:select on {sparql_node}")

        for property_node in shapes_graph.objects(shape, SH.property):
            constraints += 1
            predicates = {p for p, _ in shapes_graph.predicate_objects(property_node)}
            unknown = predicates - allowed_property_predicates
            if unknown:
                raise RuntimeError(
                    f"Unsupported SHACL property predicate(s) on {property_node}: "
                    + ", ".join(sorted(map(str, unknown)))
                )
            if shapes_graph.value(property_node, SH.path) is None:
                raise RuntimeError(f"Missing sh:path on {property_node}")
            if shapes_graph.value(property_node, SH.hasValue) is None:
                raise RuntimeError(f"Missing sh:hasValue on {property_node}")

        if constraints == 0:
            raise RuntimeError(f"A6 NodeShape has no supported constraint: {shape}")


def run_shacl_native(data_graph: Graph, shapes_graph: Graph) -> Tuple[str, List[Dict[str, Any]]]:
    """Execute the exact SHACL feature subset used by GEARM-A6."""
    _native_shacl_supported(shapes_graph)
    findings: List[Dict[str, Any]] = []

    for shape in shapes_graph.subjects(RDF.type, SH.NodeShape):
        target = shapes_graph.value(shape, SH.targetNode)
        severity = shapes_graph.value(shape, SH.severity) or SH.Violation
        inv = _shape_invariant(shapes_graph, shape)

        for sparql_node in shapes_graph.objects(shape, SH.sparql):
            query = str(shapes_graph.value(sparql_node, SH.select))
            message = str(shapes_graph.value(sparql_node, SH.message) or "SHACL-SPARQL violation")
            rows = list(data_graph.query(query, initBindings={"this": target}))
            if rows:
                findings.append(
                    {
                        "shape": str(shape),
                        "invariant": inv,
                        "severity": str(severity),
                        "message": message,
                        "constraintKind": "sparql",
                        "resultCount": len(rows),
                    }
                )

        for property_node in shapes_graph.objects(shape, SH.property):
            path = shapes_graph.value(property_node, SH.path)
            required = shapes_graph.value(property_node, SH.hasValue)
            message = str(shapes_graph.value(property_node, SH.message) or "SHACL property violation")
            if (target, path, required) not in data_graph:
                findings.append(
                    {
                        "shape": str(shape),
                        "invariant": inv,
                        "severity": str(severity),
                        "message": message,
                        "constraintKind": "hasValue",
                        "path": str(path),
                        "requiredValue": str(required),
                    }
                )

    fails = [f for f in findings if f["severity"] == str(SH.Violation)]
    return ("fail" if fails else "pass"), findings


def run_shacl_pyshacl(data_graph: Graph, shapes_graph: Graph) -> Tuple[str, List[Dict[str, Any]]]:
    if pyshacl_validate is None:
        raise RuntimeError("pyshacl is not installed; use --shacl-engine native or install pyshacl")

    conforms, report_graph, _ = pyshacl_validate(
        data_graph=data_graph,
        shacl_graph=shapes_graph,
        inference="none",
        advanced=True,
        allow_warnings=True,
        allow_infos=True,
        abort_on_first=False,
        meta_shacl=False,
    )

    findings: List[Dict[str, Any]] = []
    for result in report_graph.subjects(RDF.type, SH.ValidationResult):
        shape = report_graph.value(result, SH.sourceShape)
        severity = report_graph.value(result, SH.resultSeverity) or SH.Violation
        message = report_graph.value(result, SH.resultMessage)
        inv = _shape_invariant(shapes_graph, shape) if shape is not None else None
        findings.append(
            {
                "shape": str(shape) if shape is not None else None,
                "invariant": inv,
                "severity": str(severity),
                "message": str(message) if message is not None else "SHACL violation",
                "constraintKind": "pyshacl",
            }
        )

    fails = [f for f in findings if f["severity"] == str(SH.Violation)]
    status = "fail" if (not bool(conforms) or fails) else "pass"
    return status, findings


def choose_shacl_engine(requested: str) -> str:
    if requested == "auto":
        return "pyshacl" if pyshacl_validate is not None else "native"
    if requested == "pyshacl" and pyshacl_validate is None:
        raise RuntimeError("--shacl-engine pyshacl requested but pyshacl is not installed")
    return requested


def run_shacl(data_graph: Graph, shapes_graph: Graph, engine: str) -> Tuple[str, List[Dict[str, Any]]]:
    if engine == "native":
        return run_shacl_native(data_graph, shapes_graph)
    if engine == "pyshacl":
        return run_shacl_pyshacl(data_graph, shapes_graph)
    raise ValueError(f"Unsupported SHACL engine: {engine}")


def manifest_band(testcase: Dict[str, Any]) -> str:
    ident = testcase["id"]
    number = int(ident.rsplit("-", 1)[-1])
    if 1 <= number <= 99:
        return "positive"
    if 100 <= number <= 199:
        return "structural-invalid"
    if 200 <= number <= 299:
        return "semantic-invalid"
    if 300 <= number <= 399:
        return "profile-invalid"
    return "other"


def validate_manifest(manifest: Dict[str, Any], root: Path) -> None:
    if manifest.get("specification") != {"id": "GEARM", "version": "0.1"}:
        raise RuntimeError("testcases/manifest.json does not identify GEARM 0.1")

    seen: set[str] = set()
    for testcase in manifest.get("testcases", []):
        ident = testcase.get("id")
        if not isinstance(ident, str) or not ident.startswith("GEARM-TC-"):
            raise RuntimeError(f"Invalid testcase ID: {ident!r}")
        if ident in seen:
            raise RuntimeError(f"Duplicate testcase ID: {ident}")
        seen.add(ident)

        if "file" in testcase:
            path = testcase_root(root) / testcase["file"]
        elif "artifact" in testcase:
            path = (testcase_root(root) / testcase["artifact"]).resolve()
        else:
            raise RuntimeError(f"{ident} has neither file nor artifact")
        if not path.exists():
            raise RuntimeError(f"{ident} references missing path: {path}")

        expected = testcase.get("expected", {})
        for layer in MACHINE_LAYERS:
            value = expected.get(layer)
            if value not in ALLOWED_MACHINE_RESULTS:
                raise RuntimeError(f"{ident} has invalid expected {layer}: {value!r}")
        semantic = expected.get("semanticAssessment")
        if semantic not in ALLOWED_SEMANTIC_ASSESSMENT:
            raise RuntimeError(f"{ident} has invalid semanticAssessment: {semantic!r}")
        complete = expected.get("completeConformance")
        if complete not in ALLOWED_COMPLETE_CONFORMANCE:
            raise RuntimeError(f"{ident} has invalid completeConformance: {complete!r}")


def _resolve_testcase_path(root: Path, testcase: Dict[str, Any]) -> Path:
    if "file" in testcase:
        return testcase_root(root) / testcase["file"]
    return (testcase_root(root) / testcase["artifact"]).resolve()


def _build_shacl_data_graph(
    root: Path,
    testcase: Dict[str, Any],
    fixture_graph: Optional[Graph],
    canonical_graph: Graph,
) -> Graph:
    if "artifact" in testcase:
        artifact_path = _resolve_testcase_path(root, testcase)
        if artifact_path.suffix.lower() in {".ttl", ".turtle"}:
            return turtle_graph(artifact_path)
        raise RuntimeError(f"Unsupported SHACL artifact format for {testcase['id']}: {artifact_path}")

    if fixture_graph is None:
        raise RuntimeError(f"SHACL requested but no RDF fixture graph is available for {testcase['id']}")

    mode = testcase.get("shaclGraphMode")
    if mode is None:
        # A10 semantic graph fixtures are defined as canonical A4 + fixture.
        mode = "canonical-plus-fixture"
    if mode == "canonical-plus-fixture":
        return union_graphs(canonical_graph, fixture_graph)
    if mode == "fixture-only":
        return fixture_graph
    raise RuntimeError(f"Unsupported shaclGraphMode for {testcase['id']}: {mode}")


def run_testcase(
    root: Path,
    testcase: Dict[str, Any],
    general_schema: Dict[str, Any],
    profile_schema: Dict[str, Any],
    canonical_graph: Graph,
    shapes_graph: Graph,
    shacl_engine: str,
) -> Dict[str, Any]:
    expected = testcase["expected"]
    path = _resolve_testcase_path(root, testcase)

    actual: Dict[str, str] = {layer: "not-run" for layer in MACHINE_LAYERS}
    messages: Dict[str, List[str]] = {layer: [] for layer in MACHINE_LAYERS}
    violations: List[Dict[str, Any]] = []
    document: Optional[Any] = None
    fixture_graph: Optional[Graph] = None

    needs_json = any(expected[layer] != "not-run" for layer in ("jsonSchema", "jsonLdExpansion", "profileSchema"))
    if needs_json:
        try:
            document = load_json(path)
        except Exception as exc:
            # JSON parse failure is a structural failure. Current A10/A11 fixtures
            # are valid JSON, but this behaviour keeps the runner fail-closed.
            messages["jsonSchema"].append(f"JSON parse error: {exc}")
            if expected["jsonSchema"] != "not-run":
                actual["jsonSchema"] = "fail"
            if expected["profileSchema"] != "not-run":
                actual["profileSchema"] = "fail"

    if expected["jsonSchema"] != "not-run" and document is not None:
        status, errs = json_schema_result(document, general_schema)
        actual["jsonSchema"] = status
        messages["jsonSchema"].extend(errs)

    if expected["profileSchema"] != "not-run" and document is not None:
        status, errs = json_schema_result(document, profile_schema)
        actual["profileSchema"] = status
        messages["profileSchema"].extend(errs)

    if expected["jsonLdExpansion"] != "not-run":
        try:
            fixture_graph = jsonld_graph(path)
            actual["jsonLdExpansion"] = "pass"
        except Exception as exc:
            actual["jsonLdExpansion"] = "fail"
            messages["jsonLdExpansion"].append(str(exc))

    if expected["shacl"] != "not-run":
        try:
            if fixture_graph is None and "file" in testcase and path.suffix.lower() in {".json", ".jsonld"}:
                # SHACL may be requested even when expansion is not explicitly an
                # asserted layer. Build the RDF graph without changing the declared
                # jsonLdExpansion outcome.
                fixture_graph = jsonld_graph(path)
            data_graph = _build_shacl_data_graph(root, testcase, fixture_graph, canonical_graph)
            status, findings = run_shacl(data_graph, shapes_graph, shacl_engine)
            actual["shacl"] = status
            violations.extend(findings)
            messages["shacl"].extend(f["message"] for f in findings)
        except Exception as exc:
            raise RuntimeError(f"{testcase['id']} SHACL execution failed: {exc}") from exc

    mismatches: List[str] = []
    for layer in MACHINE_LAYERS:
        if actual[layer] != expected[layer]:
            mismatches.append(f"{layer}: actual={actual[layer]}, expected={expected[layer]}")

    expected_violation = testcase.get("failure", {}).get("expectedViolationInvariant")
    if expected_violation:
        reported = {v.get("invariant") for v in violations}
        if expected_violation not in reported:
            mismatches.append(
                f"expectedViolationInvariant not reported: {expected_violation}; "
                f"reported={sorted(x for x in reported if x)}"
            )

    try:
        report_path = str(path.resolve().relative_to(root.resolve()))
    except ValueError:
        report_path = str(path)

    return {
        "id": testcase["id"],
        "band": manifest_band(testcase),
        "kind": testcase.get("kind"),
        "path": report_path,
        "ok": not mismatches,
        "expected": {layer: expected[layer] for layer in MACHINE_LAYERS},
        "actual": actual,
        "semanticAssessmentObligation": expected.get("semanticAssessment"),
        "completeConformance": expected.get("completeConformance"),
        "violations": violations,
        "messages": messages,
        "mismatches": mismatches,
    }


def select_testcases(
    manifest: Dict[str, Any],
    case_ids: Optional[Sequence[str]],
    band: Optional[str],
) -> List[Dict[str, Any]]:
    testcases = manifest["testcases"]
    if case_ids:
        wanted = set(case_ids)
        available = {tc["id"] for tc in testcases}
        missing = sorted(wanted - available)
        if missing:
            raise RuntimeError("Unknown testcase ID(s): " + ", ".join(missing))
        testcases = [tc for tc in testcases if tc["id"] in wanted]
    if band:
        testcases = [tc for tc in testcases if manifest_band(tc) == band]
    return testcases


def suite_mode(
    root: Path,
    case_ids: Optional[Sequence[str]],
    band: Optional[str],
    shacl_engine: str,
    json_output: Optional[Path],
    quiet: bool,
) -> int:
    manifest_path = testcase_root(root) / "manifest.json"
    manifest = load_json(manifest_path)
    validate_manifest(manifest, root)

    general_schema = load_json(root / "gearm-0.1.schema.json")
    profile_schema = load_json(root / "gearm-0.1.profile.schema.json")
    canonical_graph = turtle_graph(root / "gearm-0.1.ttl")
    shapes_graph = turtle_graph(root / "gearm-0.1.shacl.ttl")

    engine = choose_shacl_engine(shacl_engine)
    if engine == "native":
        _native_shacl_supported(shapes_graph)

    selected = select_testcases(manifest, case_ids, band)
    if not selected:
        raise RuntimeError("No testcases selected")

    results = [
        run_testcase(
            root,
            testcase,
            general_schema,
            profile_schema,
            canonical_graph,
            shapes_graph,
            engine,
        )
        for testcase in selected
    ]

    matched = sum(1 for r in results if r["ok"])
    mismatched = len(results) - matched
    semantic_required = sum(1 for r in results if r["semanticAssessmentObligation"] == "required")

    report = {
        "specification": {"id": "GEARM", "version": "0.1"},
        "artifactStage": "GEARM-A12",
        "runnerVersion": RUNNER_VERSION,
        "manifestArtifactStage": manifest.get("artifactStage"),
        "manifestVersion": manifest.get("manifestVersion"),
        "shaclEngine": engine,
        "scope": {
            "caseIds": list(case_ids) if case_ids else None,
            "band": band,
        },
        "summary": {
            "testcases": len(results),
            "expectationsMatched": matched,
            "expectationMismatches": mismatched,
            "semanticAssessmentRequired": semantic_required,
            "machineRegression": "PASS" if mismatched == 0 else "FAIL",
            "completeConformance": "NOT_CLAIMED",
        },
        "results": results,
    }

    if not quiet:
        print("GEARM 0.1 machine-validation regression suite")
        print(f"Runner: GEARM-A12 {RUNNER_VERSION}")
        print(f"SHACL engine: {engine}")
        if band:
            print(f"Band: {band}")
        if case_ids:
            print("Cases: " + ", ".join(case_ids))
        print(f"Testcases: {len(results)}")
        for row in results:
            mark = "PASS" if row["ok"] else "FAIL"
            a = row["actual"]
            print(
                f"{mark:4} {row['id']}: "
                f"A7={a['jsonSchema']} JSON-LD={a['jsonLdExpansion']} "
                f"A6={a['shacl']} A8={a['profileSchema']}"
            )
            if not row["ok"]:
                for mismatch in row["mismatches"]:
                    print(f"      - {mismatch}")

        print(f"Expectation matches: {matched}/{len(results)}")
        print(f"Semantic assessment still required by manifest: {semantic_required}")
        print("Complete GEARM conformance: NOT CLAIMED")
        print("Overall machine regression: " + ("PASS" if mismatched == 0 else "FAIL"))

    if json_output:
        json_output.parent.mkdir(parents=True, exist_ok=True)
        json_output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
        if not quiet:
            print(f"JSON report: {json_output}")

    return EXIT_OK if mismatched == 0 else EXIT_EXPECTATION_MISMATCH


def list_mode(root: Path) -> int:
    manifest = load_json(testcase_root(root) / "manifest.json")
    validate_manifest(manifest, root)
    for testcase in manifest["testcases"]:
        print(f"{testcase['id']}\t{manifest_band(testcase)}\t{testcase.get('kind', '')}\t{testcase.get('description', '')}")
    return EXIT_OK


def main() -> int:
    parser = argparse.ArgumentParser(
        description="GEARM 0.1 automated machine-conformance regression runner (GEARM-A12)"
    )
    parser.add_argument("--suite", action="store_true", help="Run the packaged testcase suite (default).")
    parser.add_argument("--case", action="append", dest="case_ids", help="Run one testcase ID; repeatable.")
    parser.add_argument(
        "--band",
        choices=["positive", "structural-invalid", "semantic-invalid", "profile-invalid"],
        help="Run only one testcase band.",
    )
    parser.add_argument("--list", action="store_true", help="List packaged testcase IDs and exit.")
    parser.add_argument(
        "--shacl-engine",
        choices=["native", "pyshacl", "auto"],
        default="native",
        help="SHACL engine. 'native' is the deterministic A12 baseline; 'pyshacl' is optional; 'auto' prefers pyshacl when installed.",
    )
    parser.add_argument("--json-report", type=Path, help="Write a machine-readable JSON execution report.")
    parser.add_argument("--quiet", action="store_true", help="Suppress normal console output.")
    args = parser.parse_args()

    root = package_root()

    try:
        if args.list:
            return list_mode(root)
        return suite_mode(
            root=root,
            case_ids=args.case_ids,
            band=args.band,
            shacl_engine=args.shacl_engine,
            json_output=args.json_report,
            quiet=args.quiet,
        )
    except Exception as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return EXIT_DEPENDENCY_OR_RUNTIME


if __name__ == "__main__":
    raise SystemExit(main())