From 2134a097d955ff22a8a3218187cc54d46fb8ae7c Mon Sep 17 00:00:00 2001
From: Dylan Carroll <dylan.carroll@candelatech.com>
Date: Mon, 27 Jul 2026 15:19:05 -0700
Subject: [PATCH] http_regression_test: Added initial framework for http
 regression testing

Added the directory py-scripts/http_regression_test. This contains a
framework for running regression tests against the JSON HTTP API. Results
are evaluated against a known baseline using a per-endpoint comparator
class.

main.py:
    Contains the main request/response, saving/loading, and logging logic

comparators.py:
    Contains all of the logic for comparing the content of responses
    for correctness.
    So far, only the comparator for the resource/card endpoint has an
    implementation.

init_wrapper.py:
    A utility for executing initialization scripts before performing the
    given suite of requests.
---
 .../http_regression_test/comparators.py       | 678 ++++++++++++++++++
 .../http_regression_test/init_wrapper.py      |  90 +++
 py-scripts/http_regression_test/main.py       | 448 ++++++++++++
 3 files changed, 1216 insertions(+)
 create mode 100644 py-scripts/http_regression_test/comparators.py
 create mode 100644 py-scripts/http_regression_test/init_wrapper.py
 create mode 100644 py-scripts/http_regression_test/main.py

diff --git a/py-scripts/http_regression_test/comparators.py b/py-scripts/http_regression_test/comparators.py
new file mode 100644
index 000000000..c66977c29
--- /dev/null
+++ b/py-scripts/http_regression_test/comparators.py
@@ -0,0 +1,678 @@
+#!/usr/bin/env python3
+
+"""
+NAME:       http_regression_test/comparators.py
+
+PURPOSE:    A collection of comparators extended from a base endpoint Comparator class. Each comparator is
+            responsible for comparing the results for a given http endpoint.
+
+            Each comparator class must register which endpoint(s) it is capable of comparing, which allows
+            responses from a given test iteration to be dispatched to the correct comparator.
+"""
+
+import traceback
+from abc import ABC, abstractmethod
+
+from typing import TYPE_CHECKING, List, Optional, Union
+if TYPE_CHECKING:
+    from http_regression import Response
+
+
+class Result():
+    def __init__(self, messages: Optional[List[str]] = None):
+        self._messages = messages or []
+        self.key = -1
+
+    def __or__(self, other: 'Result') -> 'Result':
+        result_type: type = type(self) if (self.key > other.key) else type(other)
+        new_result: Result = result_type()
+
+        new_result._messages.extend(self._messages)
+        new_result._messages.extend(other._messages)
+
+        return new_result
+
+    @property
+    def message(self):
+        return "\n".join(self._messages)
+
+
+class Success(Result):
+    def __init__(self):
+        super().__init__()
+        self.key = 0
+
+
+class Warning(Result):
+    def __init__(self, message: Optional[str] = None):
+        super().__init__()
+        self.key = 1
+
+        if message is not None:
+            self._messages.append(f"(warning) {message}")
+
+
+class Failure(Result):
+    def __init__(self, message: Optional[str] = None):
+        super().__init__()
+        self.key = 1
+
+        if message is not None:
+            self._messages.append(f"(failure) {message}")
+
+#
+# Comparator Base Class
+#
+
+
+class Comparator(ABC):
+    """
+    An abstract class containing a compare() method, which compares results
+    for a particular REST endpoint.
+    """
+
+    def __init__(self, host: str, response_ver: str, baseline_ver: str):
+        self.host = host
+        self.response_ver = response_ver
+        self.baseline_ver = baseline_ver
+
+    comparator_mappings: Optional[dict] = None
+
+    @classmethod
+    def fetch_comparator_cls(cls, endpoint: str) -> type:
+        if cls.comparator_mappings is None:
+            cls._populate_comparator_mappings()
+
+        return cls.comparator_mappings[endpoint]
+
+    @classmethod
+    def _populate_comparator_mappings(cls) -> None:
+        cls.comparator_mappings = {}
+        for comparator_cls in Comparator.__subclasses__():
+            endpoints: List[str] = comparator_cls.handles_endpoints()
+
+            duplicate_keys = [k for k in endpoints if k in cls.comparator_mappings]
+            if len(duplicate_keys) > 0:
+                msg = f"Attempted to register {duplicate_keys[0]} for {comparator_cls.__name__}, " + \
+                      f"which is already registered for {cls.comparator_mappings[duplicate_keys[0]].__name__}" + \
+                      (f" (and {len(duplicate_keys) - 1} more conflicts)" if len(duplicate_keys) > 1 else "")
+                raise ValueError(msg)
+
+            cls.comparator_mappings.update({
+                endpoint: comparator_cls for endpoint in endpoints
+            })
+
+    @staticmethod
+    @abstractmethod
+    def handles_endpoints() -> List[str]:
+        """Returns a list of endpoints that the current comparator class is
+        capable of handling. Used for dispatching comparison tasks to the
+        correct Comparator class"""
+
+        pass
+
+    def compare(self, response: 'Response', reference: 'Response') -> Result:
+        try:
+            return self._compare(response, reference)
+        except Exception:
+            return Failure(f"Exception in {type(self).__name__}: {traceback.format_exc()}")
+
+    @abstractmethod
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        """Compare the given response to the given reference baseline response.
+        Both response and reference are dictionaries of json-structured date
+        representing the response of a LANforge REST AI endpoint to a single
+        query. """
+        pass
+
+
+#
+# Helpers
+#
+def zip_json(a_val, b_val) -> Union[tuple, dict, list]:
+    """
+    Given two json-structured objects (list or dict), fold them together such
+    that every leaf-element with matching keys/indices is a tuple of (a_value, b_value).
+    Values that are dict or list will be recursively zipped together.
+    """
+    if isinstance(a_val, list) and isinstance(b_val, list):
+        value = _zip_list(a_val, b_val)
+
+    elif isinstance(a_val, dict) and isinstance(b_val, dict):
+        value = _zip_dict(a_val, b_val)
+
+    else:
+        value = (a_val, b_val)
+
+    return value
+
+
+def _zip_list(a: list, b: list) -> list:
+    """Helper for zip_json that handles list elements"""
+
+    if len(a) < len(b):
+        a += [None] - (len(b) - len(a))
+    if len(b) < len(b):
+        b += [None] - (len(a) - len(b))
+
+    return [
+        zip_json(a_val, b_val)
+        for a_val, b_val in zip(a, b)
+    ]
+
+
+def _zip_dict(a: dict, b: dict) -> dict:
+    keys = set(a.keys() | b.keys())
+
+    return {
+        k: zip_json(a.get(k, None), b.get(k, None))
+        for k in keys
+    }
+
+
+def check_tolerance(value1, value2, tol=0.2) -> bool:
+    """Check if value 1 is within tol*value2 of value2"""
+    return value2*(1-tol) <= value1 <= value2*(1+tol)
+
+
+def mismatch_message(name, value1, value2):
+    return f"Value of '{name}' ({value1}) does not match expected value ({value2})."
+
+
+def tolerance_message(name, value1, value2):
+    return f"Value of '{name}' ({value1}) is not within the expected tolerances of {value2}."
+
+
+def mismatch_type_message(name, value1, value2):
+    return f"Type of '{name}' ({value1}: {type(value1)}) does not match expected type ({value2}: {type(value2)})."
+
+
+#
+# Concrete Comparator Implementations
+#
+
+# The NotImplemented classes that
+
+class DatabaseComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["database", "databases", "db"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class MloComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["mlo"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class RfgenComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["rfgen"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class AdbComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["adb"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class AdvancedPortStatsComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["wifi-stats"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class PortProbeComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["portprobe", "probe"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class AttenuatorComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["atten", "attenuator", "attenuators"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class CxGroupsComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["cxg", "cx-group", "cxgroup", "cx-groups", "cxgroups", "test-group",
+                "testgroup", "test-groups", "testgroups", "tg"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class VrComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["virtual-routers", "virtualrouters", "vr"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class VrCxComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["vr-cx", "vrcx"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class VoipComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["voip"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class VoipEndpComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["voip-endp"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class StationScanComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["scan", "scan-results", "scanresults"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class FileIOComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["fileio"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class StationsComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["stations"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class GuiCliComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["gui-cli", "gui_cli", "gui-cli", "gui-cmd", "gui_cmd", "gui-json", "gui_json"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class ChamberComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["chamber", "chambers"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class WlEndpComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["wl-endp", "wl_endp", "wlendp", "wl-ep", "wl_ep"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class WlComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["wl"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class ArmageddonComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["arm", "arm", "arm-endp", "arm-endp"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class GenericEndpComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["generic"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class DUTComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["dut"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class ProfilesComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["profile", "profiles"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class L4Comparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["layer4"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class RadioReportComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["radiostatus"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class CxComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["cx"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class EventsComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["alerts", "events"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class WifiMsgsComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["wifi-msgs"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class TextBlobsComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["text"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class PortComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["port", "ports"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class EndpComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["endp"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class ResourceComparator(Comparator):
+
+    all_known_keys = [
+        "app-id", "bps-rx-3s", "bps-tx-3s", "build date", "cli-port", "cpu", "ct-kernel", "ctrl-ip",
+        "ctrl-port", "device type", "df-boot", "df-home", "df-root", "eid", "entity id", "free mem",
+        "free swap", "gps", "hostname", "hw version", "kernel", "load", "max if-up", "max staged",
+        "mem", "phantom", "ports", "rf-path", "rx bytes", "shelf", "sta up", "swap", "sw version",
+        "tx bytes", "user", "_links",
+    ]
+
+    exact_match_keys = [
+        "app-id", "cli-port", "cpu", "ct-kernel", "ctrl-ip", "ctrl-port", "device type", "eid", "entity id",
+        "gps", "hostname", "hw version", "max if-up", "max staged", "phantom", "ports", "rf-path", "shelf",
+        "user", "_links",
+    ]
+
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["shelf", "resource"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        result = Success()
+
+        if response is None:
+            if reference is not None:
+                return Failure("Response has no value")
+            else:
+                return Success()
+
+        if response.status != reference.status:
+            result |= Failure(mismatch_message("status code", response.status, reference.status))
+
+        if response.content is None:
+            if reference.content is not None:
+                return Failure("Response missing content.")
+            else:
+                return result
+
+        # Some resource requests redirect to ports
+        if "HttpPort" in response.content["handler"]:
+            port_comparator = PortComparator(self.host, self.response_ver, self.baseline_ver)
+            return port_comparator.compare(response, reference)
+
+        if response.status != reference.status:
+            result |= Failure(mismatch_message("status code", response.status, reference.status))
+
+        resources = zip_json(
+            self.get_resource_list(response.content),
+            self.get_resource_list(reference.content)
+        )
+        for resource in resources:
+            new_result = self._compare_single_resource(resource)
+            result |= new_result
+
+        return result
+
+    def get_resource_list(self, content: dict):
+        if "resources" in content.keys():
+            return content["resources"]
+        else:
+            fields = content["resource"]
+            return [{"resource": fields}]
+
+    def _compare_single_resource(self, resource: dict) -> Result:
+        result = Success()
+
+        zipped_content = list(resource.values())[0]
+        for key, (response_val, reference_val) in zipped_content.items():
+            if response_val is None is not reference_val:
+                # Missing value
+                result |= Failure(f"'{key}' is missing from the reference.")
+
+            elif response_val is not None is reference_val:
+                # Extra value
+                result |= Warning(f"(warning) '{key}' is present in the response but not the reference.")
+
+            elif key in self.exact_match_keys:
+                # Exact match
+                if response_val != reference_val:
+                    result |= Failure(mismatch_message(key, response_val, reference_val))
+
+            elif type(response_val) is not type(reference_val):
+                result |= Failure(mismatch_type_message(key, response_val, reference_val))
+
+        if key not in self.all_known_keys:
+            result |= Warning(f"(warning) '{key}' is a new field unrecognized by the comparator.")
+
+        return result
+
+
+class CliComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["cli"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class CliFormComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["cli-form", "cli-form"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class CliJsonComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["cli-json"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class HelpComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["help"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class StatusDataComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["status-msg"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class WebSocketMessageComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["ws-msg"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class MiscComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["misc"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+class ControlComparator(Comparator):
+    @staticmethod
+    def handles_endpoints() -> List[str]:
+        return ["newsession", "endsession", "quit", "control"]
+
+    def _compare(self, response: 'Response', reference: 'Response') -> Result:
+        # TODO: Implement
+        raise NotImplementedError()
+
+
+# Called after all subclasses are defined
+Comparator._populate_comparator_mappings()
diff --git a/py-scripts/http_regression_test/init_wrapper.py b/py-scripts/http_regression_test/init_wrapper.py
new file mode 100644
index 000000000..da54e498e
--- /dev/null
+++ b/py-scripts/http_regression_test/init_wrapper.py
@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+
+"""
+NAME:       http_regression_test/init_wrapper.py
+
+PURPOSE:    Establish the preconditions for a the regression test by invoking a given initialization
+            script before invoking http_regression_test/main.py with the supplied arguments.
+
+EXAMPLE:    $ python http_regression_test/init_wrapper.py \
+                --init_script_args python http_regression_ports_init_ct_us_008.py \
+                --regression_test_args baseline 5.5.2 192.168.101.137 --save 5.5.2-baseline.json --get /port/1/all
+"""
+
+import sys
+import subprocess
+import logging
+
+# Some shenannigans to import the logger config from a cousin file
+import importlib
+from os import path
+file_dir = path.dirname(path.realpath(__file__))
+sys.path.insert(0, path.abspath(path.join(file_dir, "..")))
+
+lf_logger_config = importlib.import_module("lf_logger_config")
+logger = logging.getLogger(__name__)
+
+# Get the path to the main test script
+dir_path = path.dirname(path.realpath(__file__))
+main_path = path.join(dir_path, "main.py")
+
+
+def usage(help=False):
+    if help:
+        print("Establish the preconditions for a the regression test by invoking a given initialization\n"
+              "script before invoking http_regression_test/main.py with the supplied arguments.\n")
+
+    print("usage:\n"
+          "    init_wrapper.py --init_script_args exe args... --regression_test_args args...")
+
+    if help:
+        exit(0)
+    else:
+        exit(1)
+
+
+def main():
+    args = sys.argv
+
+    # Find argument indices
+    try:
+        init_index = args.index("--init_script_args")
+    except ValueError:
+        init_index = None
+
+    try:
+        test_index = args.index("--regression_test_args")
+    except ValueError:
+        test_index = None
+
+    # Check for help
+    if any((h in sys.argv) for h in ("-h", "--help", "--help_summary")):
+        if init_index is None:
+            usage(help=True)
+        elif any((h in sys.argv[:init_index]) for h in ("-h", "--help", "--help_summary")):
+            usage(help=True)
+
+    # Check for invalid argument arrangements
+    if init_index is None:
+        print("Missing required argument '--init_script_args'\n")
+        usage()
+
+    elif test_index is None:
+        print("Missing required argument '--regression_test_args'\n")
+        usage()
+
+    elif test_index <= init_index:
+        print("'--init_script_args' must come before '--regression_test_args'\n")
+        usage()
+
+    # Slice arguments out of sys.argv based on the found indices
+    init_args = sys.argv[init_index+1:test_index]
+    test_args = ["python", main_path] + sys.argv[test_index+1:]
+
+    # Execute both scripts with the given arguments
+    subprocess.run(init_args)
+    subprocess.run(test_args)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/py-scripts/http_regression_test/main.py b/py-scripts/http_regression_test/main.py
new file mode 100644
index 000000000..52bf96d30
--- /dev/null
+++ b/py-scripts/http_regression_test/main.py
@@ -0,0 +1,448 @@
+#!/usr/bin/env python3
+"""
+NAME:       http_regression_test/main.py
+
+PURPOSE:    Run repeated sets of queries against various endpoints of the LANforge JSON REST API and
+            compare the results to previous baseline results from known good versions in order to
+            prevent regressive changes to the API.
+
+EXAMPLE:    Create a baseline for a given system on 5.5.2 from a few GET requests to resource endpoint
+            $ python http_regression_test/main.py baseline 5.5.2 192.168.101.137 --save 5.5.2-baseline.json \
+                --get /resource/1/1 --get /resource/1/all --get /resource/1/list --get /resource/1/1?fields=eid
+
+            Evaluate a given system, now on 5.5.3, against the previous 5.5.2 baseline
+            % python http_regression_test/main.py test 5.5.3 --baseline 5.5.2-baseline.json
+
+NOTES:      The code the compares the endpoint responses exists in http_regression_test/comparators.py.
+            The comparator for a given endpoint must be andimplemented in order to receive test
+            results for requests against the corresponding endpoint.
+
+            Some endpoint response comparators expect certain preconditions for evaluating the
+            correctness of response content such as loading databases, running a known traffic
+            profile for a fixed duration, etc. Such preconditions can be established by running an
+            initialization script before the test using http_regression_test/init_wrapper.py.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import requests
+import textwrap
+import importlib
+
+from enum import Enum
+from dataclasses import dataclass
+from typing import Tuple, Optional
+from argparse import ArgumentParser, Namespace
+from comparators import Comparator, Result, Success, Warning, Failure
+
+# Some shenannigans to import the logger config from a cousin file
+import sys
+from os import path
+file_dir = path.dirname(path.realpath(__file__))
+sys.path.insert(0, path.abspath(path.join(file_dir, "..")))
+
+lf_logger_config = importlib.import_module("lf_logger_config")
+logger = logging.getLogger(__name__)
+
+
+class Method(Enum):
+    GET = "GET"
+    POST = "POST"
+    PUT = "PUT"
+    DELETE = "DELETE"
+
+    @staticmethod
+    def from_str(val: str):
+        if val in ["GET", "get"]:
+            return Method.GET
+        elif val in ["POST", "post"]:
+            return Method.POST
+        elif val in ["PUT", "put"]:
+            return Method.PUT
+        elif val in ["DELETE", "delete"]:
+            return Method.DELETE
+
+    def __str__(self):
+        return self.name
+
+
+@dataclass(frozen=True)
+class Response:
+    status: int
+    reason: str
+    content: Optional[dict]
+
+
+@dataclass(frozen=True)
+class Request:
+    uri: str
+    method: Method
+
+    _request_funcs = {
+        Method.GET:    requests.get,
+        Method.POST:   requests.post,
+        Method.PUT:    requests.put,
+        Method.DELETE: requests.delete,
+    }
+
+    def make_request(self, host: str, port: int) -> Optional[Exchange]:
+        try:
+            method_func = self._request_funcs[self.method]
+            requests_response: requests.models.Response = method_func(self.full_url(host, port))
+
+        except requests.exceptions.ConnectionError as e:
+            logger.warning(e)
+            requests_response = None
+
+        try:
+            content = requests_response.json() if requests_response else None
+        except requests.exceptions.JSONDecodeError as e:
+            logger.warning(e)
+            content = None
+
+        if requests_response is None:
+            return None
+        else:
+            response = Response(
+                status=requests_response.status_code,
+                reason=requests_response.reason,
+                content=content,
+            )
+
+            return Exchange(self, response)
+
+    def full_url(self, host: str, port: int):
+        return (
+            f"http://{host}:{port}" +
+            ("/" if self.uri[0] != "/" else "") +
+            self.uri
+        )
+
+    @property
+    def endpoint(self):
+        endp = self.uri.split("/")
+        return endp[0] if endp[0] != "" else endp[1]
+
+
+@dataclass(frozen=True)
+class Exchange:
+    request: Request
+    response: Optional[Response]
+
+    def format(self) -> dict:
+        return {
+            "uri":     self.request.uri,
+            "method":  str(self.request.method),
+            "status":  self.response.status if self.response else None,
+            "reason":  self.response.reason if self.response else None,
+            "content": self.response.content if self.response else None,
+        }
+
+    @staticmethod
+    def parse(data: dict) -> 'Exchange':
+        request = Request(
+            data["uri"],
+            Method.from_str(data["method"]),
+        )
+
+        if data["status"] is None or data["reason"] is None:
+            response = None
+
+        else:
+            response = Response(
+                data["status"],
+                data["reason"],
+                data["content"],
+            )
+
+        return Exchange(request, response)
+
+    def __iter__(self):
+        return iter((self.request, self.response))
+
+
+@dataclass(frozen=True)
+class Baseline:
+    host: str
+    port: int
+    version: str
+    metadata: Optional[dict]
+    exchanges: Tuple[Exchange]
+
+    def format(self: 'Baseline') -> dict:
+        data = {}
+        data["host"] = self.host
+        data["port"] = self.port
+        data["version"] = self.version
+
+        if (self.metadata is not None):
+            data["metadata"] = self.metadata
+
+        data["exchanges"] = list(map(Exchange.format, self.exchanges))
+
+        return data
+
+    @staticmethod
+    def parse(data: dict) -> 'Baseline':
+        return Baseline(
+            host=data["host"],
+            port=data["port"],
+            version=data["version"],
+            metadata=data["metadata"] if ("metadata" in data.keys()) else None,
+            exchanges=tuple(Exchange.parse(e) for e in data["exchanges"])
+        )
+
+
+#
+# baseline subcommand
+#
+
+def run_baseline(args: Namespace):
+    request_list: list[tuple[str, Method]] = (
+        [Request(uri, Method.GET) for uri in args.get] +
+        [Request(uri, Method.POST) for uri in args.post] +
+        [Request(uri, Method.PUT) for uri in args.put] +
+        [Request(uri, Method.DELETE) for uri in args.delete]
+    )
+
+    exchanges: tuple[Exchange] = tuple(
+        request.make_request(args.host, args.port)
+        for request in request_list
+    )
+
+    baseline = Baseline(
+        host=args.host,
+        port=args.port,
+        version=args.version,
+        metadata=args.metadata,
+        exchanges=exchanges,
+    )
+
+    data = baseline.format()
+
+    if args.save:
+        with open(args.save, "w") as f:
+            json.dump(data, f, indent=4)
+    else:
+        json.dump(data, sys.stdout, indent=4)
+
+
+#
+# test subcommand
+#
+
+LOG_WIDTH = 80
+
+
+def pretty_log(msg: str, indent=0, level: str = "info", stacklevel=2):
+    log = {"debug": logger.debug, "info": logger.info, "warning": logger.warning,
+           "error": logger.error, "critical": logger.critical}
+    log = log[level]
+
+    if "\n" in msg:
+        for line in msg.split("\n"):
+            pretty_log(line, indent=indent, level=level, stacklevel=stacklevel+1)
+    else:
+        lines = textwrap.wrap(msg, LOG_WIDTH-(4*indent))
+        for line in lines:
+            def test(s):
+                print(len(s), " : ", s)
+            log(("    "*indent + line).ljust(LOG_WIDTH), stacklevel=stacklevel)
+
+
+def log_exchange_preview(request: Request, base_version: str, version: str):
+    pretty_log("=" * LOG_WIDTH)
+
+    pretty_log(f"Baseline   : {base_version} - {request.method} - {request.uri}")
+    pretty_log(f"Result     : {version} - {request.method} - {request.uri}")
+
+
+def add_newline_indent(s):
+    return s.replace('\n', '\n    ')
+
+
+def log_exchange_response(comparator, base_response: Response, response: Response):
+    pretty_log("Baseline   :", level="debug")
+
+    pretty_log(f"{base_response.status} - {base_response.reason}", indent=1, level="debug")
+    pretty_log(f"{add_newline_indent(json.dumps(base_response.content, indent=4))}", indent=1, level="debug")
+
+    pretty_log("Result     :", level="debug")
+
+    if response is None:
+        pretty_log("None", indent=1, level="debug")
+    else:
+        pretty_log(f"{response.status} - {response.reason}", indent=1, level="debug")
+        pretty_log(f"{add_newline_indent(json.dumps(response.content, indent=4))}", indent=1, level="debug")
+
+    pretty_log(f"Comparator : {type(comparator).__name__}")
+
+
+def log_success():
+    pretty_log("SUCCESS")
+
+
+def log_warning(strict: bool, message: str = ""):
+    if strict:
+        log_failure(message)
+    else:
+        pretty_log("WARNING:")
+        pretty_log(f"{add_newline_indent(message)}", indent=1)
+
+
+def log_failure(message: str = ""):
+    pretty_log("FAILURE:")
+    pretty_log(f"{add_newline_indent(message)}", indent=1)
+
+
+def log_summary(results: dict):
+    pretty_log("=" * 80)
+    pretty_log("Summary:")
+    pretty_log(f"Successes : {results['successes']}", indent=1)
+    pretty_log(f"Warnings  : {results['warnings']}", indent=1)
+    pretty_log(f"Failures  : {results['failures']}", indent=1)
+
+
+def run_test(args: Namespace):
+    if args.baseline:
+        with open(args.baseline, "r") as f:
+            data = json.load(f)
+    else:
+        data = json.load(sys.stdin)
+
+    baseline = Baseline.parse(data)
+
+    results = {
+        "successes": 0,
+        "warnings": 0,
+        "failures": 0,
+    }
+
+    request: Request
+    baseline_response: Response
+    for request, baseline_response in baseline.exchanges:
+        log_exchange_preview(request, baseline.version, args.version)
+
+        new_response: Response = request.make_request(baseline.host, baseline.port).response
+        comparator = Comparator.fetch_comparator_cls(request.endpoint)(
+            baseline.host,
+            baseline.version,
+            args.version
+        )
+        result: Result = comparator.compare(new_response, baseline_response)
+
+        log_exchange_response(comparator, baseline_response, new_response)
+
+        if isinstance(result, Success):
+            log_success()
+            results["successes"] += 1
+        elif isinstance(result, Warning):
+            log_warning(args.strict, result.message)
+            results["warnings"] += 1
+        elif isinstance(result, Failure):
+            log_failure(result.message)
+            results["failures"] += 1
+        else:
+            logger.error(f"Unknown result type: {result}")
+
+    log_summary(results)
+
+#
+# Initialization
+#
+
+
+def parse_args() -> Namespace:
+    parser = ArgumentParser(prog="http_regression_test.py")
+    parser.add_argument('--log_level', default=None, help='Set logging level: debug | info | warning | error | critical')
+    parser.add_argument("--lf_logger_config_json", help="--lf_logger_config_json <json file> , json configuration of logger")
+
+    subparsers = parser.add_subparsers(
+        dest="mode",
+        required=True,
+        metavar="{baseline, test}",
+        help="Required, the mode to run the script in"
+    )
+
+    # baseline subcommand
+    parser_baseline = subparsers.add_parser(
+        "baseline",
+        help="Perform a series of specified HTTP requests against a given host "
+        "and serialize the responses as a JSON baseline reference."
+    )
+    parser_baseline.add_argument("version",
+                                 help="LANforge version of the host")
+
+    parser_baseline.add_argument("host",
+                                 help="Hostname or IP address of the target HTTP API")
+
+    parser_baseline.add_argument("-p", "--port", default=8080,
+                                 help="Port of the target HTTP API. Defaults to 8080 if omitted.")
+
+    parser_baseline.add_argument("-s", "--save", nargs="?", default=None,
+                                 help="Path to save the results to. If absent, "
+                                 "writes the baseline JSON to stdout.")
+
+    parser_baseline.add_argument("-c", "--copy", default=None,
+                                 help="Create a new baseline with the same "
+                                      "requests as a given baseline file.")
+
+    parser_baseline.add_argument("--metadata",
+                                 help="Optional JSON metadata such as comments,"
+                                 "configuration information, or preconditions.")
+
+    parser_baseline.add_argument("--get", action="append", default=[],
+                                 help="URIs to process with http GET requests.")
+
+    parser_baseline.add_argument("--post", action="append", default=[],
+                                 help="URIs to process with http POST requests.")
+
+    parser_baseline.add_argument("--put", action="append", default=[],
+                                 help="URIs to process with http PUT requests.")
+
+    parser_baseline.add_argument("--delete", action="append", default=[],
+                                 help="URIs to process with http DELETE requests.")
+
+    parser_baseline.set_defaults(handler=run_baseline)
+
+    # test subcommand
+    parser_test = subparsers.add_parser(
+        "test",
+        help="Given a saved baseline, perform the contained HTTP requests and "
+        "compare responses to the baseline. Successes, Warnings, and Failures "
+        "will be reported over stdout."
+    )
+
+    parser_test.add_argument("version",
+                             help="The LANforge version of the host")
+
+    parser_test.add_argument("-b", "--baseline", nargs="?",
+                             help="Path to the baseline file to compare against. "
+                                   "If absent, reads the baseline JSON from stdin.")
+    parser_test.add_argument("-S", "--strict", action="store_true",
+                             help="Treat Warning results as failures.")
+
+    parser_test.set_defaults(handler=run_test)
+
+    return parser.parse_args()
+
+
+def init_logging(args):
+    logger_config = lf_logger_config.lf_logger_config()
+    if args.log_level:
+        logger_config.set_level(level=args.log_level)
+
+    if args.lf_logger_config_json:
+        # logger_config.lf_logger_config_json = "lf_logger_config.json"
+        logger_config.lf_logger_config_json = args.lf_logger_config_json
+        logger_config.load_lf_logger_config()
+
+
+if __name__ == "__main__":
+    args = parse_args()
+
+    init_logging(args)
+
+    args.handler(args)