suite.suite¶

class otto.suite.suite.OttoSuite¶

Bases: Generic[TOptions]

Base class for otto test suites.

Subclass this and decorate with @register_suite() to register your suite as an otto test <ClassName> subcommand. OttoSuite is a plain class (not unittest.TestCase), so all standard pytest features work natively — fixtures, @pytest.mark.parametrize, markers, conftest.py, and yield-based setup/teardown.

Defining suite options¶

Define a @dataclass before the suite class, annotate each field with Annotated[T, typer.Option(help="...")] so the help text appears in otto test <ClassName> --help, then pass it as the generic argument:

from dataclasses import dataclass
from typing import Annotated

import typer

from otto.suite import OttoSuite, register_suite

@dataclass
class _Opts:
    device_type: Annotated[str, typer.Option(
        help="Kind of device under test ('router', 'switch').",
    )] = "router"

@register_suite()
class TestMyDevice(OttoSuite[_Opts]):
    """Validate device configuration."""
    Options = _Opts

Accessing options in tests¶

Request the suite_options fixture as a parameter. It provides the Options instance constructed from CLI arguments:

async def test_device_reachable(self, suite_options: _Opts) -> None:
    self.logger.info(f"Testing {suite_options.device_type}")

Parametrized tests¶

Use @pytest.mark.parametrize on test methods:

@pytest.mark.parametrize("interface", ["eth0", "eth1", "mgmt0"])
async def test_interface_up(self, interface: str) -> None:
    self.logger.info(f"Checking {interface}")
    assert True

Each parameter combination gets its own testDir with a sanitized name.

Using fixtures¶

Test methods can request any pytest fixture as a parameter. Define shared fixtures in a conftest.py alongside your suite:

# conftest.py
@pytest.fixture
async def primary_host():
    from otto.configmodule import get_host
    host = get_host("primary")
    yield host
    await host.close()

# test_device.py
async def test_with_host(self, primary_host) -> None:
    result = await primary_host.oneshot("echo hello")
    assert result.status == Status.Success

Inheriting repo-wide options¶

Create a shared base dataclass in your pylib and inherit from it to share common options across multiple suites:

# pylib/my_suites/options.py
@dataclass
class RepoOptions:
    lab_env: Annotated[str, typer.Option(
        help="Lab environment to target.",
    )] = "staging"

# tests/test_device.py
@dataclass
class _Opts(RepoOptions):                 # inherits --lab-env
    firmware: Annotated[str, typer.Option(
        help="Firmware version to validate.",
    )] = "latest"

@register_suite()
class TestDevice(OttoSuite[_Opts]):
    Options = _Opts

Both --lab-env and --firmware appear in otto test TestDevice --help.

The same RepoOptions dataclass may also be passed to @instruction(options=...) so that otto run subcommands expose the same repo-wide flags as otto test — see otto.cli.run.instruction().

Built-in autouse fixtures¶

OttoSuite provides three autouse fixtures that run for every test:

_otto_log_test_start — logs a banner marking the start of each test
_otto_test_dir — creates self.testDir with sanitized node name
_otto_monitor_events — records monitor start/end events

Per-test timeouts are enforced by the shared _otto_timeout fixture in otto.suite.timeout, registered automatically by OttoPlugin. Apply @pytest.mark.timeout(seconds) to individual tests, classes, or set a timeout class attribute on the suite.

setup_method(method=None)¶

Return type:¶: None

teardown_method(method=None)¶

Return type:¶: None

classmethod setup_class()¶

classmethod teardown_class()¶

expect(condition, msg=None)¶

Record a non-fatal expectation.

Unlike assert, a failing expect() does not stop the test. All failures are collected and reported together when the test finishes — the test is marked failed only at that point.

Use assert for preconditions that must hold before the test can continue (fatal). Use expect() for checks where you want to see all failures at once (non-fatal).

Parameters:¶

condition – Any truthy/falsy expression to evaluate.
msg – Optional human-friendly message printed alongside the auto-captured source line and locals — not a replacement.

Examples

Fatal vs non-fatal:

# Fatal — test stops here if command itself failed
assert result.status == Status.Success

# Non-fatal — records failure, test continues
self.expect("hostname" in result.output)
self.expect("interface" in result.output)
self.expect(result.retcode == 0, "unexpected retcode")

The failure report always includes the source location and caller locals. When msg is provided it appears in addition to the auto-captured source context, never replacing it:

>>> from unittest.mock import MagicMock
>>> from otto.suite.suite import OttoSuite
>>> suite = OttoSuite()
>>> suite._expect_failures = []
>>> suite.logger = MagicMock()
>>> x = 42
>>> suite.expect(x == 99, "math is broken")
>>> report = suite._expect_failures[0]
>>> "Message: math is broken" in report
True
>>> "x = 42" in report
True

Note

The auto-captured source line and locals are best-effort. Provide msg when the expression alone isn’t self-explanatory.

Return type:¶: None

async startMonitor(hosts=None, interval=datetime.timedelta(seconds=5), parsers=None, port=0, bind='127.0.0.1', db_path=None, targets=None)¶

Start metric collection from all hosts and launch the web dashboard.

Must be called with await:

url = await self.startMonitor(hosts=[host])

All hosts are polled simultaneously on each tick via asyncio.gather(). Series keys in results are "hostname/metric_label".

Parameters:¶

hosts – The RemoteHosts to monitor. Ignored when targets is provided.
interval – How often to poll the hosts. timedelta or float (seconds).
parsers – Custom metric parsers applied to all hosts. Ignored when targets is provided.
port – TCP port for the dashboard web server (0 = auto-assign).
bind – Address to bind to. Use ‘0.0.0.0’ for access from other machines.
db_path – Path for SQLite persistence. If None, data is in-memory only.
targets – Per-host MonitorTarget objects. When provided, hosts and parsers are ignored. Use this to assign different parsers to different hosts.

Return type:¶

str

Returns:¶

Dashboard URL, e.g. ‘http – //127.0.0.1:8080’.

async stopMonitor()¶

Stop metric collection and shut down the dashboard server.

Must be called with await:

await self.stopMonitor()

Return type:¶: None

async addMonitorEvent(label, color='#888888', dash='dash')¶

Record a labeled event on the live dashboard at the current time.

Has no effect if monitoring is not active. Honors a per-suite collector created by startMonitor() first, then falls back to the session-wide collector started by otto test --monitor.

Return type:¶: None

getMonitorResults()¶

Return collected metric series after stopMonitor(). Empty dict if never started.

Return type:¶: dict[str, list[tuple[datetime, float]]]

getMonitorEvents()¶

Return all recorded events after stopMonitor(). Empty list if never started.

Return type:¶: list[MonitorEvent]