203 lines
6.8 KiB
Python
203 lines
6.8 KiB
Python
from unittest.mock import patch
|
|
|
|
from starlette.testclient import TestClient
|
|
|
|
from aare.daq.server import app
|
|
|
|
|
|
def test_status_renews_gui_session_with_gui_timeout(client, mock_backend):
|
|
mock_cfg = mock_backend["cfg"]
|
|
mock_daq = mock_backend["daq"]
|
|
|
|
from aare.common.models import (
|
|
DAQStatusModel,
|
|
SessionStatus,
|
|
SessionsStateEnum,
|
|
BeamlineStateEnum,
|
|
SampleGeometryModel,
|
|
BeamlineStatus,
|
|
CrystalSize,
|
|
SampleCameraSettings,
|
|
)
|
|
from aare.common.diffraction_geometry import DiffractionGeometry
|
|
from aare.common.coordinate import Coordinate, SmargonCoordinate
|
|
|
|
geom = SampleGeometryModel(
|
|
beam_location_pxl=Coordinate(x=500, y=500),
|
|
pixel_in_mm=0.001,
|
|
aerotech=Coordinate(x=0, y=0, z=0),
|
|
aerotech_meas=Coordinate(x=0, y=0, z=0),
|
|
smargon=SmargonCoordinate(sh_mm=Coordinate(x=0, y=0, z=0), phi_deg=0.0, chi_deg=0.0),
|
|
omega_deg=0.0,
|
|
beam_size_mm=Coordinate(x=0.01, y=0.01),
|
|
)
|
|
diff = DiffractionGeometry(
|
|
energy_keV=12.0,
|
|
dtz_mm=150.0,
|
|
pixel_size_mm=0.075,
|
|
beam_center_pxl=(1000.0, 1000.0),
|
|
detector_size_pxl=(2000, 2000),
|
|
detector_description="Eiger 16M",
|
|
detector_serial_number="123",
|
|
poni_rot1_rad=0.0,
|
|
poni_rot2_rad=0.0,
|
|
)
|
|
bl_status = BeamlineStatus(
|
|
name="X06DA",
|
|
ring_current_mA=400.0,
|
|
front_light=0.0,
|
|
back_light=0.0,
|
|
cryojet_K=100.0,
|
|
shutter_open=False,
|
|
exp_shutter_open=False,
|
|
flux_ph_s=1e12,
|
|
transmission=1.0,
|
|
zoom=1.0,
|
|
sample_camera=SampleCameraSettings(exposure=0.1, gain=1.0),
|
|
commissioning_mode=False,
|
|
dtz_min=120.0,
|
|
dtz_max=1600.0,
|
|
)
|
|
session_status = SessionStatus(
|
|
session=SessionsStateEnum.Vacant,
|
|
current_pgroup="p12345",
|
|
staff=True,
|
|
)
|
|
|
|
mock_daq.status = DAQStatusModel(
|
|
geom=geom,
|
|
diffraction=diff,
|
|
bl=bl_status,
|
|
state=BeamlineStateEnum.Maintenance,
|
|
busy=False,
|
|
session=session_status,
|
|
crystal_size=CrystalSize(x=0, y=0, z=0),
|
|
)
|
|
mock_cfg.pgroup = "p12345"
|
|
mock_cfg.session_state.return_value = SessionsStateEnum.OwnedByYou
|
|
mock_cfg.get_open_gui_sessions.return_value = []
|
|
mock_cfg.GUI_SESSION_EXPIRE_SECONDS = 10
|
|
|
|
with patch("aare.daq.server.cfg", mock_cfg), patch("aare.daq.server.daq", mock_daq):
|
|
response = client.get("/status", headers={"Authorization": "Bearer fake-token"})
|
|
|
|
assert response.status_code == 200
|
|
mock_cfg.touch_gui_session.assert_called_once_with(
|
|
session=123,
|
|
username="testuser",
|
|
staff=True,
|
|
expiry_sec=10,
|
|
)
|
|
|
|
|
|
def test_status_hides_open_guis_for_non_staff(mock_backend):
|
|
with patch("aare.daq.auth.parse_token") as mock_parse:
|
|
from aare.daq.auth import TokenData
|
|
from aare.common.models import (
|
|
DAQStatusModel,
|
|
SessionStatus,
|
|
SessionsStateEnum,
|
|
BeamlineStateEnum,
|
|
SampleGeometryModel,
|
|
BeamlineStatus,
|
|
CrystalSize,
|
|
SampleCameraSettings,
|
|
OpenGuiSessionInfo,
|
|
)
|
|
from aare.common.diffraction_geometry import DiffractionGeometry
|
|
from aare.common.coordinate import Coordinate, SmargonCoordinate
|
|
|
|
mock_parse.return_value = TokenData(
|
|
sub="user1",
|
|
staff=False,
|
|
pgroups=["p12345"],
|
|
session=123,
|
|
)
|
|
|
|
geom = SampleGeometryModel(
|
|
beam_location_pxl=Coordinate(x=500, y=500),
|
|
pixel_in_mm=0.001,
|
|
aerotech=Coordinate(x=0, y=0, z=0),
|
|
aerotech_meas=Coordinate(x=0, y=0, z=0),
|
|
smargon=SmargonCoordinate(sh_mm=Coordinate(x=0, y=0, z=0), phi_deg=0.0, chi_deg=0.0),
|
|
omega_deg=0.0,
|
|
beam_size_mm=Coordinate(x=0.01, y=0.01),
|
|
)
|
|
diff = DiffractionGeometry(
|
|
energy_keV=12.0,
|
|
dtz_mm=150.0,
|
|
pixel_size_mm=0.075,
|
|
beam_center_pxl=(1000.0, 1000.0),
|
|
detector_size_pxl=(2000, 2000),
|
|
detector_description="Eiger 16M",
|
|
detector_serial_number="123",
|
|
poni_rot1_rad=0.0,
|
|
poni_rot2_rad=0.0,
|
|
)
|
|
bl_status = BeamlineStatus(
|
|
name="X06DA",
|
|
ring_current_mA=400.0,
|
|
front_light=0.0,
|
|
back_light=0.0,
|
|
cryojet_K=100.0,
|
|
shutter_open=False,
|
|
exp_shutter_open=False,
|
|
flux_ph_s=1e12,
|
|
transmission=1.0,
|
|
zoom=1.0,
|
|
sample_camera=SampleCameraSettings(exposure=0.1, gain=1.0),
|
|
commissioning_mode=False,
|
|
dtz_min=120.0,
|
|
dtz_max=1600.0,
|
|
)
|
|
|
|
mock_backend["daq"].status = DAQStatusModel(
|
|
geom=geom,
|
|
diffraction=diff,
|
|
bl=bl_status,
|
|
state=BeamlineStateEnum.Maintenance,
|
|
busy=False,
|
|
session=SessionStatus(session=SessionsStateEnum.Vacant, current_pgroup="p12345", staff=False),
|
|
crystal_size=CrystalSize(x=0, y=0, z=0),
|
|
)
|
|
mock_backend["cfg"].pgroup = "p12345"
|
|
mock_backend["cfg"].session_state.return_value = SessionsStateEnum.OwnedByElse
|
|
mock_backend["cfg"].get_open_gui_sessions.return_value = [
|
|
OpenGuiSessionInfo(session=1, username="staff1", last_seen_ts=1.0, staff=True)
|
|
]
|
|
mock_backend["cfg"].get_gui_session.return_value = OpenGuiSessionInfo(
|
|
session=123,
|
|
username="user1",
|
|
last_seen_ts=2.0,
|
|
staff=False,
|
|
)
|
|
|
|
with TestClient(app) as local_client:
|
|
response = local_client.get("/status", headers={"Authorization": "Bearer fake-token"})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()["open_guis"]
|
|
assert len(payload) == 1
|
|
assert payload[0]["session"] == 123
|
|
assert payload[0]["username"] == "user1"
|
|
|
|
|
|
def test_admin_gui_sessions_includes_baton_holder_flag(client, mock_backend):
|
|
from aare.common.models import OpenGuiSessionInfo
|
|
|
|
mock_cfg = mock_backend["cfg"]
|
|
mock_cfg.get_open_gui_sessions.return_value = [
|
|
OpenGuiSessionInfo(session=11, username="user1", last_seen_ts=1.0, staff=False, holds_baton=False),
|
|
OpenGuiSessionInfo(session=22, username="holder", last_seen_ts=2.0, staff=True, holds_baton=True),
|
|
]
|
|
|
|
with patch("aare.daq.server.cfg", mock_cfg):
|
|
print(mock_backend["cfg"], id(mock_backend["cfg"]))
|
|
import aare.daq.server as server
|
|
print(server.cfg, id(server.cfg))
|
|
response = client.get("/admin/gui_sessions", headers={"Authorization": "Bearer fake-token"})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload[0]["holds_baton"] is False
|
|
assert payload[1]["holds_baton"] is True |