add tests for statistics calculations (#15)

services/process_data/__init__.py

@@ -61,8 +61,12 @@ def calculate_workflow_run_durations(self, workflow: models.Workflow):
     def calculate_statistics(self, durations: List[int]) -> dict:
         mean_duration = statistics.mean(durations) if durations != [] else None
         median_duration = statistics.median(durations) if durations != [] else None
-        stdev_duration = statistics.stdev(durations) if len(durations) > 1 else 0
-        deciles = statistics.quantiles(durations, n=10) if len(durations) > 1 else None
+        stdev_duration = statistics.pstdev(durations) if len(durations) > 1 else None
+        deciles = (
+            statistics.quantiles(durations, n=10, method="inclusive")
+            if len(durations) > 1
+            else None
+        )
         p90_duration = deciles[-1] if deciles else None
         return dict(
             mean_duration=mean_duration,

services/process_data/tests/test_process_data.py

@@ -0,0 +1,52 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from services.process_data import ProcessDataService
+
+
+class TestProcessData(object):
+    @pytest.mark.parametrize(
+        "durations,expected",
+        [
+            (
+                [],
+                dict(
+                    mean_duration=None,
+                    median_duration=None,
+                    stdev_duration=None,
+                    p90_duration=None,
+                ),
+            ),
+            (
+                [1],
+                dict(
+                    mean_duration=1,
+                    median_duration=1,
+                    stdev_duration=None,
+                    p90_duration=None,
+                ),
+            ),
+            (
+                [2, 2],
+                dict(
+                    mean_duration=2,
+                    median_duration=2,
+                    stdev_duration=0,
+                    p90_duration=2,
+                ),
+            ),
+            (
+                [2, 3, 4, 5],
+                dict(
+                    mean_duration=3.5,
+                    median_duration=3.5,
+                    stdev_duration=1.118033988749895,
+                    p90_duration=4.7,
+                ),
+            ),
+        ],
+    )
+    def test_calculate_statistics(self, durations, expected):
+        process_data = ProcessDataService(MagicMock())
+        assert process_data.calculate_statistics(durations) == expected