Validate RPC request parameters against schema. Args: method: Method name (e.g., 'runSingleTest') args: Method arguments (object or array) Raises: ValidationError: If request doesn't match schema KeyError: If method not found
(
self, method: str, args: dict[str, Any] | list[Any] | Any
)
| 130 | raise RuntimeError(f"Failed to fetch RPC schema: {e}") |
| 131 | |
| 132 | def validate_request( |
| 133 | self, method: str, args: dict[str, Any] | list[Any] | Any |
| 134 | ) -> None: |
| 135 | """ |
| 136 | Validate RPC request parameters against schema. |
| 137 | |
| 138 | Args: |
| 139 | method: Method name (e.g., 'runSingleTest') |
| 140 | args: Method arguments (object or array) |
| 141 | |
| 142 | Raises: |
| 143 | ValidationError: If request doesn't match schema |
| 144 | KeyError: If method not found in schema |
| 145 | """ |
| 146 | if not self.schema: |
| 147 | raise RuntimeError("Schema not loaded") |
| 148 | |
| 149 | if method not in self.schema.methods: |
| 150 | raise KeyError( |
| 151 | f"Method '{method}' not found in schema. Available: {list(self.schema.methods.keys())}" |
| 152 | ) |
| 153 | |
| 154 | method_info = self.schema.methods[method] |
| 155 | |
| 156 | # Validate against expected parameters |
| 157 | # Note: The RPC system unwraps single-element arrays, so we validate the object directly |
| 158 | if not isinstance(args, dict): |
| 159 | raise ValidationError( |
| 160 | f"Expected object for {method}, got {type(args).__name__}" |
| 161 | ) |
| 162 | |
| 163 | # Type narrowing: after isinstance check, args is guaranteed to be dict |
| 164 | args_dict = cast(dict[str, Any], args) |
| 165 | |
| 166 | # Check required parameters |
| 167 | expected_params: set[str] = {p.name for p in method_info.params} |
| 168 | provided_params: set[str] = set(args_dict.keys()) |
| 169 | |
| 170 | # For now, just warn about mismatches (can be made stricter) |
| 171 | missing: set[str] = expected_params - provided_params |
| 172 | extra: set[str] = provided_params - expected_params |
| 173 | |
| 174 | errors: list[str] = [] |
| 175 | if missing: |
| 176 | errors.append(f"Missing required parameters: {missing}") |
| 177 | if extra: |
| 178 | errors.append(f"Unknown parameters: {extra}") |
| 179 | |
| 180 | if errors: |
| 181 | raise ValidationError( |
| 182 | f"Validation errors for {method}: {'; '.join(errors)}" |
| 183 | ) |
| 184 | |
| 185 | def get_method_schema(self, method: str) -> MethodInfo: |
| 186 | """Get schema for a specific method""" |