1 files changed, 145 insertions, 0 deletions
diff --git a/subplot/subplot.py b/subplot/subplot.py
new file mode 100644
index 0000000..08472bd
--- /dev/null
+++ b/subplot/subplot.py
@@ -0,0 +1,145 @@
+import logging
+import os
+import yaml
+
+
+def fixme(ctx, **kwargs):
+    assert 0
+
+
+def create_vm(ctx):
+    QemuSystem = globals()["QemuSystem"]
+    srcdir = globals()["srcdir"]
+
+    MiB = 1024 ** 2
+    GiB = 1024 ** 3
+
+    cfg = yaml.safe_load(open(os.path.join(srcdir, "test.cfg")))
+    name = cfg["name"]
+    base_image = cfg["base_image"]
+    username = cfg["username"]
+    cpus = cfg["cpus"]
+    memory = cfg["memory"] * MiB
+
+    # We use a hard-coded test key that we have in the source tree.
+    pubkey = open(os.path.join(srcdir, "ssh", "id.pub")).read().strip()
+
+    # We hard code the disk image size, since we don't expect scenarios to have
+    # any specific size needs, and also, the qcow2 image format is only as big
+    # as the data put into the disk, so we can choose a size that fits all.
+    disk_size = 10 * GiB
+
+    logging.info("starting a VM using qemu-system")
+    logging.info(f"  name    : {name}")
+    logging.info(f"  image   : {base_image}")
+    logging.info(f"  disk    : {disk_size}")
+    logging.info(f"  pubkey  : {pubkey}")
+    logging.info(f"  memory  : {memory}")
+    logging.info(f"  cpus    : {cpus}")
+    logging.info(f"  username: {username}")
+
+    qemu = QemuSystem(name, base_image, disk_size, pubkey)
+    qemu.set_memory(memory)
+    qemu.set_vcpus(cpus)
+    qemu.set_username(username)
+    qemu.start()
+
+    logging.debug("waiting for SSH to be ready")
+    if qemu.wait_for_ssh():
+        logging.debug("SSH is ready")
+    else:
+        logging.error("SSH did not get ready")
+        assert 0
+    logging.info("a qemu-system VM is up and running and accessible over SSH")
+    ctx["qemu"] = qemu
+
+
+def destroy_vm(ctx):
+    logging.debug(f"destroying qemu running")
+    qemu = ctx["qemu"]
+    qemu.stop()
+
+
+def run_true_on_host(ctx):
+    qemu = ctx["qemu"]
+    qemu.ssh(["/bin/true"])
+
+
+def use_role_in_playbook(ctx, role=None):
+    empty_playbook = {
+        "hosts": "test-host",
+        "remote_user": "debian",  # FIXME: don't hardcode this
+        "become": True,
+        "roles": [],
+    }
+    playbook = ctx.get("playbook", dict(empty_playbook))
+    playbook["roles"].append(role)
+    ctx["playbook"] = playbook
+
+
+def set_vars_file(ctx, filename=None):
+    get_file = globals()["get_file"]
+    data = get_file(filename)
+    with open("vars.yml", "wb") as f:
+        f.write(data)
+
+
+def try_playbook(ctx):
+    runcmd_run = globals()["runcmd_run"]
+    assert_ne = globals()["assert_ne"]
+    srcdir = globals()["srcdir"]
+
+    qemu = ctx["qemu"]
+
+    with open("hosts", "w") as f:
+        f.write("test-host\n")
+
+    if not os.path.exists("vars.yml"):
+        with open("vars.yml", "w") as f:
+            yaml.safe_dump({}, stream=f)
+
+    playbook = [ctx["playbook"]]
+    assert_ne(playbook, None)
+    with open("playbook.yml", "w") as f:
+        yaml.safe_dump(playbook, stream=f)
+
+    ssh_opts = [
+        "-ouserknownhostsfile=/dev/null",
+        "-ostricthostkeychecking=accept-new",
+        "-i",
+        os.path.join(srcdir, "ssh", "id"),
+        f"-p{qemu.get_port()}",
+    ]
+
+    env = dict(os.environ)
+    env["ANSIBLE_SSH_ARGS"] = " ".join(ssh_opts)
+    env["ANSIBLE_LOG"] = "ansible.log"
+    env["ANSIBLE_ROLES_PATH"] = os.path.join(srcdir, "roles")
+
+    argv = [
+        "ansible-playbook",
+        "-i",
+        "hosts",
+        f"-e@vars.yml",
+        "-eansible_ssh_host=localhost",
+        f"-eansible_ssh_port={qemu.get_port()}",
+        "playbook.yml",
+    ]
+
+    runcmd_run(ctx, argv, env=env)
+
+
+def command_fails(ctx):
+    runcmd_exit_code_is_nonzero = globals()["runcmd_exit_code_is_nonzero"]
+    runcmd_exit_code_is_nonzero(ctx)
+
+
+def run_playbook(ctx):
+    runcmd_exit_code_is_zero = globals()["runcmd_exit_code_is_zero"]
+    try_playbook(ctx)
+    runcmd_exit_code_is_zero(ctx)
+
+
+def xstdout_contains(ctx, text=None):
+    runcmd_stdout_contains = globals()["runcmd_stdout_contains"]
+    runcmd_stdout_contains(ctx, text=text)