diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..094729e --- /dev/null +++ b/.editorconfig @@ -0,0 +1,21 @@ +root = true + +[*] +end_of_line = lf +insert_final_newline = true +trim_trailing_whitespace = true +charset = utf-8 + +[*.py] +indent_style = space +indent_size = 4 + +[*.{yml,yaml}] +indent_style = space +indent_size = 2 + +[*.md] +trim_trailing_whitespace = false + +[Makefile] +indent_style = tab diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..979a9c1 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,24 @@ +# Auto detect text files and perform LF normalization +* text=auto + +# Explicitly declare text files +*.md text diff=markdown +*.txt text +*.csv text +*.yml text +*.yaml text +*.json text +*.xml text +*.html text diff=html +*.css text diff=css + +# Denote binary files +*.png binary +*.jpg binary +*.jpeg binary +*.gif binary +*.ico binary +*.pdf binary +*.zip binary +*.gz binary +*.tar binary diff --git a/README.md b/README.md index a199fc5..f1a3b11 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,5 @@ -# Open Connectivity Foundatation | IoT Data Models -© 2016-2022 Open Connectivity Foundatation. All rights reserved. +# Open Connectivity Foundation | IoT Data Models +© 2016-2022 Open Connectivity Foundation. All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: diff --git a/scripts/test_validate_models.py b/scripts/test_validate_models.py new file mode 100644 index 0000000..8144227 --- /dev/null +++ b/scripts/test_validate_models.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +"""Tests for validate_models.py""" + +import json +import os +import tempfile +import pytest +from validate_models import ( + ModelValidator, + ValidationReport, + ValidationIssue, + find_model_files, +) + + +@pytest.fixture +def tmp_dir(): + with tempfile.TemporaryDirectory() as d: + yield d + + +def write_json(directory, filename, data): + path = os.path.join(directory, filename) + with open(path, "w") as f: + json.dump(data, f) + return path + + +class TestValidationReport: + def test_add_error(self): + report = ValidationReport() + report.add_issue(ValidationIssue("f.json", "/", "error", "test", "msg")) + assert report.total_errors == 1 + + def test_add_warning(self): + report = ValidationReport() + report.add_issue(ValidationIssue("f.json", "/", "warning", "test", "msg")) + assert report.total_warnings == 1 + + +class TestModelValidator: + def test_invalid_json(self, tmp_dir): + path = os.path.join(tmp_dir, "bad.json") + with open(path, "w") as f: + f.write("{invalid}") + v = ModelValidator(tmp_dir) + v.validate_file("bad.json") + assert v.report.total_errors >= 1 + + def test_missing_swagger_fields(self, tmp_dir): + write_json(tmp_dir, "api.json", {"swagger": "2.0"}) + v = ModelValidator(tmp_dir) + v.validate_file("api.json") + errs = [i for i in v.report.issues if i.category == "required_field"] + assert len(errs) >= 1 + + def test_valid_swagger(self, tmp_dir): + write_json(tmp_dir, "api.json", { + "swagger": "2.0", + "info": {"title": "Test", "version": "1.0.0"}, + "paths": {} + }) + v = ModelValidator(tmp_dir) + v.validate_file("api.json") + assert v.report.total_errors == 0 + + def test_invalid_type(self, tmp_dir): + write_json(tmp_dir, "model.json", { + "type": "object", + "properties": { + "name": {"type": "invalid_type"} + } + }) + v = ModelValidator(tmp_dir) + v.validate_file("model.json") + errs = [i for i in v.report.issues if i.category == "type"] + assert len(errs) >= 1 + + def test_array_missing_items(self, tmp_dir): + write_json(tmp_dir, "model.json", { + "type": "object", + "properties": { + "tags": {"type": "array"} + } + }) + v = ModelValidator(tmp_dir) + v.validate_file("model.json") + warns = [i for i in v.report.issues if "items" in i.message] + assert len(warns) >= 1 + + def test_required_prop_not_defined(self, tmp_dir): + write_json(tmp_dir, "model.json", { + "type": "object", + "required": ["name"], + "properties": {} + }) + v = ModelValidator(tmp_dir) + v.validate_file("model.json") + errs = [i for i in v.report.issues if "Required property" in i.message] + assert len(errs) >= 1 + + def test_naming_convention(self, tmp_dir): + write_json(tmp_dir, "model.json", { + "type": "object", + "properties": { + "BadName": {"type": "string"} + } + }) + v = ModelValidator(tmp_dir, naming_convention="camel") + v.validate_file("model.json") + infos = [i for i in v.report.issues if i.category == "naming"] + assert len(infos) >= 1 + + +class TestFindModelFiles: + def test_finds_json(self, tmp_dir): + write_json(tmp_dir, "model.json", {}) + files = find_model_files(tmp_dir) + assert len(files) == 1 + + def test_ignores_hidden(self, tmp_dir): + write_json(tmp_dir, ".hidden.json", {}) + files = find_model_files(tmp_dir) + assert len(files) == 0 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/scripts/validate_models.py b/scripts/validate_models.py new file mode 100644 index 0000000..abe3f3c --- /dev/null +++ b/scripts/validate_models.py @@ -0,0 +1,402 @@ +#!/usr/bin/env python3 +"""JSON Schema and Swagger/OpenAPI Validator for IoT Data Models. + +Validates JSON model definitions for conformance to OpenAPI/Swagger specs, +checks required fields, type consistency, reference resolution, naming +conventions, and version compatibility. +""" + +import argparse +import json +import os +import re +import sys +from collections import defaultdict +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Set, Tuple + + +@dataclass +class ValidationIssue: + """A single validation issue.""" + file_path: str + json_path: str + severity: str # "error", "warning", "info" + category: str + message: str + + +@dataclass +class ValidationReport: + """Overall validation report.""" + issues: List[ValidationIssue] = field(default_factory=list) + files_checked: int = 0 + total_errors: int = 0 + total_warnings: int = 0 + total_info: int = 0 + models_found: int = 0 + + def add_issue(self, issue: ValidationIssue): + self.issues.append(issue) + if issue.severity == "error": + self.total_errors += 1 + elif issue.severity == "warning": + self.total_warnings += 1 + else: + self.total_info += 1 + + +# Naming convention patterns +CAMEL_CASE_RE = re.compile(r'^[a-z][a-zA-Z0-9]*$') +PASCAL_CASE_RE = re.compile(r'^[A-Z][a-zA-Z0-9]*$') +SNAKE_CASE_RE = re.compile(r'^[a-z][a-z0-9_]*$') +KEBAB_CASE_RE = re.compile(r'^[a-z][a-z0-9-]*$') + +# Required fields for OpenAPI/Swagger schemas +SWAGGER_REQUIRED_ROOT = ["swagger", "info", "paths"] +OPENAPI3_REQUIRED_ROOT = ["openapi", "info", "paths"] +INFO_REQUIRED = ["title", "version"] + + +class ModelValidator: + """Validates IoT data model JSON/Swagger files.""" + + def __init__(self, base_dir: str, naming_convention: str = "camel"): + self.base_dir = Path(base_dir) + self.report = ValidationReport() + self.all_definitions: Dict[str, Dict] = {} + self.naming_convention = naming_convention + self._naming_re = { + "camel": CAMEL_CASE_RE, + "pascal": PASCAL_CASE_RE, + "snake": SNAKE_CASE_RE, + "kebab": KEBAB_CASE_RE, + }.get(naming_convention, CAMEL_CASE_RE) + + def validate_file(self, file_path: str) -> None: + """Validate a single JSON model file.""" + full_path = self.base_dir / file_path + if not full_path.exists(): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="", + severity="error", category="file", + message=f"File not found: {file_path}" + )) + return + + self.report.files_checked += 1 + try: + with open(full_path, "r", encoding="utf-8") as f: + data = json.load(f) + except json.JSONDecodeError as e: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="", + severity="error", category="parse", + message=f"Invalid JSON: {e}" + )) + return + + if not isinstance(data, dict): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/", + severity="error", category="structure", + message="Root element must be an object" + )) + return + + # Detect schema type + if "swagger" in data: + self._validate_swagger(file_path, data) + elif "openapi" in data: + self._validate_openapi3(file_path, data) + else: + # Treat as generic JSON schema or model definition + self._validate_generic_model(file_path, data) + + def _validate_swagger(self, file_path: str, data: Dict) -> None: + """Validate Swagger 2.0 document.""" + for req_field in SWAGGER_REQUIRED_ROOT: + if req_field not in data: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/", + severity="error", category="required_field", + message=f"Missing required Swagger field: {req_field}" + )) + + if "info" in data: + self._validate_info(file_path, data["info"]) + + if "definitions" in data: + self._validate_definitions(file_path, data["definitions"], "/definitions") + + if "paths" in data: + self._validate_paths(file_path, data["paths"]) + + self._check_version_format(file_path, data.get("swagger", ""), "swagger") + + def _validate_openapi3(self, file_path: str, data: Dict) -> None: + """Validate OpenAPI 3.x document.""" + for req_field in OPENAPI3_REQUIRED_ROOT: + if req_field not in data: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/", + severity="error", category="required_field", + message=f"Missing required OpenAPI field: {req_field}" + )) + + if "info" in data: + self._validate_info(file_path, data["info"]) + + if "components" in data and "schemas" in data["components"]: + self._validate_definitions(file_path, data["components"]["schemas"], "/components/schemas") + + if "paths" in data: + self._validate_paths(file_path, data["paths"]) + + self._check_version_format(file_path, data.get("openapi", ""), "openapi") + + def _validate_info(self, file_path: str, info: Dict) -> None: + """Validate info section.""" + for req in INFO_REQUIRED: + if req not in info: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/info", + severity="error", category="required_field", + message=f"Missing required info field: {req}" + )) + + version = info.get("version", "") + if version and not re.match(r'^\d+\.\d+(\.\d+)?', version): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/info/version", + severity="warning", category="version", + message=f"Version '{version}' doesn't follow semver format" + )) + + def _validate_definitions(self, file_path: str, definitions: Dict, base_path: str) -> None: + """Validate model definitions/schemas.""" + for name, schema in definitions.items(): + self.report.models_found += 1 + path = f"{base_path}/{name}" + + # Check naming convention + self._check_naming(file_path, name, path) + + if isinstance(schema, dict): + # Check required type field + if "type" not in schema and "$ref" not in schema and "allOf" not in schema and "oneOf" not in schema and "anyOf" not in schema: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=path, + severity="warning", category="type", + message=f"Schema '{name}' has no type specified" + )) + + # Validate properties + if "properties" in schema: + self._validate_properties(file_path, schema["properties"], f"{path}/properties", schema.get("required", [])) + + # Check $ref resolution + self._check_refs(file_path, schema, path) + + # Store for cross-file reference checking + self.all_definitions[name] = schema + + def _validate_properties(self, file_path: str, properties: Dict, base_path: str, required_list: List) -> None: + """Validate schema properties.""" + for prop_name, prop_schema in properties.items(): + prop_path = f"{base_path}/{prop_name}" + + self._check_naming(file_path, prop_name, prop_path) + + if isinstance(prop_schema, dict): + # Check type consistency + prop_type = prop_schema.get("type") + if prop_type and prop_type not in ["string", "number", "integer", "boolean", "array", "object", "null"]: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=prop_path, + severity="error", category="type", + message=f"Invalid type '{prop_type}' for property '{prop_name}'" + )) + + # Array items check + if prop_type == "array" and "items" not in prop_schema: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=prop_path, + severity="warning", category="type", + message=f"Array property '{prop_name}' missing 'items' definition" + )) + + # Check $ref + self._check_refs(file_path, prop_schema, prop_path) + + # Check required fields exist in properties + for req_prop in required_list: + if req_prop not in properties: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=base_path, + severity="error", category="required_field", + message=f"Required property '{req_prop}' not defined in properties" + )) + + def _validate_paths(self, file_path: str, paths: Dict) -> None: + """Validate API paths.""" + for path_str, path_item in paths.items(): + if not path_str.startswith("/"): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=f"/paths/{path_str}", + severity="error", category="structure", + message=f"Path must start with '/': {path_str}" + )) + + if isinstance(path_item, dict): + for method in ["get", "post", "put", "delete", "patch", "options", "head"]: + if method in path_item: + operation = path_item[method] + if isinstance(operation, dict): + if "responses" not in operation: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=f"/paths/{path_str}/{method}", + severity="warning", category="required_field", + message=f"Operation {method.upper()} {path_str} missing 'responses'" + )) + + def _check_refs(self, file_path: str, schema: Dict, path: str) -> None: + """Check $ref references recursively.""" + if "$ref" in schema: + ref = schema["$ref"] + if ref.startswith("#/"): + # Local reference - extract definition name + parts = ref.split("/") + def_name = parts[-1] if parts else "" + # Will be validated in cross_validate + elif not ref.startswith(("http://", "https://")): + # File reference + ref_file = ref.split("#")[0] + if ref_file and not (self.base_dir / ref_file).exists(): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=path, + severity="error", category="reference", + message=f"Referenced file not found: {ref_file}" + )) + + for key, value in schema.items(): + if isinstance(value, dict): + self._check_refs(file_path, value, f"{path}/{key}") + elif isinstance(value, list): + for i, item in enumerate(value): + if isinstance(item, dict): + self._check_refs(file_path, item, f"{path}/{key}/{i}") + + def _check_naming(self, file_path: str, name: str, path: str) -> None: + """Check naming convention compliance.""" + if not self._naming_re.match(name): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path=path, + severity="info", category="naming", + message=f"Name '{name}' doesn't follow {self.naming_convention} convention" + )) + + def _check_version_format(self, file_path: str, version: str, spec_type: str) -> None: + """Validate spec version format.""" + if spec_type == "swagger" and version != "2.0": + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/swagger", + severity="warning", category="version", + message=f"Unexpected Swagger version: {version} (expected 2.0)" + )) + elif spec_type == "openapi" and not version.startswith("3."): + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/openapi", + severity="warning", category="version", + message=f"Unexpected OpenAPI version: {version} (expected 3.x)" + )) + + def _validate_generic_model(self, file_path: str, data: Dict) -> None: + """Validate a generic JSON model (not Swagger/OpenAPI).""" + self.report.models_found += 1 + if "properties" in data: + self._validate_properties(file_path, data["properties"], "/properties", data.get("required", [])) + if "type" in data: + if data["type"] not in ["string", "number", "integer", "boolean", "array", "object", "null"]: + self.report.add_issue(ValidationIssue( + file_path=file_path, json_path="/type", + severity="error", category="type", + message=f"Invalid root type: {data['type']}" + )) + + +def find_model_files(base_dir: str) -> List[str]: + """Find all JSON model files in directory.""" + json_files = [] + for root, dirs, files in os.walk(base_dir): + # Skip node_modules, .git, etc. + dirs[:] = [d for d in dirs if d not in ('.git', 'node_modules', '__pycache__', 'scripts')] + for f in files: + if f.endswith(".json") and not f.startswith("."): + rel = os.path.relpath(os.path.join(root, f), base_dir) + json_files.append(rel) + return sorted(json_files) + + +def print_report(report: ValidationReport) -> None: + """Print formatted validation report.""" + print("\n" + "=" * 60) + print("MODEL VALIDATION REPORT") + print("=" * 60) + print(f"Files checked: {report.files_checked}") + print(f"Models found: {report.models_found}") + print(f"Errors: {report.total_errors}") + print(f"Warnings: {report.total_warnings}") + print(f"Info: {report.total_info}") + print() + + by_category = defaultdict(list) + for issue in report.issues: + by_category[issue.category].append(issue) + + for category, issues in sorted(by_category.items()): + print(f"[{category}]") + for issue in issues: + icon = {"error": "E", "warning": "W", "info": "I"}.get(issue.severity, "?") + print(f" [{icon}] {issue.file_path} {issue.json_path}: {issue.message}") + print() + + +def main(): + """Main entry point.""" + parser = argparse.ArgumentParser(description="Validate IoT data model JSON files") + parser.add_argument("directory", nargs="?", default=".", help="Base directory to scan") + parser.add_argument("--files", nargs="+", help="Specific JSON files to check") + parser.add_argument("--naming", choices=["camel", "pascal", "snake", "kebab"], + default="camel", help="Naming convention to enforce") + parser.add_argument("--strict", action="store_true", help="Treat warnings as errors") + args = parser.parse_args() + + base_dir = os.path.abspath(args.directory) + validator = ModelValidator(base_dir, naming_convention=args.naming) + + if args.files: + model_files = args.files + else: + model_files = find_model_files(base_dir) + + if not model_files: + print("No JSON model files found.") + sys.exit(0) + + print(f"Validating {len(model_files)} model file(s)...") + for mf in model_files: + validator.validate_file(mf) + + print_report(validator.report) + + if validator.report.total_errors > 0: + sys.exit(1) + if args.strict and validator.report.total_warnings > 0: + sys.exit(1) + sys.exit(0) + + +if __name__ == "__main__": + main()