| 1 | #!/usr/bin/env python3 |
| 2 | """Validate map.yaml against JSON Schema with additional custom rules. |
| 3 | |
| 4 | This validator uses JSON Schema for structure validation and adds custom |
| 5 | checks for rules that can't be expressed in JSON Schema, such as: |
| 6 | - Nodes with integration_placeholder children can omit edit_url |
| 7 | - No duplicate edit_urls |
| 8 | |
| 9 | Path reconstruction rule (for ingest): |
| 10 | - Nodes WITH items array → label is the path segment (they define hierarchy) |
| 11 | - Nodes WITHOUT items → leaves that belong to their parent's path |
| 12 | |
| 13 | Exit codes: |
| 14 | 0 - Validation passed |
| 15 | 1 - Validation failed |
| 16 | """ |
| 17 | |
| 18 | import sys |
| 19 | import json |
| 20 | from pathlib import Path |
| 21 | from typing import List, Dict, Any, Tuple |
| 22 | |
| 23 | try: |
| 24 | from ruamel.yaml import YAML |
| 25 | except ImportError: |
| 26 | print("ERROR: ruamel.yaml is required. Install with: pip install ruamel.yaml") |
| 27 | sys.exit(1) |
| 28 | |
| 29 | try: |
| 30 | import jsonschema |
| 31 | from jsonschema import Draft7Validator |
| 32 | except ImportError: |
| 33 | print("ERROR: jsonschema is required. Install with: pip install jsonschema") |
| 34 | sys.exit(1) |
| 35 | |
| 36 | |
| 37 | class MapValidationError: |
| 38 | def __init__(self, path: str, message: str): |
| 39 | self.path = path |
| 40 | self.message = message |
| 41 | |
| 42 | def __str__(self): |
| 43 | return f"[{self.path}] {self.message}" |
| 44 | |
| 45 | |
| 46 | def load_schema(schema_path: str) -> dict: |
| 47 | """Load JSON Schema from file.""" |
| 48 | with open(schema_path, "r", encoding="utf-8") as f: |
| 49 | return json.load(f) |
| 50 | |
| 51 | |
| 52 | def load_yaml(yaml_path: str) -> dict: |
| 53 | """Load YAML file.""" |
| 54 | yaml = YAML(typ="safe") |
| 55 | with open(yaml_path, "r", encoding="utf-8") as f: |
| 56 | return yaml.load(f) |
| 57 | |
| 58 | |
| 59 | def format_schema_error(error: jsonschema.ValidationError) -> str: |
| 60 | """Format a JSON Schema validation error nicely.""" |
| 61 | path = ( |
| 62 | ".".join(str(p) for p in error.absolute_path) if error.absolute_path else "root" |
| 63 | ) |
| 64 | # For oneOf/anyOf, jsonschema often reports a generic message like |
| 65 | # "is not valid under any of the given schemas" and puts the real |
| 66 | # problems into error.context. Surface the most relevant sub-error(s) |
| 67 | # so users see actionable messages without needing debug output. |
| 68 | validator = getattr(error, "validator", None) |
| 69 | if validator in ("oneOf", "anyOf") and getattr(error, "context", None): |
| 70 | suberrors = list(error.context) |
| 71 | |
| 72 | def _path_depth(e: jsonschema.ValidationError) -> int: |
| 73 | try: |
| 74 | return len(list(e.absolute_path)) |
| 75 | except Exception: |
| 76 | return 0 |
| 77 | |
| 78 | # Prefer the deepest (most specific) sub-error. |
| 79 | suberrors.sort(key=_path_depth, reverse=True) |
| 80 | primary = suberrors[0] |
| 81 | sub_path = ( |
| 82 | ".".join(str(p) for p in primary.absolute_path) |
| 83 | if primary.absolute_path |
| 84 | else path |
| 85 | ) |
| 86 | # Collect up to a couple of distinct messages for context. |
| 87 | messages = [primary.message] |
| 88 | for sub in suberrors[1:3]: |
| 89 | if sub.message not in messages: |
| 90 | messages.append(sub.message) |
| 91 | details = "; ".join(messages) |
| 92 | return f"[{sub_path}] {details} (while validating {validator} at {path})" |
| 93 | return f"[{path}] {error.message}" |
| 94 | |
| 95 | |
| 96 | def check_has_integration_placeholder(items: List[Any]) -> bool: |
| 97 | """Check if a node's items contain an integration_placeholder.""" |
| 98 | if not isinstance(items, list): |
| 99 | return False |
| 100 | return any( |
| 101 | isinstance(item, dict) and item.get("type") == "integration_placeholder" |
| 102 | for item in items |
| 103 | ) |
| 104 | |
| 105 | |
| 106 | def check_duplicate_edit_urls( |
| 107 | node: Any, path: str, edit_urls: Dict[str, str], errors: List[MapValidationError] |
| 108 | ) -> None: |
| 109 | """Recursively check for duplicate edit_urls.""" |
| 110 | if not isinstance(node, dict): |
| 111 | return |
| 112 | |
| 113 | # Skip integration placeholders |
| 114 | if node.get("type") == "integration_placeholder": |
| 115 | return |
| 116 | |
| 117 | # Check meta |
| 118 | meta = node.get("meta", {}) |
| 119 | if isinstance(meta, dict): |
| 120 | label = meta.get("label", "???") |
| 121 | node_path = f"{path}/{label}" if path else label |
| 122 | edit_url = meta.get("edit_url") |
| 123 | |
| 124 | if edit_url and isinstance(edit_url, str): |
| 125 | if edit_url in edit_urls: |
| 126 | errors.append( |
| 127 | MapValidationError( |
| 128 | node_path, |
| 129 | f"Duplicate edit_url: '{edit_url}' (first seen at {edit_urls[edit_url]})", |
| 130 | ) |
| 131 | ) |
| 132 | else: |
| 133 | edit_urls[edit_url] = node_path |
| 134 | |
| 135 | # Recurse into children |
| 136 | items = node.get("items", []) |
| 137 | if isinstance(items, list): |
| 138 | for item in items: |
| 139 | check_duplicate_edit_urls(item, node_path, edit_urls, errors) |
| 140 | |
| 141 | |
| 142 | def check_integration_placeholder_rule( |
| 143 | node: Any, path: str, errors: List[MapValidationError] |
| 144 | ) -> None: |
| 145 | """ |
| 146 | Check that leaf nodes have edit_url. |
| 147 | |
| 148 | Custom rule: |
| 149 | - Structural nodes (with children) may omit edit_url. |
| 150 | - Leaf nodes (without children) must provide edit_url. |
| 151 | """ |
| 152 | if not isinstance(node, dict): |
| 153 | return |
| 154 | |
| 155 | # Skip integration placeholders themselves |
| 156 | if node.get("type") == "integration_placeholder": |
| 157 | return |
| 158 | |
| 159 | meta = node.get("meta", {}) |
| 160 | if not isinstance(meta, dict): |
| 161 | return |
| 162 | |
| 163 | label = meta.get("label", "???") |
| 164 | node_path = f"{path}/{label}" if path else label |
| 165 | edit_url = meta.get("edit_url") |
| 166 | items = node.get("items", []) |
| 167 | |
| 168 | has_items = isinstance(items, list) and len(items) > 0 |
| 169 | |
| 170 | # If edit_url is missing, only structural category nodes are allowed. |
| 171 | if edit_url is None: |
| 172 | if not has_items: |
| 173 | errors.append( |
| 174 | MapValidationError( |
| 175 | node_path, |
| 176 | "Missing 'edit_url' field (only allowed for structural nodes with children)", |
| 177 | ) |
| 178 | ) |
| 179 | |
| 180 | # Recurse into children |
| 181 | if isinstance(items, list): |
| 182 | for item in items: |
| 183 | check_integration_placeholder_rule(item, node_path, errors) |
| 184 | |
| 185 | |
| 186 | def validate_with_schema(data: dict, schema: dict) -> Tuple[bool, List[str]]: |
| 187 | """Validate data against JSON Schema.""" |
| 188 | validator = Draft7Validator(schema) |
| 189 | errors = [] |
| 190 | |
| 191 | for error in validator.iter_errors(data): |
| 192 | errors.append(format_schema_error(error)) |
| 193 | |
| 194 | return len(errors) == 0, errors |
| 195 | |
| 196 | |
| 197 | def validate_custom_rules(data: dict) -> Tuple[bool, List[MapValidationError]]: |
| 198 | """Apply custom validation rules not expressible in JSON Schema.""" |
| 199 | errors: List[MapValidationError] = [] |
| 200 | edit_urls: Dict[str, str] = {} |
| 201 | |
| 202 | # Guard against non-dict YAML root |
| 203 | if not isinstance(data, dict): |
| 204 | return False, [ |
| 205 | MapValidationError( |
| 206 | "root", f"YAML root must be a dictionary, got {type(data).__name__}" |
| 207 | ) |
| 208 | ] |
| 209 | |
| 210 | sidebar = data.get("sidebar", []) |
| 211 | if not isinstance(sidebar, list): |
| 212 | return False, [MapValidationError("root", "sidebar must be a list")] |
| 213 | |
| 214 | # Check for duplicate edit_urls |
| 215 | for node in sidebar: |
| 216 | check_duplicate_edit_urls(node, "", edit_urls, errors) |
| 217 | |
| 218 | # Check integration placeholder rule |
| 219 | for node in sidebar: |
| 220 | check_integration_placeholder_rule(node, "", errors) |
| 221 | |
| 222 | return len(errors) == 0, errors |
| 223 | |
| 224 | |
| 225 | def main(): |
| 226 | """Main validation routine.""" |
| 227 | script_dir = Path(__file__).parent |
| 228 | yaml_path = script_dir / "map.yaml" |
| 229 | schema_path = script_dir / "map.schema.json" |
| 230 | |
| 231 | if not yaml_path.exists(): |
| 232 | print(f"ERROR: {yaml_path} not found") |
| 233 | sys.exit(1) |
| 234 | |
| 235 | if not schema_path.exists(): |
| 236 | print(f"ERROR: {schema_path} not found") |
| 237 | sys.exit(1) |
| 238 | |
| 239 | print("Validating map.yaml...") |
| 240 | print() |
| 241 | |
| 242 | # Load files |
| 243 | try: |
| 244 | data = load_yaml(str(yaml_path)) |
| 245 | schema = load_schema(str(schema_path)) |
| 246 | except Exception as e: |
| 247 | print(f"ERROR loading files: {e}") |
| 248 | sys.exit(1) |
| 249 | |
| 250 | # Guard against non-dict YAML root |
| 251 | if not isinstance(data, dict): |
| 252 | print(f"❌ Validation FAILED:\n") |
| 253 | print(f" • YAML root must be a dictionary, got {type(data).__name__}") |
| 254 | sys.exit(1) |
| 255 | |
| 256 | all_errors = [] |
| 257 | |
| 258 | # Validate against JSON Schema |
| 259 | schema_valid, schema_errors = validate_with_schema(data, schema) |
| 260 | if not schema_valid: |
| 261 | all_errors.append("Schema validation errors:") |
| 262 | all_errors.extend(f" • {err}" for err in schema_errors) |
| 263 | |
| 264 | # Apply custom rules |
| 265 | custom_valid, custom_errors = validate_custom_rules(data) |
| 266 | if not custom_valid: |
| 267 | if all_errors: |
| 268 | all_errors.append("") |
| 269 | all_errors.append("Custom rule violations:") |
| 270 | all_errors.extend(f" • {err}" for err in custom_errors) |
| 271 | |
| 272 | # Report results |
| 273 | if all_errors: |
| 274 | print("❌ Validation FAILED:\n") |
| 275 | print("\n".join(all_errors)) |
| 276 | sys.exit(1) |
| 277 | else: |
| 278 | print("✅ Validation PASSED") |
| 279 | print(f" - Validated against schema: {schema_path.name}") |
| 280 | print(f" - All custom rules satisfied") |
| 281 | sys.exit(0) |
| 282 | |
| 283 | |
| 284 | if __name__ == "__main__": |
| 285 | main() |