canonical-ubuntu-qa team mailing list archive

[Tim Andersson <mp+460847@xxxxxxxxxxxxxxxxxx>]

Amended based on all your comments and ready for re-review :)

Diff comments:

> diff --git a/charms/focal/autopkgtest-web/webcontrol/sqlite-writer b/charms/focal/autopkgtest-web/webcontrol/sqlite-writer
> new file mode 100755
> index 0000000..4ba8bdf
> --- /dev/null
> +++ b/charms/focal/autopkgtest-web/webcontrol/sqlite-writer
> @@ -0,0 +1,138 @@
> +#!/usr/bin/python3
> +
> +import configparser
> +import json
> +import logging
> +import os
> +import socket
> +import sqlite3
> +import urllib.parse
> +
> +import amqplib.client_0_8 as amqp
> +from helpers.utils import init_db
> +
> +EXCHANGE_NAME = "sqlite-write-me.fanout"
> +
> +config = None
> +db_con = None
> +
> +INSERT_INTO_KEYS = [
> +    "test_id",
> +    "run_id",
> +    "version",
> +    "triggers",
> +    "duration",
> +    "exitcode",
> +    "requester",
> +    "env",
> +    "uuid",
> +]
> +
> +
> +def amqp_connect():
> +    """Connect to AMQP server"""
> +
> +    cp = configparser.ConfigParser()
> +    cp.read(os.path.expanduser("~ubuntu/autopkgtest-cloud.conf"))
> +    amqp_uri = cp["amqp"]["uri"]
> +    parts = urllib.parse.urlsplit(amqp_uri, allow_fragments=False)
> +    amqp_con = amqp.Connection(
> +        parts.hostname, userid=parts.username, password=parts.password
> +    )
> +    logging.info(
> +        "Connected to AMQP server at %s@%s" % (parts.username, parts.hostname)
> +    )
> +
> +    return amqp_con
> +
> +
> +def db_connect():
> +    """Connect to SQLite DB"""
> +    cp = configparser.ConfigParser()
> +    cp.read(os.path.expanduser("~ubuntu/autopkgtest-cloud.conf"))
> +
> +    db_con = init_db(cp["web"]["database"])
> +
> +    return db_con
> +
> +
> +def check_msg(queue_msg):
> +    required_keys = set(

oh doi, I added INSERT_INTO_KEYS after this and forgot to refactor :D

> +        [
> +            "test_id",
> +            "run_id",
> +            "version",
> +            "triggers",
> +            "duration",
> +            "exitcode",
> +            "requester",
> +            "env",
> +            "uuid",
> +        ]
> +    )
> +    queue_keys = set(queue_msg.keys())
> +    if required_keys == queue_keys:
> +        return True
> +    return False
> +
> +
> +def process_message(msg, db_con):
> +    # We want to ack and re-send messages if insert fails?
> +    body = msg.body
> +    if isinstance(body, bytes):
> +        body = body.decode("UTF-8", errors="replace")
> +    info = json.loads(body)
> +    logging.info("Message is: \n%s" % json.dumps(info, indent=2))
> +    if not check_msg(info):
> +        logging.error(
> +            "Message has incorrect keys! Ignoring\n%s"
> +            % json.dumps(info, indent=2)
> +        )
> +        msg.channel.basic_ack(msg.delivery_tag)
> +        return
> +    # insert into db
> +    sqlite3.paramstyle = "named"
> +    with db_con:
> +        c = db_con.cursor()
> +        # change this to column names
> +        c.execute(
> +            (
> +                "INSERT INTO result(test_id, run_id, version, triggers, duration, "
> +                "exitcode, requester, env, uuid) VALUES (:test_id, :run_id, "
> +                ":version, :triggers, :duration, :exitcode, :requester, :env, :uuid)"
> +            ),
> +            {
> +                "test_id": info["test_id"],
> +                "run_id": info["run_id"],
> +                "version": info["version"],
> +                "triggers": info["triggers"],
> +                "duration": info["duration"],
> +                "exitcode": info["exitcode"],
> +                "requester": info["requester"],
> +                "env": info["env"],
> +                "uuid": info["uuid"],
> +            },
> +        )
> +    logging.info("Inserted the following entry into the db:\n%s" % body)
> +
> +    msg.channel.basic_ack(msg.delivery_tag)
> +
> +
> +if __name__ == "__main__":
> +    logging.basicConfig(level=logging.INFO)
> +    db_con = db_connect()
> +    amqp_con = amqp_connect()
> +    status_ch = amqp_con.channel()
> +    status_ch.access_request("/complete", active=True, read=True, write=False)
> +    status_ch.exchange_declare(
> +        EXCHANGE_NAME, "fanout", durable=True, auto_delete=False
> +    )
> +    queue_name = "sqlite-writer-listener-%s" % socket.getfqdn()
> +    status_ch.queue_declare(queue_name, durable=True, auto_delete=False)
> +    status_ch.queue_bind(queue_name, EXCHANGE_NAME, queue_name)
> +    logging.info("Listening to requests on %s" % queue_name)
> +    status_ch.basic_consume(
> +        "", callback=lambda msg: process_message(msg, db_con)
> +    )
> +    while status_ch.callbacks:
> +        status_ch.wait()


-- 
https://code.launchpad.net/~andersson123/autopkgtest-cloud/+git/autopkgtest-cloud/+merge/460847
Your team Canonical's Ubuntu QA is requested to review the proposed merge of ~andersson123/autopkgtest-cloud:d-r-d-a-r-merging into autopkgtest-cloud:master.