Add test/mock config for MQ

Dockerfile

@@ -6,6 +6,8 @@ RUN pip install -r /requirements.txt
 COPY Collectors /app
 WORKDIR /app
 
+COPY tests /tests
+
 EXPOSE 9930/udp
 EXPOSE 8000/tcp
 CMD [ "/app/DetailedCollector.py", "/configs/connection.conf" ]

docker-compose.yml

@@ -17,3 +17,8 @@ services:
 #    ports:
 #     - "9301:9301/udp"
 #    restart: always
+  mock-mq:
+    # https://github.com/tsv1/amqp-mock
+    image: tsv1/amqp-mock
+    profiles:
+      - mock

tests/mqmock.md

@@ -0,0 +1,31 @@
+
+Retrieving a set of records from RabbitMQ
+```
+curl 'https://xrd-rabbitmq.gracc-prod.chtc.io/api/queues/xrd-mon/xrd.push.external/get' \
+  -X POST \
+  --user user:hunter2 \
+  --data-raw '{"vhost":"xrd-mon","name":"xrd.push.external","truncate":"50000","ackmode":"ack_requeue_true","encoding":"auto","count":"1000"}' \
+  --output records.out
+```
+
+Pull the payload fields and make into a JSON Lines file
+```
+jq --raw '.[].payload' records.out > test1.jsonl
+```
+
+Configure collector to use the mock MQ
+```
+[AMQP]
+url = amqp://user:password@mock-mq
+```
+
+Start the services, including the mock MQ container
+```
+docker compose --profile mock up
+```
+
+Exec into the collector container and send test data to the mock MQ
+```
+docker compose exec -it detailed_collector /bin/sh
+/tests/mqmock.py /tests/test1.jsonl
+```

tests/mqmock.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python
+
+import argparse
+import json
+import sys
+import time
+import requests
+
+QUEUE_IN = "xrd.push"
+EXCHANGES_OUT = [
+    "xrd.detailed",
+    "xrd.wlcg_format",
+    "xrd.tcp_exchange",
+    "xrd.cache.stash",
+    "xrd.cache.wlcg",
+    "xrd.tpc",
+    "xrd.tpc.wlcg",
+]
+
+
+def main():
+    args = parse_args()
+
+    # Send input records into mock MQ
+    rec_count = 0
+    for line in args.input:
+        rec_count += 1
+        data = json.loads(line)
+
+        msg = {
+            "value": data,
+            "exchange": "amq.default",
+            "routing_key": QUEUE_IN,
+            "properties": None,
+        }
+
+        resp = requests.post(f"{args.mock_mq_url}/queues/{QUEUE_IN}/messages", json=msg)
+        resp.raise_for_status()
+
+    print(f"Sent {rec_count} records to mock MQ", file=sys.stderr)
+    print(f"Waiting {args.wait}s for processing", file=sys.stderr)
+    time.sleep(args.wait)
+
+    # Retrieve records
+    for ex in sorted(EXCHANGES_OUT):
+        print(f"Getting messages from exchange {ex}", file=sys.stderr)
+        resp = requests.get(f"{args.mock_mq_url}/exchanges/{ex}/messages")
+        resp.raise_for_status()
+
+        for rec in resp.json():
+            # Remove random value
+            del rec["id"]
+
+            # Remove WLCG non-deterministic values
+            if rec["exchange"] == "xrd.wlcg_format":
+                del rec["value"]["metadata"]["_id"]
+                del rec["value"]["metadata"]["timestamp"]
+                del rec["value"]["unique_id"]
+
+            print(json.dumps(rec, sort_keys=True))
+
+        # Delete messages from exchange
+        resp = requests.delete(f"{args.mock_mq_url}/exchanges/{ex}/messages")
+        resp.raise_for_status()
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--mock-mq-url", default="http://mock-mq")
+    parser.add_argument("--wait", default=5)
+    parser.add_argument(
+        "input", type=argparse.FileType("r"), help="records in JSONL format"
+    )
+
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    main()