adding test for CI

2026-06-08 13:21:28 +02:00
parent 48ac088900
commit 5e575a020c
7 changed files with 202 additions and 23 deletions
@@ -0,0 +1,21 @@
+name: CI
+
+on:
+  push:
+    branches: [ "main" ]
+
+jobs:
+  build-and-test:
+    runs-on: linmidas
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: Configure project
+        run: |
+          cmake -B build
+      - name: Build
+        run: |
+          cmake --build build
+      - name: Run tests
+        run: |
+          
@@ -67,6 +67,7 @@ get8 {
    disconnect;
 }

+
 set {
    OutTerminator = "\r\n";
    out "GET /hidden.htm?M0\%3AO\$2=%s HTTP/1.1";
@@ -75,4 +76,4 @@ set {
    out "Content-Length: 0";
    out "";
    disconnect;
-}
+}
@@ -10,3 +10,10 @@ record(scalcout, "$(DEVICE_NAME)_outlet_getter") {
 	field(FLNK, "$(DEVICE_NAME)_get_outlet_1")
 }

+record(scalcout, "$(DEVICE_NAME)_outlet_init") {
+    field(DTYP, "stream")
+	field(PINI, "YES")
+	field(CALC, "0")
+	field(OUT, "@$(DB)command.proto get$(OUTLET_NUMBER)($(HOST_NAME)) $(PORT)")
+	field(FLNK, "$(DEVICE_NAME)_init_1")
+}
@@ -1,5 +1,6 @@
 record(stringout, "$(DEVICE_NAME)_set_$(PARAMETER_NAME)_$(PARAMETER_ID)") {
    field(DTYP, "stream")
+    field(UDF, "0")
    field(PINI, "NO")
    field(OUT, "@$(DB)command.proto set($(HOST_NAME), $(PARAMETER_ID)) $(PORT)")
 }
@@ -8,3 +9,10 @@ record(stringin, "$(DEVICE_NAME)_get_$(PARAMETER_NAME)_$(PARAMETER_ID)") {
    field(INP, "$(DEVICE_NAME)_outlet_getter.$(SCALCOUT_FIELD_INPUT) NPP MS")
    field(FLNK, "$(DEVICE_NAME)_get_$(PARAMETER_NAME)_$(FORWARDED_ID)")
 }
+
+record(stringout, "$(DEVICE_NAME)_init_$(PARAMETER_ID)") {
+    field(DOL, "$(DEVICE_NAME)_outlet_init.$(SCALCOUT_FIELD_INPUT) NPP MS")
+    field(OUT, "$(DEVICE_NAME)_set_$(PARAMETER_NAME)_$(PARAMETER_ID) NPP")
+    field(OMSL, "closed_loop")
+    field(FLNK, "$(DEVICE_NAME)_init_$(FORWARDED_ID)")
+}
@@ -13,4 +13,4 @@ file feedback_records.db
    pattern
    { CONFIG_PARAMETER_NAME, OUTLET_NUMBER, GETTER_OUTPUT_PARAMETER_NAME}
    { "maxOutlet", 4, "outlet"}
-}
+}
@@ -1,6 +1,8 @@
 import socket
 import datetime
 import random
+import signal
+import time

 port=55555

@@ -22,14 +24,17 @@ def readAscii(connexion_client : socket):
    message = ""

    chunk = ""
-    while chunk != b'\r\n':
+    while chunk != b'\r\n' and keepAlive:
+        if not keepAlive: 
+            return
        chunk = connexion_client.recv(1024)
        message += chunk.decode()

-    if message[15] == '\n' :
-        outlet_index = ord(message[23]) - ord('0')
+    if len(message) != 58 :
+        print(message)
+        outlet_index = ord(message[22]) - ord('1')
        print(outlet_state[outlet_index])
-        match message[26]:
+        match message[25]:
            case 'n':
                outlet_state[outlet_index] = 'On'
                print("powering on " + str(outlet_index))
@@ -38,9 +43,9 @@ def readAscii(connexion_client : socket):
                print("powering off " + str(outlet_index))
            case 'e':
                outlet_state[outlet_index] = 'Rst'
-                print("restarting " + str(outlet_index))
+                print("restarting " + str(outlet_index + 1))
                outlet_restart_time[outlet_index] = random.randint(2,5)
-    if message[15] == '?':
+    if len(message) == 58:
        time = datetime.datetime.now().strftime('%d %B %Y %H:%M:%S')
        connexion_client.send(bytes("HTTP/1.0 200 OK\r\n" +
                                    "Server: lwIP/1.3.2 (http://www.sics.se/~adam/lwip/)\r\n"+
@@ -70,13 +75,28 @@ def readAscii(connexion_client : socket):
                                    "</body></html> "                                  
                                    
                                    , "UTF-8"))
+keepAlive = True;
+
+def turnOff():
+    keepAlive = False
+
+signal.signal(signal.SIGTERM, turnOff )


 socket_listener.bind(('', port))
 socket_listener.listen()

-while(True):
+def turnOff():
+    keepAlive = False
+    socket_listener.close()
+
+while(keepAlive):
    connexion_client, address_client = socket_listener.accept()
+    socket_listener.settimeout(1)
    readAscii(connexion_client)
    updateSocketRestartTimer()
    connexion_client.close()
+socket_listener.close()
+
+
+
@@ -1,20 +1,142 @@
 from caproto.threading.client import Context
+from caproto.sync.client import read, write

+from time import sleep

-def test_main():
-    assert(0 == 1)
+delay = 1

 class CaprotoWrapper:
-    def __init__(self, timeout = 3.0):
-        self._pvs = {}
-        self._ctx = Context()
-    
-    def connect_pv(self, pvname):
-        if pvname in self._pvs:
-            return
-        pv = self._ctx.get_pvs(pvname)
-        self._pvs[pvname] = pv
+    def raw_read(pvname):
+        return read(pvname).data[0].decode('utf-8')
+
+    def get_pv_value(pvname, expected_value):
+        timeout = 10
+        while(expected_value != CaprotoWrapper.raw_read(pvname) and timeout > 0):
+            timeout -=1
+            sleep(0.5)
+        
+        return CaprotoWrapper.raw_read(pvname)
+
+    def set_pv_value(pvname, value):
+        write(pvname, value, notify=True)
+        timeout = 10
+        while(value != CaprotoWrapper.raw_read(pvname) and timeout > 0):
+            timeout -= 1
+            sleep(0.5)
+        
+
+def test_turn_on():
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_1", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_1", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_1", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_2", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_2", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_2", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_3", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_3", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_3", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_4", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_4", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_4", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_5", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_5", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_5", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_6", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_6", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_6", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_7", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_7", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_7", "On") == "On")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_8", "On")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_8", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_8", "On") == "On")
+
+
+
+def test_turn_off():
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_1", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_1", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_1", "Off") == "Off")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_2", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_2", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_2", "Off") == "Off")
+
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_3", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_3", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_3", "Off") == "Off")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_4", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_4", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_4", "Off") == "Off")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_5", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_5", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_5", "Off") == "Off")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_6", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_6", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_6", "Off") == "Off")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_7", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_7", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_7", "Off") == "Off")
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_8", "Off")
+    
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_set_outlet_8", "Off") == "Off")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_8", "Off") == "Off")
+
+
+
+def test_restart():
+
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_1", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_1", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_2", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_2", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_3", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_3", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_4", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_4", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_5", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_5", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_6", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_6", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_7", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_7", "Rst") == "Rst")
+    CaprotoWrapper.set_pv_value("ePowerSwitch_set_outlet_8", "Restart")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_8", "Rst") == "Rst")
+
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_1", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_2", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_3", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_4", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_5", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_6", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_7", "On") == "On")
+    assert(CaprotoWrapper.get_pv_value("ePowerSwitch_get_outlet_8", "On") == "On")

-    def get_pv_value(self, pvname):
-        self.connect_pv(pvname)
-        pv = self._pvs[pvname]