x01dcandClaude Opus 5.5
c1cc2cbc8e
CI for csaxs_bec / test (push) Successful in 3m17s
flomni.rt_<TAB> now lists the whole real-time system: - feedback_* -> rt_feedback_* - laser_tracker_on/off/show_all -> rt_laser_tracker_*, laser_parameters_tweak_tracking -> rt_laser_tracker_tweak - show_signal_strength_interferometer -> rt_signal_strength (old name kept as alias) - realtime_system -> rt_settings; laser_parameters_* setters moved there as rt_settings.tracking_show / tracking_set_target_y/z / tracking_set_interferometer_threshold / tracking_set_psd_threshold_low - rt_on / rt_off unchanged Docs, callers (x_ray_eye_align, e2e bootstrap) and tests updated. TODO note for porting the scheme to LamNI and OMNY in OMNY_shared/AI_docs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
4.8 KiB
Python
109 lines
4.8 KiB
Python
"""Tests for the baseline device flomni_loop_settings (devices/omny/flomni_loop_settings.py): the loop settings of
|
|
the flOMNI web HMI stored with every scan. A small local HTTP server stands in for the web HMI (/api/settings)."""
|
|
|
|
import json
|
|
import math
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import pytest
|
|
|
|
from csaxs_bec.devices.omny import flomni_loop_settings as fls
|
|
from csaxs_bec.devices.omny.flomni_loop_settings import SETTINGS, FlomniLoopSettings
|
|
|
|
TOKEN = "device-token-0123456789abcdef"
|
|
|
|
SNAP = {
|
|
"instrument": "flOMNI", "definition": "/home/control/OMNY/flOMNI/FeedbackLoop/loopsettings/flomni.json",
|
|
"definition_sha1": "a" * 40, "scan": {"mode": 0, "active": False, "text": ""},
|
|
"settings": [
|
|
{"name": "fzp_enable", "value": 1, "unsaved": False, "consistent": True},
|
|
{"name": "fzp_range_V_x", "value": 1.5, "unsaved": True, "consistent": True},
|
|
{"name": "trigger_tolerance_nm", "value": 10.0, "unsaved": False, "consistent": False},
|
|
{"name": "new_setting_later", "value": 3.0, "unsaved": False, "consistent": True},
|
|
],
|
|
"rules": [{"name": "r1", "severity": "error", "ok": False}, {"name": "r2", "severity": "warning", "ok": False}],
|
|
}
|
|
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
hits = 0
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
_Handler.hits += 1
|
|
code, body = (200, SNAP) if self.headers.get("Authorization") == "Bearer " + TOKEN else (401, {"detail": "no"})
|
|
data = json.dumps(body).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
|
|
@pytest.fixture
|
|
def url():
|
|
_Handler.hits = 0
|
|
srv = HTTPServer(("127.0.0.1", 0), _Handler)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
yield f"http://127.0.0.1:{srv.server_address[1]}"
|
|
srv.shutdown()
|
|
|
|
|
|
def test_read_stores_settings(url):
|
|
dev = FlomniLoopSettings(name="flomni_loop_settings", url=url, token=TOKEN)
|
|
r = dev.read()
|
|
assert r["flomni_loop_settings_available"]["value"] == 1
|
|
assert r["flomni_loop_settings_fzp_enable"]["value"] == 1.0
|
|
assert r["flomni_loop_settings_fzp_range_V_x"]["value"] == 1.5
|
|
assert math.isnan(r["flomni_loop_settings_move_beyond_um_x"]["value"]) # not in the answer
|
|
assert r["flomni_loop_settings_unsaved"]["value"] == 1
|
|
assert r["flomni_loop_settings_inconsistent"]["value"] == 1
|
|
assert r["flomni_loop_settings_rules_violated"]["value"] == 1 # error rules only
|
|
assert r["flomni_loop_settings_definition"]["value"].endswith("sha1 " + "a" * 40)
|
|
assert "new_setting_later" in r["flomni_loop_settings_message"]["value"]
|
|
snap = json.loads(r["flomni_loop_settings_snapshot_json"]["value"])
|
|
assert snap["settings"]["fzp_enable"]["value"] == 1 and "profiles" not in snap
|
|
assert snap["settings"]["new_setting_later"]["value"] == 3.0 # later settings: in the JSON
|
|
assert [r["name"] for r in snap["rules_violated"]] == ["r1", "r2"]
|
|
assert set(SETTINGS) <= {k.replace("flomni_loop_settings_", "") for k in r}
|
|
|
|
|
|
def test_unavailable_does_not_raise_and_backs_off(url, monkeypatch):
|
|
dev = FlomniLoopSettings(name="ls", url=url, token="wrong-token-xxxxxxxxxxxx")
|
|
r = dev.read()
|
|
assert r["ls_available"]["value"] == 0 and "401" in r["ls_message"]["value"]
|
|
assert math.isnan(r["ls_fzp_enable"]["value"]) and r["ls_unsaved"]["value"] == -1
|
|
hits = _Handler.hits
|
|
dev.read() # within the back-off: no new request
|
|
assert _Handler.hits == hits
|
|
monkeypatch.setattr(fls, "RETRY_AFTER_FAILURE_S", 0.0)
|
|
dev.read()
|
|
assert _Handler.hits == hits + 1
|
|
|
|
|
|
def test_unreachable():
|
|
dev = FlomniLoopSettings(name="ls", url="http://127.0.0.1:9", token=TOKEN, timeout=0.5)
|
|
r = dev.read()
|
|
assert r["ls_available"]["value"] == 0 and "not reachable" in r["ls_message"]["value"]
|
|
|
|
|
|
def test_token_file_and_snapshot(url, tmp_path):
|
|
tok = tmp_path / "token"
|
|
tok.write_text(TOKEN + "\n")
|
|
dev = FlomniLoopSettings(name="ls", url=url, token_file=str(tok))
|
|
assert dev.snapshot()["fzp_range_V_x"] == 1.5
|
|
|
|
|
|
def test_api_methods_for_realtime_system(url):
|
|
"""api_get / api_post: the RPC used by flomni.rt_settings (token only in the device config)."""
|
|
dev = FlomniLoopSettings(name="ls", url=url, token=TOKEN)
|
|
assert dev.api_get("/api/settings")["instrument"] == "flOMNI"
|
|
with pytest.raises(Exception, match="not allowed"):
|
|
dev.api_get("/api/params") # only the loop-settings endpoints
|
|
with pytest.raises(Exception, match="not allowed"):
|
|
dev.api_post("/api/loop/stop", {})
|
|
assert "api_get" in FlomniLoopSettings.USER_ACCESS and "set" not in FlomniLoopSettings.USER_ACCESS
|