Update postprocessing and add fake test server for ONCat

Author: rosswhitfield

Dockerfile.autoreducer

@@ -25,8 +25,12 @@ RUN ls ${DATA_TARBALL}
 RUN mkdir /SNS
 RUN cd /SNS && tar xzf ${DATA_TARBALL}
 
+# install pyoncat
+COPY tests/python3-pyoncat-2.1-1.noarch.rpm /tmp/
+RUN dnf install -y /tmp/python3-pyoncat-2.1-1.noarch.rpm
+
 # install postprocessing
-RUN dnf install -y https://github.com/neutrons/post_processing_agent/releases/download/v4.0.0/postprocessing-4.0.0-1.el9.noarch.rpm
+RUN dnf install -y https://github.com/neutrons/post_processing_agent/releases/download/v4.1.0/postprocessing-4.1.0-1.el9.noarch.rpm
 
 # Install dependencies
 ARG PLOTLY_VERSION=5
@@ -40,17 +44,13 @@ RUN pip3 install git+https://github.com/neutrons/plot_publisher.git@main
 
 # Install specific plotly version based on build arg
 RUN if [ "$PLOTLY_VERSION" = "5" ]; then \
-        pip3 install plotly==5.17.0; \
+    pip3 install plotly==5.17.0; \
     elif [ "$PLOTLY_VERSION" = "6" ]; then \
-        pip3 install plotly==6.0.0; \
+    pip3 install plotly==6.0.0; \
     else \
-        pip3 install plotly==5.17.0; \
+    pip3 install plotly==5.17.0; \
     fi
 
-# add fake ONCat ingest scripts
-RUN touch /opt/postprocessing/scripts/oncat_ingest.py && \
-    touch /opt/postprocessing/scripts/oncat_reduced_ingest.py
-
 # create startup script
 # add updating certificates to add self-signed certificate for local testing
 RUN echo "#!/bin/bash" > /usr/bin/run_postprocessing && \

Dockerfile.oncat

@@ -0,0 +1,5 @@
+FROM python:3-alpine
+COPY tests/oncat_server.py .
+RUN mkdir -p /SNS/ARCS/IPTS-27800/nexus && \
+    touch /SNS/ARCS/IPTS-27800/nexus/ARCS_214583.nxs.h5
+CMD ["python", "oncat_server.py"]

docker-compose.yml

@@ -224,6 +224,11 @@ services:
       activemq:
         condition: service_healthy
 
+  oncat:
+    build:
+      context: .
+      dockerfile: Dockerfile.oncat
+
   autoheal:
     restart: always
     image: willfarrell/autoheal

tests/configuration/post_process_consumer.conf

@@ -30,5 +30,7 @@
     "publisher_password": "workflow",
     "system_mem_limit_perc": 5,
     "mem_check_interval_sec": 0.01,
-    "task_time_limit_minutes": 0.1
+    "task_time_limit_minutes": 0.1,
+    "oncat_url": "http://oncat:8000",
+    "oncat_api_token": "test-token"
 }

tests/oncat_server.py

@@ -0,0 +1,80 @@
+import json
+import logging
+import os
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+logging.basicConfig(filename="oncat_server.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+
+
+class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):
+    """Fake ONCat server for testing purposes"""
+
+    def do_POST(self):
+        if self.headers.get("Authorization") != "Bearer test-token":
+            self.send_response(403)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            response = {"error": "Forbidden"}
+            self.wfile.write(json.dumps(response).encode("utf-8"))
+            return
+
+        if self.path.startswith("/api/datafiles/"):
+            location = self.path.replace("/api/datafiles", "").replace("/ingest", "")
+            logging.info("Received datafile ingest request for %s", location)
+        elif self.path.startswith("/api/reductions/"):
+            location = self.path.replace("/api/reductions", "").replace("/ingest", "")
+            logging.info("Received reduction ingest request for %s", location)
+        else:
+            self.send_response(400)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            logging.error("Unknown endpoint: %s", self.path)
+            response = {"error": "Unknown endpoint"}
+            self.wfile.write(json.dumps(response).encode("utf-8"))
+            return
+
+        if not os.path.isfile(location):
+            self.send_response(404)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            logging.error("File not found: %s", location)
+            response = {"error": "File not found"}
+            self.wfile.write(json.dumps(response).encode("utf-8"))
+            return
+
+        try:
+            # assume the format /facility/instrument/experiment/nexus/instrument_run_number.nxs.h5
+            # the real server would be more complex
+            _, facility, instrument, experiment, *_, filename = location.split("/")
+            run_number = int(filename.split("_")[-1].split(".")[0])
+        except Exception:
+            self.send_response(400)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            logging.error("Invalid path format: %s", location)
+            response = {"error": "Invalid path format"}
+            self.wfile.write(json.dumps(response).encode("utf-8"))
+            return
+
+        # Send response
+        self.send_response(200)
+        self.send_header("Content-type", "application/json")
+        self.end_headers()
+        response = {
+            "location": location,
+            "facility": facility,
+            "instrument": instrument,
+            "experiment": experiment,
+            "indexed": {"run_number": run_number},
+        }
+        self.wfile.write(json.dumps(response).encode("utf-8"))
+
+
+def run(server_class=HTTPServer, handler_class=SimpleHTTPRequestHandler, port=8000):
+    server_address = ("", port)
+    httpd = server_class(server_address, handler_class)
+    httpd.serve_forever()
+
+
+if __name__ == "__main__":
+    run()