diff --git a/trace_agg_unit_test/tests/test_core.py b/trace_agg_unit_test/tests/test_core.py new file mode 100644 index 0000000..0f93498 --- /dev/null +++ b/trace_agg_unit_test/tests/test_core.py @@ -0,0 +1,62 @@ +import pytest +import pandas as pd +from pathlib import Path + +# Das getestete Modul +import trace_agg + +def test_run_aggregation_tests_basic(tmp_path): + # Beispielhafte CSV-Datei mit TraceEvent-Daten erstellen + data_path = tmp_path / "trace_sample.csv" + df = pd.DataFrame([ + {"timestamp": 0.0, "event_type": "clocksource", "value": 1.0}, + {"timestamp": 0.1, "event_type": "clocksource", "value": 2.0}, + {"timestamp": 0.2, "event_type": "clocksource_switch", "value": 3.0} + ]) + df.to_csv(data_path, index=False) + + # Funktion aus trace_agg ausführen + results = trace_agg.run_aggregation_tests([str(data_path)]) + + # Ergebnistyp prüfen + assert isinstance(results, dict) + assert str(data_path) in results + + file_result = results[str(data_path)] + assert "sum" in file_result + assert "status" in file_result + + # Summenprüfung: Summe im Testdatenframe sollte 6.0 sein + assert pytest.approx(file_result["sum"], rel=1e-6) == df["value"].sum() + assert file_result["status"] in ("passed", "failed") + +def test_run_aggregation_tests_multiple_files(tmp_path): + file_paths = [] + # Zwei CSV-Dateien mit Testdaten erzeugen + for i in range(2): + path = tmp_path / f"trace_{i}.csv" + df = pd.DataFrame([ + {"timestamp": i, "event_type": "clocksource", "value": i + 1.0}, + {"timestamp": i + 0.1, "event_type": "clocksource_switch", "value": i + 2.0} + ]) + df.to_csv(path, index=False) + file_paths.append(str(path)) + + results = trace_agg.run_aggregation_tests(file_paths) + + # Ergebnisse müssen für alle Dateien vorhanden sein + assert isinstance(results, dict) + assert set(map(str, file_paths)) == set(results.keys()) + + for fp in file_paths: + res = results[fp] + assert "sum" in res and "status" in res + sum_expected = pd.read_csv(fp)["value"].sum() + assert pytest.approx(res["sum"], rel=1e-6) == sum_expected + assert res["status"] == "passed" + + +def test_run_aggregation_tests_invalid_input(): + # Wenn eine ungültige Datei übergeben wird, soll ein Fehler oder leeres Ergebnis folgen + with pytest.raises(Exception): + trace_agg.run_aggregation_tests(["/nonexistent/path.csv"]) \ No newline at end of file