Add trace_agg_unit_test/src/trace_agg_unit_test/core.py

trace_agg_unit_test/src/trace_agg_unit_test/core.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+import os
+import pandas as pd
+from pathlib import Path
+from typing import List, Dict, Any
+
+import trace_agg
+
+
+def run_aggregation_tests(trace_files: List[str]) -> Dict[str, Any]:
+    """Führt Unit-Tests für die Aggregationslogik von trace_agg aus.
+
+    Args:
+        trace_files: Liste von Pfaden zu CSV-Dateien mit Trace-Event-Daten.
+
+    Returns:
+        Ein Dictionary mit Testergebnissen pro Datei, einschließlich Status und Summen.
+    """
+    results: Dict[str, Any] = {}
+
+    for file_path in trace_files:
+        file_name = os.path.basename(file_path)
+        file_result = {"status": "unknown", "sum_value": None, "error": None}
+        try:
+            if not os.path.exists(file_path):
+                raise FileNotFoundError(f"File not found: {file_path}")
+
+            df = pd.read_csv(file_path)
+            required_cols = {"timestamp", "event_type", "value"}
+            if not required_cols.issubset(df.columns):
+                raise ValueError(f"Missing required columns in {file_name}")
+
+            # Rufe Aggregationslogik aus dem getesteten Modul auf
+            try:
+                agg_result = trace_agg.aggregate_trace(df)
+            except AttributeError:
+                raise RuntimeError("trace_agg.aggregate_trace() not implemented or import failed")
+
+            if not isinstance(agg_result, (dict, pd.DataFrame)):
+                raise TypeError("Invalid aggregation result type.")
+
+            # Prüfe, ob Summen stimmen (vereinfachter Test)
+            original_sum = float(df["value"].sum())
+            if isinstance(agg_result, dict) and "value_sum" in agg_result:
+                agg_sum = float(agg_result["value_sum"])
+            elif isinstance(agg_result, pd.DataFrame) and "value" in agg_result.columns:
+                agg_sum = float(agg_result["value"].sum())
+            else:
+                raise ValueError("Aggregation result missing sum value.")
+
+            passed = abs(original_sum - agg_sum) < 1e-6
+            file_result["sum_value"] = {"original": original_sum, "aggregated": agg_sum}
+            file_result["status"] = "passed" if passed else "failed"
+
+        except Exception as exc:  # noqa: BLE001
+            file_result["status"] = "error"
+            file_result["error"] = str(exc)
+
+        results[file_name] = file_result
+
+    return results