From 23e36ccb1d04589da16e0c0c507c8c1aaf7836e8 Mon Sep 17 00:00:00 2001 From: kparasch Date: Tue, 22 Sep 2026 14:30:49 +0200 Subject: [PATCH 1/3] version bump --- pySC/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pySC/version.py b/pySC/version.py index 3a384fb..8c118e6 100644 --- a/pySC/version.py +++ b/pySC/version.py @@ -1 +1 @@ -__version__ = "1.5.4" +__version__ = "1.5.5" From e6d3f718a51b7f02bbb32881c5592ac9d63e99a5 Mon Sep 17 00:00:00 2001 From: kparasch Date: Thu, 1 Oct 2026 21:24:00 +0200 Subject: [PATCH 2/3] Server: allow variable to have slashes and write tests. --- pySC/control_system/magnet_server.py | 6 +- pySC/control_system/rf_server.py | 6 +- pySC/control_system/server.py | 2 +- tests/control_system/test_server.py | 99 ++++++++++++++++++++++++++++ 4 files changed, 108 insertions(+), 5 deletions(-) create mode 100644 tests/control_system/test_server.py diff --git a/pySC/control_system/magnet_server.py b/pySC/control_system/magnet_server.py index c9f1b67..a1505ee 100644 --- a/pySC/control_system/magnet_server.py +++ b/pySC/control_system/magnet_server.py @@ -3,7 +3,9 @@ def magnet_server(conn, signal, SC): variable = signal.split(' ')[1] command = signal[:3] - server, device, prop = variable.strip().split('/') + server, _, device_prop = variable.strip().partition('/') + device, _, prop = device_prop.rpartition('/') + control = '/'.join([device, prop]) if control not in SC.magnet_settings.controls.keys(): raise pySCServerError @@ -17,4 +19,4 @@ def magnet_server(conn, signal, SC): SC.magnet_settings.set(control, value) send_int(conn, 1) # 1 for set ok else: - raise pySCServerError \ No newline at end of file + raise pySCServerError diff --git a/pySC/control_system/rf_server.py b/pySC/control_system/rf_server.py index 93090f2..79a637f 100644 --- a/pySC/control_system/rf_server.py +++ b/pySC/control_system/rf_server.py @@ -3,7 +3,9 @@ def rf_server(conn, signal, SC): variable = signal.split(' ')[1] command = signal[:3] - server, device, prop = variable.strip().split('/') + server, _, device_prop = variable.strip().partition('/') + device, _, prop = device_prop.rpartition('/') + device = device.lower() if device in SC.rf_settings.systems: if command == 'GET': @@ -25,4 +27,4 @@ def rf_server(conn, signal, SC): SC.rf_settings.systems[device].set_frequency(value) send_int(conn, 1) # 1 for set ok else: - raise pySCServerError \ No newline at end of file + raise pySCServerError diff --git a/pySC/control_system/server.py b/pySC/control_system/server.py index dc2d778..bc4c494 100644 --- a/pySC/control_system/server.py +++ b/pySC/control_system/server.py @@ -55,7 +55,7 @@ def start_server(SC: "SimulatedCommissioning" , port : int = 13131, refresh_rate if len(signal) > 2 and signal[:3] in ['GET', 'SET']: variable = signal.split(' ')[1] - server, device, prop = variable.strip().split('/') + server = variable.strip().partition('/')[0] if variable == 'ORBIT/INJECTION/MODE': command = signal[:3] diff --git a/tests/control_system/test_server.py b/tests/control_system/test_server.py new file mode 100644 index 0000000..7b2107a --- /dev/null +++ b/tests/control_system/test_server.py @@ -0,0 +1,99 @@ +"""Regression coverage for device names in the socket server dispatch path.""" + +import struct +from types import SimpleNamespace +from unittest.mock import MagicMock, Mock, call + +import pytest + +from pySC.control_system import server +from pySC.control_system.send_receive import INT_SIZE + + +class RequestsComplete(Exception): + """Stop the server once all test requests have been handled.""" + + +@pytest.fixture +def serve_request(monkeypatch): + def run(signal, sc): + conn = MagicMock() + conn.__enter__.return_value = conn + conn.recv.return_value = signal.encode() + listener = MagicMock() + listener.__enter__.return_value = listener + listener.accept.side_effect = [ + (conn, ("127.0.0.1", 12345)), + RequestsComplete(), + ] + monkeypatch.setattr(server.socket, "socket", Mock(return_value=listener)) + monkeypatch.setattr(server, "atexit", SimpleNamespace(register=Mock())) + # Keep the refresh deadline fixed so these tests need no real-time waits. + monkeypatch.setattr(server, "time", SimpleNamespace(time=lambda: 0.0)) + sc.bpm_system = SimpleNamespace(capture_orbit=Mock(return_value=([], []))) + + with pytest.raises(RequestsComplete): + server.start_server(sc) + return conn + + return run + + +def assert_response(conn, command, value): + if command == "GET": + payload = struct.pack("=d", value) + expected = [ + call((3).to_bytes(INT_SIZE, "big")), + call(len(payload).to_bytes(INT_SIZE, "big")), + call(payload), + ] + else: + expected = [call((1).to_bytes(INT_SIZE, "big"))] + assert conn.sendall.call_args_list == expected + + +@pytest.mark.regression +@pytest.mark.parametrize("device", ["q1", "cell/q1", "ring/cell/q1"]) +@pytest.mark.parametrize("command", ["GET", "SET"]) +def test_magnet_device_names(serve_request, device, command): + control = f"{device}/B2" + settings = SimpleNamespace( + controls={control: object()}, get=Mock(return_value=2.5), set=Mock() + ) + sc = SimpleNamespace(magnet_settings=settings) + signal = f"{command} MAGNET/{control}" + if command == "SET": + signal += " 3.5" + + conn = serve_request(signal, sc) + + if command == "GET": + settings.get.assert_called_once_with(control) + settings.set.assert_not_called() + else: + settings.set.assert_called_once_with(control, 3.5) + settings.get.assert_not_called() + assert_response(conn, command, 2.5) + + +@pytest.mark.regression +@pytest.mark.parametrize("device", ["cavity", "cell/cavity", "ring/cell/cavity"]) +@pytest.mark.parametrize("command", ["GET", "SET"]) +@pytest.mark.parametrize("prop", ["VOLTAGE", "PHASE", "FREQUENCY"]) +def test_rf_device_names(serve_request, device, command, prop): + values = {"voltage": 2.5, "phase": 0.25, "frequency": 500e6} + setters = {f"set_{name}": Mock() for name in values} + system = SimpleNamespace(**values, **setters) + sc = SimpleNamespace(rf_settings=SimpleNamespace(systems={device: system})) + signal = f"{command} RF/{device}/{prop}" + if command == "SET": + signal += " 3.5" + + conn = serve_request(signal, sc) + + for name, setter in setters.items(): + if command == "SET" and name == f"set_{prop.lower()}": + setter.assert_called_once_with(3.5) + else: + setter.assert_not_called() + assert_response(conn, command, values[prop.lower()]) From 65ba20eb47f9385b6f89dddf2a172516f00ed680 Mon Sep 17 00:00:00 2001 From: kparasch Date: Thu, 1 Oct 2026 21:24:27 +0200 Subject: [PATCH 3/3] version bump --- pySC/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pySC/version.py b/pySC/version.py index 8c118e6..d0113b3 100644 --- a/pySC/version.py +++ b/pySC/version.py @@ -1 +1 @@ -__version__ = "1.5.5" +__version__ = "1.5.6"