Skip to content

Plugin implementation¤

pytest_addoption ¤

pytest_addoption(parser: Parser) -> None

Register argparse-style options and ini-style config values.

This function is called once at the beginning of a test run.

Performs the following action:

  • Get or create the terminal reporting group in the parser.
  • Add the --collect-report option to the group.
  • Add the --collect-log option to the group.
  • Add the --collect-url option to the group.
  • Add the --collect-log-url option to the group.

See pytest.hookspec.pytest_addoption.

Source code in src/pytest_broadcaster/plugin.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def pytest_addoption(parser: pytest.Parser) -> None:
    """Register argparse-style options and ini-style config values.

    This function is called once at the beginning of a test run.

    Performs the following action:

    - Get or create the `terminal reporting` group in the parser.
    - Add the `--collect-report` option to the group.
    - Add the `--collect-log` option to the group.
    - Add the `--collect-url` option to the group.
    - Add the `--collect-log-url` option to the group.

    See [pytest.hookspec.pytest_addoption][_pytest.hookspec.pytest_addoption].
    """
    group = parser.getgroup(
        name="terminal reporting",
        description="pytest-broadcaster plugin options",
    )
    group.addoption(
        "--collect-report",
        action="store",
        metavar="path",
        default=None,
        help="Path to JSON output file holding collected items.",
    )
    group.addoption(
        "--collect-log",
        action="store",
        metavar="path",
        default=None,
        help="Path to JSON Lines output file where events are logged to.",
    )
    group.addoption(
        "--collect-url",
        action="store",
        metavar="url",
        default=None,
        help="URL to send collected items to.",
    )
    group.addoption(
        "--collect-log-url",
        action="store",
        metavar="url",
        default=None,
        help="URL to send events to.",
    )

pytest_configure ¤

pytest_configure(config: Config) -> None

Perform initial plugin configuration.

This function is called once after command line options have been parsed.

Perform the following actions:

  • Skip if workerinput is present, which means we are in a worker process.
  • Create a JSONFile destination if the JSON output file path is present.
  • Create a JSONLinesFile destination if the JSON Lines output file path is present.
  • Create an HTTPWebhook destination if the URL is present.
  • Create an HTTPWebhook destination if the URL for the JSON Lines output file is present.
  • Let the user add their own destinations if they want to.
  • Create the default reporter.
  • Let the user set the reporter if they want to.
  • Create, open and register the plugin instance.
  • Store the plugin instance in the config object.

See pytest.hookspec.pytest_configure.

Source code in src/pytest_broadcaster/plugin.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def pytest_configure(config: pytest.Config) -> None:
    """Perform initial plugin configuration.

    This function is called once after command line options have been parsed.

    Perform the following actions:

    - Skip if workerinput is present, which means we are in a worker process.
    - Create a JSONFile destination if the JSON output file path is present.
    - Create a JSONLinesFile destination if the JSON Lines output file path is present.
    - Create an HTTPWebhook destination if the URL is present.
    - Create an HTTPWebhook destination if the URL for the JSON Lines output file is present.
    - Let the user add their own destinations if they want to.
    - Create the default reporter.
    - Let the user set the reporter if they want to.
    - Create, open and register the plugin instance.
    - Store the plugin instance in the config object.

    See [pytest.hookspec.pytest_configure][_pytest.hookspec.pytest_configure].
    """
    # Skip if pytest-xdist worker
    if hasattr(config, "workerinput"):
        return

    # Create publishers
    destinations: list[Destination] = []

    if json_path := config.option.collect_report:
        destinations.append(JSONFile(json_path))

    if json_lines_path := config.option.collect_log:
        destinations.append(JSONLinesFile(json_lines_path))

    if json_url := config.option.collect_url:
        destinations.append(HTTPWebhook(json_url, emit_events=False, emit_result=True))

    if json_lines_url := config.option.collect_log_url:
        destinations.append(
            HTTPWebhook(json_lines_url, emit_events=True, emit_result=False)
        )

    def add_destination(destination: Destination) -> None:
        destinations.append(destination)

    # Let the user add their own destinations if they want to
    config.hook.pytest_broadcaster_add_destination(add=add_destination)

    # Create default reporter
    reporter_to_use: Reporter = DefaultReporter()

    def set_reporter(reporter: Reporter) -> None:
        nonlocal reporter_to_use
        reporter_to_use = reporter

    # Let the user set the reporter if they want to
    config.hook.pytest_broadcaster_set_reporter(set=set_reporter)

    # Create plugin instance.
    plugin = PytestBroadcasterPlugin(
        config=config,
        reporter=reporter_to_use,
        publishers=destinations,
    )
    # Open the plugin
    plugin.open()
    # Register the plugin with the plugin manager.
    config.pluginmanager.register(plugin)
    setattr(config, __PLUGIN_ATTR__, plugin)

pytest_unconfigure ¤

pytest_unconfigure(config: Config) -> None

Perform final plugin teardown.

This function is called once after all test are executed and before test process is exited.

See pytest.hookspec.pytest_unconfigure.

Perform the following actions:

  • Extract the plugin instance from the config object.
  • Close the plugin instance.
  • Delete the plugin instance from the config object.
Source code in src/pytest_broadcaster/plugin.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def pytest_unconfigure(config: pytest.Config) -> None:
    """Perform final plugin teardown.

    This function is called once after all test are executed and before test process is
    exited.

    See [pytest.hookspec.pytest_unconfigure][_pytest.hookspec.pytest_unconfigure].

    Perform the following actions:

    - Extract the plugin instance from the config object.
    - Close the plugin instance.
    - Delete the plugin instance from the config object.
    """
    plugin: PytestBroadcasterPlugin | None = getattr(config, __PLUGIN_ATTR__, None)
    if plugin:
        plugin.close()
        config.pluginmanager.unregister(plugin)
        delattr(config, __PLUGIN_ATTR__)

PytestBroadcasterPlugin ¤

PytestBroadcasterPlugin(
    config: Config,
    reporter: Reporter,
    publishers: list[Destination],
)

A pytest plugin to log collection to a line-based JSON file.

Methods:

Name Description
close

Close the plugin instance.
open

Open the plugin instance.
pytest_collectreport

Collector finished collecting a node.
pytest_exception_interact

Collector encountered an error.
pytest_runtest_logfinish

Pytest calls this function after running the runtest protocol for a single item.
pytest_runtest_logreport

Process the TestReport produced for each of the setup, call and teardown runtest steps of a test case.
pytest_sessionfinish

Write a session end event.
pytest_sessionstart

Write a session start event.
pytest_terminal_summary

Add a section to terminal summary reporting.
pytest_warning_recorded

Process a warning captured during the session.
Source code in src/pytest_broadcaster/plugin.py 
185
186
187
188
189
190
191
192
193
194
195
def __init__(
    self,
    config: pytest.Config,
    reporter: Reporter,
    publishers: list[Destination],
) -> None:
    """Create a new pytest broadcaster plugin."""
    self.config = config
    self.publishers = publishers
    self.reporter = reporter
    self.stack = ExitStack()

close ¤

close() -> None

Close the plugin instance.

Perform the following actions:

  • Close the JSON Lines output file (if any).
  • Write the results to the JSON output file (if any)
Source code in src/pytest_broadcaster/plugin.py 
215
216
217
218
219
220
221
222
223
224
225
def close(self) -> None:
    """Close the plugin instance.

    Perform the following actions:

    - Close the JSON Lines output file (if any).
    - Write the results to the JSON output file (if any)
    """
    if result := self.reporter.make_session_result():
        self._write_result(result)
    self.stack.close()

open ¤

open() -> None

Open the plugin instance.

Perform the following actions:

  • Skip if there is no JSON Lines output
  • Raise an error if the JSON Lines output file is already open.
  • Ensure the parent directory of JSON Lines output file exists.
  • Open the JSON Lines output file in write mode (erasing any previous content)
Source code in src/pytest_broadcaster/plugin.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def open(self) -> None:
    """Open the plugin instance.

    Perform the following actions:

    - Skip if there is no JSON Lines output
    - Raise an error if the JSON Lines output file is already open.
    - Ensure the parent directory of JSON Lines output file exists.
    - Open the JSON Lines output file in write mode (erasing any previous content)
    """
    for publisher in self.publishers:
        try:
            self.stack.enter_context(publisher)
        except Exception as e:  # noqa: PERF203, BLE001
            warnings.warn(
                f"Failed to open publisher: {publisher} - {e!r}", stacklevel=1
            )

pytest_collectreport ¤

pytest_collectreport(report: CollectReport) -> None

Collector finished collecting a node.

See pytest.hookspec.pytest_collectreport.

Source code in src/pytest_broadcaster/plugin.py 
281
282
283
284
285
286
287
288
289
def pytest_collectreport(self, report: pytest.CollectReport) -> None:
    """Collector finished collecting a node.

    See [pytest.hookspec.pytest_collectreport][_pytest.hookspec.pytest_collectreport].
    """
    # Skip if the report failed.
    if report.failed:
        return
    self._write_event(self.reporter.make_collect_report(report))

pytest_exception_interact ¤

pytest_exception_interact(
    node: Item | Collector,
    call: CallInfo[Any],
    report: TestReport | CollectReport,
) -> None

Collector encountered an error.

See pytest.hookspec.pytest_exception_interact.

Source code in src/pytest_broadcaster/plugin.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def pytest_exception_interact(
    self,
    node: pytest.Item | pytest.Collector,
    call: pytest.CallInfo[Any],
    report: pytest.TestReport | pytest.CollectReport,
) -> None:
    """Collector encountered an error.

    See [pytest.hookspec.pytest_exception_interact][_pytest.hookspec.pytest_exception_interact].
    """
    # Skip if the report is not a test report.
    if isinstance(report, pytest.TestReport):
        return
    self._write_event(self.reporter.make_error_message(report, call))

pytest_runtest_logfinish ¤

pytest_runtest_logfinish(
    nodeid: str, location: tuple[str, int | None, str]
) -> None

Pytest calls this function after running the runtest protocol for a single item.

See pytest.hookspec.pytest_runtest_logfinish.

Source code in src/pytest_broadcaster/plugin.py 
298
299
300
301
302
303
304
305
def pytest_runtest_logfinish(
    self, nodeid: str, location: tuple[str, int | None, str]
) -> None:
    """Pytest calls this function after running the runtest protocol for a single item.

    See [pytest.hookspec.pytest_runtest_logfinish][_pytest.hookspec.pytest_runtest_logfinish].
    """
    self._write_event(self.reporter.make_test_case_end(nodeid))

pytest_runtest_logreport ¤

pytest_runtest_logreport(report: TestReport) -> None

Process the TestReport produced for each of the setup, call and teardown runtest steps of a test case.

See pytest.hookspec.pytest_runtest_logreport.

Source code in src/pytest_broadcaster/plugin.py 
291
292
293
294
295
296
def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
    """Process the [TestReport][pytest.TestReport] produced for each of the setup, call and teardown runtest steps of a test case.

    See [pytest.hookspec.pytest_runtest_logreport][_pytest.hookspec.pytest_runtest_logreport].
    """
    self._write_event(self.reporter.make_test_case_step(report))

pytest_sessionfinish ¤

pytest_sessionfinish(exitstatus: int) -> None

Write a session end event.

This function is called after whole test run finished, right before returning the exit status to the system.

See pytest.hookspec.pytest_sessionfinish

Source code in src/pytest_broadcaster/plugin.py 
237
238
239
240
241
242
243
244
245
def pytest_sessionfinish(self, exitstatus: int) -> None:
    """Write a session end event.

    This function is called after whole test run finished, right before returning
    the exit status to the system.

    See [pytest.hookspec.pytest_sessionfinish][_pytest.hookspec.pytest_sessionfinish]
    """
    self._write_event(self.reporter.make_session_end(exitstatus))

pytest_sessionstart ¤

pytest_sessionstart() -> None

Write a session start event.

This function is called after the Session object has been created and before performing collection and entering the run test loop.

See pytest.hookspec.pytest_sessionstart.

Source code in src/pytest_broadcaster/plugin.py 
227
228
229
230
231
232
233
234
235
def pytest_sessionstart(self) -> None:
    """Write a session start event.

    This function is called after the [Session object][pytest.Session] has been
    created and before performing collection and entering the run test loop.

    See [pytest.hookspec.pytest_sessionstart][_pytest.hookspec.pytest_sessionstart].
    """
    self._write_event(self.reporter.make_session_start())

pytest_terminal_summary ¤

pytest_terminal_summary(
    terminalreporter: TerminalReporter,
) -> None

Add a section to terminal summary reporting.

See pytest.hookspec.pytest_terminal_summary.

Source code in src/pytest_broadcaster/plugin.py 
307
308
309
310
311
312
313
314
def pytest_terminal_summary(self, terminalreporter: TerminalReporter) -> None:
    """Add a section to terminal summary reporting.

    See [pytest.hookspec.pytest_terminal_summary][_pytest.hookspec.pytest_terminal_summary].
    """
    for publisher in self.publishers:
        if summary := publisher.summary():
            terminalreporter.write_sep("-", f"generated report log file: {summary}")

pytest_warning_recorded ¤

pytest_warning_recorded(
    warning_message: WarningMessage,
    when: Literal["config", "collect", "runtest"],
    nodeid: str,
    location: tuple[str, int, str] | None,
) -> None

Process a warning captured during the session.

See pytest.hookspec.pytest_warning_recorded.

Source code in src/pytest_broadcaster/plugin.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def pytest_warning_recorded(
    self,
    warning_message: warnings.WarningMessage,
    when: Literal["config", "collect", "runtest"],
    nodeid: str,
    location: tuple[str, int, str] | None,
) -> None:
    """Process a warning captured during the session.

    See [pytest.hookspec.pytest_warning_recorded][_pytest.hookspec.pytest_warning_recorded].
    """
    self._write_event(
        self.reporter.make_warning_message(
            warning_message=warning_message,
            when=when,
            nodeid=nodeid,
        )
    )