Store post requests' data for further processing.

Also drop the pydantic message type.
We currently don't make use of this in the ROS2 C++ driver.

schunk_egu_egk_gripper_dummy/schunk_egu_egk_gripper_dummy/main.py

@@ -3,7 +3,6 @@
 from fastapi import FastAPI, Request, Form
 from fastapi.middleware.cors import CORSMiddleware
 from typing import Optional
-from pydantic import BaseModel
 
 # Components
 dummy = Dummy()
@@ -20,21 +19,14 @@
 )
 
 
-class Update(BaseModel):
-    inst: str
-    value: str
-    elem: Optional[int] = None
-    callback: Optional[str] = None
-
-
 @server.post("/adi/update.json")
 async def post(
     inst: str = Form(...),
     value: str = Form(...),
     elem: Optional[int] = Form(None),
     callback: Optional[str] = Form(None),
 ):
-    msg = Update(inst=inst, value=value, elem=elem, callback=callback)
+    msg = {"inst": inst, "value": value, "elem": elem, "callback": callback}
     return dummy.post(msg)
 
 

schunk_egu_egk_gripper_dummy/src/dummy.py

@@ -3,6 +3,7 @@
 import os
 from pathlib import Path
 import json
+import string
 
 
 class Dummy(object):
@@ -55,6 +56,17 @@ def _run(self) -> None:
         print("Done")
 
     def post(self, msg: dict) -> dict:
+        if "inst" not in msg and "value" not in msg:
+            return {"result": 1}
+        if msg["inst"] not in self.data:
+            return {"result": 1}
+        if not all(digit in string.hexdigits for digit in msg["value"]):
+            return {"result": 1}
+
+        if msg["inst"] == self.plc_output:
+            self.plc_output_buffer = bytearray(bytes.fromhex(msg["value"]))
+        else:
+            self.data[msg["inst"]] = [msg["value"]]
         return {"result": 0}
 
     def get_info(self, query: dict[str, str]) -> dict:

schunk_egu_egk_gripper_dummy/tests/test_requests.py

@@ -62,9 +62,36 @@ def test_dummy_survives_invalid_data_requests():
     assert dummy.get_data(query) == expected
 
 
-def test_dummy_responds_correctly_to_post_requests():
+def test_dummy_stores_post_requests():
     dummy = Dummy()
-    inst = "0x0048"
-    data = {"inst": inst, "value": "01"}
-    expected = {"result": 0}
-    assert dummy.post(data) == expected
+
+    # Using the plc command variable
+    msg = "00112233445566778899AABBCCDDEEFF"
+    data = {"inst": dummy.plc_output, "value": msg}
+    dummy.post(data)
+    assert dummy.get_plc_output() == [msg]
+
+    # Using general variables
+    msg = "AABBCCDD"
+    inst = "0x0238"
+    data = {"inst": inst, "value": msg}
+    dummy.post(data)
+    assert dummy.data[inst] == [msg]
+
+
+def test_dummy_rejects_invalid_post_requests():
+    dummy = Dummy()
+    valid_data = "AABBCCDD"
+    valid_inst = "0x0238"
+    data = {"inst": valid_inst, "value": valid_data}
+    assert dummy.post(data) == {"result": 0}
+
+    invalid_data = "hello:)"
+    valid_inst = "0x0238"
+    data = {"inst": valid_inst, "value": invalid_data}
+    assert dummy.post(data) == {"result": 1}
+
+    valid_data = "AABBCCDD"
+    invalid_inst = "0x9999"
+    data = {"inst": invalid_inst, "value": valid_data}
+    assert dummy.post(data) == {"result": 1}