Accessible objects and listeners

Test Junkie gives you two ways into test results: a structured object graph you can inspect after the run finishes, and a real-time event listener you can hook into as each test completes. Use the object graph for post-run reporting; use listeners for immediate notifications.

Post-run: the object graph

After runner.run() returns, call runner.get_executed_suites() to get a list of SuiteObjects. From each suite, call suite.get_test_objects() to get TestObjects. From each test, call test_obj.metrics.get_metrics() to get per-variant status data:

report.pypython

from test_junkie.runner import Runner
from test_junkie.constants import TestCategory

runner = Runner([CheckoutSuite, LoginSuite])
runner.run()

for suite in runner.get_executed_suites():
    for test_obj in suite.get_test_objects():
        metrics = test_obj.metrics.get_metrics()
        # metrics is nested: {suite_param: {test_param: data}}
        for _, by_param in metrics.items():
            for _, data in by_param.items():
                if data["status"] == TestCategory.FAIL:
                    slack.post(
                        f"{test_obj.get_function_name()} failed: "
                        + str(data["exceptions"][-1])
                    )

What's in the metrics dict

The data dict for each variant contains:

status — one of TestCategory.FAIL, TestCategory.ERROR, TestCategory.SUCCESS, TestCategory.SKIP, TestCategory.CANCEL, or TestCategory.IGNORE
exceptions — list of exception objects raised during all attempts (including retries)
tracebacks — list of formatted traceback strings, one per exception

report.pypython

from test_junkie.constants import TestCategory

# check for both failures and unexpected errors
for _, by_param in metrics.items():
    for _, data in by_param.items():
        status = data["status"]
        if status in (TestCategory.FAIL, TestCategory.ERROR):
            exceptions = data["exceptions"]
            tracebacks = data["tracebacks"]
            # your reporting here, not ours
            jira.create_issue(str(exceptions[-1]), tracebacks[-1])

Real-time events: Listener

If you need to react as each test completes — not after the whole run — subclass Listener and override the events you care about. Pass the class to listener= on @Suite:

listeners.pypython

from test_junkie.listener import Listener

class AlertListener(Listener):

    def on_failure(self, **kwargs):
        test_obj = kwargs["properties"]["jm"]["jto"]
        exception = kwargs["exception"]
        slack.post(
            f"FAIL: {test_obj.get_function_name()} — {exception}"
        )

    def on_error(self, **kwargs):
        test_obj = kwargs["properties"]["jm"]["jto"]
        exception = kwargs["exception"]
        trace = kwargs["trace"]
        pagerduty.alert(str(exception), trace)

checkout_suite.pypython

@Suite(listener=AlertListener)
class CheckoutSuite:
    ...

@Suite(listener=AlertListener)
class LoginSuite:
    ...

Listener vs. post-run objects: when to use which

Use a Listener when you need immediate notification as each test finishes — Slack alerts, live dashboards. Use the post-run object graph when you need to aggregate across all tests before acting — custom reports, Jira bulk-create, CI gate decisions.

Available listener events

The full set of events you can override:

listener_reference.pypython

class MyListener(Listener):
    # test-level events
    def on_success(self, **kwargs): ...
    def on_failure(self, **kwargs): ...
    def on_error(self, **kwargs): ...
    def on_skip(self, **kwargs): ...
    def on_cancel(self, **kwargs): ...
    def on_ignore(self, **kwargs): ...
    def on_in_progress(self, **kwargs): ...
    def on_complete(self, **kwargs): ...
    # suite-level events
    def on_before_class_failure(self, **kwargs): ...
    def on_before_class_error(self, **kwargs): ...
    def on_after_class_failure(self, **kwargs): ...
    def on_after_class_error(self, **kwargs): ...
    def on_class_skip(self, **kwargs): ...
    def on_class_complete(self, **kwargs): ...

Internal reporting: persisting results to a database

Listeners are well-suited for writing results to a database as each test completes. The pattern is: create a run record before the runner starts (generating a run ID), then use the listener to insert a result row per test, referencing that run ID as a foreign key. This gives you a queryable history of every run and every result.

1. Schema

A minimal two-table structure — one row per run, one row per test variant result:

db.pypython

import sqlite3, uuid

def init_db(path="results.db"):
    con = sqlite3.connect(path)
    con.executescript("""
        CREATE TABLE IF NOT EXISTS test_runs (
            id        TEXT PRIMARY KEY,
            started_at TEXT
        );
        CREATE TABLE IF NOT EXISTS test_results (
            id             INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id         TEXT REFERENCES test_runs(id),
            suite          TEXT,
            test           TEXT,
            parameter      TEXT,
            suite_parameter TEXT,
            status         TEXT,
            exception      TEXT,
            traceback      TEXT
        );
    """)
    con.commit()
    return con

def create_run(con):
    run_id = str(uuid.uuid4())
    con.execute("INSERT INTO test_runs (id, started_at) VALUES (?, datetime('now'))", (run_id,))
    con.commit()
    return run_id

2. Listener

Test Junkie instantiates the listener class itself when building each suite, so you can't inject a run ID through the constructor. Use class-level variables for shared state — set them before the runner starts and every suite's listener instance reads the same values:

db_listener.pypython

from test_junkie.listener import Listener

class DbListener(Listener):
    con = None     # set before runner.run()
    run_id = None  # set before runner.run()

    def _insert(self, kwargs, status, exception=None, trace=None):
        props = kwargs["properties"]
        test_obj = props["jm"]["jto"]
        suite_obj = props["jm"]["jso"]
        DbListener.con.execute(
            """INSERT INTO test_results
               (run_id, suite, test, parameter, suite_parameter, status, exception, traceback)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                DbListener.run_id,
                suite_obj.get_class_name(),
                test_obj.get_function_name(),
                str(props["test_meta"]["parameter"]),
                str(props["suite_meta"]["parameter"]),
                status,
                str(exception) if exception else None,
                trace,
            )
        )
        DbListener.con.commit()

    def on_success(self, **kwargs):
        self._insert(kwargs, status="pass")

    def on_failure(self, **kwargs):
        self._insert(kwargs, status="fail",
                     exception=kwargs["exception"], trace=kwargs["trace"])

    def on_error(self, **kwargs):
        self._insert(kwargs, status="error",
                     exception=kwargs["exception"], trace=kwargs["trace"])

    def on_skip(self, **kwargs):
        self._insert(kwargs, status="skip")

3. Wiring it together

Set the class-level state before the runner starts, then pass the class (not an instance) to each suite via listener=:

run.pypython

from db import init_db, create_run
from db_listener import DbListener
from test_junkie.runner import Runner

# set class-level state before the runner instantiates the listener
DbListener.con = init_db("results.db")
DbListener.run_id = create_run(DbListener.con)

runner = Runner([CheckoutSuite, LoginSuite])
runner.run()
print(f"Run {DbListener.run_id} complete. Query results.db to report.")

suites.pypython

from db_listener import DbListener

@Suite(listener=DbListener)
class CheckoutSuite: ...

@Suite(listener=DbListener)
class LoginSuite: ...

Why class variables and not instance variables

Test Junkie calls MyListener(class_meta=...) to create a new listener instance per suite — you don't control that call. Class-level variables are shared across all instances of the class, so every suite's listener reads the same run_id and con. Set them before runner.run() and they're available to all.

Custom reports from the post-run object graph

For reports that need the full picture — totals, pass rates, aggregated tracebacks — the post-run object graph is simpler than accumulating state in a listener. Iterate once after runner.run() returns and generate whatever format you need:

report.pypython

from test_junkie.constants import TestCategory

def build_report(runner):
    rows = []
    for suite in runner.get_executed_suites():
        for test_obj in suite.get_test_objects():
            metrics = test_obj.metrics.get_metrics()
            for suite_param, by_param in metrics.items():
                for test_param, data in by_param.items():
                    rows.append({
                        "suite":         suite.get_class_name(),
                        "test":          test_obj.get_function_name(),
                        "owner":         test_obj.get_owner(),
                        "component":     test_obj.get_component(),
                        "suite_param":   suite_param,
                        "test_param":    test_param,
                        "status":        data["status"],
                        "exception":     str(data["exceptions"][-1])
                                         if data["exceptions"] else None,
                    })
    return rows

runner.run()
results = build_report(runner)

passed  = [r for r in results if r["status"] == TestCategory.SUCCESS]
failed  = [r for r in results if r["status"] == TestCategory.FAIL]
errored = [r for r in results if r["status"] == TestCategory.ERROR]
print(f"{len(passed)} passed, {len(failed)} failed, {len(errored)} errored")

# write to any format — CSV, HTML, JSON, markdown
import json
with open("report.json", "w") as f:
    json.dump(results, f, indent=2)

Combining both: listener writes live, object graph builds the summary

The two approaches compose cleanly. Use the listener for real-time notification and the object graph for the post-run summary — they operate on the same underlying data:

run.pypython

DbListener.con = init_db()
DbListener.run_id = create_run(DbListener.con)

runner = Runner([CheckoutSuite, LoginSuite])
runner.run()

# DB already has every row — listener wrote them in real time
# now generate the summary from the object graph
results = build_report(runner)
send_summary_email(
    run_id=DbListener.run_id,
    passed=len([r for r in results if r["status"] == TestCategory.SUCCESS]),
    failed=len([r for r in results if r["status"] == TestCategory.FAIL]),
)