41 lines
1.7 KiB
Python
41 lines
1.7 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pytest
|
|
import integer_buckets
|
|
|
|
|
|
def test_integer_buckets():
|
|
"""Testet die deterministische Integer-Bucket-Aggregation gegen Beispieltestdaten."""
|
|
data_path = Path(__file__).parent / "data" / "synthetic_trace.json"
|
|
assert data_path.exists(), f"Testdaten nicht gefunden: {data_path}"
|
|
|
|
with open(data_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
# Validierung: alle Records entsprechen dem erwarteten Datenmodell
|
|
for rec in data:
|
|
assert isinstance(rec.get("bucket_id"), int), "bucket_id muss int sein"
|
|
assert isinstance(rec.get("value"), int), "value muss int sein"
|
|
assert isinstance(rec.get("timestamp"), str), "timestamp muss String sein (ISO-Datetime)"
|
|
|
|
# Führe Aggregation über integer_buckets-Modul aus
|
|
result_first = integer_buckets.aggregate(data)
|
|
result_second = integer_buckets.aggregate(data)
|
|
|
|
# Prüfung: deterministisch identische Ergebnisse
|
|
assert result_first == result_second, "Integer-Bucket-Ergebnisse sind nicht deterministisch"
|
|
|
|
# Prüfung: Float-Vergleichskonsistenz
|
|
result_float_compare = integer_buckets.aggregate([
|
|
dict(bucket_id=int(r["bucket_id"]), value=float(r["value"]), timestamp=r["timestamp"])
|
|
for r in data
|
|
])
|
|
assert len(result_float_compare) == len(result_first), "Float-Vergleich hat abweichende Länge"
|
|
|
|
# Keine Off-by-One oder Rundungsfehler in Grenzen zulässig
|
|
for i, (a, b) in enumerate(zip(result_first, result_float_compare)):
|
|
assert int(a["bucket_id"]) == int(b["bucket_id"]), f"Bucket-ID-Abweichung bei Index {i}"
|
|
assert round(a["value"]) == round(b["value"]), f"Wertabweichung bei Index {i}"
|
|
|
|
return True
|