nixos/tests/archiver-appliance: test manual SCAN sampling

nixos/tests/archiver-appliance/test_script.py

@@ -1,13 +1,13 @@
 import json
 import time
-from typing import Any, List, Mapping
+from typing import Any, Dict, List
 
 # Use Any here, instead of the recursive "JSON",
 # because recursion is not yet supported
-JSON = str | int | float | bool | None | Mapping[str, Any] | List[Any]
+JSON = str | int | float | bool | None | Dict[str, Any] | List[Any]
 
 
-def get(uri: str) -> List[Any]:
+def get(uri: str):
     return json.loads(
         server.succeed(
             "curl -sSf "
@@ -17,7 +17,7 @@ def get(uri: str) -> List[Any]:
     )
 
 
-def post(uri: str, data: JSON) -> JSON:
+def post(uri: str, data: JSON):
     encoded_data = json.dumps(data)
     return json.loads(
         server.succeed(
@@ -31,12 +31,15 @@ def post(uri: str, data: JSON) -> JSON:
     )
 
 
-def check_pv_archived(pv_name: str):
-    pv_status = get(f"/mgmt/bpl/getPVStatus?pv={pv_name}")
-    return pv_status[0]["status"] == "Being archived"
+def pv_status(pv_name: str) -> List[Dict[str, Any]]:
+    return get(f"/mgmt/bpl/getPVStatus?pv={pv_name}")
 
 
-def get_data(pv_name: str):
+def check_pv_archived(pv_name: str) -> bool:
+    return pv_status(pv_name)[0]["status"] == "Being archived"
+
+
+def get_data(pv_name: str) -> List[Dict[str, Any]]:
     raw_data = get(f"/retrieval/data/getData.json?pv={pv_name}")
     return raw_data[0]["data"]
 
@@ -65,7 +68,7 @@ def caput(pv_name: str, value: str):
         "/mgmt/bpl/archivePV",
         [
             {"pv": "aiExample"},
-            {"pv": "calcExample"},
+            {"pv": "calcExample", "samplingmethod": "SCAN", "samplingperiod": "2"},
             {"pv": "static"},
             {"pv": "staticDeadband"},
             {"pv": "staticProcessed"},
@@ -214,7 +217,7 @@ def static_deadband_has_more_data(_):
         delay_sum = 0
 
         for i in range(1, 6):
-            data[i]["val"] == 0, "value of staticProcessed should not change"
+            assert data[i]["val"] == 0, "value of staticProcessed should not change"
             delay_sum += data[i]["secs"] - previous_secs
             previous_secs = data[i]["secs"]
 
@@ -243,14 +246,47 @@ def waveform_has_data(_):
         assert data[0]["val"] == ["1,2,3,4,5"], "waveform datapoint is incorrect"
 
 with subtest("non existing record"):
-    pv_status = get("/mgmt/bpl/getPVStatus?pv=nonExisting")
-    assert pv_status != "Being archived", "nonExisting record shouldn't be archived"
+    assert (
+        pv_status("nonExisting")[0]["status"] != "Being archived"
+    ), "nonExisting record shouldn't be archived"
 
     never_connected_pvs = get("/mgmt/bpl/getNeverConnectedPVs")
 
     assert len(never_connected_pvs) == 1, "only 1 PVs should never have been connected"
     assert never_connected_pvs[0]["pvName"] == "nonExisting", "wrong PV never connected"
 
-# TODO: check archiving with manual sampling period
-# TODO: check /arch, check consolidation into mts, sts
+with subtest("manual sampling period"):
+    with subtest("wait until calcExample is being archived"):
+        retry(lambda _: check_pv_archived("calcExample"))
+
+    # Somehow these ones has capital P
+    assert (
+        pv_status("calcExample")[0]["samplingPeriod"] == "2.0"
+    ), "wrong sampling period returned in status"
+
+    pv_type_info = get("/mgmt/bpl/getPVTypeInfo?pv=calcExample")
+    assert (
+        pv_type_info["samplingMethod"] == "SCAN"
+    ), "wrong sampling method returned in type info"
+    assert (
+        pv_type_info["samplingPeriod"] == "2.0"
+    ), "wrong sampling period returned in type info"
+
+    with subtest("json of calcExample is valid"):
+        data = get_data("calcExample")
+
+        previous_secs = data[0]["secs"]
+        delay_sum = 0
+
+        for i in range(1, 6):
+            delay_sum += data[i]["secs"] - previous_secs
+            previous_secs = data[i]["secs"]
+
+        mean_delay = delay_sum / 5
+
+        assert round(mean_delay) == 2, "calcExample should be archived every two second"
+
 # TODO: check pause/play
+# TODO: check /arch, check consolidation into mts, sts
+# TODO: check controlling PV
+# TODO: check different policies