Add sanity_check_tool/src/sanity_check_tool/core.py

sanity_check_tool/src/sanity_check_tool/core.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+import json
+from typing import Any, Dict, List
+
+
+class SanityCheckError(Exception):
+    """Custom exception for sanity check processing errors."""
+    pass
+
+
+def _validate_run_data_structure(run_data: Dict[str, Any]) -> None:
+    required_keys = {"runs", "sanity_checks"}
+    if not isinstance(run_data, dict):
+        raise SanityCheckError("run_data must be a dictionary.")
+    if not required_keys.issubset(run_data.keys()):
+        missing = required_keys - run_data.keys()
+        raise SanityCheckError(f"run_data missing required keys: {', '.join(missing)}")
+    if not isinstance(run_data["runs"], list):
+        raise SanityCheckError("'runs' must be a list.")
+    if not isinstance(run_data["sanity_checks"], list):
+        raise SanityCheckError("'sanity_checks' must be a list.")
+
+
+def perform_sanity_checks(run_data: Dict[str, Any]) -> Dict[str, int]:
+    """Führt einfache Sanity-Checks auf Run-Daten aus und gibt strukturierte Ergebnisse zurück.
+
+    Args:
+        run_data: JSON-ähnliche Struktur mit Feldern 'runs' und 'sanity_checks'.
+
+    Returns:
+        Dict mit Zählern für verschiedene Sanity-Checks.
+    """
+    _validate_run_data_structure(run_data)
+
+    runs: List[Dict[str, Any]] = run_data.get("runs", [])
+
+    missing_pairs = 0
+    broken_ids = 0
+    empty_fields = 0
+    clock_switch_count = 0
+
+    # Fehlende write_pre/write_post-Paare
+    write_pre_events = [r for r in runs if r.get("event_type") == "write_pre"]
+    write_post_events = [r for r in runs if r.get("event_type") == "write_post"]
+    diff = abs(len(write_pre_events) - len(write_post_events))
+    missing_pairs = diff
+
+    # Gebrochene corr_id-Ketten
+    seen_ids = set()
+    for run in runs:
+        cid = run.get("corr_id")
+        if cid is None:
+            broken_ids += 1
+        elif cid in seen_ids:
+            broken_ids += 1
+        else:
+            seen_ids.add(cid)
+
+    # Leere Felder prüfen
+    for run in runs:
+        for key, val in run.items():
+            if val in (None, "", []):
+                empty_fields += 1
+
+    # Clocksource-Switch-Events zählen
+    for run in runs:
+        if run.get("event_type") == "clocksource_switch":
+            clock_switch_count += 1
+
+    results: Dict[str, int] = {
+        "missing_pairs": missing_pairs,
+        "broken_ids": broken_ids,
+        "empty_fields": empty_fields,
+        "clock_switch_count": clock_switch_count,
+    }
+
+    # CI-Assertion: Sicherstellen, dass alle Werte int sind
+    assert all(isinstance(v, int) for v in results.values()), "All result values must be integers"
+
+    return results