Python Test Stand

test_stand.py

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Tuple, Union


class TestStand(ABC):
    """
    A simple test stand for running async assertions multiple times.

    Example:
        ```python
        class MyTestStand(TestStand):
            async def validate(self, **kwargs) -> "TestStand.ValidateRes":
                actual = await some_async_function(kwargs["param"])
                expected = "expected value"
                return self.compare(actual, expected, "Check if actual matches expected")
        await MyTestStand(runs=5).run(param="test")

        # ✅ DONE: 0/5 failed.
        ```
    """

    @dataclass
    class Validated:
        actual: object
        expected: object
        message: str  # log message
        ctx: dict | None  # log context

    ValidateRes = Union[Validated, Tuple[Validated, ...]]

    log_success = False
    log = print

    def __init__(self, runs: int = 10):
        self.runs = runs

    @abstractmethod
    async def validate(self, **kwargs) -> "TestStand.ValidateRes":
        """
        Implement in subclass, perform assertions and return results.
        A single `validate` call can return one or more assertions.
        """

    async def run(self, **kwargs):
        failed_count = 0
        started_at_total = datetime.now(timezone.utc)
        started_at_run = started_at_total
        for run_ix in range(self.runs):
            started_at = datetime.now(timezone.utc)
            time_diff_total = (started_at - started_at_total).total_seconds()
            time_diff_run = (started_at - started_at_run).total_seconds()
            started_at_run = started_at
            self.log(f"⌛ Run {run_ix+1}/{self.runs}, +{time_diff_total:.3f}s (+{time_diff_run:.3f}s)")

            try:
                validated: "TestStand.ValidateRes" = await self.validate(**kwargs)
            except Exception as e:
                validated = self.error("Uncaught run exception", ctx={"exception": str(e)})
            if not isinstance(validated, tuple):
                validated = (validated,)

            if wrong_return := next((item for item in validated if not isinstance(item, TestStand.Validated)), None):
                self.log(f"  ‼️ ERROR: Invalid return type: {type(wrong_return).__name__}")
                failed_count = self.runs
                break

            for item_ix, item in enumerate(validated):
                if item.actual != item.expected:
                    failed_count += 1
                    log = "  🟡 FAILURE: "
                elif self.log_success:
                    log = "  🟢 SUCCESS: "
                else:
                    continue
                log += f"Item {item_ix+1}/{len(validated)} ({item.message or '…'})"
                if not (item.actual is False and item.expected is True):
                    log += f": {item.actual}, expected: {item.expected}"
                self.log(log)
                ctx = item.ctx or {}
                for key, value in ctx.items():
                    self.log(f"    🗒️ {key}: {value}")

        self.log(f"{'✅' if not failed_count else '❌'} DONE: {failed_count}/{self.runs} failed.")

    @staticmethod
    def compare(actual: object, expected: object, message: str = "", ctx: dict | None = None) -> "TestStand.Validated":
        return TestStand.Validated(actual=actual, expected=expected, message=message, ctx=ctx)

    @staticmethod
    def assertion(condition: bool, message: str = "", ctx: dict | None = None) -> "TestStand.Validated":
        return TestStand.Validated(actual=condition, expected=True, message=message, ctx=ctx)

    @staticmethod
    def error(message: str, ctx: dict | None = None) -> "TestStand.Validated":
        return TestStand.Validated(actual=False, expected=True, message=message, ctx=ctx)