Add bandwidth_analysis/tests/test_core.py

bandwidth_analysis/tests/test_core.py

@@ -0,0 +1,73 @@
+import json
+import pandas as pd
+import pytest
+from bandwidth_analysis import core
+
+
+@pytest.fixture
+def baseline_data():
+    # Simulierte Basisdaten
+    return pd.DataFrame({
+        'segment': ['A', 'B', 'C'],
+        'bandwidth': [100, 200, 300],
+        'retry_tail': [0.01, 0.02, 0.03]
+    })
+
+
+@pytest.fixture
+def test_data():
+    # Simulierte Testdaten
+    return pd.DataFrame({
+        'segment': ['A', 'B', 'C'],
+        'bandwidth': [110, 250, 270],
+        'retry_tail': [0.012, 0.018, 0.04]
+    })
+
+
+def test_analyze_bandwidth_nominal(baseline_data, test_data):
+    result = core.analyze_bandwidth(baseline_data, test_data)
+
+    assert isinstance(result, dict)
+    assert all(k in result for k in ['bandwidth_change', 'retry_tail_change', 'hotspot_segments'])
+
+    # Erwartete Richtung der Veränderung prüfen
+    assert result['bandwidth_change'] > 0  # durchschnittliche Erhöhung
+    assert isinstance(result['retry_tail_change'], float)
+    assert isinstance(result['hotspot_segments'], list)
+
+
+def test_analyze_bandwidth_dict_input():
+    baseline_dict = {
+        'segment': ['X', 'Y'],
+        'bandwidth': [100, 200],
+        'retry_tail': [0.05, 0.1]
+    }
+    test_dict = {
+        'segment': ['X', 'Y'],
+        'bandwidth': [120, 180],
+        'retry_tail': [0.04, 0.12]
+    }
+    result = core.analyze_bandwidth(baseline_dict, test_dict)
+
+    assert isinstance(result, dict)
+    assert isinstance(result['bandwidth_change'], float)
+    assert isinstance(result['retry_tail_change'], float)
+
+
+def test_invalid_input_raises():
+    with pytest.raises((TypeError, ValueError)):
+        core.analyze_bandwidth(None, None)
+
+
+def test_hotspot_detection(baseline_data, test_data):
+    result = core.analyze_bandwidth(baseline_data, test_data)
+    # Hotspot sollte mindestens einen betroffenen Abschnitt enthalten
+    assert any(seg in ['A', 'B', 'C'] for seg in result['hotspot_segments'])
+
+
+def test_output_serializable(baseline_data, test_data):
+    result = core.analyze_bandwidth(baseline_data, test_data)
+    # Ergebnis muss JSON-serialisierbar sein
+    json_str = json.dumps(result)
+    restored = json.loads(json_str)
+    assert restored.keys() == result.keys()