Merge ~maas-committers/maas-ci/+git/system-tests:MAASENG-1717-Automated-Image-Testing-feature-branch into ~maas-committers/maas-ci/+git/system-tests:master

[Alexsander de Souza]

automated image testing

adds the capability of:
- building custom images using packer-maas
- testing the deployment of custom images

includes Temporal workflows to build, test and report the results

Co-authored-by: Jack Lloyd-Walters <email address hidden>

[Jack Lloyd-Walters (lloydwaltersj)]

+1 on the merge once all branches are in

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b MAASENG-1717-Automated-Image-Testing-feature-branch lp:~maas-committers/maas-ci/+git/system-tests into -b master lp:~maas-committers/maas-ci/+git/system-tests

STATUS: SUCCESS
COMMIT: db3f8c2f2a2aff73a2d7a3e2e8f89d1b8fcf11c6

review: Approve

02b5fbe... by Jack Lloyd-Walters on 2023-08-11

rebase changes and merge again

[MAAS Lander (maas-lander)]

UNIT TESTS
-b MAASENG-1717-Automated-Image-Testing-feature-branch lp:~maas-committers/maas-ci/+git/system-tests into -b master lp:~maas-committers/maas-ci/+git/system-tests

STATUS: SUCCESS
COMMIT: 02b5fbe61ed5bfe5bffa33775938b9af25486261

review: Approve

diff --git a/.gitignore b/.gitignore
index e71b819..3ed1283 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,16 +1,16 @@
-*.egg-info
-.vscode
+__pycache__
 .idea
-.tox
 .mypy_cache
+.tox
+.vscode
+*.egg-info
+base_config.yaml
+build-*.log
 build/
-__pycache__
 config.yaml
 credentials.yaml
-base_config.yaml
 image_mapping.yaml
+images/
 junit*.xml
 sosreport
 systemtests*.log
-images/
-build/
diff --git a/image_mapping.yaml.sample b/image_mapping.yaml.sample
index d23a1fc..72b2c35 100644
--- a/image_mapping.yaml.sample
+++ b/image_mapping.yaml.sample
@@ -5,7 +5,7 @@
 # An example of a mapping is:
 # images:
 #    $IMAGE_NAME:
-#        url: $IMAGE_URL
+#        filename: $IMAGE_FILENAME
 #        filetype: $IMAGE_FILETYPE
 #        architecture: $IMAGE_ARCH
 #        osystem: $IMAGE_OSYSTEM
@@ -17,7 +17,7 @@
 
 images:
   centos7:
-    url: centos7.tar.gz
+    filename: centos7.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: centos
@@ -25,7 +25,7 @@ images:
     packer_template: centos7
     ssh_username: centos
   centos8:
-    url: centos8.tar.gz
+    filename: centos8.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: centos
@@ -33,7 +33,7 @@ images:
     packer_template: centos8
     ssh_username: centos
   centos8-stream:
-    url: centos8-stream.tar.gz
+    filename: centos8-stream.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: centos
@@ -41,7 +41,7 @@ images:
     packer_template: centos8-stream
     ssh_username: centos
   rhel7:
-    url: rhel7.tar.gz
+    filename: rhel7.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: rhel
@@ -50,7 +50,7 @@ images:
     source_iso: rhel-server-7.9-x86_64-dvd.iso
     ssh_username: cloud-user
   rhel8:
-    url: rhel8.tar.gz
+    filename: rhel8.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: rhel
@@ -59,7 +59,7 @@ images:
     source_iso: rhel-8.6-x86_64-dvd.iso
     ssh_username: cloud-user
   rhel9:
-    url: rhel9.tar.gz
+    filename: rhel9.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: rhel
@@ -68,7 +68,7 @@ images:
     source_iso: rhel-baseos-9.1-x86_64-dvd.iso
     ssh_username: cloud-user
   rocky8:
-    url: rocky8.tar.gz
+    filename: rocky8.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: custom
@@ -77,7 +77,7 @@ images:
     base_image: "rhel/8"
     ssh_username: cloud-user
   rocky9:
-    url: rocky9.tar.gz
+    filename: rocky9.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: custom
@@ -86,7 +86,7 @@ images:
     base_image: "rhel/9"
     ssh_username: cloud-user
   sles12:
-    url: sles12.tar.gz
+    filename: sles12.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: suse
@@ -95,7 +95,7 @@ images:
     source_iso: SLES12-SP5-JeOS.x86_64-12.5-OpenStack-Cloud-GM.qcow2
     ssh_username: sles
   sles15:
-    url: sles15.tar.gz
+    filename: sles15.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: suse
@@ -104,7 +104,7 @@ images:
     source_iso: SLE-15-SP4-Full-x86_64-GM-Media1.iso
     ssh_username: sles
   esxi6:
-    url: vmware-esxi-6.dd.gz
+    filename: vmware-esxi-6.dd.gz
     filetype: ddgz
     architecture: amd64/generic
     osystem: esxi
@@ -113,7 +113,7 @@ images:
     source_iso: VMware-VMvisor-Installer-6.7.0.update03-14320388.x86_64.iso
     ssh_username: root
   esxi7:
-    url: vmware-esxi-7.dd.gz
+    filename: vmware-esxi-7.dd.gz
     filetype: ddgz
     architecture: amd64/generic
     osystem: esxi
@@ -122,7 +122,7 @@ images:
     source_iso: VMware-VMvisor-Installer-7.0U3g-20328353.x86_64.iso
     ssh_username: root
   esxi8:
-    url: vmware-esxi-8.dd.gz
+    filename: vmware-esxi-8.dd.gz
     filetype: ddgz
     architecture: amd64/generic
     osystem: esxi
@@ -131,7 +131,7 @@ images:
     source_iso: VMware-VMvisor-Installer-8.0b-21203435.x86_64.iso
     ssh_username: root
   ubuntu:
-    url: ubuntu-cloudimg.tar.gz
+    filename: ubuntu-cloudimg.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: custom
@@ -139,7 +139,7 @@ images:
     packer_template: ubuntu
     packer_target: custom-cloudimg.tar.gz
   ubuntu-flat:
-    url: ubuntu-flat.tar.gz
+    filename: ubuntu-flat.tar.gz
     filetype: tgz
     architecture: amd64/generic
     osystem: custom
@@ -147,7 +147,7 @@ images:
     packer_template: ubuntu
     packer_target: custom-ubuntu.tar.gz
   ubuntu-lvm:
-    url: ubuntu-lvm.tar.gz
+    filename: ubuntu-lvm.tar.gz
     filetype: ddgz
     architecture: amd64/generic
     osystem: custom
diff --git a/setup.py b/setup.py
index f6d6ae4..b6c9b32 100644
--- a/setup.py
+++ b/setup.py
@@ -1,6 +1,7 @@
 from setuptools import find_packages, setup
 
 install_requires = (
+    'jenkinsapi',
     'netaddr',
     'paramiko',
     'pytest-dependency',
@@ -12,6 +13,7 @@ install_requires = (
     'requests',
     'retry',
     'ruamel.yaml',
+    'temporalio'
 )
 
 
diff --git a/systemtests/api.py b/systemtests/api.py
index ec76b0e..dde94dd 100644
--- a/systemtests/api.py
+++ b/systemtests/api.py
@@ -78,6 +78,7 @@ class BootSource(TypedDict):
 # TODO: Expand these to TypedDict matching API response structure
 
 Subnet = Dict[str, Any]
+Interface = Dict[str, Any]
 RackController = Dict[str, Any]
 RegionController = Dict[str, Any]
 IPRange = Dict[str, Any]
@@ -256,6 +257,7 @@ class AuthenticatedAPIClient:
         architecture: str,
         filetype: str,
         image_file_path: str,
+        base_image: str | None = None,
     ) -> None:
         cmd = [
             "boot-resources",
@@ -266,6 +268,8 @@ class AuthenticatedAPIClient:
             f"filetype={filetype}",
             f"content@={image_file_path}",
         ]
+        if base_image:
+            cmd.append(f"base_image={base_image}")
         self.execute(cmd, json_output=False)
 
     def import_boot_resources(self) -> str:
@@ -716,6 +720,38 @@ class AuthenticatedAPIClient:
             + [f"{k}={v}" for k, v in options.items()]
         )
 
+    def create_interface(
+        self, machine: Machine, network_type: str, options: dict[str, str] = {}
+    ) -> Interface:
+        """bond, bridge,"""
+        interface: Interface = self.execute(
+            ["interfaces", f"create-{network_type}", machine["system_id"]]
+            + [f"{k}={v}" for k, v in options.items()]
+        )
+        return interface
+
+    def delete_interface(self, machine: Machine, interface: Interface) -> str:
+        result: str = self.execute(
+            ["interface", "delete", machine["systed_id"], str(interface["id"])],
+            json_output=False,
+        )
+        return result
+
+    def read_interfaces(self, machine: Machine) -> list[Interface]:
+        result: list[Interface] = self.execute(
+            ["interfaces", "read", machine["system_id"]]
+        )
+        return result
+
+    def update_interface(
+        self, machine: Machine, interface: Interface, options: dict[str, str]
+    ) -> Interface:
+        updated_interface: Interface = self.execute(
+            ["interface", "update", machine["system_id"], str(interface["id"])]
+            + [f"{k}={v}" for k, v in options.items()]
+        )
+        return updated_interface
+
 
 class QuietAuthenticatedAPIClient(AuthenticatedAPIClient):
     """An Authenticated API Client that is quiet."""
diff --git a/systemtests/conftest.py b/systemtests/conftest.py
index a069d84..6acabf7 100644
--- a/systemtests/conftest.py
+++ b/systemtests/conftest.py
@@ -358,7 +358,9 @@ def pytest_generate_tests(metafunc: Metafunc) -> None:
         metafunc.parametrize("instance_config", instance_config, ids=str, indirect=True)
 
     if "image_to_test" in metafunc.fixturenames:
-        if images_to_test := [image for image in generate_images(cfg) if image.url]:
+        if images_to_test := [
+            image for image in generate_images(cfg) if image.filename
+        ]:
             metafunc.parametrize(
                 "image_to_test", images_to_test, ids=str, indirect=True
             )
diff --git a/systemtests/fixtures.py b/systemtests/fixtures.py
index 7521c7d..e53ba45 100644
--- a/systemtests/fixtures.py
+++ b/systemtests/fixtures.py
@@ -763,7 +763,9 @@ def dns_tester(
 
 
 @pytest.fixture(scope="session")
-def packer_main(config: dict[str, Any]) -> Optional[Iterator[PackerMain]]:
+def packer_main(
+    request: pytest.FixtureRequest, config: dict[str, Any]
+) -> Optional[Iterator[PackerMain]]:
     """Set up a new LXD container with Packer installed."""
     packer_config = config.get("packer-maas", {})
     repo = packer_config.get("git-repo")
@@ -787,6 +789,7 @@ def packer_main(config: dict[str, Any]) -> Optional[Iterator[PackerMain]]:
         proxy_env=proxy_env,
         file_store=config.get("file-store", {}),
         debug=packer_config.get("verbosity", ""),
+        root_path=request.config.rootpath,
     )
     main.setup()
     yield main
diff --git a/systemtests/git_build.py b/systemtests/git_build.py
index 342fa0c..3803322 100644
--- a/systemtests/git_build.py
+++ b/systemtests/git_build.py
@@ -5,6 +5,7 @@ from contextlib import closing
 from functools import partial
 from pathlib import Path
 from subprocess import CalledProcessError
+from textwrap import dedent
 from timeit import Timer
 from typing import TYPE_CHECKING, Any, Callable
 from urllib.request import urlopen
@@ -33,6 +34,7 @@ class GitBuild:
         self._repos = repo
         self._branch = branch
         self._clone_path = clone_path
+        self._set_apt_proxy()
 
     @property
     def clone_path(self) -> str:
@@ -46,6 +48,18 @@ class GitBuild:
     def logger(self, logger: Logger) -> None:
         self._instance.logger = logger
 
+    def _set_apt_proxy(self) -> None:
+        if proxy := self._env.get("http_proxy"):
+            conf = self._instance.files["/etc/apt/apt.conf.d/99-proxy.conf"]
+            conf.write(
+                dedent(
+                    f"""\
+                    Acquire::http::Proxy "{proxy}";
+                    Acquire::https::Proxy "{proxy}";
+                    """
+                )
+            )
+
     def apt_update(self) -> None:
         """Update APT indices, fix broken dpkg."""
         self._instance.quietly_execute(
diff --git a/systemtests/image_builder/test_packer.py b/systemtests/image_builder/test_packer.py
index 3bf5836..3619ff2 100644
--- a/systemtests/image_builder/test_packer.py
+++ b/systemtests/image_builder/test_packer.py
@@ -19,7 +19,10 @@ class TestPackerMAASConfig:
         assert readme.exists(), f"README.md not found in {packer_main.clone_path}"
 
     def test_build_image(
-        self, testlog: Logger, packer_main: PackerMain, image_to_build: TestableImage
+        self,
+        testlog: Logger,
+        packer_main: PackerMain,
+        image_to_build: TestableImage,
     ) -> None:
         # tell mypy we have this under control
         assert image_to_build.packer_template is not None
@@ -28,12 +31,12 @@ class TestPackerMAASConfig:
         image = packer_main.build_image(
             image_to_build.packer_template,
             image_to_build.packer_target,
-            image_to_build.filename,
+            image_to_build.packer_filename,
             image_to_build.source_iso,
         )
         assert image is not None
         img_file = packer_main._instance.files[image]
         assert img_file.exists(), f"failed to produce the expected image ({img_file})"
 
-        if image_to_build.url is not None:
-            packer_main.upload_image(img_file, image_to_build.url)
+        if image_to_build.filename:
+            packer_main.upload_image(img_file, image_to_build.filename)
diff --git a/systemtests/image_config.py b/systemtests/image_config.py
index 4f7a0e0..d92bff2 100644
--- a/systemtests/image_config.py
+++ b/systemtests/image_config.py
@@ -21,7 +21,7 @@ EXTENSION_MAP = {
 @dataclass(frozen=True)
 class TestableImage:
     name: str
-    url: str | None
+    filename: str
     filetype: str = "targz"
     architecture: str = "amd64/generic"
     osystem: str = "ubuntu"
@@ -48,7 +48,7 @@ class TestableImage:
         )
 
     @property
-    def filename(self) -> str:
+    def packer_filename(self) -> str:
         ext = EXTENSION_MAP[self.filetype]
         if self.packer_template is None:
             return f"{self.name}.{ext}"
diff --git a/systemtests/packer.py b/systemtests/packer.py
index 693d6a4..03beb5e 100644
--- a/systemtests/packer.py
+++ b/systemtests/packer.py
@@ -29,6 +29,7 @@ class PackerMain(GitBuild):
         file_store: dict[str, Any],
         proxy_env: dict[str, str] | None,
         debug: str | None,
+        root_path: Path,
     ) -> None:
         super().__init__(
             packer_repo,
@@ -40,8 +41,14 @@ class PackerMain(GitBuild):
         )
         self.default_debug = debug or ""
         self.file_store = file_store
+        self.root_path = root_path
 
     def setup(self) -> None:
+        if "http_proxy" in self._env:
+            sudoers = self._instance.files["/etc/sudoers.d/50-preserve-proxy"]
+            sudoers.write(
+                'Defaults env_keep += "ftp_proxy http_proxy https_proxy no_proxy"'
+            )
         self.apt_source_add(
             "packer",
             "https://apt.releases.hashicorp.com",
@@ -101,8 +108,14 @@ class PackerMain(GitBuild):
         source_iso: str | None,
     ) -> str | None:
         env = self._env.copy()
+        env["SUDO"] = "sudo -E"
+        log_file = f"build-{packer_template}-{packer_target or 'all'}.log"
+        env["PACKER_LOG"] = "on"
+        env["PACKER_LOG_PATH"] = f"{self.clone_path}/{log_file}"
         if source_iso:
             env["ISO"] = self.download_image(source_iso)
+        if proxy := env.get("https_proxy"):
+            env["KS_PROXY"] = f'--proxy="{proxy}"'
         cmd: list[str] = [
             "eatmydata",
             "make",
@@ -110,12 +123,16 @@ class PackerMain(GitBuild):
             f"{self.clone_path}/{packer_template}",
             f"{packer_target or 'all'}",
         ]
-        runtime = self.timed(
-            self._instance.execute,
-            command=cmd,
-            environment=env,
-        )
-        self.logger.info(f"Image built in {runtime:.2f}s")
+        try:
+            runtime = self.timed(
+                self._instance.execute,
+                command=cmd,
+                environment=env,
+            )
+            self.logger.info(f"Image built in {runtime:.2f}s")
+        finally:
+            build_log = self._instance.files[env["PACKER_LOG_PATH"]]
+            build_log.pull(str(self.root_path / log_file))
         return f"{self.clone_path}/{packer_template}/{img_filename}"
 
     def __repr__(self) -> str:
diff --git a/systemtests/state.py b/systemtests/state.py
index 36b89ba..7ca5be8 100644
--- a/systemtests/state.py
+++ b/systemtests/state.py
@@ -10,9 +10,8 @@ from urllib.parse import urljoin, urlparse
 import pytest
 from retry import retry
 
-from systemtests.image_config import TestableImage
-from systemtests.packer import UnknowStorageBackendError
-
+from .image_config import TestableImage
+from .packer import UnknowStorageBackendError
 from .region import get_rack_controllers
 from .utils import waits_for_event_after
 
diff --git a/systemtests/tests_per_machine/test_machine.py b/systemtests/tests_per_machine/test_machine.py
index c4995b3..7c11328 100644
--- a/systemtests/tests_per_machine/test_machine.py
+++ b/systemtests/tests_per_machine/test_machine.py
@@ -11,6 +11,7 @@ from ..utils import (
     assert_machine_in_machines,
     assert_machine_not_in_machines,
     release_and_redeploy_machine,
+    report_feature_tests,
     ssh_execute_command,
     wait_for_machine,
     wait_for_machine_to_power_off,
@@ -27,7 +28,7 @@ if TYPE_CHECKING:
     from ..machine_config import MachineConfig
 
 
-@test_steps("enlist", "metadata", "commission", "deploy", "rescue")
+@test_steps("enlist", "metadata", "commission", "deploy", "test_image", "rescue")
 def test_full_circle(
     maas_api_client: AuthenticatedAPIClient,
     machine_config: MachineConfig,
@@ -147,21 +148,47 @@ def test_full_circle(
     yield
 
     if image_to_test:
-        testable_layouts = ["flat", "lvm", "bcache"]
-        for storage_layout in testable_layouts:
-            testlog.info(f"Testing storage layout: {storage_layout}")
-            passed = False
-            try:
+        testable_configs: dict[str, dict[str, str]] = {
+            "bond": {"parents": "1"},
+            "bridge": {},
+        }
+        for network_config, network_options in testable_configs.items():
+            with report_feature_tests(testlog, f"network layout {network_config}"):
                 with release_and_redeploy_machine(
-                    maas_api_client, machine, timeout=TIMEOUT
-                ) as redeployed:
-                    maas_api_client.create_storage_layout(
-                        redeployed, storage_layout, {}
+                    maas_api_client,
+                    machine,
+                    osystem=deploy_osystem,
+                    oseries=deploy_oseries,
+                    timeout=TIMEOUT,
+                ):
+                    interface = maas_api_client.create_interface(
+                        machine, network_config, network_options
                     )
-                    passed = True
-            finally:
-                status = "PASSED" if passed else "FAILED"
-                testlog.info(f"Storage layout: {storage_layout} {status}")
+                assert interface in maas_api_client.read_interfaces(machine)
+                with release_and_redeploy_machine(
+                    maas_api_client,
+                    machine,
+                    osystem=deploy_osystem,
+                    oseries=deploy_oseries,
+                    timeout=TIMEOUT,
+                ):
+                    maas_api_client.delete_interface(machine, interface)
+                assert interface not in maas_api_client.read_interfaces(machine)
+        testable_layouts = ["flat", "lvm", "bcache"]
+        for storage_layout in testable_layouts:
+            with report_feature_tests(
+                testlog, f"storage layout {storage_layout}"
+            ), release_and_redeploy_machine(
+                maas_api_client,
+                machine,
+                osystem=deploy_osystem,
+                oseries=deploy_oseries,
+                timeout=TIMEOUT,
+            ):
+                # release the machine, add a new storage layout,
+                # assert the machine can redeploy
+                maas_api_client.create_storage_layout(machine, storage_layout, {})
+    yield
 
     if deploy_osystem == "windows" or (
         deploy_osystem == "custom" and deploy_oseries.startswith("esxi")
diff --git a/systemtests/utils.py b/systemtests/utils.py
index 66ebc8b..b412813 100644
--- a/systemtests/utils.py
+++ b/systemtests/utils.py
@@ -9,6 +9,7 @@ import time
 from contextlib import contextmanager
 from dataclasses import dataclass
 from logging import Logger
+from subprocess import CalledProcessError
 from typing import Iterator, Optional, TypedDict, Union
 
 import paramiko
@@ -300,32 +301,51 @@ def assert_machine_not_in_machines(
 def release_and_redeploy_machine(
     maas_api_client: api.AuthenticatedAPIClient,
     machine: api.Machine,
+    osystem: str,
+    oseries: str | None = None,
     timeout: int = 60 * 40,
 ) -> Iterator[api.Machine]:
-    name, osystem = machine["name"], machine["osystem"]
     try:
         maas_api_client.release_machine(machine)
-        wait_for_machine(
+        yield wait_for_machine(
             maas_api_client,
             machine,
             status="Ready",
             abort_status="Releasing failed",
-            machine_id=name,
             timeout=timeout,
         )
-        yield machine
     finally:
-        maas_api_client.deploy_machine(machine, osystem=osystem)
+        maas_api_client.deploy_machine(
+            machine, osystem=osystem, distro_series=oseries or osystem
+        )
         wait_for_machine(
             maas_api_client,
             machine,
             status="Deployed",
             abort_status="Failed deployment",
-            machine_id=name,
             timeout=timeout,
         )
 
 
+@contextmanager
+def report_feature_tests(testlog: Logger, feature_name: str) -> Iterator[Logger]:
+    """Return a context manager for reporting on a feature.
+    Ensures we always report a paas/fail state, irrespective of errors.
+    """
+    feature_status = False
+    feature_logger = testlog.getChild(feature_name)
+    feature_logger.info("Starting test")
+    try:
+        yield feature_logger
+        feature_status = True
+    except CalledProcessError as exc:
+        feature_logger.exception(exc.stderr)
+    except Exception as e:
+        feature_logger.exception(e)
+    finally:
+        feature_logger.info("PASSED" if feature_status else "FAILED")
+
+
 @dataclass
 class IPRange:
     start: ipaddress.IPv4Address
diff --git a/temporal/README.md b/temporal/README.md
new file mode 100644
index 0000000..3817166
--- /dev/null
+++ b/temporal/README.md
@@ -0,0 +1,88 @@
+# Temporal workflows for OS Image Testing
+
+Here be dragons.
+(Well, maybe not quite)
+
+Contained are the set of scripts required to take a supported image in the [PackerMAAS](https://github.com/canonical/packer-maas/tree/main) repository, build and test it's capabilities on a set MAAS version, and report the results of those tests to a [results area](https://github.com/maas/MAAS-Image-Results) ready to be consumed by documentation.
+
+## Workflows
+
+We distribute four workflows, each with a correspondingly named worker that should be ran to execute that workflow. 
+
+- `image_building_workflow` - Builds an image according to the makefile listed in PackerMAAS.
+- `image_testing_workflow` - Tests an image against `tests_per_mahcine` in this repo, 
+- `image_reporting_workflow` - Compiles the results of the two above workflows into YAML, exporting it to the remote store.
+- `e2e_workflow` - Orchestrates the above as child workflows. Additionally performs some some mild pre-processing for the `image_reporting` workflow.
+
+## Execution
+
+Connect all four workers to a running temporal server instance. An image test can then be requested with a single call to `e2e_workflow`, such as:
+```bash
+temporal workflow start -t e2e_tests --type e2e_workflow -w 'centos_tests' -i '{"image_name": ["centos7", "centos8"], "maas_snap_channel": "3.3/stable", "jenkins_url": $jenkins_url, "jenkins_user": $jenkins_user, "jenkins_pass": $jenkins_pass}'
+```
+
+The `e2e_workflow` will then call it's children workflows as required to test the requested images.
+
+### Parameters
+
+#### Required
+
+- `image_name` - The name, or list of names, of images to test.
+
+- Jenkins details
+
+    - `jenkins_url` - The url of the Jenkins server where image tests are located.
+
+    - `jenkins_user` - The username to use to login to the Jenkins server.
+
+    - `jenkins_pass` - The password to use to login to the Jenkins server.
+
+#### Optional
+
+- Filepaths
+
+    - `image_mapping` - The filepath of the image mapping YAML distributed as part of MAAS-Integration-CI, defaults as `image_mapping.yaml` in the current working directory.
+
+    - `repo_location` - The filepath of the location where the image results repo is to be cloned.
+
+- Test instances
+
+    - `maas_snap_channel` - The snap channel to use when installing MAAS in image tests, defaults as `latest/edge`.
+
+    - `system_test_repo` - The url of the system-tests repo to use for building and testing images, defaults as `https://git.launchpad.net/~maas-committers/maas-ci/+git/system-tests`.
+
+    - `system_test_branch` - The branch in the system-test repo to use for building and tetsing images, defaults as `master`.
+
+    - `packer_maas_repo` - The url of the PackerMAAS repo to use for building images, defaults as `https://github.com/canonical/packer-maas.git`.
+
+    - `packer_maas_branch` - The branch in the PackerMAas repo to use for building images, defaults as `main`.
+
+    - `parallel_tests` - A flag to request a single image test build for all images, rather than a test build per image, defaults as `False`.
+
+    - `overwite_results` - A flag to request new results overwrite old results rather than combining with them, defaults as `False`.
+
+- Retries
+
+    - `max_retry_attempts` - How many times workflow activities should retry before throwing an exception, defaults as `10`
+
+    - `heartbeat_delay` - How many seconds between heartbeats for long running workflow activities, defaults as `15`
+
+- Timeouts
+
+    - Timeouts given are in seconds, and are passed to temporal as [`start_to_close`](https://www.temporal.io/blog/activity-timeouts), which defines the maximum execution time of a single invocation.
+
+    - `default_timeout` - How long a workflow activity can run before being timed out, defaults as `300`. This is used in place of any timeouts below that are not set.
+
+    - `jenkins_login_timeout` - How long we wait to log into the Jenkins server.
+
+    - `return_status_timeout` - How long we wait for an activity to fetch the status of a Jenkins build.
+
+    - `get_results_timeout` - How long we wait for the results of a Jenkins build to be available.
+
+    - `fetch_results_timeout` - How long we wait for an activity to fetch the results of a Jenkins build, and perform some operation on them.
+
+    - `log_details_timeout` - How long we wait for an activity to fetch logs from a Jenkins build, and perform some operation on them.
+
+    - `request_build_timeout` - How long we wait for an activity to request a Jenkins build.
+
+    - `build_complete_timeout` - How long we wait for a Jenkins build to complete, defaults as `7200`.
diff --git a/temporal/build_results.py b/temporal/build_results.py
new file mode 100644
index 0000000..f98eed8
--- /dev/null
+++ b/temporal/build_results.py
@@ -0,0 +1,395 @@
+from __future__ import annotations
+
+import re
+import subprocess
+from collections import defaultdict
+from contextlib import contextmanager
+from dataclasses import dataclass
+from functools import cached_property
+from typing import Any, Iterator
+
+from common_tasks import cleanup_files
+
+
+class TestStatus:
+    # failure
+    FAILED = 0
+    REGRESSION = 1
+    # successes
+    PASSED = 10
+    FIXED = 11
+    # no known state
+    UNKNOWN = 100
+
+    def __init__(self, state: str | None = None, code: int | None = None) -> None:
+        if state is None and code is None:
+            s, c = "UNKNOWN", self.UNKNOWN
+        elif state is None and code is not None:
+            s, c = self._code_to_state_(code), code
+        elif state is not None and code is None:
+            s, c = state, self._state_to_code_(state)
+        elif state is not None and code is not None:
+            s, c = state, code
+        self._state_, self._code_ = s, c
+
+    def __str__(self) -> str:
+        return f"{self._state_} {self._code_}"
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    @cached_property
+    def _code_state_map_(self) -> dict[int, str]:
+        return {
+            getattr(self, attr): attr for attr in dir(self) if not attr.startswith("_")
+        }
+
+    @cached_property
+    def _state_code_map_(self) -> dict[str, int]:
+        return {v: k for k, v in self._code_state_map_.items()}
+
+    def _code_to_state_(self, code: int) -> str:
+        return self._code_state_map_.get(code, "UNKNOWN")
+
+    def _state_to_code_(self, state: str) -> int:
+        return self._state_code_map_.get(state.upper(), self.UNKNOWN)
+
+    def _is_positive_state_(self, state: str) -> bool:
+        return self._is_positive_code_(self._state_to_code_(state))
+
+    def _is_positive_code_(self, code: int) -> bool:
+        return False if code == self.UNKNOWN else code >= self.PASSED
+
+    @property
+    def _is_positive_(self) -> bool:
+        return self._is_positive_code_(self._code_)
+
+    @property
+    def _has_custom_state_(self) -> bool:
+        return (self._state_to_code_(self._state_) == self.UNKNOWN) and (
+            self._state_ != "UNKNOWN"
+        )
+
+    def to_dict(self) -> dict[str, str | int]:
+        return {"state": self._state_, "code": self._code_}
+
+    def __add__(self, other: Any) -> TestStatus:
+        if not isinstance(other, TestStatus):
+            return self
+        newcode = min(self._code_, other._code_)
+        custom_states = [self._has_custom_state_, other._has_custom_state_]
+        if all(custom_states):
+            newstate = self._state_ + "; " + other._state_
+        elif any(custom_states):
+            newstate = self._state_ if self._has_custom_state_ else other._state_
+        else:
+            newstate = self._code_to_state_(newcode)
+        return TestStatus(newstate, newcode)
+
+    def __radd__(self, other: Any) -> TestStatus:
+        if isinstance(other, TestStatus):
+            return self + other
+        return self
+
+    def __iadd__(self, other: Any) -> TestStatus:
+        if isinstance(other, TestStatus):
+            return self + other
+        return self
+
+
+@dataclass
+class FeatureStatus:
+    name: str = ""
+    state: bool = False
+    readable_state: str | dict[str, Any] = "Failed"
+    info: str = "Could not complete test"
+
+    def __str__(self) -> str:
+        return "\n - ".join([f"{self.name}: {self.readable_state}", self.info])
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            self.name: {
+                "state": "passed" if self.state else "failed",
+                "summary": self.readable_state,
+                "info": self.info,
+            }
+        }
+
+    def __add__(self, other: FeatureStatus) -> FeatureStatus:
+        if not other.state:
+            return self
+        elif not self.state:
+            return other
+        if self.name != other.name:
+            raise Exception(f"{other} does not correspond to the same feature!")
+        return FeatureStatus(
+            name=self.name,
+            state=self.state or other.state,
+            readable_state=self.readable_state,
+            info=self.info,
+        )
+
+
+class ImageTestResults:
+    def __init__(
+        self,
+        image: str = "",
+        maas_version: list[str] = [],
+        packer_version: list[str] = [],
+        readable_state: str = "",
+        tested_arches: list[str] = [],
+        prerequisites: list[str] = [],
+    ) -> None:
+        self.image = image
+        self.maas_version = maas_version
+        self.readable_state = readable_state
+        self.tested_arches = tested_arches
+        self.packer_version = packer_version
+        self.prerequisites = prerequisites
+
+    @property
+    def _feature_dicts_(self) -> dict[str, Any]:
+        out: dict[str, Any] = {}
+        for feature in self._results_:
+            out |= getattr(self, feature).to_dict()
+        return out
+
+    @property
+    def _features_(self) -> list[str]:
+        """Return a short summary of all test results of all features
+        for MAAS Image tests."""
+        return [getattr(self, feature) for feature in self._results_]
+
+    @property
+    def _results_(self) -> list[str]:
+        """Return a list of all features whose results have been collected"""
+        return list(set(self.__dict__) - set(ImageTestResults().__dict__))
+
+    def __str__(self) -> str:
+        return "\n".join(
+            [f"{self.image}: {self.readable_state}"]
+            + [str(feature) for feature in self._features_]
+        )
+
+    @property
+    def state(self) -> str:
+        """Image test state, short pass/fail result as a single bianry string.
+        results formatted as:
+        0b00000{storage}{network}{deploy}"""
+        byte = sum(
+            2**i * getattr(result, "state", 0)
+            for i, result in enumerate(self._results_)
+        )
+        return f"{byte:08b}"
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            self.image: {
+                "summary": self.readable_state,
+                "maas_version": self.maas_version,
+                "architectures": list(self.tested_arches),
+                "packer_versions": self.packer_version,
+                "prerequisites": list(self.prerequisites),
+            }
+            | self._feature_dicts_
+        }
+
+    def from_dict(self, fromdict: dict[str, Any]) -> ImageTestResults:
+        image, details = tuple(fromdict.items())[0]
+        results = ImageTestResults(
+            image=image,
+            maas_version=details.get("maas_version", []),
+            packer_version=details.get("packer_versions", []),
+            readable_state=details.get("summary", ""),
+            tested_arches=details.get("architectures", []),
+            prerequisites=details.get("prerequisites", []),
+        )
+        for key in list(results.to_dict().values())[0].keys():
+            details.pop(key)
+        for feature, feature_dict in details.items():
+            setattr(
+                results,
+                feature,
+                FeatureStatus(
+                    name=feature,
+                    state=feature_dict["state"] == "passed",
+                    readable_state=feature_dict["summary"],
+                    info=feature_dict["info"],
+                ),
+            )
+        return results
+
+    def __add__(self, other: ImageTestResults) -> ImageTestResults:
+        if self.image != other.image:
+            raise Exception(f"{other} does not correspond to the same image!")
+        # return itself if the other failed
+        if not int(other.state, 2) & 1:
+            return self
+        elif not int(self.state, 2) & 1:
+            return other
+
+        def force_set(var: str | list[Any] | set[Any]) -> set[Any]:
+            return set([var]) if isinstance(var, str) else set(var)
+
+        def combine_sets(
+            var: str | list[Any] | set[Any], var2: str | list[Any] | set[Any]
+        ) -> list[Any]:
+            return list(force_set(var).union(force_set(var2)))
+
+        combined_state = TestStatus(state=self.readable_state) + TestStatus(
+            state=other.readable_state
+        )
+        results = ImageTestResults(
+            image=self.image,
+            maas_version=combine_sets(self.maas_version, other.maas_version),
+            packer_version=combine_sets(self.packer_version, other.packer_version),
+            readable_state=combined_state._state_,
+            tested_arches=combine_sets(self.tested_arches, other.tested_arches),
+            prerequisites=combine_sets(self.prerequisites, other.prerequisites),
+        )
+        for feature in set(self._results_).union(set(other._results_)):
+            setattr(
+                results,
+                feature,
+                getattr(self, feature, FeatureStatus())
+                + getattr(self, feature, FeatureStatus()),
+            )
+        return results
+
+
+def todict(nested: defaultdict[str, Any] | dict[str, Any]) -> dict[str, Any]:
+    for k, v in nested.items():
+        if isinstance(v, dict):
+            nested[k] = todict(v)
+    return dict(nested)
+
+
+def nested_dict() -> defaultdict[str, Any]:
+    return defaultdict(nested_dict)
+
+
+def feature_dict_summary(
+    feature_dict: dict[str, dict[str, list[str]]]
+) -> tuple[bool, dict[str, list[str]], str]:
+    # /artificial data for testing
+    states = set(feature_dict.keys())
+    failed = set(feature_dict["FAILED"].keys())
+    passed = set(feature_dict["PASSED"].keys())
+    unknown: set[str] = set()
+    for unknown_states in states - {"PASSED", "FAILED"}:
+        unknown |= set(feature_dict[unknown_states].keys())
+
+    # overall pass fail for the entire feature
+    state = not (len(failed) or len(unknown))
+    # overall pass fail for each value of the feature
+    summary: dict[str, list[str]] = {}
+    if full_pass := passed - (failed | unknown):
+        summary["PASS"] = list(full_pass)
+    if full_fail := failed - (passed | unknown):
+        summary["FAIL"] = list(full_fail)
+    if partial_fail := (passed & failed) | unknown:
+        summary["PARTIAL"] = list(partial_fail)
+    # specific pass fail for each value of the feature
+    info = []
+    for fstate, fvalue in feature_dict.items():
+        info.extend(
+            [fstate.lower()]
+            + [f" - {layout}: {', '.join(arch)}" for layout, arch in fvalue.items()]
+        )
+    return state, summary, "\n".join(info)
+
+
+def scan_log_for_feature(
+    feature_name: str, arches: dict[str, Any]
+) -> dict[str, dict[str, list[str]]]:
+    tested = nested_dict()
+    """ Matches the two ways we can show test results:
+        'storage layout flat: PASSED'
+        'Storage layout: bcache - FAILED'
+        returns the feature (flat, bcache) and result (PASSED, FAILED)
+    """
+    versioning_match = r":?\s(\w+):?\s(?:\-\s)?([A-Z]{4,})"
+    feature_match = re.compile(f"{feature_name}{versioning_match}", flags=re.IGNORECASE)
+    for arch_name, arch in arches.items():
+        arch_log = "\n".join(arch["log"])
+        for feature, state in feature_match.findall(arch_log):
+            if feature not in tested[state]:
+                tested[state][feature] = []
+            tested[state][feature].append(arch_name)
+    return todict(tested)
+
+
+def determine_feature_state(
+    feature_name: str, arches: dict[str, Any]
+) -> tuple[bool, dict[str, list[str]], str] | None:
+    if feature_tested := scan_log_for_feature(feature_name, arches):
+        return feature_dict_summary(feature_tested)
+    return None
+
+
+def execute(
+    command: list[str], cwd: str | None = None
+) -> subprocess.CompletedProcess[str]:
+    """Execute a command"""
+    __tracebackhide__ = True
+    return subprocess.run(
+        command,
+        capture_output=True,
+        check=True,
+        encoding="utf-8",
+        errors="backslashreplace",
+        cwd=cwd,
+    )
+
+
+@contextmanager
+def checkout_and_commit(
+    branch: str,
+    commit_message: str,
+    base_branch: str | None = None,
+    add_file: str | list[str] | None = None,
+    cwd: str | None = None,
+) -> Iterator[None]:
+    branches = execute(["git", "branch", "-a"], cwd=cwd).stdout
+    branch_base = base_branch or ("main" if "main" in branches else "master")
+    current_branch = execute(["git", "rev-parse", "--abbrev-ref HEAD"], cwd=cwd).stdout
+
+    # ensure we're up to date with the base branch first
+    if current_branch != branch_base:
+        execute(["git", "checkout", branch_base], cwd=cwd)
+        execute(["git", "pull"], cwd=cwd)
+        current_branch = branch_base
+
+    # navigate to the correct branch
+    if current_branch != branch:
+        if branch in branches:
+            execute(["git", "checkout", branch], cwd=cwd)
+            try:
+                execute(["git", "pull"], cwd=cwd)
+            except Exception as e:
+                print(e)
+        else:
+            execute(["git", "checkout", "-b", branch], cwd=cwd)
+
+    yield
+
+    if cwd and add_file:
+        cleanup_files(cwd, preserve=add_file)
+
+    # if the previous commit matches the one we want to make, combine them
+    reset = False
+    while (
+        execute(["git", "show-branch", "--no-name", "HEAD~1"], cwd=cwd).stdout
+        == f"{commit_message}"
+    ):
+        execute(["git", "reset", "--hard", "HEAD~1"], cwd=cwd)
+        reset = True
+
+    # add files and commit
+    execute(["git", "add", "."], cwd=cwd)
+    execute(["git", "commit", "-m", f'"{commit_message}"'], cwd=cwd)
+    if reset:
+        execute(["git", "push", "-f"], cwd=cwd)
+    else:
+        execute(["git", "push"], cwd=cwd)
diff --git a/temporal/common_tasks.py b/temporal/common_tasks.py
new file mode 100644
index 0000000..8fc6011
--- /dev/null
+++ b/temporal/common_tasks.py
@@ -0,0 +1,293 @@
+import argparse
+import asyncio
+import os
+import sys
+from dataclasses import dataclass
+from datetime import timedelta
+from time import sleep
+from typing import Any
+
+import yaml
+from temporalio import activity, workflow
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+with workflow.unsafe.imports_passed_through():
+    from jenkinsapi.build import Artifact, Build  # type:ignore[import]
+    from jenkinsapi.jenkins import Jenkins  # type:ignore[import]
+    from jenkinsapi.job import Job  # type:ignore[import]
+
+
+# Workflow parameter class
+@dataclass
+class workflow_parameters:
+    jenkins_url: str
+    jenkins_user: str
+    jenkins_pass: str
+    job_name: str = ""
+    build_num: int = -1
+
+    # retry stuff
+    max_retry_attempts: int = 10
+    heartbeat_delay: int = 15
+
+    # default timeout to be used if none available
+    default_timeout: int = 300
+    # how long should we wait to login
+    jenkins_login_timeout: int = -1
+    # how long should we wait for the build to complete
+    return_status_timeout: int = -1
+    # how long should we wait to get build results?
+    fetch_results_timeout: int = -1
+    # how long should we wait for log scanning to occur?
+    log_details_timeout: int = -1
+    # how long should we wait for this build to be requested
+    request_build_timeout: int = -1
+    # how long should we wait for the build to complete
+    build_complete_timeout: int = 7200
+    # how long should we wait for the results to be available
+    get_results_timeout: int = -1
+
+    # return the default timeout if the set timeout is not applicable
+    def gettimeout(self, timeout_name: str = "") -> timedelta:
+        if (timeout := self.__dict__.get(timeout_name, 0)) > 0:
+            return timedelta(seconds=timeout)
+        return timedelta(seconds=self.default_timeout)
+
+
+# common functions
+
+
+def cleanup_files(file_path: str, preserve: str | list[str] | None = None) -> None:
+    if os.path.exists(file_path):
+        files = os.listdir(file_path)
+        files.remove(".git")
+        if preserve:
+            for preserved_file in aslist(preserve):
+                this_file = os.path.basename(preserved_file)
+                if this_file in files:
+                    files.remove(this_file)
+        if files:
+            print(f"Removing: {files}")
+            for cleanup in files:
+                os.remove(f"{file_path}/{cleanup}")
+
+
+def aslist(to_list: str | list[Any]) -> list[Any]:
+    if isinstance(to_list, list):
+        return to_list
+    return [to_list] if to_list else []
+
+
+def get_server(params: workflow_parameters) -> Jenkins:
+    return Jenkins(
+        params.jenkins_url,
+        username=params.jenkins_user,
+        password=params.jenkins_pass,
+        timeout=params.gettimeout("jenkins_login_timeout").seconds,
+        max_retries=params.max_retry_attempts,
+    )
+
+
+def get_job(
+    params: workflow_parameters,
+    job_name: str | None = None,
+) -> Job:
+    return get_server(params).get_job(job_name or params.job_name)
+
+
+def get_build(
+    params: workflow_parameters,
+    job_name: str | None = None,
+    build_num: int | None = None,
+) -> Build:
+    job = get_job(params, job_name=job_name)
+    if (num := build_num or params.build_num) >= 0:
+        return job.get_build(num)
+    return job.get_last_build()
+
+
+def get_params(
+    params: workflow_parameters,
+    job_name: str | None = None,
+    build_num: int | None = None,
+) -> dict[str, Any]:
+    build = get_build(params, job_name=job_name, build_num=build_num)
+    return build.get_params()  # type: ignore
+
+
+def get_logs(
+    params: workflow_parameters,
+    job_name: str | None = None,
+    build_num: int | None = None,
+) -> dict[str, str]:
+    # attempt utf-8. If that doesn't work, try utf-16
+    def decode_artifact_data(artifact: Artifact) -> str:
+        data = artifact.get_data()
+        try:
+            return str(data, encoding="utf-8")
+        except Exception as e:
+            print(e)
+            return str(data, encoding="utf-16")
+
+    build = get_build(params, job_name=job_name, build_num=build_num)
+    logs = {
+        name.split(".")[-2]: decode_artifact_data(artifact)
+        for name, artifact in build.get_artifact_dict().items()
+        if ".log" in name
+    }
+    return logs
+
+
+def get_config(
+    params: workflow_parameters,
+    job_name: str | None = None,
+    build_num: int | None = None,
+) -> dict[str, Any]:
+    build = get_build(params, job_name=job_name, build_num=build_num)
+    return yaml.safe_load(  # type: ignore
+        [
+            artifact.get_data()
+            for name, artifact in build.get_artifact_dict().items()
+            if "config.yaml" in name
+        ][0]
+    )
+
+
+def get_results(
+    params: workflow_parameters,
+    job_name: str | None = None,
+    build_num: int | None = None,
+) -> dict[str, Any]:
+    build = get_build(params, job_name=job_name, build_num=build_num)
+    results = build.get_resultset()
+    return {k: v.__dict__ for k, v in results.items()}
+
+
+def request_build(
+    params: workflow_parameters, job_params: dict[str, Any], job_name: str | None = None
+) -> int:
+    server = get_server(params)
+    last_build = int(server.get_job(params.job_name or job_name).get_last_buildnumber())
+    server.build_job(params.job_name, job_params)
+    return last_build + 1
+
+
+# common activities
+
+
[e-mail]
+async def check_jenkins_reachable(params: workflow_parameters) -> bool:
+    server = get_server(params)
+    return bool(server and (server.version != "0.0"))
+
+
[e-mail]
+async def check_build_has_results(params: workflow_parameters) -> bool:
+    build = get_build(params)
+    return bool(build.has_resultset())
+
+
[e-mail]
+async def fetch_build_status(params: workflow_parameters) -> str:
+    build = get_build(params)
+    while build.is_running():
+        sleep(params.heartbeat_delay)
+        activity.heartbeat("Awaiting build finish")
+    return str(build.get_status())
+
+
[e-mail]
+async def fetch_build_and_result(
+    params: workflow_parameters,
+) -> dict[str, dict[str, str]]:
+    build = get_build(params)
+    while not build.has_resultset():
+        sleep(params.heartbeat_delay)
+        activity.heartbeat("Awaiting build results")
+    return {k: {"status": v.status} for k, v in build.get_resultset().items()}
+
+
[e-mail]
+async def await_build_exists(params: workflow_parameters) -> None:
+    job = get_job(params)
+    while not job.is_queued_or_running():
+        sleep(params.heartbeat_delay)
+        activity.heartbeat("Awaiting job start")
+    build = None
+    while True:
+        try:
+            if build is None:
+                build = get_build(params)
+            if build.is_running():
+                break
+        except Exception as e:
+            activity.heartbeat(f"Could not fetch build: {e}")
+        sleep(params.heartbeat_delay)
+        activity.heartbeat("Awaiting build running")
+
+
[e-mail]
+async def await_build_complete(params: workflow_parameters) -> None:
+    build = get_build(params)
+    while build.is_running():
+        sleep(params.heartbeat_delay)
+        activity.heartbeat("Awaiting job completion")
+
+
+# workers
+
+
+def worker_url(argv: list[str]) -> str:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "temporal_url",
+        type=str,
+        default="localhost:7233",
+        help="url of the temporal server",
+    )
+    args = parser.parse_args(argv)
+    return str(args.temporal_url)
+
+
+async def worker_main(
+    interrupt_event: asyncio.Event,
+    temporal_url: str,
+    task_queue: str,
+    workflows: list[Any],
+    activities: list[Any],
+) -> None:
+    client = await Client.connect(temporal_url)
+    async with Worker(
+        client,
+        task_queue=task_queue.lower().replace(" ", "_"),
+        workflows=workflows,
+        activities=activities,
+    ):
+        print(
+            f"{task_queue} worker started, ctrl+c to exit".capitalize().replace(
+                "_", " "
+            )
+        )
+        await interrupt_event.wait()
+
+
+def start_worker(task_queue: str, workflows: list[Any], activities: list[Any]) -> None:
+    temporal_url = worker_url(sys.argv[1:])
+    interrupt_event = asyncio.Event()
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(
+            worker_main(
+                interrupt_event, temporal_url, task_queue, workflows, activities
+            )
+        )
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        interrupt_event.clear()
+        loop.run_until_complete(interrupt_event.wait())
+    finally:
+        loop.run_until_complete(loop.shutdown_asyncgens())
+        loop.close()
diff --git a/temporal/e2e_worker.py b/temporal/e2e_worker.py
new file mode 100644
index 0000000..97c5260
--- /dev/null
+++ b/temporal/e2e_worker.py
@@ -0,0 +1,10 @@
+from common_tasks import start_worker
+from e2e_workflow import activities as e2e_activities
+from e2e_workflow import workflows as e2e_workflows
+
+if __name__ == "__main__":
+    start_worker(
+        task_queue="e2e_tests",
+        workflows=e2e_workflows,
+        activities=e2e_activities,
+    )
diff --git a/temporal/e2e_workflow.py b/temporal/e2e_workflow.py
new file mode 100644
index 0000000..82f2ba7
--- /dev/null
+++ b/temporal/e2e_workflow.py
@@ -0,0 +1,206 @@
+import re
+from dataclasses import dataclass
+from typing import Any
+
+from build_results import nested_dict, todict
+from common_tasks import aslist, get_logs, workflow_parameters
+from image_building_workflow import image_building_param, image_building_workflow
+from image_reporting_workflow import image_reporting_param, image_reporting_workflow
+from image_testing_workflow import image_testing_param, image_testing_workflow
+from temporalio import activity, workflow
+from temporalio.common import RetryPolicy
+
+
+@dataclass
+class e2e_workflow_params(workflow_parameters):
+    image_name: str | list[str] = ""
+    image_mapping: str = (
+        "image_mapping.yaml"  # this needs to be accessible to the worker
+    )
+
+    system_test_repo: str = (
+        "https://git.launchpad.net/~maas-committers/maas-ci/+git/system-tests"
+    )
+    system_test_branch: str = "master"
+    packer_naas_repo: str = "https://github.com/canonical/packer-maas.git"
+    packer_maas_branch: str = "main"
+
+    maas_snap_channel: str = "latest/edge"
+
+    repo_location: str = "image_results_repo"
+
+    overwrite_results: bool = False
+    # reccommended to leave this false until the rescue issue at CI is fixed
+    parallel_tests: bool = False
+
+
[e-mail]
+async def fetch_packer_version_from_logs(
+    params: e2e_workflow_params,
+) -> dict[str, Any]:
+    logs = get_logs(params, job_name="maas-automated-image-builder")
+    packer_details = nested_dict()
+    for image in aslist(params.image_name):
+        packer_details[image]["packer_version"] = ""
+        packer_details[image]["prerequisites"] = []
+        # fetch the build log for this image
+        if log := [v for k, v in logs.items() if image in k]:
+            # fetch the packer version
+            if search := re.search(r"Packer version\: ((\d+\.\d+)\.\d+)", log[0]):
+                long_version, _ = search.groups()
+                packer_details[image]["packer_version"] = long_version
+            else:
+                packer_details[image]["packer_version"] = ""
+            # search for prerequisites
+    return todict(packer_details)
+
+
[e-mail]
+async def fetch_image_details(params: dict[str, Any]) -> dict[str, Any]:
+    details: dict[str, Any] = {}
+    for image in aslist(params["images"]):
+        image_packer_details = params.get("packer_details", {}).get(image, {})
+        image_test_details = params.get("image_results", {}).get(image, {})
+        details[image] = {
+            "built": image not in params.get("failed_images", []),
+            "tested": bool(image_test_details),
+            "build_num": params.get("build_num", -1),
+            "test_num": image_test_details.get("build_num"),
+            "packer_version": image_packer_details.get("packer_version", "0.0"),
+            "prerequisites": image_packer_details.get("prerequisites", []),
+        }
+    return details
+
+
[e-mail]
+class e2e_workflow:
+    @workflow.run
+    async def run(self, params: e2e_workflow_params) -> None:
+        # build images
+        image_building_results: dict[str, Any] = await workflow.execute_child_workflow(
+            image_building_workflow,
+            image_building_param(
+                # building parameters
+                image_name=params.image_name,
+                image_mapping=params.image_mapping,
+                system_test_repo=params.system_test_repo,
+                system_test_branch=params.system_test_branch,
+                packer_naas_repo=params.packer_naas_repo,
+                packer_maas_branch=params.packer_maas_branch,
+                # jenkins stuff
+                jenkins_url=params.jenkins_url,
+                jenkins_user=params.jenkins_user,
+                jenkins_pass=params.jenkins_pass,
+                # timeouts and retry
+                max_retry_attempts=params.max_retry_attempts,
+                heartbeat_delay=params.heartbeat_delay,
+                default_timeout=params.default_timeout,
+                jenkins_login_timeout=params.jenkins_login_timeout,
+                return_status_timeout=params.return_status_timeout,
+                fetch_results_timeout=params.fetch_results_timeout,
+                log_details_timeout=params.log_details_timeout,
+                request_build_timeout=params.request_build_timeout,
+                build_complete_timeout=params.build_complete_timeout,
+                get_results_timeout=params.get_results_timeout,
+            ),
+            task_queue="image_building",
+            id=f"Building: {','.join(params.image_name)}",
+        )
+        # images that failed or succeeded to be built
+        params.build_num = image_building_results.get("build_num", -1)
+        images_built = image_building_results["image_results"]
+        failed_images = [image for image, built in images_built.items() if not built]
+        passed_images = [image for image, built in images_built.items() if built]
+        # get the packer version and prerequisites
+        packer_details = await workflow.execute_activity(
+            fetch_packer_version_from_logs,
+            params,
+            start_to_close_timeout=params.gettimeout("log_details_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        # get all of the images that were built
+        image_testing_results: dict[str, Any] = {}
+        if images_to_test := passed_images:
+            # test images
+            # if we are testing images in parallel, this list will have one entry.
+            for image_test_group in (
+                [images_to_test] if params.parallel_tests else images_to_test
+            ):
+                try:
+                    image_testing_results |= await workflow.execute_child_workflow(
+                        image_testing_workflow,
+                        image_testing_param(
+                            # testing parameters
+                            image_name=image_test_group,
+                            system_test_repo=params.system_test_repo,
+                            system_test_branch=params.system_test_branch,
+                            maas_snap_channel=params.maas_snap_channel,
+                            parallel_tests=params.parallel_tests,
+                            # jenkins stuff
+                            jenkins_url=params.jenkins_url,
+                            jenkins_user=params.jenkins_user,
+                            jenkins_pass=params.jenkins_pass,
+                            # timeouts and retry
+                            max_retry_attempts=params.max_retry_attempts,
+                            heartbeat_delay=params.heartbeat_delay,
+                            default_timeout=params.default_timeout,
+                            jenkins_login_timeout=params.jenkins_login_timeout,
+                            return_status_timeout=params.return_status_timeout,
+                            fetch_results_timeout=params.fetch_results_timeout,
+                            log_details_timeout=params.log_details_timeout,
+                            request_build_timeout=params.request_build_timeout,
+                            build_complete_timeout=params.build_complete_timeout,
+                            get_results_timeout=params.get_results_timeout,
+                        ),
+                        task_queue="image_testing",
+                        id=f"Testing: {','.join(aslist(image_test_group))}",
+                    )
+                except Exception as e:
+                    workflow.logger.exception(f"Could not test {image_test_group}: {e}")
+
+        # populate image details from test results
+        image_details = await workflow.execute_activity(
+            fetch_image_details,
+            {
+                "images": params.image_name,
+                "packer_details": packer_details,
+                "failed_images": failed_images,
+                "build_num": params.build_num,
+                "image_results": image_testing_results,
+            },
+            start_to_close_timeout=params.gettimeout("log_details_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+
+        # report image results
+        await workflow.execute_child_workflow(
+            image_reporting_workflow,
+            image_reporting_param(
+                # reporting parameters
+                image_details=image_details,
+                repo_location=params.repo_location,
+                overwrite_results=params.overwrite_results,
+                maas_snap_channel=params.maas_snap_channel,
+                # jenkins stuff
+                jenkins_url=params.jenkins_url,
+                jenkins_user=params.jenkins_user,
+                jenkins_pass=params.jenkins_pass,
+                # timeouts and retry
+                max_retry_attempts=params.max_retry_attempts,
+                heartbeat_delay=params.heartbeat_delay,
+                default_timeout=params.default_timeout,
+                jenkins_login_timeout=params.jenkins_login_timeout,
+                return_status_timeout=params.return_status_timeout,
+                fetch_results_timeout=params.fetch_results_timeout,
+                log_details_timeout=params.log_details_timeout,
+                request_build_timeout=params.request_build_timeout,
+                build_complete_timeout=params.build_complete_timeout,
+                get_results_timeout=params.get_results_timeout,
+            ),
+            task_queue="image_reporting",
+            id=f"Reporting: {','.join(params.image_name)}",
+        )
+
+
+activities = [fetch_packer_version_from_logs, fetch_image_details]
+workflows = [e2e_workflow]
diff --git a/temporal/image_building_worker.py b/temporal/image_building_worker.py
new file mode 100644
index 0000000..885f578
--- /dev/null
+++ b/temporal/image_building_worker.py
@@ -0,0 +1,10 @@
+from common_tasks import start_worker
+from image_building_workflow import activities as image_build_activities
+from image_building_workflow import workflows as image_build_workflows
+
+if __name__ == "__main__":
+    start_worker(
+        task_queue="image_building",
+        workflows=image_build_workflows,
+        activities=image_build_activities,
+    )
diff --git a/temporal/image_building_workflow.py b/temporal/image_building_workflow.py
new file mode 100644
index 0000000..586f1f4
--- /dev/null
+++ b/temporal/image_building_workflow.py
@@ -0,0 +1,165 @@
+import re
+from dataclasses import dataclass
+from typing import Any
+
+import yaml
+from common_tasks import (
+    aslist,
+    await_build_complete,
+    await_build_exists,
+    check_jenkins_reachable,
+    fetch_build_and_result,
+    fetch_build_status,
+    request_build,
+    workflow_parameters,
+)
+from temporalio import activity, workflow
+from temporalio.common import RetryPolicy
+
+
+@dataclass
+class image_building_param(workflow_parameters):
+    image_name: str | list[str] = ""  # allow builk image building if desired
+    image_mapping: str = (
+        "image_mapping.yaml"  # this needs to be accessible to the worker
+    )
+
+    job_name: str = "maas-automated-image-builder"
+    build_num: int = -1
+
+    # job details with default values we may want to change
+    system_test_repo: str = (
+        "https://git.launchpad.net/~maas-committers/maas-ci/+git/system-tests"
+    )
+    system_test_branch: str = "master"
+    packer_naas_repo: str = "https://github.com/canonical/packer-maas.git"
+    packer_maas_branch: str = "main"
+
+
[e-mail]
+async def request_images_built(params: image_building_param) -> int:
+    """Start an image testing job, returning the job number."""
+    job_params: dict[str, Any] = {
+        "IMAGE_NAMES": ",".join(image for image in aslist(params.image_name)),
+        "SYSTEMTESTS_GIT_REPO": params.system_test_repo,
+        "SYSTEMTESTS_GIT_BRANCH": params.system_test_branch,
+        "PACKER_MAAS_GIT_REPO": params.packer_naas_repo,
+        "PACKER_MAAS_GIT_BRANCH": params.packer_maas_branch,
+    }
+    return request_build(params, job_params)
+
+
[e-mail]
+async def fetch_image_mapping(
+    params: image_building_param,
+) -> dict[str, dict[str, Any]]:
+    with open(params.image_mapping, "r") as fh:
+        image_cfg: dict[str, Any] = yaml.safe_load(fh)
+    return image_cfg
+
+
[e-mail]
+async def fetch_image_built_status(params: dict[str, Any]) -> dict[str, bool]:
+    results: dict[str, dict[str, str]] = params["results"]
+    mapping: dict[str, dict[str, Any]] = params["mapping"]
+    image_built_results: dict[str, bool] = {}
+
+    for image in params["image"]:
+        this_image = mapping["images"].get(image, {})
+        oseries = this_image.get("oseries")
+        osystem = mapping["images"].get(image, {}).get("osystem")
+        image_name = f"{osystem}/{oseries}"
+        status = False
+        for test_name, test_result in results.items():
+            if re.search(rf"test_build_image.*{image_name}", test_name):
+                if test_result["status"] in ["FIXED", "PASSED"]:
+                    status = True
+                    break
+        image_built_results[image] = status
+    return image_built_results
+
+
[e-mail]
+class image_building_workflow:
+    @workflow.run
+    async def run(
+        self, params: image_building_param
+    ) -> dict[str, int | dict[str, bool]]:
+        # await an open connection to the server
+        await workflow.execute_activity(
+            check_jenkins_reachable,
+            params,
+            start_to_close_timeout=params.gettimeout("jenkins_login_timeout"),
+        )
+        # only attempt to build the image once
+        params.build_num = await workflow.execute_activity(
+            request_images_built,
+            params,
+            start_to_close_timeout=params.gettimeout("request_build_timeout"),
+        )
+        # try multiple times to get the results or status
+        await workflow.execute_activity(
+            await_build_exists,
+            params,
+            start_to_close_timeout=params.gettimeout("request_build_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        await workflow.execute_activity(
+            await_build_complete,
+            params,
+            start_to_close_timeout=params.gettimeout("build_complete_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        # return a default failure state if the build was aborted
+        build_status = await workflow.execute_activity(
+            fetch_build_status,
+            params,
+            start_to_close_timeout=params.gettimeout("build_complete_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        image_results: dict[str, bool] = {k: False for k in aslist(params.image_name)}
+        if build_status.lower() != "aborted":
+            try:
+                # return pass/fail status for image/images being built
+                results = await workflow.execute_activity(
+                    fetch_build_and_result,
+                    params,
+                    start_to_close_timeout=params.gettimeout("get_results_timeout"),
+                    retry_policy=RetryPolicy(
+                        maximum_attempts=params.max_retry_attempts
+                    ),
+                )
+                # these should never require a retry
+                mapping = await workflow.execute_activity(
+                    fetch_image_mapping,
+                    params,
+                    start_to_close_timeout=params.gettimeout(),
+                )
+                image_results = await workflow.execute_activity(
+                    fetch_image_built_status,
+                    {
+                        "results": results,
+                        "image": aslist(params.image_name),
+                        "mapping": mapping,
+                    },
+                    start_to_close_timeout=params.gettimeout("return_status_timeout"),
+                )
+            except Exception as e:
+                workflow.logger.exception(e)
+        return {
+            "build_num": params.build_num,
+            "image_results": image_results,
+        }
+
+
+activities = [
+    check_jenkins_reachable,
+    await_build_exists,
+    await_build_complete,
+    request_images_built,
+    fetch_build_status,
+    fetch_build_and_result,
+    fetch_image_mapping,
+    fetch_image_built_status,
+]
+workflows = [image_building_workflow]
diff --git a/temporal/image_reporting_worker.py b/temporal/image_reporting_worker.py
new file mode 100644
index 0000000..bd1f08b
--- /dev/null
+++ b/temporal/image_reporting_worker.py
@@ -0,0 +1,10 @@
+from common_tasks import start_worker
+from image_reporting_workflow import activities as image_reporting_activities
+from image_reporting_workflow import workflows as image_reporting_workflows
+
+if __name__ == "__main__":
+    start_worker(
+        task_queue="image_reporting",
+        workflows=image_reporting_workflows,
+        activities=image_reporting_activities,
+    )
diff --git a/temporal/image_reporting_workflow.py b/temporal/image_reporting_workflow.py
new file mode 100644
index 0000000..05d5b63
--- /dev/null
+++ b/temporal/image_reporting_workflow.py
@@ -0,0 +1,450 @@
+import copy
+import os
+import re
+from dataclasses import dataclass
+from typing import Any
+
+import yaml
+from build_results import (
+    FeatureStatus,
+    ImageTestResults,
+    TestStatus,
+    checkout_and_commit,
+    determine_feature_state,
+    execute,
+)
+from common_tasks import (
+    check_jenkins_reachable,
+    get_build,
+    get_config,
+    get_logs,
+    get_results,
+    workflow_parameters,
+)
+from temporalio import activity, workflow
+from temporalio.common import RetryPolicy
+
+STEPS_TO_PARSE = ["deploy", "test_image"]
+
+
+@dataclass
+class image_reporting_param(workflow_parameters):
+    image_details: None | dict[str, Any] = None
+
+    job_name: str = "maas-automated-image-tester"
+
+    repo_location: str = "image_results_repo"
+
+    maas_snap_channel: str = "latest/edge"
+
+    overwrite_results: bool = False
+
+
+@dataclass
+class Filtered_Results:
+    # image: arch: step: data
+    data: dict[str, Any] = {}
+
+    def _add_image_(self, image: str) -> None:
+        if image not in self.data:
+            self.data[image] = {}
+        if "state" not in self.data[image]:
+            self.data[image]["state"] = TestStatus()
+
+    def add_result(
+        self, image: str, arch: str, step: str, data: dict[str, Any], status: TestStatus
+    ) -> None:
+        self._add_image_(image)
+        if arch not in self.data[image]:
+            self.data[image][arch] = {}
+        self.data[image][arch][step] = data
+        self.data[image]["state"] += status
+
+    def to_dict(self) -> dict[str, Any]:
+        data = copy.deepcopy(self.data)
+        # convert statuses to dicts
+        for image, image_data in data.items():
+            status: TestStatus = image_data["state"]
+            data[image]["state"] = status.to_dict()
+        # return
+        return data
+
+
+def image_from_osytem_oseries(
+    params: image_reporting_param,
+    osystem: str,
+    oseries: str,
+    job_name: str | None = None,
+    build_num: str | int | None = None,
+) -> str:
+    cfg = get_config(
+        params, job_name=job_name, build_num=int(build_num) if build_num else None
+    )
+    images = cfg.get("image-tests", {})
+    return [
+        str(k)
+        for k, v in images.items()
+        if v["osystem"] == osystem and v["oseries"] == oseries
+    ][0]
+
+
[e-mail]
+async def get_test_numbers(params: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    parameters = image_reporting_param(**params["params"])
+    image_details: dict[str, Any] = params["image_details"]
+
+    test_details: dict[str, dict[str, str | bool]] = {}
+    test_numbers = list(set(details["test_num"] for details in image_details.values()))
+    for test_num in test_numbers:
+        if test_num:
+            this_test = get_build(parameters, build_num=int(test_num))
+            test_details[str(test_num)] = {
+                "status": str(this_test.get_status()),
+                "has_results": bool(this_test.has_resultset()),
+            }
+    return test_details
+
+
[e-mail]
+async def fetch_maas_version_from_logs(
+    params: dict[str, Any],
+) -> dict[str, dict[str, str]]:
+    """MAAS version from a test log: ie: ["3.5","3.5.0~alpha1-14542-g.6d2c926d8"]"""
+    parameters = image_reporting_param(**params["params"])
+    tests: list[str] = params["tests"]
+
+    maas_snap_info = str(execute(["snap", "info", "maas"]).stdout)
+    long_version, short_version = ("", "")
+    if search := re.search(
+        rf"{parameters.maas_snap_channel}\:\s+((\d+\.\d+)\.\d+[^\s]+)", maas_snap_info
+    ):
+        long_version, short_version = search.groups()
+
+    versions: dict[str, dict[str, str]] = {
+        "None": {"short": short_version, "long": long_version},
+    }
+    for test in tests:
+        test_logs = get_logs(parameters, build_num=int(test))
+        log = [v for k, v in test_logs.items() if k == "env_builder"][0]
+        if search := re.search(
+            r"maas\-client\: \|maas\s+((\d+\.\d+)\.\d+[^\s]+).*canonical\*", log
+        ):
+            long_version, short_version = search.groups()
+            versions[test] = {"short": short_version, "long": long_version}
+            continue
+        raise Exception("Cannot determine MAAS version.")
+    return versions
+
+
[e-mail]
+async def filter_test_results(params: dict[str, Any]) -> dict[str, Any]:
+    parameters = image_reporting_param(**params["params"])
+    test_num: str = params["test_num"]
+    filtered_result = Filtered_Results()
+    log = (
+        get_logs(parameters, build_num=int(test_num))
+        .get("tests_per_machine", "")
+        .split("\n")
+    )
+    results = get_results(parameters, build_num=int(test_num))
+    for test_name, test_result in results.items():
+        if "test_full_circle" not in test_name:
+            continue
+        if search := re.search(r"\[(.*)\.(.*)\-(.*)\/(.*)\-(.*)\]", test_name):
+            machine, arch, osystem, oseries, step = search.groups()
+            if step.lower() not in STEPS_TO_PARSE:
+                continue
+
+            image = image_from_osytem_oseries(
+                parameters, osystem, oseries, build_num=int(test_num)
+            )
+
+            this_status = TestStatus(test_result["status"])
+            this_result = {
+                "result": test_result,
+                "state": this_status.to_dict(),
+                "error": test_result["errorDetails"],
+                "error_trace": test_result["errorStackTrace"],
+                "log": [line for line in log if test_result["name"] in line],
+            }
+            filtered_result.add_result(image, arch, step, this_result, this_status)
+    # pack the results status so it is serialisable
+    return filtered_result.to_dict()
+
+
[e-mail]
+async def parse_test_results(params: dict[str, Any]) -> dict[str, Any]:
+    maas_version: str = params["maas_version"]
+    image_details: dict[str, Any] = params["image_details"]
+    filtered_results: dict[str, Any] = params["results"]
+    results: dict[str, Any] = {}
+
+    def get_step_from_results(
+        image_results: dict[str, Any], step: str
+    ) -> dict[str, Any]:
+        arches = set(image_results.keys()) - {"state"}
+        return {
+            arch: image_results[arch].get(step)
+            for arch in arches
+            if step in image_results[arch]
+        }
+
+    for image, this_image_result in filtered_results.items():
+        this_image_details: dict[str, Any] = image_details[image]
+        packer_version: str = this_image_details["packer_version"]
+        prereq: list[str] = this_image_details["prerequisites"]
+        arches = set(this_image_result.keys()) - {"state"}
+        image_results = ImageTestResults(
+            image=image,
+            maas_version=[maas_version],
+            readable_state=this_image_result["state"]["state"],
+            tested_arches=list(arches),
+            packer_version=[packer_version],
+            prerequisites=prereq,
+        )
+
+        # check for the deployment state
+        if deployed := get_step_from_results(this_image_result, "deploy"):
+            # Image deployment
+            if deploy_state := sum(
+                TestStatus(**arch["state"]) for arch in deployed.values()
+            ):
+                deployable = FeatureStatus(
+                    name="Deployable",
+                    state=deploy_state._is_positive_,
+                    readable_state=deploy_state._state_,
+                    info="All machines deployed"
+                    if deploy_state._is_positive_
+                    else "; ".join(
+                        f"{name}:{arch['error']}"
+                        for name, arch in deployed.items()
+                        if arch["error"]
+                    ),
+                )
+                image_results.deployable = deployable  # type: ignore[attr-defined]
+        # check to see if we did any tests of the image after it deployed
+        if image_tests := get_step_from_results(this_image_result, "test_image"):
+            # storage configuration
+            if storage_state := determine_feature_state("storage layout", image_tests):
+                state, readable, info = storage_state
+                storage_conf = FeatureStatus(
+                    "Storage Configuration",
+                    state=state,
+                    readable_state=readable,
+                    info=info,
+                )
+                image_results.storage_conf = storage_conf  # type:ignore[attr-defined]
+            # network configuration
+            if network_state := determine_feature_state("network layout", image_tests):
+                state, readable, info = network_state
+                net_conf = FeatureStatus(
+                    "Network Configuration",
+                    state=state,
+                    readable_state=readable,
+                    info=info,
+                )
+                image_results.net_conf = net_conf  # type:ignore[attr-defined]
+        # add to image results list
+        results |= image_results.to_dict()
+    return results
+
+
[e-mail]
+async def parse_failed_images(params: dict[str, Any]) -> dict[str, Any]:
+    maas_version: dict[str, dict[str, str]] = params["maas_version"]
+    image_details: dict[str, Any] = params["image_details"]
+    passed_images: list[str] = params["passed_images"]
+    results: dict[str, Any] = {}
+
+    default_maas_version = maas_version["None"]
+
+    # report on images that failed one of the steps
+    for image, details in image_details.items():
+        # don't report on images we've already recovered test statuses for
+        if image in passed_images:
+            continue
+
+        test_num = str(details["test_num"])
+
+        readable_state = "Unkown Error"
+        if not details["built"]:
+            readable_state = "Could not build image"
+        elif not details["tested"]:
+            readable_state = "Could not test image"
+        results |= ImageTestResults(
+            image=image,
+            maas_version=[maas_version.get(test_num, default_maas_version)["short"]],
+            readable_state=readable_state,
+            packer_version=[details["packer_version"]],
+            prerequisites=details["prerequisites"],
+        ).to_dict()
+    return results
+
+
[e-mail]
+async def post_test_results(params: dict[str, Any]) -> None:
+    image_results: dict[str, Any] = params["image_results"]
+    maas_version: dict[str, dict[str, str]] = params["maas_version"]
+    repo_location: str = params["repo_location"]
+    image_details: dict[str, Any] = params["image_details"]
+    overwrite_results: bool = params["overwrite_results"]
+    # clone the results repo
+    if not os.path.exists(repo_location):
+        execute(
+            [
+                "git",
+                "clone",
+                "https://github.com/maas/MAAS-Image-Results",
+                repo_location,
+            ]
+        )
+
+    # read the combined results
+    combined_results: dict[str, dict[str, Any]] = {"images": {}}
+    combined_results_path = f"{repo_location}/image_results.yaml"
+    with open(combined_results_path, "r") as result_file:
+        if old_results := yaml.safe_load(result_file):
+            combined_results = old_results
+
+    test_nums = set()
+    # write the results for each image
+    for image, image_results in params["image_results"].items():
+        this_result_path = f"{repo_location}/{image}.yaml"
+        results: ImageTestResults = ImageTestResults().from_dict({image: image_results})
+        details: dict[str, Any] = image_details[image]
+        this_test_num: str = str(details["test_num"])
+        default_maas_version = maas_version["None"]
+        this_maas_version: str = maas_version.get(this_test_num, default_maas_version)[
+            "long"
+        ]
+        test_nums.add(int(this_test_num))
+
+        with checkout_and_commit(
+            branch=image,
+            commit_message=f"{image} results: {this_maas_version} - {this_test_num}",
+            add_file=this_result_path,
+            cwd=repo_location,
+        ):
+            if os.path.exists(this_result_path) and not overwrite_results:
+                with open(this_result_path, "r") as result_file:
+                    if old_results := yaml.safe_load(result_file):
+                        results += ImageTestResults().from_dict(old_results)
+
+            if combined_results["images"]:
+                combined_results["images"] |= results.to_dict()
+            else:
+                combined_results["images"] = results.to_dict()
+
+            with open(this_result_path, "w") as result_file:
+                yaml.safe_dump(results.to_dict(), result_file)
+
+    tested_builds = (
+        f"{min(test_nums)} - {max(test_nums)}" if len(test_nums) > 1 else f"{test_nums}"
+    )
+
+    # write the combined results to main
+    with checkout_and_commit(
+        branch="main",
+        commit_message=f"Combined results: {tested_builds}",
+        add_file=combined_results_path,
+        cwd=repo_location,
+    ), open(combined_results_path, "w") as result_file:
+        yaml.safe_dump(combined_results, result_file)
+
+
[e-mail]
+class image_reporting_workflow:
+    @workflow.run
+    async def run(self, params: image_reporting_param) -> None:
+        if not params.image_details:
+            raise Exception("No Image details provided")
+        # await an open connection to the server
+        await workflow.execute_activity(
+            check_jenkins_reachable,
+            params,
+            start_to_close_timeout=params.gettimeout("jenkins_login_timeout"),
+        )
+        test_numbers = await workflow.execute_activity(
+            get_test_numbers,
+            {
+                "image_details": params.image_details,
+                "params": params,
+            },
+            start_to_close_timeout=params.gettimeout("log_details_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        maas_versions = await workflow.execute_activity(
+            fetch_maas_version_from_logs,
+            {
+                "params": params,
+                "tests": list(test_numbers.keys()),
+            },
+            start_to_close_timeout=params.gettimeout("log_details_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        results_to_report: dict[str, Any] = {}
+        for test_num, test_details in test_numbers.items():
+            # if the tests completed and results are available.
+            if (
+                test_details["status"].lower() != "aborted"
+                and test_details["has_results"]
+            ):
+                results = await workflow.execute_activity(
+                    filter_test_results,
+                    {"params": params, "test_num": test_num},
+                    start_to_close_timeout=params.gettimeout("fetch_results_timeout"),
+                    retry_policy=RetryPolicy(
+                        maximum_attempts=params.max_retry_attempts
+                    ),
+                )
+                default_maas_version = maas_versions["None"]
+                results_to_report |= await workflow.execute_activity(
+                    parse_test_results,
+                    {
+                        "maas_version": maas_versions.get(
+                            test_num, default_maas_version
+                        )["short"],
+                        "image_details": params.image_details,
+                        "results": results,
+                    },
+                    start_to_close_timeout=params.gettimeout("fetch_results_timeout"),
+                    retry_policy=RetryPolicy(
+                        maximum_attempts=params.max_retry_attempts
+                    ),
+                )
+        # add any images that didn't test
+        results_to_report |= await workflow.execute_activity(
+            parse_failed_images,
+            {
+                "image_details": params.image_details,
+                "maas_version": maas_versions,
+                "passed_images": list(results_to_report.keys()),
+            },
+            start_to_close_timeout=params.gettimeout("fetch_results_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        # only try to upload once.
+        await workflow.execute_activity(
+            post_test_results,
+            {
+                "image_results": results_to_report,
+                "maas_version": maas_versions,
+                "repo_location": params.repo_location,
+                "image_details": params.image_details,
+                "overwrite_results": params.overwrite_results,
+            },
+            start_to_close_timeout=params.gettimeout("fetch_results_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=1),
+        )
+
+
+activities = [
+    check_jenkins_reachable,
+    get_test_numbers,
+    fetch_maas_version_from_logs,
+    filter_test_results,
+    parse_test_results,
+    parse_failed_images,
+    post_test_results,
+]
+workflows = [image_reporting_workflow]
diff --git a/temporal/image_testing_worker.py b/temporal/image_testing_worker.py
new file mode 100644
index 0000000..f28bb23
--- /dev/null
+++ b/temporal/image_testing_worker.py
@@ -0,0 +1,10 @@
+from common_tasks import start_worker
+from image_testing_workflow import activities as image_test_activities
+from image_testing_workflow import workflows as image_test_workflows
+
+if __name__ == "__main__":
+    start_worker(
+        task_queue="image_testing",
+        workflows=image_test_workflows,
+        activities=image_test_activities,
+    )
diff --git a/temporal/image_testing_workflow.py b/temporal/image_testing_workflow.py
new file mode 100644
index 0000000..a587b4b
--- /dev/null
+++ b/temporal/image_testing_workflow.py
@@ -0,0 +1,100 @@
+from dataclasses import dataclass
+from typing import Any
+
+from common_tasks import (
+    aslist,
+    await_build_complete,
+    await_build_exists,
+    check_jenkins_reachable,
+    fetch_build_status,
+    request_build,
+    workflow_parameters,
+)
+from temporalio import activity, workflow
+from temporalio.common import RetryPolicy
+
+
+@dataclass
+class image_testing_param(workflow_parameters):
+    image_name: str | list[str] = ""  # allow builk image building if desired
+
+    job_name: str = (
+        "maas-automated-image-tester"  # Need to check which job actually does this
+    )
+    build_num: int = -1
+
+    # job details with default values we may want to change
+    system_test_repo: str = (
+        "https://git.launchpad.net/~maas-committers/maas-ci/+git/system-tests"
+    )
+    system_test_branch: str = "master"
+
+    maas_snap_channel: str = "latest/edge"
+
+    parallel_tests: bool = False
+
+
[e-mail]
+async def request_images_test(params: image_testing_param) -> int:
+    """Start an image testing job, returning the job number."""
+    job_params: dict[str, Any] = {
+        "IMAGE_NAMES": ",".join(image for image in aslist(params.image_name)),
+        "SYSTEMTESTS_GIT_REPO": params.system_test_repo,
+        "SYSTEMTESTS_GIT_BRANCH": params.system_test_branch,
+        "MAAS_SNAP_CHANNEL": params.maas_snap_channel,
+    }
+    return request_build(params, job_params)
+
+
[e-mail]
+class image_testing_workflow:
+    @workflow.run
+    async def run(self, params: image_testing_param) -> dict[str, Any]:
+        # await an open connection to the server
+        await workflow.execute_activity(
+            check_jenkins_reachable,
+            params,
+            start_to_close_timeout=params.gettimeout("jenkins_login_timeout"),
+        )
+        # test the image, only trigger once
+        params.build_num = await workflow.execute_activity(
+            request_images_test,
+            params,
+            start_to_close_timeout=params.gettimeout("request_build_timeout"),
+        )
+        # try multiple times to get the results or status
+        await workflow.execute_activity(
+            await_build_exists,
+            params,
+            start_to_close_timeout=params.gettimeout("request_build_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        await workflow.execute_activity(
+            await_build_complete,
+            params,
+            start_to_close_timeout=params.gettimeout("build_complete_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+        # return a default failure state if the build was aborted
+        build_status = await workflow.execute_activity(
+            fetch_build_status,
+            params,
+            start_to_close_timeout=params.gettimeout("build_complete_timeout"),
+            retry_policy=RetryPolicy(maximum_attempts=params.max_retry_attempts),
+        )
+
+        # return the image details in the correct format
+        return {
+            image: {"build_num": params.build_num, "build_status": build_status}
+            for image in aslist(params.image_name)
+        }
+
+
+activities = [
+    check_jenkins_reachable,
+    request_images_test,
+    await_build_exists,
+    await_build_complete,
+    fetch_build_status,
+]
+workflows = [image_testing_workflow]
diff --git a/tox.ini b/tox.ini
index e4efc18..f347a7b 100644
--- a/tox.ini
+++ b/tox.ini
@@ -66,8 +66,8 @@ description=Reformat Python code and README.md
 deps= -rrequirements.txt
 skip_install = true
 commands=
-  isort --profile black systemtests utils
-  black systemtests utils
+  isort --profile black systemtests utils temporal
+  black systemtests utils temporal
   cog -r README.md
 
 [testenv:lint]
@@ -76,10 +76,10 @@ deps= -rrequirements.txt
 allowlist_externals=sh
 skip_install = true
 commands=
-  isort --profile black --check-only systemtests utils
-  black --check systemtests utils
+  isort --profile black --check-only systemtests utils temporal
+  black --check systemtests utils temporal
   cog --verbosity=0 --check README.md
-  flake8 systemtests utils
+  flake8 systemtests utils temporal
   sh -c 'git ls-files \*.yaml\* | xargs -r yamllint'
 
 [testenv:mypy]
@@ -95,6 +95,7 @@ deps=
   types-netaddr
 commands=
   mypy -p systemtests -p utils --install-types
+  mypy temporal
 
 [testenv:generate_config]
 description=Generate config.yaml
diff --git a/utils/gen_config.py b/utils/gen_config.py
index 3a1a4cd..4ea3e5e 100755
--- a/utils/gen_config.py
+++ b/utils/gen_config.py
@@ -144,10 +144,14 @@ def main(argv: list[str]) -> int:
     packer_group.add_argument(
         "--packer-repo",
         type=str,
+        metavar="REPOS",
         help="Which git repository to use to get Packer from",
     )
     packer_group.add_argument(
-        "--packer-branch", type=str, help="Which git branch use to get Packer"
+        "--packer-branch",
+        type=str,
+        metavar="BRANCH",
+        help="Which git branch use to get Packer",
     )
     packer_group.add_argument(
         "--packer-container-image",
@@ -318,7 +322,7 @@ def main(argv: list[str]) -> int:
             # if running custom image tests, only use compatible machines
             target_arches = (
                 args.architecture
-                if not args.image_tests
+                if "image-tests" not in config
                 else [image["architecture"] for image in config["image-tests"].values()]
             )
             # Filter out machines with architectures not matching specified ones.
@@ -333,12 +337,12 @@ def main(argv: list[str]) -> int:
             machines["hardware"] = {
                 name: details
                 for name, details in hardware.items()
-                if name not in args.machine
+                if name in args.machine
             }
 
-    if args.vm_machine:
+    if vms:
         # Filter out VMs with name not listed in specified vm_machines
-        if vms:
+        if args.vm_machine:
             vms["instances"] = {
                 vm_name: vm_config
                 for vm_name, vm_config in vms["instances"].items()