Coverage for src / tracekit / schemas / __init__.py: 84%

1"""JSON Schema definitions for TraceKit configuration types. 

2 

3This module provides JSON Schema definitions for validating various 

4configuration file types used in TraceKit, including packet formats, 

5device mappings, bus configurations, and protocol definitions. 

6 

7 

8Example: 

9 >>> from tracekit.schemas import load_schema, validate_config 

10 >>> schema = load_schema("packet_format") 

11 >>> validate_config(config_dict, "packet_format") 

12 True 

13""" 

14 

15from __future__ import annotations 

16 

17import json 

18from pathlib import Path 

19from typing import Any 

20 

21from tracekit.config.schema import ( 

22 ConfigSchema, 

23 get_schema_registry, 

24 register_schema, 

25 validate_against_schema, 

26) 

27 

28__all__ = [ 

29 "SCHEMA_NAMES", 

30 "get_schema_path", 

31 "load_schema", 

32 "register_builtin_schemas", 

33 "validate_config", 

34] 

35 

36# Schema type names 

37SCHEMA_NAMES = [ 

38 "packet_format", 

39 "device_mapping", 

40 "bus_configuration", 

41 "protocol_definition", 

42] 

43 

44 

45def get_schema_path(schema_name: str) -> Path: 

46 """Get the file path for a schema definition. 

47 

48 Args: 

49 schema_name: Schema name (e.g., "packet_format"). 

50 

51 Returns: 

52 Path to the JSON schema file. 

53 

54 Raises: 

55 ValueError: If schema name is not recognized. 

56 FileNotFoundError: If schema file does not exist. 

57 """ 

58 if schema_name not in SCHEMA_NAMES: 

59 raise ValueError(f"Unknown schema name: {schema_name}. Available schemas: {SCHEMA_NAMES}") 

60 

61 schema_dir = Path(__file__).parent 

62 schema_file = schema_dir / f"{schema_name}.json" 

63 

64 if not schema_file.exists(): 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true

65 raise FileNotFoundError(f"Schema file not found: {schema_file}") 

66 

67 return schema_file 

68 

69 

70def load_schema(schema_name: str) -> dict[str, Any]: 

71 """Load a JSON schema definition from disk. 

72 

73 Args: 

74 schema_name: Schema name (e.g., "packet_format"). 

75 

76 Returns: 

77 JSON schema dictionary. 

78 """ 

79 schema_path = get_schema_path(schema_name) 

80 

81 with open(schema_path) as f: 

82 result: dict[str, Any] = json.load(f) 

83 return result 

84 

85 

86def validate_config( 

87 config: dict[str, Any], 

88 schema_name: str, 

89 *, 

90 strict: bool = False, 

91) -> bool: 

92 """Validate a configuration dictionary against a schema. 

93 

94 Args: 

95 config: Configuration dictionary to validate. 

96 schema_name: Schema name (e.g., "packet_format"). 

97 strict: If True, fail on additional properties. 

98 

99 Returns: 

100 True if validation passes. 

101 

102 Example: 

103 >>> config = {"name": "test", "packet": {...}, "header": {...}} 

104 >>> validate_config(config, "packet_format") 

105 True 

106 """ 

107 # Use the existing validation system from config.schema 

108 return validate_against_schema(config, schema_name, strict=strict) 

109 

110 

111def register_builtin_schemas() -> None: 

112 """Register all built-in configuration schemas. 

113 

114 This function registers the following schemas: 

115 - packet_format: Binary packet format configurations 

116 - device_mapping: Device ID to name mappings 

117 - bus_configuration: Parallel bus configurations 

118 - protocol_definition: Protocol DSL definitions 

119 

120 

121 

122 The schemas are registered with the global SchemaRegistry and can 

123 be accessed via validate_against_schema() or get_schema_registry(). 

124 """ 

125 registry = get_schema_registry() 

126 

127 # Load and register each schema 

128 for schema_name in SCHEMA_NAMES: 

129 # Skip if already registered 

130 if registry.has_schema(schema_name): 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true

131 continue 

132 

133 schema_dict = load_schema(schema_name) 

134 

135 # Extract version from $id if present 

136 version = "1.0.0" # default 

137 if "$id" in schema_dict: 137 ↛ 145line 137 didn't jump to line 145 because the condition on line 137 was always true

138 # Extract version from URI like .../v1.0.0.json 

139 schema_id = schema_dict["$id"] 

140 if "/v" in schema_id: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 version_part = schema_id.split("/v")[-1] 

142 version = version_part.replace(".json", "") 

143 

144 # Create ConfigSchema object 

145 config_schema = ConfigSchema( 

146 name=schema_name, 

147 version=version, 

148 schema=schema_dict, 

149 description=schema_dict.get("description", ""), 

150 uri=schema_dict.get("$id"), 

151 ) 

152 

153 # Register with global registry 

154 register_schema(config_schema, set_default=True) 

155 

156 

157# Auto-register schemas on module import 

158register_builtin_schemas()