From 3283f5ad5a635d0f4837629fda1c301fd335bf6a Mon Sep 17 00:00:00 2001 From: szeka9 Date: Sun, 12 Jul 2026 15:42:59 +0200 Subject: [PATCH] Add soak testing and refactor system test harness Add a soak test to evaluate system stability under sustained load, helping detect issues such as heap fragmentation and gradual performance degradation. Restructure the system test modules to reduce the boilerplate required when adding new test types. Improve generated plots by visualizing heap usage distributions with box plots. Normalize test conditions for consistent timing: - refactor logging of heap measurements, use fixed sampling intervals, and store heap usage metrics in files - normalize all measurements to 1-second intervals using interpolation - measure idle heap before and after each test --- Makefile | 39 ++- tests/.pylintrc | 2 + tests/system/__init__.py | 0 .../{http_dimensioning => }/app_base.py | 6 +- .../{http_dimensioning => }/app_multipart.py | 10 +- tests/system/device.py | 127 ++++++++ tests/system/http_dimensioning/__init__.py | 0 tests/system/http_dimensioning/boot.py | 52 ++- tests/system/http_dimensioning/test.py | 296 +---------------- tests/system/http_dimensioning/utils.py | 100 ------ tests/system/http_soak/__init__.py | 0 tests/system/http_soak/boot.py | 79 +++++ tests/system/http_soak/test.py | 103 ++++++ .../{http_dimensioning => }/http_user.py | 9 +- tests/system/load_test.py | 260 +++++++++++++++ tests/system/summary.py | 307 ++++++++++++++++++ tests/unit/test_buffer.py | 2 +- tests/unit/test_helpers.py | 2 +- tests/unit/test_http.py | 4 +- tests/unit/test_http_file_server.py | 4 +- tests/unit/test_http_multipart.py | 2 +- 21 files changed, 990 insertions(+), 414 deletions(-) create mode 100644 tests/system/__init__.py rename tests/system/{http_dimensioning => }/app_base.py (67%) rename tests/system/{http_dimensioning => }/app_multipart.py (77%) create mode 100644 tests/system/device.py create mode 100644 tests/system/http_dimensioning/__init__.py delete mode 100644 tests/system/http_dimensioning/utils.py create mode 100644 tests/system/http_soak/__init__.py create mode 100644 tests/system/http_soak/boot.py create mode 100755 tests/system/http_soak/test.py rename tests/system/{http_dimensioning => }/http_user.py (93%) create mode 100644 tests/system/load_test.py create mode 100644 tests/system/summary.py diff --git a/Makefile b/Makefile index b91a0b8..72f9c6a 100644 --- a/Makefile +++ b/Makefile @@ -11,6 +11,7 @@ DOCS_DIR := docs PKG := pyrobusta +PROJECT_ROOT := $(shell pwd) MICROPY_ROOT := external/micropython MPY_CROSS := $(MICROPY_ROOT)/mpy-cross/build/mpy-cross MICROPYTHON := $(MICROPY_ROOT)/ports/unix/build-standard/micropython @@ -253,7 +254,7 @@ black: .PHONY: unit-test unit-test: @echo "Running unit tests" - @python3 -m unittest -v + @python3 -m unittest discover -s tests/unit -v # ----------------------------- # Run all static checkers @@ -310,23 +311,49 @@ test-device: stage-test #clean-device upload # Performance testing # ================================================ +# ----------------------------- +# Run HTTP soak tests +# ----------------------------- +.PHONY: perf-test-http-soak +perf-test-http-soak: + @mpremote $(DEVICE) soft-reset + mpremote $(DEVICE) cp $(PT_DIR)/app_base.py :app_base.py + mpremote $(DEVICE) cp $(PT_DIR)/app_multipart.py :app_multipart.py + mpremote $(DEVICE) cp $(PT_DIR)/http_soak/boot.py :boot.py + cd $(PT_DIR) && python -m http_soak.test \ + "$(DEVICE)" "$(DEVICE_IP)" "$(DEVICE_NAME)" \ + "$(PROJECT_ROOT)/$(DOCS_DIR)/soak" + # ----------------------------- # Run HTTP dimensioning tests # ----------------------------- .PHONY: perf-test-http-dimensioning perf-test-http-dimensioning: @mpremote $(DEVICE) soft-reset - mpremote $(DEVICE) cp $(PT_DIR)/http_dimensioning/app_base.py :app_base.py - mpremote $(DEVICE) cp $(PT_DIR)/http_dimensioning/app_multipart.py :app_multipart.py + mpremote $(DEVICE) cp $(PT_DIR)/app_base.py :app_base.py + mpremote $(DEVICE) cp $(PT_DIR)/app_multipart.py :app_multipart.py mpremote $(DEVICE) cp $(PT_DIR)/http_dimensioning/boot.py :boot.py - @mpremote $(DEVICE) reset - $(PT_DIR)/http_dimensioning/test.py "$(DEVICE)" "$(DEVICE_IP)" "$(DEVICE_NAME)" + cd $(PT_DIR) && python -m http_dimensioning.test \ + "$(DEVICE)" "$(DEVICE_IP)" "$(DEVICE_NAME)" \ + "$(PROJECT_ROOT)/$(DOCS_DIR)/dimensioning" # ----------------------------- # Run all performance tests # ----------------------------- .PHONY: perf-test-device -perf-test-device: perf-test-http-dimensioning +perf-test-device: perf-test-http-dimensioning perf-test-http-soak + +# ----------------------------- +# Regenerate plots +# ----------------------------- +.PHONY: perf-test-regenerate-plots +perf-test-regenerate-plots: + @for type in dimensioning soak; do \ + [ -d docs/$$type/esp32_c3 ] && \ + python3 tests/system/summary.py ESP32-C3 docs/$$type; \ + [ -d docs/$$type/esp32_s3 ] && \ + python3 tests/system/summary.py ESP32-S3 docs/$$type; \ + done # ================================================ # Utilities for TLS diff --git a/tests/.pylintrc b/tests/.pylintrc index 10e5469..533feb6 100644 --- a/tests/.pylintrc +++ b/tests/.pylintrc @@ -6,3 +6,5 @@ disable=W0212, R0904, R0902, R0801 + +init-hook='import os, sys; tests = os.path.join(os.getcwd(), "tests"); [sys.path.append(d) for d, _, _ in os.walk(tests)];' \ No newline at end of file diff --git a/tests/system/__init__.py b/tests/system/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/system/http_dimensioning/app_base.py b/tests/system/app_base.py similarity index 67% rename from tests/system/http_dimensioning/app_base.py rename to tests/system/app_base.py index b8feb4c..2829609 100644 --- a/tests/system/http_dimensioning/app_base.py +++ b/tests/system/app_base.py @@ -1,10 +1,12 @@ +# pylint: disable=E0401 from pyrobusta.protocol.http import HttpEngine -def chunk_handler(_, chunk): +def chunk_handler(_, chunk: bytes): if not chunk: # Received terminating chunk/part return "text/plain", "OK" - pass # process chunk data as needed + # + return None def load(): diff --git a/tests/system/http_dimensioning/app_multipart.py b/tests/system/app_multipart.py similarity index 77% rename from tests/system/http_dimensioning/app_multipart.py rename to tests/system/app_multipart.py index 65859cb..6c554d2 100644 --- a/tests/system/http_dimensioning/app_multipart.py +++ b/tests/system/app_multipart.py @@ -1,7 +1,8 @@ +# pylint: disable=E0401 from pyrobusta.protocol.http import HttpEngine -def multipart_response(num_responses, part_size): +def multipart_response(num_responses: int, part_size: int): i = 0 def response_generator(): @@ -14,19 +15,20 @@ def response_generator(): return response_generator -def multipart_producer(http_ctx, _): +def multipart_producer(http_ctx: HttpEngine, _): part_count = int(http_ctx.headers.get("x-part-count", 1)) part_size = int(http_ctx.headers.get("x-part-size", 1024)) return "multipart/form-data", multipart_response(part_count, part_size) -def multipart_handler(http_ctx, _): +def multipart_handler(http_ctx: HttpEngine, _): if ( http_ctx.headers.get("content-type", "").startswith("multipart/form-data") and http_ctx.mp_is_last ): return "text/plain", "OK" - pass # process part data as needed + # + return None def load(): diff --git a/tests/system/device.py b/tests/system/device.py new file mode 100644 index 0000000..546c21b --- /dev/null +++ b/tests/system/device.py @@ -0,0 +1,127 @@ +import subprocess +import tempfile +import socket +import time + + +class Device: + def __init__( + self, device_id: str, device_ip: str, device_name: str, base_config: dict + ): + self.device_id = device_id + self.device_ip = device_ip + self.device_name = device_name + self.base_config = base_config + self.current_config = {} + + self.validate_device_ip() + + def apply_base_config(self): + self.apply_config(self.base_config) + + def apply_config(self, config: dict): + """ + Apply device configuration with mpremote + """ + self.current_config = dict(config) + + subprocess.run(["mpremote", self.device_id, "soft-reset"], check=True) + + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8") as tmp: + config_lines = subprocess.run( + ["mpremote", self.device_id, "cat", ":/pyrobusta.env"], + check=True, + stdout=subprocess.PIPE, + text=True, + ).stdout.splitlines() + + current_config = {} + + for line in config_lines: + line = line.rstrip("\r\n").split("#")[0] + if not line.strip(): + continue + parts = line.split("=", 1) + key = parts[0].strip() + value = parts[1].strip().strip("'").strip('"') + current_config[key] = value + + current_config.update(config) + + tmp.write( + "\n".join([f"{key}={value}" for key, value in current_config.items()]) + + "\n" + ) + tmp.flush() + + subprocess.run( + ["mpremote", self.device_id, "cp", tmp.name, ":/pyrobusta.env"], + check=True, + ) + subprocess.run(["mpremote", self.device_id, "reset"], check=True) + time.sleep(5) # Allow the device to initialize + + def get_mem_params(self): + """ + Determine SRAM and buffer settings of a device + """ + sram_bytes = int( + subprocess.run( + [ + "mpremote", + self.device_id, + "exec", + "import gc\nprint(gc.mem_free() + gc.mem_alloc())", + ], + check=True, + stdout=subprocess.PIPE, + text=True, + ).stdout + ) + + ( + send_buf_min_bytes, + send_buf_max_bytes, + recv_buf_min_bytes, + recv_buf_max_bytes, + con_overhead, + ) = [ + int(i) + for i in subprocess.run( + [ + "mpremote", + self.device_id, + "exec", + ( + "from pyrobusta.server.http_server import HttpServer\n" + "print( HttpServer.SEND_BUF_MIN_BYTES,\n" + "HttpServer.SEND_BUF_MAX_BYTES,\n" + "HttpServer.RECV_BUF_MIN_BYTES,\n" + "HttpServer.RECV_BUF_MAX_BYTES,\n" + "HttpServer.CON_OVERHEAD_BYTES,\n" + ")" + ), + ], + check=True, + stdout=subprocess.PIPE, + text=True, + ).stdout.split() + ] + + buffer_small = send_buf_min_bytes + recv_buf_min_bytes + con_overhead + buffer_large = send_buf_max_bytes + recv_buf_max_bytes + con_overhead + return sram_bytes, buffer_small, buffer_large + + def validate_device_ip(self): + """ + Check if a device IP is valid + """ + try: + socket.inet_aton(self.device_ip) + except socket.error as exc: + raise ValueError(f"Invalid device address: {self.device_ip}") from exc + + def get_host(self): + proto = "https" if self.current_config["tls"] else "http" + port = 4443 if self.current_config["tls"] else 8080 + return f"{proto}://{self.device_ip}:{port}" diff --git a/tests/system/http_dimensioning/__init__.py b/tests/system/http_dimensioning/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/system/http_dimensioning/boot.py b/tests/system/http_dimensioning/boot.py index 3b3d0cb..c3883fc 100644 --- a/tests/system/http_dimensioning/boot.py +++ b/tests/system/http_dimensioning/boot.py @@ -1,31 +1,59 @@ +# pylint: disable=E0401,C0415,E1101,E0611 # This file is executed on every boot (including wake-boot from deepsleep) -import machine import asyncio +import os +import time from gc import mem_alloc, collect +import machine -import pyrobusta.server.http_server as http_server +from pyrobusta.server import http_server from pyrobusta.protocol.http import HttpEngine from pyrobusta.connectivity import wifi from pyrobusta.utils.config import get_config, CONF_HTTP_MULTIPART -TS_DURATION = 100 -MEM_TIME_SERIES = [0] * TS_DURATION +LOG_FILE = "heap_usage.csv" + +SAMPLE_PERIOD = 5 # seconds +FLUSH_PERIOD = 12 # Flush every minute @HttpEngine.route("/heap/time-series", "GET") -def time_series(*_): - return "application/json", MEM_TIME_SERIES +def time_series(http_ctx: HttpEngine, _): + http_ctx.set_response_header( + b"content-length", str(os.stat(LOG_FILE)[6]).encode("ascii") + ) + http_ctx.set_response_header(b"content-type", b"text/csv") + http_ctx.terminate(200) + # pylint: disable=R1732 + http_ctx.resp_handler = open(LOG_FILE, "rb", encoding="utf-8") async def mem_usage(): - i = 0 + flush_counter = 0 + try: + os.remove(LOG_FILE) + except OSError: + pass collect() - while True: - collect() - MEM_TIME_SERIES[i] = mem_alloc() - await asyncio.sleep(1) - i = (i + 1) % TS_DURATION + with open(LOG_FILE, "a", encoding="utf-8") as log: + start_ts = time.ticks_ms() + log.write(f"0,{mem_alloc()}\n") + + next_sample = time.ticks_add(start_ts, SAMPLE_PERIOD * 1000) + + while True: + delay = time.ticks_diff(next_sample, time.ticks_ms()) + if delay > 0: + await asyncio.sleep_ms(delay) + + collect() + elapsed = time.ticks_diff(time.ticks_ms(), start_ts) // 1000 + log.write(f"{elapsed},{mem_alloc()}\n") + flush_counter = (flush_counter + 1) % FLUSH_PERIOD + if not flush_counter: + log.flush() + next_sample = time.ticks_add(next_sample, SAMPLE_PERIOD * 1000) async def main(): diff --git a/tests/system/http_dimensioning/test.py b/tests/system/http_dimensioning/test.py index 0ce1bd2..f69cba4 100755 --- a/tests/system/http_dimensioning/test.py +++ b/tests/system/http_dimensioning/test.py @@ -25,35 +25,15 @@ summary table of performance metrics for each configuration. """ -from flask import json -from gevent import os -import gevent.monkey - -gevent.monkey.patch_all() - -import urllib3 - -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - -import gevent -import requests import sys -import socket -import subprocess -import tempfile import math - -from time import sleep, monotonic -from locust.env import Environment -from locust.stats import stats_printer - -from utils import generate_measurement_table, generate_plot -from http_user import DefaultUser, FilesApiUser, MultipartUser +from load_test import run_test +from device import Device # --------------------------- # Test configuration settings # --------------------------- -LOAD_TEST_DURATION = 60 +TEST_DURATION_MINUTES = 5 base_config = { "socket_max_con": 1, @@ -68,59 +48,7 @@ } -def get_mem_params(device): - """ - Determine SRAM and buffer settings of a device - """ - sram_bytes = int( - subprocess.run( - [ - "mpremote", - device, - "exec", - "import gc\nprint(gc.mem_free() + gc.mem_alloc())", - ], - check=True, - stdout=subprocess.PIPE, - text=True, - ).stdout - ) - - ( - send_buf_min_bytes, - send_buf_max_bytes, - recv_buf_min_bytes, - recv_buf_max_bytes, - con_overhead, - ) = [ - int(i) - for i in subprocess.run( - [ - "mpremote", - device, - "exec", - ( - "from pyrobusta.server.http_server import HttpServer\n" - "print( HttpServer.SEND_BUF_MIN_BYTES,\n" - "HttpServer.SEND_BUF_MAX_BYTES,\n" - "HttpServer.RECV_BUF_MIN_BYTES,\n" - "HttpServer.RECV_BUF_MAX_BYTES,\n" - "HttpServer.CON_OVERHEAD_BYTES,\n" - ")" - ), - ], - check=True, - stdout=subprocess.PIPE, - text=True, - ).stdout.split() - ] - - buffer_small = send_buf_min_bytes + recv_buf_min_bytes + con_overhead - buffer_large = send_buf_max_bytes + recv_buf_max_bytes + con_overhead - return sram_bytes, buffer_small, buffer_large - - -def get_test_config(sram_bytes, buffer_small, buffer_large): +def get_test_config(sram_bytes: int, buffer_small: int, buffer_large: int): socket_counts = (1, 2, 4) def round_up_sig(x, sig=3): @@ -178,215 +106,27 @@ def round_up_sig(x, sig=3): return test_config -# ------------ -# Test helpers -# ------------ - - -def apply_mpremote_config(config, device): - subprocess.run(["mpremote", device, "soft-reset"], check=True) - - with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8") as tmp: - config_lines = subprocess.run( - ["mpremote", device, "cat", ":/pyrobusta.env"], - check=True, - stdout=subprocess.PIPE, - text=True, - ).stdout.splitlines() - - current_config = {} - - for line in config_lines: - line = line.rstrip("\r\n").split("#")[0] - if not line.strip(): - continue - parts = line.split("=", 1) - key = parts[0].strip() - value = parts[1].strip().strip("'").strip('"') - current_config[key] = value - - current_config.update(config) - - tmp.write( - "\n".join([f"{key}={value}" for key, value in current_config.items()]) - + "\n" - ) - tmp.flush() - - subprocess.run( - ["mpremote", device, "cp", tmp.name, ":/pyrobusta.env"], check=True - ) - subprocess.run(["mpremote", device, "reset"], check=True) - sleep(15) - - -def validate_device_ip(device_ip): - try: - socket.inet_aton(device_ip) - except socket.error: - print(f"Invalid device address: {device_ip}") - sys.exit(1) - - -# ------------ -# Test methods -# ------------ - - -def load_test(config, device_ip): - proto = "https" if config["tls"] else "http" - port = 4443 if config["tls"] else 8080 - host = f"{proto}://{device_ip}:{port}" - max_con = config.get("socket_max_con", 1) - - user_classes = [DefaultUser] - if config.get("http_multipart", False): - user_classes = [MultipartUser] - if config.get("http_files_api", False): - user_classes = [FilesApiUser] - - env = Environment( - user_classes=user_classes, - host=host, - ) - - runner = env.create_local_runner() - runner.start( - user_count=max_con, - spawn_rate=max_con, - ) - - start_time = monotonic() - while monotonic() - start_time < LOAD_TEST_DURATION: - gevent.sleep(5) - - print( - f"users={runner.user_count} " - f"state={runner.state} " - f"requests={env.stats.total.num_requests} " - f"failures={env.stats.total.num_failures}" - ) - - runner.quit() - - total = env.stats.total - stats = { - "ok": total.num_requests, - "errors": total.num_failures, - "avg_ms": total.avg_response_time, - "min_ms": total.min_response_time, - "max_ms": total.max_response_time, - "rps": total.current_rps, - "p95_ms": total.get_response_time_percentile(0.95), - "p99_ms": total.get_response_time_percentile(0.99), - } - - try: - sleep(5) - usage = requests.get( - f"{host}/heap/time-series", - verify=False, - timeout=5, - headers={"Connection": "close"}, - ).json() - - print(f"Measured: {usage}") - except Exception as e: - print(f"WARNING - exception: {e}") - return 0, [], stats - - idle_threshold = usage[0] * 0.01 - idle_last_idx = 0 - for i in range(len(usage)): - idle_last_idx = i - if i > 0 and usage[i] - usage[i - 1] > idle_threshold: - break - idle = round(sum(usage[:idle_last_idx]) / (idle_last_idx)) - - return idle, usage, stats - - -def test_config_delta(device_name, device_ip, base_config, config_delta={}): - target_config = dict(base_config) - target_config.update(config_delta) - - if base_config == target_config and config_delta: - print("Base config matches target config, skipping.") - return None, None - - if not config_delta: - print(f"Measure with base config: {base_config}") - target_config = base_config - else: - print(f"Measure with target config: {target_config}") - pass - - apply_mpremote_config(target_config, device_name) - idle, usage, stats = load_test(target_config, device_ip) - return idle, usage, stats - - -if __name__ == "__main__": +def main(): device_id = sys.argv[1] # mpremote id e.g. a1 (/dev/ttyACM1) device_ip = sys.argv[2] device_name = sys.argv[3] + output_path = sys.argv[4] - if not device_id or not device_ip or not device_name: + if not device_id or not device_ip or not device_name or not output_path: raise ValueError( - "Invalid arguments.\nUsage: test.py device_id device_ip device_name" + "Invalid arguments.\nUsage: test.py device_id device_ip device_name output_path" ) - validate_device_ip(device_ip) - apply_mpremote_config(base_config, device_id) - - idle, usage, stats = test_config_delta(device_id, device_ip, base_config) + dev = Device(device_id, device_ip, device_name, base_config) - base_measurement = { - "id": "base", - "idle": idle, - "usage": usage, - "stats": stats, - "config": base_config, - } - generate_plot(base_measurement, device_name) - - measurements = [base_measurement] - sram_bytes, buffer_small, buffer_large = get_mem_params(device_id) - test_config = get_test_config(sram_bytes, buffer_small, buffer_large) - for case in test_config: - delta_cnt = 0 - for i, delta in enumerate(test_config[case]): - load_idle, load_usage, load_stats = test_config_delta( - device_id, device_ip, base_config, delta - ) - if load_usage and load_stats: - delta_cnt += 1 - m = { - "id": f"{case}_{delta_cnt:03d}", - "idle": load_idle, - "usage": load_usage, - "stats": load_stats, - "config": base_config | delta, - } - measurements.append(m) - generate_plot(m, device_name) - - target_dir = device_name.replace("-", "_").lower() - if target_dir not in os.listdir("./docs/dimensioning/"): - os.mkdir(f"./docs/dimensioning/{target_dir}") - - with open(f"docs/dimensioning/{target_dir}/measurements.json", "w") as f: - json.dump(measurements, f, indent=4) - - table = generate_measurement_table( - measurements, - excluded_keys={ - "http_port", - "https_port", - "http_served_paths", - "log_level", - "http_files_api", - }, + run_test( + output_path, + dev, + get_test_config, + None, + TEST_DURATION_MINUTES, ) - print(table) + +if __name__ == "__main__": + main() diff --git a/tests/system/http_dimensioning/utils.py b/tests/system/http_dimensioning/utils.py deleted file mode 100644 index 1ebaa42..0000000 --- a/tests/system/http_dimensioning/utils.py +++ /dev/null @@ -1,100 +0,0 @@ -import os -import matplotlib.pyplot as plt - -# ----------------------------------- -# Helpers for processing measurements -# ----------------------------------- - - -def generate_measurement_table(measurements, excluded_keys={}): - """ - Generate a table in markdown format for measurement data. - """ - base = measurements[0] - config_keys = set(base.get("config", {}).keys()) - - for m in measurements[1:]: - config_keys.update(m.get("config", {}).keys()) - - config_keys = sorted(k for k in config_keys if k not in excluded_keys) - headers = ["id"] + config_keys + ["footprint_bytes"] - rows = [] - base_cfg = dict(base["config"]) - base_row = [base["id"]] - - # Base config - for key in config_keys: - base_row.append(base_cfg.get(key, "")) - - base_row.append(base["idle"]) - rows.append(base_row) - - # Measurements - for m in measurements[1:]: - row = [m["id"]] - for key in config_keys: - row.append(m["config"].get(key, "")) - - row.append(m["idle"]) - rows.append(row) - - header_line = "| " + " | ".join(headers) + " |" - separator = "| " + " | ".join("---" for _ in headers) + " |" - body = ["| " + " | ".join(str(v) for v in row) + " |" for row in rows] - return "\n".join([header_line, separator] + body) - - -def generate_plot(measurement, device_name): - """ - Visualize time series data of heap usage with additional annotation. - """ - id = measurement["id"] - - try: - last_idx = measurement["usage"].index(0, 1) - except ValueError: - last_idx = len(measurement["usage"]) - - xticks = list(range(last_idx)) - - fig, ax = plt.subplots() - p = ax.plot(xticks, measurement["usage"][0:last_idx]) - plt.title(f"Heap usage over time - {device_name}\n{id}") - plt.grid(True) - plt.xlabel("Time [s]") - plt.ylabel("Heap usage [bytes]") - plt.xticks(xticks) - plt.ylim(0, max(measurement["usage"][0:last_idx]) * 1.2) - for i, label in enumerate(ax.xaxis.get_ticklabels()): - if i % 5 == 0: - label.set_visible(True) - else: - label.set_visible(False) - - # Additional annotation - load_stats = measurement.get("stats", {}) - concurrency = measurement["config"].get("socket_max_con", "n/a") - - annotation_text = "\n".join( - f"{key}: {value if not isinstance(value, float) else (f'{value:.0f}' if key.endswith('_ms') else f'{value:.2f}')}" - for key, value in load_stats.items() - ) - annotation_text += f"\nconcurrency: {concurrency}" - - plt.gca().text( - 0.95, - 0.05, - annotation_text, - transform=plt.gca().transAxes, - fontsize=9, - verticalalignment="bottom", - horizontalalignment="right", - bbox=dict(boxstyle="round", alpha=0.3), - ) - - target_dir = device_name.replace("-", "_").lower() - if target_dir not in os.listdir("./docs/dimensioning/"): - os.mkdir(f"./docs/dimensioning/{target_dir}") - - plt.savefig(f"./docs/dimensioning/{target_dir}/{id}.png") - plt.clf() diff --git a/tests/system/http_soak/__init__.py b/tests/system/http_soak/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/system/http_soak/boot.py b/tests/system/http_soak/boot.py new file mode 100644 index 0000000..a190ff3 --- /dev/null +++ b/tests/system/http_soak/boot.py @@ -0,0 +1,79 @@ +# pylint: disable=E0401,C0415,E1101,E0611 +# This file is executed on every boot (including wake-boot from deepsleep) +import asyncio +import os +import time +from gc import mem_alloc, collect +import machine + +from pyrobusta.server import http_server +from pyrobusta.protocol.http import HttpEngine +from pyrobusta.connectivity import wifi + +from pyrobusta.utils.config import get_config, CONF_HTTP_MULTIPART + +LOG_FILE = "heap_usage.csv" + +SAMPLE_PERIOD = 10 # seconds +FLUSH_PERIOD = 6 # Flush every minute + + +@HttpEngine.route("/heap/time-series", "GET") +def time_series(http_ctx: HttpEngine, _): + http_ctx.set_response_header( + b"content-length", str(os.stat(LOG_FILE)[6]).encode("ascii") + ) + http_ctx.set_response_header(b"content-type", b"text/csv") + http_ctx.terminate(200) + # pylint: disable=R1732 + http_ctx.resp_handler = open(LOG_FILE, "rb", encoding="utf-8") + + +async def mem_usage(): + flush_counter = 0 + try: + os.remove(LOG_FILE) + except OSError: + pass + collect() + with open(LOG_FILE, "a", encoding="utf-8") as log: + start_ts = time.ticks_ms() + log.write(f"0,{mem_alloc()}\n") + + next_sample = time.ticks_add(start_ts, SAMPLE_PERIOD * 1000) + + while True: + delay = time.ticks_diff(next_sample, time.ticks_ms()) + if delay > 0: + await asyncio.sleep_ms(delay) + + collect() + elapsed = time.ticks_diff(time.ticks_ms(), start_ts) // 1000 + log.write(f"{elapsed},{mem_alloc()}\n") + flush_counter = (flush_counter + 1) % FLUSH_PERIOD + if not flush_counter: + log.flush() + next_sample = time.ticks_add(next_sample, SAMPLE_PERIOD * 1000) + + +async def main(): + server = http_server.HttpServer() + connected = wifi.initialize() + if connected and not machine.reset_cause() == machine.SOFT_RESET: + import app_base + + app_base.load() + + if get_config(CONF_HTTP_MULTIPART): + import app_multipart + + app_multipart.load() + + await server.start_socket_server() + asyncio.create_task(mem_usage()) + + while True: + await asyncio.sleep(1) + + +asyncio.run(main()) diff --git a/tests/system/http_soak/test.py b/tests/system/http_soak/test.py new file mode 100755 index 0000000..af9daca --- /dev/null +++ b/tests/system/http_soak/test.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +""" +This test performs soak tests while measuring the heap usage +and other performance metrics with different configurations. +The test is designed to run against a device running the boot.py + +Tests are performed with the Locust load testing framework, simulating +multiple concurrent users accessing the device's HTTP server with +different request patterns. + +The test workflow includes: +1. Applying a configuration to the device using mpremote. + +2. Performing a soak test with Locust, simulating concurrent + users making requests to the device. + +3. Measuring the heap usage of the device during the soak test + using a dedicated endpoint. + +4. Collecting performance metrics such as response times and + request rates from Locust. + +5. Visualizing the results with time series plots of heap usage and a + summary table of performance metrics for each configuration. +""" + +import sys +import math + +from http_user import DefaultUser, FilesApiUser, MultipartUser +from load_test import run_test +from device import Device + +# --------------------------- +# Test configuration settings +# --------------------------- +TEST_DURATION_MINUTES = 60 + +base_config = { + "socket_max_con": 4, + "http_mem_cap": 0.05, + "http_multipart": True, + "http_files_api": True, + "tls": False, + "http_port": 8080, + "https_port": 4443, + "log_level": "info", + "http_served_paths": "/lib/pyrobusta /www", +} + + +def get_test_config(sram_bytes: int, _, buffer_large: int): + socket_counts = (4,) + + def round_up_sig(x, sig=3): + """ + Round up with significant digits. + round_up_sig(0.0012345, 2) == 0.0013 + round_up_sig(0.0012345, 3) == 0.00124 + """ + if x == 0: + return 0 + factor = 10 ** (sig - int(math.floor(math.log10(abs(x)))) - 1) + return math.ceil(x * factor) / factor + + test_config = { + "tls": [ + { + "http_mem_cap": round_up_sig((buffer_large / sram_bytes) * max_con, 3), + "tls": True, + "socket_max_con": max_con, + } + for max_con in socket_counts + ], + } + print(test_config) + return test_config + + +def main(): + device_id = sys.argv[1] # mpremote id e.g. a1 (/dev/ttyACM1) + device_ip = sys.argv[2] + device_name = sys.argv[3] + output_path = sys.argv[4] + + if not device_id or not device_ip or not device_name or not output_path: + raise ValueError( + "Invalid arguments.\nUsage: test.py device_id device_ip device_name output_path" + ) + + dev = Device(device_id, device_ip, device_name, base_config) + + run_test( + output_path, + dev, + get_test_config, + [DefaultUser, FilesApiUser, MultipartUser], + TEST_DURATION_MINUTES, + ) + + +if __name__ == "__main__": + main() diff --git a/tests/system/http_dimensioning/http_user.py b/tests/system/http_user.py similarity index 93% rename from tests/system/http_dimensioning/http_user.py rename to tests/system/http_user.py index 7dd4776..b2ef607 100644 --- a/tests/system/http_dimensioning/http_user.py +++ b/tests/system/http_user.py @@ -35,7 +35,7 @@ def post_chunked(self): part_count = 10 part_size = 256 chunked_data = b"" - for i in range(part_count): + for _ in range(part_count): chunked_data += b"%X\r\n" % part_size chunked_data += b"X" * part_size + b"\r\n" chunked_data += b"0\r\n\r\n" @@ -51,7 +51,7 @@ def post_chunked(self): ) print( self.client.base_url - + "/test/stream (chunked; parts=%d, size=%d)" % (part_count, part_size), + + f"/test/stream (chunked; parts={part_count}, size={part_size})", response.status_code, response.elapsed.total_seconds(), ) @@ -78,7 +78,7 @@ def get_multipart(self): ) print( self.client.base_url - + "/test/multipart (parts=%d, size=%d)" % (part_count, part_size), + + f"/test/multipart (parts={part_count}, size={part_size})", response.status_code, response.elapsed.total_seconds(), ) @@ -110,8 +110,7 @@ def post_multipart(self): ) print( self.client.base_url - + "/test/multipart (multipart; parts=%d, size=%d)" - % (part_count, part_size), + + f"/test/multipart (multipart; parts={part_count}, size={part_size})", response.status_code, response.elapsed.total_seconds(), ) diff --git a/tests/system/load_test.py b/tests/system/load_test.py new file mode 100644 index 0000000..bf45781 --- /dev/null +++ b/tests/system/load_test.py @@ -0,0 +1,260 @@ +# pylint: disable=C0413,C0411 +import gevent +import gevent.monkey + +gevent.monkey.patch_all() + +import os +import json +from time import monotonic + +import requests +from locust.env import Environment + +from device import Device +from http_user import DefaultUser, FilesApiUser, MultipartUser +from summary import generate_measurement_table, generate_plot + +import urllib3 + +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +TEST_IDLE_PERIOD = 30 + + +def interpolate_time_series(usage_data: list): + """ + Interpolate time-series data to normalize to 1-second intervals + """ + usage = [] + prev_ts = None + prev_mem = None + + for u in usage_data: + ts = u[0] + mem = u[1] + + if prev_ts is None: + usage.append(mem) + else: + dt = ts - prev_ts + assert dt > 0 + + for i in range(1, dt): + usage.append(round(prev_mem + (mem - prev_mem) * i / dt)) + + usage.append(mem) + + prev_ts = ts + prev_mem = mem + return usage + + +def determine_idle_heap(usage_data: list): + """ + Determine idle heap usage + """ + idle_threshold = usage_data[0] * 0.01 + idle_last_idx = 0 + for i, _ in enumerate(usage_data): + idle_last_idx = i + if i > 0 and usage_data[i] - usage_data[i - 1] > idle_threshold: + break + return round(sum(usage_data[:idle_last_idx]) / (idle_last_idx)) + + +def load_test(config: dict, device: Device, user_classes: list, duration_minutes: int): + """ + Run load test + """ + host = device.get_host() + max_con = config.get("socket_max_con", 1) + + if not user_classes: + user_classes = [DefaultUser] + if config.get("http_multipart", False): + user_classes = [MultipartUser] + if config.get("http_files_api", False): + user_classes = [FilesApiUser] + + env = Environment( + user_classes=user_classes, + host=host, + ) + + gevent.sleep(TEST_IDLE_PERIOD) # For idle measurement + runner = env.create_local_runner() + runner.start( + user_count=max_con, + spawn_rate=max_con, + ) + + end_time = monotonic() + duration_minutes * 60 - 2 * TEST_IDLE_PERIOD + while True: + remaining = end_time - monotonic() + if remaining <= 0: + break + gevent.sleep(min(10, remaining)) + print( + f"users={runner.user_count} " + f"state={runner.state} " + f"requests={env.stats.total.num_requests} " + f"failures={env.stats.total.num_failures}" + ) + + runner.quit() + + gevent.sleep( + TEST_IDLE_PERIOD + 10 + ) # For idle measurement and heap usage data collection + + total = env.stats.total + stats = { + "ok": total.num_requests, + "errors": total.num_failures, + "avg_ms": total.avg_response_time, + "min_ms": total.min_response_time, + "max_ms": total.max_response_time, + "rps": total.current_rps, + "p95_ms": total.get_response_time_percentile(0.95), + "p99_ms": total.get_response_time_percentile(0.99), + } + + try: + usage_ts = requests.get( + f"{host}/heap/time-series", + verify=False, + timeout=5, + headers={"Connection": "close"}, + ) + if usage_ts.headers.get("Content-Type", "").startswith("text/csv"): + usage = interpolate_time_series( + [ + (int(u.split(",", 1)[0]), int(u.split(",", 1)[1])) + for u in usage_ts.text.splitlines() + ] + ) + else: + raise ValueError( + f"Unexpected Content-Type: {usage_ts.headers.get('Content-Type')}" + ) + + print(f"Measured: {usage}") + except Exception as e: # pylint: disable=W0718 + # Catch all exceptions without halting test execution + print(f"WARNING - exception: {e}") + return 0, [], stats + + return determine_idle_heap(usage), usage, stats + + +def test_config_delta( + device: Device, + config_delta: dict, + duration_minutes: int, + user_classes: list, +): + """ + Test a configuration delta + """ + target_config = dict(device.base_config) + target_config.update(config_delta) + + if device.base_config == target_config and config_delta: + print("Base config matches target config, skipping.") + return None, None + + if not config_delta: + print(f"Measure with base config: {device.base_config}") + target_config = device.base_config + else: + print(f"Measure with target config: {target_config}") + + device.apply_config(target_config) + idle, usage, stats = load_test( + target_config, device, user_classes, duration_minutes + ) + return idle, usage, stats + + +# pylint: disable=R0914 +def run_test( + output_path: str, + device: Device, + config_getter: callable, + user_classes: list = None, + duration_minutes: int = 5, +): + # -------------------------------------------- + # Test base configuration + # -------------------------------------------- + + device.apply_base_config() + + idle, usage, stats = test_config_delta( + device, + {}, + duration_minutes=duration_minutes, + user_classes=user_classes, + ) + + base_measurement = { + "id": "base", + "idle": idle, + "usage": usage, + "stats": stats, + "config": device.base_config, + } + measurements = [base_measurement] + generate_plot(base_measurement, device.device_name, output_path) + + # -------------------------------------------- + # Test configuration delta + # -------------------------------------------- + + sram_bytes, buffer_small, buffer_large = device.get_mem_params() + test_config = config_getter(sram_bytes, buffer_small, buffer_large) + for case_id, case in test_config.items(): + delta_cnt = 0 + for _, delta in enumerate(case): + load_idle, load_usage, load_stats = test_config_delta( + device, + delta, + duration_minutes=duration_minutes, + user_classes=user_classes, + ) + if load_usage and load_stats: + delta_cnt += 1 + m = { + "id": f"{case_id}_{delta_cnt:03d}", + "idle": load_idle, + "usage": load_usage, + "stats": load_stats, + "config": device.base_config | delta, + } + measurements.append(m) + generate_plot(m, device.device_name, output_path) + + # -------------------------------------------- + # Export measurements + # -------------------------------------------- + target_dir = device.device_name.replace("-", "_").lower() + if target_dir not in os.listdir(output_path): + os.mkdir(f"{output_path}/{target_dir}") + + with open( + f"{output_path}/{target_dir}/measurements.json", "w", encoding="utf-8" + ) as f: + json.dump(measurements, f, indent=4) + + print( + generate_measurement_table( + measurements, + excluded_keys={ + "http_port", + "https_port", + "http_served_paths", + "log_level", + }, + ) + ) diff --git a/tests/system/summary.py b/tests/system/summary.py new file mode 100644 index 0000000..a7459d3 --- /dev/null +++ b/tests/system/summary.py @@ -0,0 +1,307 @@ +import os +import statistics +import sys +import json + +import matplotlib.pyplot as plt + +from matplotlib.gridspec import GridSpec +from matplotlib.figure import Figure +from matplotlib.ticker import FuncFormatter + +# ----------------------------------- +# Helpers for processing measurements +# ----------------------------------- + + +def generate_measurement_table(measurements: list, excluded_keys: list): + """ + Generate a table in markdown format for measurement data. + """ + base = measurements[0] + config_keys = set(base.get("config", {}).keys()) + + for m in measurements[1:]: + config_keys.update(m.get("config", {}).keys()) + + if not excluded_keys: + excluded_keys = [] + + config_keys = sorted(k for k in config_keys if k not in excluded_keys) + headers = ["id"] + config_keys + ["footprint_bytes"] + rows = [] + base_cfg = dict(base["config"]) + base_row = [base["id"]] + + # Base config + for key in config_keys: + base_row.append(base_cfg.get(key, "")) + + base_row.append(base["idle"]) + rows.append(base_row) + + # Measurements + for m in measurements[1:]: + row = [m["id"]] + for key in config_keys: + row.append(m["config"].get(key, "")) + + row.append(m["idle"]) + rows.append(row) + + header_line = "| " + " | ".join(headers) + " |" + separator = "| " + " | ".join("---" for _ in headers) + " |" + body = ["| " + " | ".join(str(v) for v in row) + " |" for row in rows] + return "\n".join([header_line, separator] + body) + + +def get_y_axis_parameters(data: list): + step = 4 * 1024 + ymin = min(data) + ymax = max(data) + + # Round to fixed-interval boundaries + ymin = ((ymin - step) // step) * step + ymax = ((ymax + step) // step + 1) * step + + # Generate consistent ticks + yticks = list(range(int(ymin), int(ymax + 1), 1024)) + return ymin, ymax, yticks + + +def generate_main_plot(gs: GridSpec, fig: Figure, measurement: dict, device_name: str): + usage = list(measurement["usage"]) + ax = fig.add_subplot(gs[:, 0]) + ax.set_title(f"Heap usage over time - {device_name} - [{measurement["id"]}]") + ax.grid(True) + + # ---------------------------- + # X Axis + # ---------------------------- + ticks_interval = 1 + xlabel = "Time [s]" + + for interval in [600, 300, 60, 30, 15, 10, 5]: + if len(usage) >= interval * 6: + ticks_interval = interval + break + + if ticks_interval >= 60: + xlabel = "Time [min]" + + expected = (len(usage) - 1) // ticks_interval * ticks_interval + 1 + usage = usage[:expected] + total_time = len(usage) # Assuming each index represents 1 second + + ax.plot(range(total_time), usage) + ax.set_xlabel(xlabel) + + xticks = list(range(0, total_time, ticks_interval)) + ax.set_xticks(xticks) + ax.set_xticklabels( + xticks if xlabel == "Time [s]" else [f"{x // 60}" for x in xticks] + ) + ax.set_xlim(0, total_time - 1) + + # ---------------------------- + # Y Axis + # ---------------------------- + # Rounded limits with a minimum range + ax.set_ylabel("Heap usage [KiB]") + + ymin, ymax, yticks = get_y_axis_parameters(usage) + yticks = yticks[:: max(1, 2 * ((len(yticks) // 8) // 2))] + ax.set_yticks(yticks) + ax.set_ylim(ymin, ymax) + ax.margins(y=0.05) + + ax.yaxis.set_major_formatter(FuncFormatter(lambda x, _: f"{x / 1024:.0f}")) + + # Reference lines + ax.axhline(min(usage), linestyle=":", linewidth=1) + ax.axhline(max(usage), linestyle=":", linewidth=1) + + +def generate_stats_annotation(gs: GridSpec, fig: Figure, measurement: dict): + stats_ax = fig.add_subplot(gs[0, 1]) + stats_ax.axis("off") + + load_stats = measurement.get("stats", {}) + concurrency = measurement["config"].get("socket_max_con", "n/a") + + annotation_text = "\n".join( + [ + f"HTTP - 200 OK {load_stats['ok']}", + f"HTTP errors {load_stats['errors']}", + f"Avg latency {load_stats['avg_ms']:.0f} ms", + f"Min latency {load_stats['min_ms']:.0f} ms", + f"Max latency {load_stats['max_ms']:.0f} ms", + f"RPS {load_stats['rps']:.2f}", + f"P95 {load_stats['p95_ms']:.0f} ms", + f"P99 {load_stats['p99_ms']:.0f} ms", + f"Concurrency {concurrency}", + ] + ) + + stats_ax.text( + 0.03, + 0.98, + "Test summary", + fontsize=9, + fontweight="bold", + ha="left", + va="bottom", + transform=stats_ax.transAxes, + ) + + stats_ax.text( + 0.03, + 0.88, + annotation_text, + fontsize=8, + family="monospace", + ha="left", + va="top", + transform=stats_ax.transAxes, + bbox={ + "boxstyle": "round,pad=0.45", + "facecolor": "#fafafa", + "edgecolor": "0.75", + "linewidth": 0.8, + }, + ) + + +def generate_heap_dist_plot(gs: GridSpec, fig: Figure, measurement: dict): + box_ax = fig.add_subplot(gs[1, 1]) + + usage = list(measurement["usage"]) + box_ax.boxplot( + usage, + orientation="vertical", + widths=0.35, + patch_artist=True, + showfliers=False, + whis=(0, 100), + ) + + # Hide top and right border + box_ax.spines["top"].set_visible(False) + box_ax.spines["right"].set_visible(False) + + # ---------------------------- + # Statistics + # ---------------------------- + minimum = min(usage) + q1 = statistics.quantiles(usage, n=4)[0] + median = statistics.median(usage) + q3 = statistics.quantiles(usage, n=4)[2] + maximum = max(usage) + + labels = ( + f"max {maximum/1024:.1f}\n" + f"Q3 {q3/1024:.1f}\n" + f"med {median/1024:.1f}\n" + f"Q1 {q1/1024:.1f}\n" + f"min {minimum/1024:.1f}" + ) + + box_ax.text( + 1, + 0.4, + labels, + transform=box_ax.transAxes, + ha="right", + va="top", + fontsize=8, + family="monospace", + bbox={ + "boxstyle": "round,pad=0.3", + "facecolor": "white", + "alpha": 0.8, + }, + ) + + # ---------------------------- + # Y Axis + # ---------------------------- + box_ax.set_ylabel("Heap distribution [KiB]") + + ymin, ymax, yticks = get_y_axis_parameters(usage) + yticks = yticks[:: max(1, 2 * ((len(yticks) // 5) // 2))] + box_ax.set_ylim(ymin, ymax) + box_ax.set_yticks(yticks) + box_ax.margins(y=0.05) + + box_ax.tick_params( + axis="y", + which="both", + direction="in", + pad=-20, # move labels inward + length=4, + ) + box_ax.get_yticklabels()[0].set_visible(False) + box_ax.yaxis.set_major_formatter(FuncFormatter(lambda x, _: f"{x / 1024:.0f}")) + box_ax.grid(True, axis="y", alpha=0.3) + + # ---------------------------- + # X Axis + # ---------------------------- + box_ax.set_xticks([]) + box_ax.tick_params(axis="x", bottom=False) + + +def generate_plot(measurement: list, device_name: str, output_path: str): + """ + Visualize time series data of heap usage with additional annotation. + """ + + fig = plt.figure(figsize=(9, 4.5)) + gs = GridSpec( + 2, + 2, + width_ratios=[2.5, 1], + height_ratios=[1, 1.5], + figure=fig, + wspace=0.1, + hspace=0.15, + ) + + generate_main_plot(gs, fig, measurement, device_name) + generate_stats_annotation(gs, fig, measurement) + generate_heap_dist_plot(gs, fig, measurement) + + target_dir = device_name.replace("-", "_").lower() + if target_dir not in os.listdir(output_path): + os.mkdir(f"{output_path}/{target_dir}") + + fig.savefig(f"{output_path}/{target_dir}/{measurement["id"]}.png", dpi=150) + plt.close(fig) + + +def main(): + device_name = sys.argv[1] + output_path = sys.argv[2] + + target_dir = device_name.replace("-", "_").lower() + + with open(f"{output_path}/{target_dir}/measurements.json", encoding="utf-8") as f: + measurements = json.loads(f.read()) + + for m in measurements: + generate_plot(m, device_name, output_path) + + table = generate_measurement_table( + measurements, + excluded_keys={ + "http_port", + "https_port", + "http_served_paths", + "log_level", + }, + ) + print(table) + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_buffer.py b/tests/unit/test_buffer.py index 75c7d1e..c48800a 100644 --- a/tests/unit/test_buffer.py +++ b/tests/unit/test_buffer.py @@ -1,6 +1,6 @@ import unittest -from .utils import load_module +from utils import load_module class BufferTestBase(unittest.TestCase): diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index a17ed72..fc0834f 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -2,7 +2,7 @@ from unittest.mock import patch from os import getcwd -from .utils import load_module +from utils import load_module class TestHelpers(unittest.TestCase): diff --git a/tests/unit/test_http.py b/tests/unit/test_http.py index 6cb3833..34b49db 100644 --- a/tests/unit/test_http.py +++ b/tests/unit/test_http.py @@ -4,8 +4,8 @@ from unittest import mock from unittest.mock import patch, mock_open -from .utils import stat_factory -from .http_base import TestHttpBase +from utils import stat_factory +from http_base import TestHttpBase class TestWebStateMachineHelpers(TestHttpBase): diff --git a/tests/unit/test_http_file_server.py b/tests/unit/test_http_file_server.py index a72c545..21ce9aa 100644 --- a/tests/unit/test_http_file_server.py +++ b/tests/unit/test_http_file_server.py @@ -2,8 +2,8 @@ from unittest.mock import patch, mock_open, call -from .utils import stat_factory -from .http_base import TestHttpBase +from utils import stat_factory +from http_base import TestHttpBase def patch_os_stat(stat_is_file=True): diff --git a/tests/unit/test_http_multipart.py b/tests/unit/test_http_multipart.py index 2c0af5c..06246e1 100644 --- a/tests/unit/test_http_multipart.py +++ b/tests/unit/test_http_multipart.py @@ -3,7 +3,7 @@ from unittest import mock -from .http_base import TestHttpBase +from http_base import TestHttpBase class TestMultipartStateMachine(TestHttpBase):