canonical-ubuntu-qa team mailing list archive

[Tim Andersson <mp+463208@xxxxxxxxxxxxxxxxxx>]

Tim Andersson has proposed merging ~andersson123/autopkgtest-cloud:pull-amqp-push-amqp into autopkgtest-cloud:master.

Requested reviews:
  Canonical's Ubuntu QA (canonical-ubuntu-qa)

For more details, see:
https://code.launchpad.net/~andersson123/autopkgtest-cloud/+git/autopkgtest-cloud/+merge/463208
-- 
Your team Canonical's Ubuntu QA is requested to review the proposed merge of ~andersson123/autopkgtest-cloud:pull-amqp-push-amqp into autopkgtest-cloud:master.

diff --git a/charms/focal/autopkgtest-cloud-worker/autopkgtest-cloud/tools/pull-amqp b/charms/focal/autopkgtest-cloud-worker/autopkgtest-cloud/tools/pull-amqp
new file mode 100755
index 0000000..949fa6d
--- /dev/null
+++ b/charms/focal/autopkgtest-cloud-worker/autopkgtest-cloud/tools/pull-amqp
@@ -0,0 +1,67 @@
+#!/usr/bin/python3
+
+import amqplib.client_0_8 as amqp
+import configparser
+import re
+import argparse
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--regex",
+        "-r",
+        default=None,
+        help="filter queue message by this regex"
+    )
+    parser.add_argument(
+        "--queue-name",
+        "-q",
+        required=True,
+        help="Queue name"
+    )
+    parser.add_argument(
+        "--empty",
+        "-e",
+        action="store_true",
+        help="If this flag is passed, all queue messages will be emptied during the script",
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = parse_args()
+    cp = configparser.ConfigParser()
+    with open("/home/ubuntu/rabbitmq.cred", "r") as f:
+        cp.read_string("[rabbit]\n" + f.read().replace('"', ""))
+    amqp_con = amqp.Connection(
+        cp["rabbit"]["RABBIT_HOST"],
+        cp["rabbit"]["RABBIT_USER"],
+        cp["rabbit"]["RABBIT_PASSWORD"],
+    )
+    if args.regex is not None:
+        filter_re = re.compile(args.regex.encode("UTF-8"), re.DOTALL)
+    items = []
+    with amqp_con.channel() as ch:
+        while True:
+            r = ch.basic_get(args.queue_name)
+            if r is None:
+                break
+            body = r.body.encode("UTF-8") if isinstance(r.body, str) else r.body
+            if args.regex is not None:
+                if filter_re.search(body):
+                    items.append(body)
+                    if args.empty:
+                        ch.basic_ack(r.delivery_tag)
+            else:
+                items.append(body)
+                if args.empty:
+                    ch.basic_ack(r.delivery_tag)
+    for item in items:
+        print(item)
+
+
+
+if __name__ == "__main__":
+    main()
diff --git a/charms/focal/autopkgtest-cloud-worker/autopkgtest-cloud/tools/push-amqp b/charms/focal/autopkgtest-cloud-worker/autopkgtest-cloud/tools/push-amqp
new file mode 100755
index 0000000..c112277
--- /dev/null
+++ b/charms/focal/autopkgtest-cloud-worker/autopkgtest-cloud/tools/push-amqp
@@ -0,0 +1,51 @@
+#!/usr/bin/python3
+
+import amqplib.client_0_8 as amqp
+import configparser
+import argparse
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--dry-run",
+        "-d",
+        action="store_true",
+        help="Dry run",
+    )
+    parser.add_argument(
+        "--queue-name",
+        "-q",
+        required=True,
+        help="Queue name"
+    )
+    parser.add_argument(
+        "--message",
+        "-m",
+        required=True,
+        help="amqp message to publish",
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = parse_args()
+    cp = configparser.ConfigParser()
+    with open("/home/ubuntu/rabbitmq.cred", "r") as f:
+        cp.read_string("[rabbit]\n" + f.read().replace('"', ""))
+    amqp_con = amqp.Connection(
+        cp["rabbit"]["RABBIT_HOST"],
+        cp["rabbit"]["RABBIT_USER"],
+        cp["rabbit"]["RABBIT_PASSWORD"],
+    )
+    if not args.dry_run:
+        with amqp_con.channel() as ch:
+            ch.basic_publish(
+                amqp.Message(
+                    args.message, delivery_mode=2
+                ),
+                routing_key=args.queue_name,
+            )
+    else:
+        print(f"Would publish {args.message} to {args.queue_name}")