#!/usr/bin/env python3 """GEDRM 0.1 automated conformance runner. Validates: 1. JSON/JSON-LD structure against gedrm-0.1.schema.json 2. RDF expansion using the local GEDRM JSON-LD context 3. SHACL semantic conformance against gedrm-0.1.shacl.ttl If pyshacl is installed, it is used as the authoritative SHACL engine. If it is unavailable, the runner uses a deterministic GEDRM-0.1 core fallback covering the constraints present in the current core SHACL file. The fallback exists so the repository can run offline; CI SHOULD install pyshacl. """ from __future__ import annotations import argparse, json, sys from pathlib import Path from typing import List, Tuple import jsonschema from rdflib import Graph, Namespace, RDF from rdflib.namespace import RDFS GEDRM = Namespace('https://specs.kivanura.org/gedrm#') try: from pyshacl import validate as pyshacl_validate HAVE_PYSHACL = True except Exception: HAVE_PYSHACL = False def load_json(path: Path): return json.loads(path.read_text(encoding='utf-8')) def localize_context(doc, context_path: Path): clone = json.loads(json.dumps(doc)) if '@context' in clone: clone['@context'] = load_json(context_path)['@context'] return clone def to_graph(path: Path, context_path: Path) -> Graph: doc = localize_context(load_json(path), context_path) g = Graph() g.parse(data=json.dumps(doc), format='json-ld') return g def type_closure(g: Graph, ont: Graph, node, target_class) -> bool: direct = set(g.objects(node, RDF.type)) if target_class in direct: return True # RDFS subclass closure from ontology. frontier = list(direct) seen = set(direct) while frontier: cls = frontier.pop() for parent in ont.objects(cls, RDFS.subClassOf): if getattr(parent, 'startswith', None) and str(parent).startswith('http'): if parent == target_class: return True if parent not in seen: seen.add(parent); frontier.append(parent) return False def fallback_core_validate(data: Graph, ont: Graph) -> Tuple[bool, List[str]]: errs: List[str] = [] def nodes_of(cls): return set(data.subjects(RDF.type, cls)) def count(s,p): return len(set(data.objects(s,p))) for s in nodes_of(GEDRM.SystemOfRecord): if count(s,GEDRM.recordsAuthoritativeStateFor) < 1: errs.append(f'{s}: SystemOfRecord requires recordsAuthoritativeStateFor minCount 1') for s in nodes_of(GEDRM.AuthoritativeSource): if count(s,GEDRM.isAuthoritativeSourceFor) < 1: errs.append(f'{s}: AuthoritativeSource requires isAuthoritativeSourceFor minCount 1') for s in nodes_of(GEDRM.DataProduct): vals=set(data.objects(s,GEDRM.hasAccountableOwner)) if not vals: errs.append(f'{s}: DataProduct requires hasAccountableOwner minCount 1') for s in nodes_of(GEDRM.DataContract): if count(s,GEDRM.definesCommitmentFor) < 1: errs.append(f'{s}: DataContract requires definesCommitmentFor minCount 1') for s in nodes_of(GEDRM.DataControl): vals=set(data.objects(s,GEDRM.hasControlOwner)) if not vals: errs.append(f'{s}: DataControl requires hasControlOwner minCount 1') for v in vals: if not type_closure(data,ont,v,GEDRM.ControlOwner): errs.append(f'{s}: hasControlOwner {v} is not a ControlOwner') for s in nodes_of(GEDRM.ControlEvidence): controls=set(data.objects(s,GEDRM.evidenceForControl)) if len(controls)!=1: errs.append(f'{s}: ControlEvidence requires exactly one evidenceForControl') for v in controls: if not type_closure(data,ont,v,GEDRM.DataControl): errs.append(f'{s}: evidenceForControl {v} is not a DataControl') executions=set(data.objects(s,GEDRM.evidenceForExecution)) if len(executions)>1: errs.append(f'{s}: evidenceForExecution maxCount 1') for v in executions: if not type_closure(data,ont,v,GEDRM.ControlExecution): errs.append(f'{s}: evidenceForExecution {v} is not a ControlExecution') for ex in data.subjects(GEDRM.producesEvidence,s): if type_closure(data,ont,ex,GEDRM.ControlExecution) and ex not in executions: errs.append(f'{s}: execution {ex} produces evidence but is not identified by evidenceForExecution') for s in nodes_of(GEDRM.GoldenRecord): if count(s,GEDRM.represents)!=1: errs.append(f'{s}: GoldenRecord requires exactly one represents') for s in nodes_of(GEDRM.DerivedData): vals=set(data.objects(s,GEDRM.derivedFrom)) if not vals: errs.append(f'{s}: DerivedData requires derivedFrom minCount 1') for v in vals: if not type_closure(data,ont,v,GEDRM.DataObject): errs.append(f'{s}: derivedFrom {v} is not a DataObject') for s in nodes_of(GEDRM.DataProcessor): if count(s,GEDRM.processesOnBehalfOf)<1: errs.append(f'{s}: DataProcessor requires processesOnBehalfOf minCount 1') for s in nodes_of(GEDRM.ProcessingActivity): for v in data.objects(s,GEDRM.hasDataProcessor): if not type_closure(data,ont,v,GEDRM.DataProcessor): errs.append(f'{s}: hasDataProcessor {v} is not a DataProcessor') for v in data.objects(s,GEDRM.hasDataController): if not type_closure(data,ont,v,GEDRM.DataController): errs.append(f'{s}: hasDataController {v} is not a DataController') # The core SHACL missing-controller check is Warning severity; do not fail conformance. # Requirement traceability for p in (GEDRM.derivedFromRequirement, GEDRM.satisfiesRequirement): for s,o in data.subject_objects(p): if not type_closure(data,ont,o,GEDRM.RequirementReference): errs.append(f'{s}: {p.split("#")[-1]} target {o} is not a RequirementReference') return not errs, errs def shacl_validate_file(data_graph: Graph, shapes_path: Path, ont_path: Path): ont = Graph().parse(ont_path, format='turtle') if HAVE_PYSHACL: shapes = Graph().parse(shapes_path, format='turtle') conforms, report_graph, report_text = pyshacl_validate( data_graph, shacl_graph=shapes, ont_graph=ont, inference='rdfs', advanced=True, allow_warnings=True, allow_infos=True, ) return bool(conforms), [report_text.strip()] if not conforms else [] return fallback_core_validate(data_graph, ont) def run(root: Path, verbose=False) -> int: schema_path=root/'gedrm-0.1.schema.json' context_path=root/'gedrm-0.1.context.jsonld' shapes_path=root/'gedrm-0.1.shacl.ttl' ont_path=root/'gedrm-0.1.ttl' manifest=load_json(root/'testcases/manifest.json') schema=load_json(schema_path) validator=jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker()) total=passed=0 failures=[] def schema_ok(path): errors=sorted(validator.iter_errors(load_json(path)), key=lambda e:list(e.path)) return not errors, [e.message for e in errors] for folder, expect_schema, expect_shacl in [ ('valid', True, True), ('invalid', True, False), ('schema-invalid', False, None) ]: for path in sorted((root/'testcases'/folder).glob('*.jsonld')): total += 1 sok, serrs=schema_ok(path) actual_shacl=None; sherrs=[] if sok and expect_shacl is not None: try: graph=to_graph(path, context_path) actual_shacl, sherrs=shacl_validate_file(graph, shapes_path, ont_path) except Exception as e: sherrs=[f'RDF/SHACL execution error: {e}'] actual_shacl=False ok=(sok==expect_schema) and (expect_shacl is None or actual_shacl==expect_shacl) if ok: passed += 1 else: failures.append(path.name) status='PASS' if ok else 'FAIL' print(f'[{status}] {folder:14} {path.name}') if verbose and serrs: for e in serrs: print(' schema:',e) if verbose and sherrs: for e in sherrs: print(' shacl:',e.replace('\n','\n ')) engine='pyshacl' if HAVE_PYSHACL else 'GEDRM-0.1 fallback core evaluator' print('\nGEDRM 0.1 conformance summary') print(f' SHACL engine : {engine}') print(f' Passed : {passed}/{total}') if failures: print(' Failed : ' + ', '.join(failures)) return 1 print(' Result : PASS') return 0 def main(): ap=argparse.ArgumentParser() ap.add_argument('--root', type=Path, default=Path(__file__).resolve().parents[1], help='GEDRM 0.1 directory') ap.add_argument('-v','--verbose',action='store_true') args=ap.parse_args() sys.exit(run(args.root,args.verbose)) if __name__=='__main__': main()