1. import json
  2. from jsonschema import validate, ValidationError
  4. def flag_json_anomalies(json_data, schema):
  5. """
  6. Flags anomalies in JSON responses based on a provided schema.
  8. Args:
  9. json_data (str): The JSON data as a string.
  10. schema (dict): The JSON schema to validate against.
  12. Returns:
  13. list: A list of anomaly messages. Returns an empty list if no anomalies are found.
  14. """
  15. anomalies = []
  17. try:
  18. data = json.loads(json_data)
  19. validate(instance=data, schema=schema)
  20. except json.JSONDecodeError as e:
  21. anomalies.append(f"JSON Decode Error: {str(e)}")
  22. return anomalies
  23. except ValidationError as e:
  24. anomalies.append(f"Schema Validation Error: {str(e)}")
  25. except Exception as e:
  26. anomalies.append(f"Unexpected Error: {str(e)}")
  28. return anomalies
  30. if __name__ == '__main__':
  31. # Example usage
  32. schema = {
  33. "type": "object",
  34. "properties": {
  35. "timestamp": {"type": "string"},
  36. "status": {"type": "string", "enum": ["success", "failure", "warning"]},
  37. "data": {"type": "object"},
  38. "metric_value": {"type": "number"}
  39. },
  40. "required": ["timestamp", "status", "metric_value"]
  41. }
  43. # Valid JSON
  44. valid_json = '{"timestamp": "2023-10-27T10:00:00Z", "status": "success", "data": {"cpu_usage": 50}, "metric_value": 0.75}'
  45. anomalies = flag_json_anomalies(valid_json, schema)
  46. print(f"Valid JSON Anomalies: {anomalies}")
  48. # Invalid JSON (missing required field)
  49. invalid_json_missing_field = '{"timestamp": "2023-10-27T10:00:00Z", "status": "success", "data": {"cpu_usage": 50}}'
  50. anomalies = flag_json_anomalies(invalid_json_missing_field, schema)
  51. print(f"Invalid JSON Anomalies (Missing Field): {anomalies}")
  53. # Invalid JSON (wrong data type)
  54. invalid_json_wrong_type = '{"timestamp": "2023-10-27T10:00:00Z", "status": "success", "data": {"cpu_usage": "50"}, "metric_value": 0.75}'
  55. anomalies = flag_json_anomalies(invalid_json_wrong_type, schema)
  56. print(f"Invalid JSON Anomalies (Wrong Type): {anomalies}")
  58. #Invalid JSON (invalid enum value)
  59. invalid_json_invalid_enum = '{"timestamp": "2023-10-27T10:00:00Z", "status": "unknown", "data": {"cpu_usage": 50}, "metric_value": 0.75}'
  60. anomalies = flag_json_anomalies(invalid_json_invalid_enum, schema)
  61. print(f"Invalid JSON Anomalies (Invalid Enum): {anomalies}")
  63. #Invalid JSON (invalid format)
  64. invalid_json_invalid_format = '{"timestamp": "2023-10-27 10:00:00", "status": "success", "data": {"cpu_usage": 50}, "metric_value": 0.75}'
  65. anomalies = flag_json_anomalies(invalid_json_invalid_format, schema)
  66. print(f"Invalid JSON Anomalies (Invalid Format): {anomalies}")

Add your comment