nixos/tests/phoebus/save-and-restore: init

nixos/tests/all-tests.nix

@@ -19,4 +19,5 @@ in {
   ca-gateway = handleTest ./ca-gateway.nix {};
   phoebus-alarm = handleTest ./phoebus/alarm.nix {};
   phoebus-olog = handleTest ./phoebus/olog.nix {};
+  phoebus-save-and-restore = handleTest ./phoebus/save-and-restore.nix {};
 }

nixos/tests/phoebus/save-and-restore.nix

@@ -0,0 +1,32 @@
+{
+  lib,
+  epnixLib,
+  ...
+}: {
+  name = "phoebus-save-and-restore-simple-check";
+  meta.maintainers = with epnixLib.maintainers; [minijackson];
+
+  nodes = {
+    server = {
+      services.phoebus-save-and-restore = {
+        enable = true;
+        openFirewall = true;
+      };
+
+      nixpkgs.config.allowUnfreePredicate = pkg:
+        builtins.elem (lib.getName pkg) [
+          # Elasticsearch can be used as an SSPL-licensed software, which is
+          # not open-source. But as we're using it run tests, not exposing
+          # any service, this should be fine.
+          "elasticsearch"
+        ];
+
+      # Else OOM
+      virtualisation.memorySize = 2047;
+    };
+
+    client = {};
+  };
+
+  testScript = builtins.readFile ./save-and-restore.py;
+}

nixos/tests/phoebus/save-and-restore.py

@@ -0,0 +1,163 @@
+import json
+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 | Dict[str, Any] | List[Any]
+
+root_node_id = "44bef5de-e8e6-4014-af37-b8f6c8a939a2"
+user = "myself"
+
+
+def get(uri: str):
+    return json.loads(
+        client.succeed(
+            "curl -sSf " "-H 'Accept: application/json' " f"'http://server:8080{uri}'"
+        )
+    )
+
+
+def put(uri: str, data: JSON):
+    encoded_data = json.dumps(data)
+    return json.loads(
+        client.succeed(
+            "curl -sSf "
+            "-X PUT "
+            "-H 'Content-Type: application/json' "
+            "-H 'Accept: application/json' "
+            f"'http://server:8080{uri}' "
+            f"--data '{encoded_data}'"
+        )
+    )
+
+
+def delete(uri: str):
+    client.succeed(
+        "curl -sSf "
+        "-X DELETE "
+        "-H 'Content-Type: application/json' "
+        "-H 'Accept: application/json' "
+        f"'http://server:8080{uri}'"
+    )
+
+
+start_all()
+
+server.wait_for_unit("phoebus-save-and-restore.service")
+server.wait_for_open_port(8080)
+
+client.wait_for_unit("multi-user.target")
+
+with subtest("Default root node is created"):
+    node = get(f"/node/{root_node_id}")
+    assert node["uniqueId"] == root_node_id
+
+subnode_id: str
+
+with subtest("We can create a subnode"):
+    result = put(
+        f"/node?parentNodeId={root_node_id}",
+        {
+            "name": "subnode",
+            "description": "A test subnode",
+            "userName": user,
+        },
+    )
+    subnode_id = result["uniqueId"]
+    # Check that it is really added
+    node = get(f"/node/{subnode_id}")
+    parent_node = get(f"/node/{subnode_id}/parent")
+    assert parent_node["uniqueId"] == root_node_id
+
+config_id: str
+
+with subtest("We can create a config"):
+    result = put(
+        f"/config?parentNodeId={subnode_id}",
+        {
+            "configurationNode": {
+                "name": "test configuration",
+                "userName": user,
+            },
+            "configurationData": {
+                "pvList": [
+                    {
+                        "pvName": "double",
+                    },
+                    {
+                        "pvName": "string",
+                    },
+                    {
+                        "pvName": "intarray",
+                    },
+                    {
+                        "pvName": "stringarray",
+                    },
+                    {
+                        "pvName": "enum",
+                    },
+                    {
+                        "pvName": "table",
+                    },
+                ]
+            },
+        },
+    )
+    config_id = result["configurationNode"]["uniqueId"]
+    config = get(f"/config/{config_id}")
+    assert config["uniqueId"] == config_id
+
+
+def vtype(name: str, typ: str, value: Any) -> Dict[str, Any]:
+    return {
+        "configPv": {
+            "pvName": name,
+            "readOnly": False,
+        },
+        "value": {
+            "type": {
+                "name": typ,
+                "version": 1,
+            },
+            "value": value,
+            "alarm": {
+                "severity": "NONE",
+                "status": "NONE",
+                "name": "NO_ALARM",
+            },
+            "time": {"unixSec": 1664550284, "nanoSec": 870687555},
+            "display": {
+                "lowDisplay": 0.0,
+                "highDisplay": 0.0,
+                "units": "",
+            },
+        },
+    }
+
+
+snapshot_id: str
+
+with subtest("We can create a snapshot"):
+    result = put(
+        f"/snapshot?parentNodeId={config_id}",
+        {
+            "snapshotNode": {
+                "name": "test snapshot",
+                "userName": user,
+            },
+            "snapshotData": {
+                "snapshotItems": [
+                    vtype("double", "VDouble", 42.0),
+                    vtype("string", "VString", "hello"),
+                    vtype("intarray", "VIntArray", [1, 2, 3]),
+                    vtype("stringarray", "VStringArray", ["you", "and", "me"]),
+                ],
+            },
+        },
+    )
+    snapshot_id = result["snapshotNode"]["uniqueId"]
+    snapshot = get(f"/snapshot/{snapshot_id}")
+    assert config["uniqueId"] == config_id
+
+with subtest("We can delete a node"):
+    print(delete(f"/node/{subnode_id}"))