From b4df0805e0854cfc50897f3dc01442ab80196756 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 2 Sep 2026 18:53:57 +0100 Subject: [PATCH 1/4] Added API endpoints to perform rsync transfer of the OTF directory and to trigger the request --- src/murfey/instrument_server/api.py | 85 +++++++++++++++++++++++++++++ src/murfey/server/api/instrument.py | 85 +++++++++++++++++++++++++++++ src/murfey/util/route_manifest.yaml | 16 ++++++ 3 files changed, 186 insertions(+) diff --git a/src/murfey/instrument_server/api.py b/src/murfey/instrument_server/api.py index 1b987b8c3..117993ec2 100644 --- a/src/murfey/instrument_server/api.py +++ b/src/murfey/instrument_server/api.py @@ -425,6 +425,91 @@ def get_possible_otf_dirs( return candidates +class OTFDirectoryUploadInfo(BaseModel): + dir_path: Path # Full client-side path to OTF directory + visit_path: str # Path fragment from rsync module to visit directory + destination_dir: str = "setup" # Folder in visit directory to save to + + +@router.post("/instruments/{instrument_name}/sessions/{session_id}/upload_otf_dir") +def upload_otf_dir( + instrument_name: str, + session_id: MurfeySessionID, + otf_dir_info: OTFDirectoryUploadInfo, +): + # Sanitise incoming values + otf_dir_path = sanitise(str(otf_dir_info.dir_path)) + visit_path = sanitise(otf_dir_info.visit_path) + destination_dir = sanitise(otf_dir_info.destination_dir) + + # Load machine config and other needed properties + machine_config_url_path = url_path_for( + "session_control.router", + "machine_info_by_instrument", + instrument_name=sanitise_nonpath(instrument_name), + ) + machine_config: dict[str, Any] = requests.get( + f"{_get_murfey_url()}{machine_config_url_path}", + headers={"Authorization": f"Bearer {tokens[session_id]}"}, + ).json() + + # Validate that file passed is from the gain reference directory + otf_dir = machine_config.get("gain_reference_directory", "") + if not otf_dir_path.startswith(otf_dir): + raise ValueError( + "Gain reference file does not originate from the gain reference directory " + f"{otf_dir!r}" + ) + + # Construct the rsync path to use + # Return the rsync URL if set, otherwise assume you are syncing via Murfey + rsync_url = urlparse( + str(machine_config["rsync_url"]) + if machine_config.get("rsync_url", "") + else _get_murfey_url() + ) + rsync_module = machine_config.get("rsync_module", "data") + rsync_path = f"{rsync_url.hostname}::{rsync_module}/{visit_path}/{destination_dir}" + + # Construct the expected destination path on the server side + rsync_basepath: str | None = machine_config.get("rsync_basepath") + if not rsync_basepath: + logger.error(f"No rsync base path was configured for {instrument_name}") + return {"success": False} + destination_path = ( + f"{rsync_basepath}/{visit_path}/{destination_dir}/" + f"{secure_filename(otf_dir_info.dir_path.name)}" + ) + + # Run rsync subprocess to transfer OTF directory and contents + cmd = [ + "rsync", + "-a", # Sync folder contents + ] + if rsync_chmod := machine_config.get("rsync_chmod"): + cmd.append(f"--chmod={rsync_chmod}") + cmd.extend([posix_path(Path(otf_dir_path)), rsync_path]) + process = subprocess.run( + cmd, + capture_output=True, + text=True, + ) + if process.returncode: + logger.warning( + f"Failed to transfer OTF directory {otf_dir_path!r} to {f'{destination_path}'!r} \n" + f"Executed the following command: {' '.join(cmd)!r} \n" + f"Returned the following error: \n" + f"{process.stderr}" + ) + return {"success": False} + + # Return the full path as part of the message + return { + "success": True, + "destination_path": str(destination_path), + } + + @router.get( "/instruments/{instrument_name}/sessions/{session_id}/possible_gain_references" ) diff --git a/src/murfey/server/api/instrument.py b/src/murfey/server/api/instrument.py index 7f97423a8..75c95e53e 100644 --- a/src/murfey/server/api/instrument.py +++ b/src/murfey/server/api/instrument.py @@ -2,7 +2,9 @@ import asyncio import datetime +import json import logging +import os from pathlib import Path from typing import Annotated, Any, List, Optional from urllib.parse import quote @@ -408,6 +410,89 @@ async def get_possible_otf_dirs( return data +class OTFDirectoryUploadRequest(BaseModel): + dir_path: Path + + +@router.post("/sessions/{session_id}/upload_otf_dir") +async def request_otf_dir_upload( + session_id: MurfeySessionID, + otf_dir_request: OTFDirectoryUploadRequest, + db=murfey_db, +): + # Load session information from database + murfey_session = db.exec(select(Session).where(Session.id == session_id)).one() + visit_name = murfey_session.visit + instrument_name = murfey_session.instrument_name + + # Load machine config + machine_config = get_machine_config(instrument_name=instrument_name)[ + instrument_name + ] + + # Default data to return + data: dict[str, Any] = {"success": False} + + # Load the rsync basepath + rsync_basepath = machine_config.rsync_basepath + if rsync_basepath is None: + log.error(f"No rsync basepath was configured for instrument {instrument_name}") + return data + + # Construct the partial and full paths to the server-side visit directory + visit_path = f"{datetime.datetime.now().year}/{visit_name}" + visit_dir = rsync_basepath / visit_path + if not visit_dir.exists(): # Check previous year in case of rollover + visit_dir_prev = visit_dir + visit_path = f"{datetime.datetime.now().year - 1}/{visit_name}" + visit_dir = rsync_basepath / visit_path + if not visit_dir.exists(): + log.error( + "Unable to find visit directory under " + f"{str(visit_dir_prev)} or {str(visit_dir)}" + ) + return data + + # Ensure that the OTF destination directory exists + otf_dir = visit_dir / machine_config.gain_directory_name + otf_dir.mkdir(exist_ok=True) + os.chmod(otf_dir, mode=machine_config.mkdir_chmod) # Set permissions + + if machine_config.instrument_server_url: + async with aiohttp.ClientSession() as clientsession: + url_path = url_path_for( + "api.router", + "upload_otf_dir", + instrument_name=instrument_name, + session_id=session_id, + ) + payload = { + "dir_path": str(otf_dir_request.dir_path), + "visit_path": visit_path, + "destination_dir": machine_config.gain_directory_name, + } + + # REMOVE AFTER TESTING + log.info( + "Submitting the following payload to instrument server: \n" + f"{json.dumps(payload, indent=2, default=str)}" + ) + + async with clientsession.post( + f"{machine_config.instrument_server_url}{url_path}", + json=payload, + headers={ + "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}" + }, + ) as resp: + data = await resp.json() + else: + log.error( + f"No instrument server URL was configured for instrument {instrument_name}" + ) + return data + + @router.get( "/instruments/{instrument_name}/sessions/{session_id}/possible_gain_references" ) diff --git a/src/murfey/util/route_manifest.yaml b/src/murfey/util/route_manifest.yaml index 878772a8d..7c319c344 100644 --- a/src/murfey/util/route_manifest.yaml +++ b/src/murfey/util/route_manifest.yaml @@ -139,6 +139,15 @@ murfey.instrument_server.api.router: type: int methods: - GET + - path: /instruments/{instrument_name}/sessions/{session_id}/upload_otf_dir + function: upload_otf_dir + path_params: + - name: instrument_name + type: str + - name: session_id + type: int + methods: + - POST - path: /instruments/{instrument_name}/sessions/{session_id}/possible_gain_references function: get_possible_gain_references path_params: @@ -561,6 +570,13 @@ murfey.server.api.instrument.router: type: int methods: - GET + - path: /instrument_server/sessions/{session_id}/upload_otf_dir + function: request_otf_dir_upload + path_params: + - name: session_id + type: int + methods: + - POST - path: /instrument_server/instruments/{instrument_name}/sessions/{session_id}/possible_gain_references function: get_possible_gain_references path_params: From 26fa2b98be5ca0f3b8a091fbde543e88f7a1666f Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 2 Sep 2026 19:14:31 +0100 Subject: [PATCH 2/4] Added test for the OTF directory rsync endpoint --- tests/instrument_server/test_api.py | 112 ++++++++++++++++++++++++++++ 1 file changed, 112 insertions(+) diff --git a/tests/instrument_server/test_api.py b/tests/instrument_server/test_api.py index f15d89ef8..a1cadb967 100644 --- a/tests/instrument_server/test_api.py +++ b/tests/instrument_server/test_api.py @@ -170,6 +170,118 @@ def test_get_possible_otf_dirs( ) +@pytest.mark.parametrize( + "test_params", + ( + # Rsync URL settings + ("http://1.1.1.1",), # When rsync_url is provided + ("",), # When rsync_url is blank + (None,), # When rsync_url not provided + ), +) +def test_upload_otf_dir( + mocker: MockerFixture, + test_params: tuple[Optional[str]], + tmp_path: Path, +): + # Unpack test parameters and define other ones + (rsync_url_setting,) = test_params + server_url = "https://murfey.server.test" + instrument_name = "murfey" + session_id = 1 + + # Mock out objects + mock_request = mocker.patch("murfey.instrument_server.api.requests") + mock_get_server_url = mocker.patch("murfey.instrument_server.api._get_murfey_url") + mock_subprocess = mocker.patch("murfey.instrument_server.api.subprocess") + mocker.patch("murfey.instrument_server.api.tokens", {session_id: ANY}) + + # Create a mock machine config base on the test params + rsync_module = "data" + rsync_basepath = tmp_path / "data" + otf_dir = "C:/ProgramData/SIM/OTFs" + mock_machine_config = { + "rsync_module": rsync_module, + "rsync_basepath": str(rsync_basepath), + "gain_reference_directory": otf_dir, + } + if rsync_url_setting is not None: + mock_machine_config["rsync_url"] = rsync_url_setting + + # Assign expected values to the mock objects + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = mock_machine_config + mock_request.get.return_value = mock_response + mock_get_server_url.return_value = server_url + mock_subprocess.run.return_value = MagicMock(returncode=0) + + # Construct payload and pass request to function + otf_folder_name = "OTFs-123456" + otf_dir_path = f"{otf_dir}/{otf_folder_name}" + visit_path = "2025/aa00000-0" + destination_dir = "setup" + payload = { + "dir_path": otf_dir_path, + "visit_path": visit_path, + "destination_dir": destination_dir, + } + + # Set up instrument server test client + client_server = set_up_test_client(session_id=session_id) + + # Poke the endpoint with the expected data + url_path = url_path_for( + "api.router", + "upload_otf_dir", + instrument_name=instrument_name, + session_id=session_id, + ) + response = client_server.post(url_path, json=payload) + + # Check that the machine config request was called + machine_config_url = url_path_for( + "session_control.router", + "machine_info_by_instrument", + instrument_name=instrument_name, + ) + mock_request.get.assert_called_once_with( + f"{server_url}{machine_config_url}", + headers={"Authorization": ANY}, + ) + + # Check that the subprocess was run with the expected arguments + # If no rsync_url key is provided, or rsync_url key is empty, + # It should default to the server URL + expected_rsync_url = ( + urlparse(server_url) if not rsync_url_setting else urlparse(rsync_url_setting) + ) + expected_rsync_path = ( + f"{expected_rsync_url.hostname}::{rsync_module}/{visit_path}/{destination_dir}" + ) + expected_rsync_cmd = [ + "rsync", + "-a", + posix_path(Path(otf_dir_path)), + expected_rsync_path, + ] + expected_destination_path = ( + rsync_basepath / visit_path / destination_dir / otf_folder_name + ) + + mock_subprocess.run.assert_called_once_with( + expected_rsync_cmd, + capture_output=True, + text=True, + ) + + # Check that the function ran through to completion successfully + assert response.json() == { + "success": True, + "destination_path": str(expected_destination_path), + } + + @pytest.mark.parametrize( "test_params", ( From 5ca95526144f48e01b75f65f4483f8788cec549e Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Thu, 3 Sep 2026 09:52:33 +0100 Subject: [PATCH 3/4] Added test for 'request_otf_dir_upload' POST endpoint --- tests/server/api/test_instrument.py | 120 +++++++++++++++++++++++++--- 1 file changed, 111 insertions(+), 9 deletions(-) diff --git a/tests/server/api/test_instrument.py b/tests/server/api/test_instrument.py index 6850768f8..f0d94ec22 100644 --- a/tests/server/api/test_instrument.py +++ b/tests/server/api/test_instrument.py @@ -1,3 +1,5 @@ +from datetime import datetime +from pathlib import Path from typing import Callable, Literal from unittest import mock from unittest.mock import AsyncMock, MagicMock @@ -207,20 +209,12 @@ def test_get_possible_otf_dirs( instrument_name=instrument_name, session_id=session_id, ) - assert ( - backend_url_path - == f"/instrument_server/instruments/{instrument_name}/sessions/{session_id}/possible_otf_dirs" - ) client_url_path = url_path_for( "api.router", "get_possible_otf_dirs", instrument_name=instrument_name, session_id=session_id, ) - assert ( - client_url_path - == f"/instruments/{instrument_name}/sessions/{session_id}/possible_otf_dirs" - ) # Poke the backend response = backend_server.get(backend_url_path) @@ -231,4 +225,112 @@ def test_get_possible_otf_dirs( ) assert response.status_code == 200 assert response.json() == json_data - pass + + +def test_request_otf_dir_upload( + mocker: MockerFixture, + tmp_path: Path, +): + # Set reusable variables here + instrument_name = "sim" + session_id = 1 + visit_name = "cm12345-6" + + instrument_server_url = "https://murfey.instrument-server.test" + access_token = "dummy" + rsync_basepath = tmp_path / "data" + otf_dir_name = "setup" + + current_year = datetime.now().year + + # Create the visit directory + visit_dir = rsync_basepath / str(current_year) / visit_name + visit_dir.mkdir(parents=True, exist_ok=True) + visit_path = visit_dir.relative_to(rsync_basepath) + + # Create the client-side OTF directory to transfer + otf_dir_client = tmp_path / "client" / "otfs" / "OTFs-123456" + otf_dir_client.mkdir(parents=True, exist_ok=True) + + # Mock the machine config + mock_machine_config = MachineConfig( + rsync_basepath=rsync_basepath, + gain_directory_name=otf_dir_name, + instrument_server_url=instrument_server_url, + ) + mock_get_machine_config = mocker.patch( + "murfey.server.api.instrument.get_machine_config", + return_value={instrument_name: mock_machine_config}, + ) + + # Mock the instrument server access token + mocker.patch( + "murfey.server.api.instrument.instrument_server_tokens", + {session_id: {"access_token": access_token}}, + ) + + # Override the database session generator + mock_session = MagicMock(instrument_name=instrument_name, visit=visit_name) + mock_query_result = MagicMock() + mock_query_result.one.return_value = mock_session + mock_db_session = MagicMock() + mock_db_session.exec.return_value = mock_query_result + + def mock_get_db_session(): + yield mock_db_session + + # Mock the client session the API is requesting from + json_data = { + "success": True, + "destination_path": str(visit_dir / "setup" / otf_dir_client.name), + } + mock_client_session, _ = mock_aiohttp_clientsession( + mocker, + method="post", + json_data=json_data, + ) + + # Set up the backend server + backend_server = set_up_test_backend_client( + session_id=session_id, + instrument_name=instrument_name, + mock_db_session=mock_get_db_session, + ) + + # Construct the URL paths for poking and sending to + backend_url_path = url_path_for( + "api.instrument.router", + "request_otf_dir_upload", + session_id=session_id, + ) + client_url_path = url_path_for( + "api.router", + "upload_otf_dir", + instrument_name=instrument_name, + session_id=session_id, + ) + + # Poke the backend + response = backend_server.post( + backend_url_path, + json={"dir_path": str(otf_dir_client)}, + ) + + # Check that 'get_machine_config' was called + mock_get_machine_config.assert_called_once() + + # Check that request was sent to instrument server with expected calls + payload = { + "dir_path": str(otf_dir_client), + "visit_path": str(visit_path), + "destination_dir": otf_dir_name, + } + mock_client_session.post.assert_called_once_with( + f"{instrument_server_url}{client_url_path}", + json=payload, + headers={"Authorization": f"Bearer {access_token}"}, + ) + + # Check that the status code and returned data are correct + assert response.status_code == 200 + assert response.json() == json_data From 400a350de87904a4369d4ec34828942a14c8cefc Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Thu, 3 Sep 2026 10:00:17 +0100 Subject: [PATCH 4/4] Check that the server-side OTF save directory was created --- tests/server/api/test_instrument.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/server/api/test_instrument.py b/tests/server/api/test_instrument.py index f0d94ec22..8fd516a73 100644 --- a/tests/server/api/test_instrument.py +++ b/tests/server/api/test_instrument.py @@ -258,7 +258,7 @@ def test_request_otf_dir_upload( gain_directory_name=otf_dir_name, instrument_server_url=instrument_server_url, ) - mock_get_machine_config = mocker.patch( + mocker.patch( "murfey.server.api.instrument.get_machine_config", return_value={instrument_name: mock_machine_config}, ) @@ -282,7 +282,7 @@ def mock_get_db_session(): # Mock the client session the API is requesting from json_data = { "success": True, - "destination_path": str(visit_dir / "setup" / otf_dir_client.name), + "destination_path": str(visit_dir / otf_dir_name / otf_dir_client.name), } mock_client_session, _ = mock_aiohttp_clientsession( mocker, @@ -316,8 +316,8 @@ def mock_get_db_session(): json={"dir_path": str(otf_dir_client)}, ) - # Check that 'get_machine_config' was called - mock_get_machine_config.assert_called_once() + # Check that the server-side OTF directory save location was created + assert (visit_dir / otf_dir_name).exists() # Check that request was sent to instrument server with expected calls payload = {