Can I pass back data to the python code calling pytest.main() where pytest.main() is executing unittests.

[PatrickFlaherty]

Can I pass back data to the python code calling pytest.main() where pytest.main() is executing unittests.
`class AllureRunner:

def run_tests_with_allure_report(self, tests_path):

    params = ['--tb=native',  f'--alluredir=allure_dir\\results', f'{tests_path}']

    # Run Unit Tests
    response = pytest.main(params)

    print("return code from calling pytest.main():", response)`

[RonnyPfannschmidt]

The second argument to the main function is a list of pytest plugin objects each of which can implement any pytest Hook

[PatrickFlaherty]

@RonnyPfannschmidt thanks for the response. I have not used Hooks before, Could you provide an example of this?
Thank you!

[RonnyPfannschmidt]

I believe there's one in the docs,in case you don't find it please Ping me so I can find it when I get back to the computer tomorrow

[PatrickFlaherty]

I will have a look, thank you @RonnyPfannschmidt .

[PatrickFlaherty]

Hi @RonnyPfannschmidt , have not found that example, would you be so kind as to post if you have a moment. Thank you.

[RonnyPfannschmidt]

https://docs.pytest.org/en/7.1.x/example/simple.html?highlight=main#freezing-pytest has a example of passing a plugin object (in this case its a python module with the hooks

pytest/src/_pytest/junitxml.py

Lines 464 to 700 in cdddd6d

	class LogXML:
	def __init__(
	self,
	logfile,
	prefix: Optional[str],
	suite_name: str = "pytest",
	logging: str = "no",
	report_duration: str = "total",
	family="xunit1",
	log_passing_tests: bool = True,
	) -> None:
	logfile = os.path.expanduser(os.path.expandvars(logfile))
	self.logfile = os.path.normpath(os.path.abspath(logfile))
	self.prefix = prefix
	self.suite_name = suite_name
	self.logging = logging
	self.log_passing_tests = log_passing_tests
	self.report_duration = report_duration
	self.family = family
	self.stats: Dict[str, int] = dict.fromkeys(
	["error", "passed", "failure", "skipped"], 0
	)
	self.node_reporters: Dict[
	Tuple[Union[str, TestReport], object], _NodeReporter
	] = {}
	self.node_reporters_ordered: List[_NodeReporter] = []
	self.global_properties: List[Tuple[str, str]] = []
	# List of reports that failed on call but teardown is pending.
	self.open_reports: List[TestReport] = []
	self.cnt_double_fail_tests = 0
	# Replaces convenience family with real family.
	if self.family == "legacy":
	self.family = "xunit1"
	def finalize(self, report: TestReport) -> None:
	nodeid = getattr(report, "nodeid", report)
	# Local hack to handle xdist report order.
	workernode = getattr(report, "node", None)
	reporter = self.node_reporters.pop((nodeid, workernode))
	for propname, propvalue in report.user_properties:
	reporter.add_property(propname, str(propvalue))
	if reporter is not None:
	reporter.finalize()
	def node_reporter(self, report: Union[TestReport, str]) -> _NodeReporter:
	nodeid: Union[str, TestReport] = getattr(report, "nodeid", report)
	# Local hack to handle xdist report order.
	workernode = getattr(report, "node", None)
	key = nodeid, workernode
	if key in self.node_reporters:
	# TODO: breaks for --dist=each
	return self.node_reporters[key]
	reporter = _NodeReporter(nodeid, self)
	self.node_reporters[key] = reporter
	self.node_reporters_ordered.append(reporter)
	return reporter
	def add_stats(self, key: str) -> None:
	if key in self.stats:
	self.stats[key] += 1
	def _opentestcase(self, report: TestReport) -> _NodeReporter:
	reporter = self.node_reporter(report)
	reporter.record_testreport(report)
	return reporter
	def pytest_runtest_logreport(self, report: TestReport) -> None:
	"""Handle a setup/call/teardown report, generating the appropriate
	XML tags as necessary.
	Note: due to plugins like xdist, this hook may be called in interlaced
	order with reports from other nodes. For example:
	Usual call order:
	-> setup node1
	-> call node1
	-> teardown node1
	-> setup node2
	-> call node2
	-> teardown node2
	Possible call order in xdist:
	-> setup node1
	-> call node1
	-> setup node2
	-> call node2
	-> teardown node2
	-> teardown node1
	"""
	close_report = None
	if report.passed:
	if report.when == "call": # ignore setup/teardown
	reporter = self._opentestcase(report)
	reporter.append_pass(report)
	elif report.failed:
	if report.when == "teardown":
	# The following vars are needed when xdist plugin is used.
	report_wid = getattr(report, "worker_id", None)
	report_ii = getattr(report, "item_index", None)
	close_report = next(
	(
	rep
	for rep in self.open_reports
	if (
	rep.nodeid == report.nodeid
	and getattr(rep, "item_index", None) == report_ii
	and getattr(rep, "worker_id", None) == report_wid
	)
	),
	None,
	)
	if close_report:
	# We need to open new testcase in case we have failure in
	# call and error in teardown in order to follow junit
	# schema.
	self.finalize(close_report)
	self.cnt_double_fail_tests += 1
	reporter = self._opentestcase(report)
	if report.when == "call":
	reporter.append_failure(report)
	self.open_reports.append(report)
	if not self.log_passing_tests:
	reporter.write_captured_output(report)
	else:
	reporter.append_error(report)
	elif report.skipped:
	reporter = self._opentestcase(report)
	reporter.append_skipped(report)
	self.update_testcase_duration(report)
	if report.when == "teardown":
	reporter = self._opentestcase(report)
	reporter.write_captured_output(report)
	self.finalize(report)
	report_wid = getattr(report, "worker_id", None)
	report_ii = getattr(report, "item_index", None)
	close_report = next(
	(
	rep
	for rep in self.open_reports
	if (
	rep.nodeid == report.nodeid
	and getattr(rep, "item_index", None) == report_ii
	and getattr(rep, "worker_id", None) == report_wid
	)
	),
	None,
	)
	if close_report:
	self.open_reports.remove(close_report)
	def update_testcase_duration(self, report: TestReport) -> None:
	"""Accumulate total duration for nodeid from given report and update
	the Junit.testcase with the new total if already created."""
	if self.report_duration == "total" or report.when == self.report_duration:
	reporter = self.node_reporter(report)
	reporter.duration += getattr(report, "duration", 0.0)
	def pytest_collectreport(self, report: TestReport) -> None:
	if not report.passed:
	reporter = self._opentestcase(report)
	if report.failed:
	reporter.append_collect_error(report)
	else:
	reporter.append_collect_skipped(report)
	def pytest_internalerror(self, excrepr: ExceptionRepr) -> None:
	reporter = self.node_reporter("internal")
	reporter.attrs.update(classname="pytest", name="internal")
	reporter._add_simple("error", "internal error", str(excrepr))
	def pytest_sessionstart(self) -> None:
	self.suite_start_time = timing.time()
	def pytest_sessionfinish(self) -> None:
	dirname = os.path.dirname(os.path.abspath(self.logfile))
	# exist_ok avoids filesystem race conditions between checking path existence and requesting creation
	os.makedirs(dirname, exist_ok=True)
	with open(self.logfile, "w", encoding="utf-8") as logfile:
	suite_stop_time = timing.time()
	suite_time_delta = suite_stop_time - self.suite_start_time
	numtests = (
	self.stats["passed"]
	+ self.stats["failure"]
	+ self.stats["skipped"]
	+ self.stats["error"]
	- self.cnt_double_fail_tests
	)
	logfile.write('<?xml version="1.0" encoding="utf-8"?>')
	suite_node = ET.Element(
	"testsuite",
	name=self.suite_name,
	errors=str(self.stats["error"]),
	failures=str(self.stats["failure"]),
	skipped=str(self.stats["skipped"]),
	tests=str(numtests),
	time="%.3f" % suite_time_delta,
	timestamp=datetime.fromtimestamp(self.suite_start_time).isoformat(),
	hostname=platform.node(),
	)
	global_properties = self._get_global_properties_node()
	if global_properties is not None:
	suite_node.append(global_properties)
	for node_reporter in self.node_reporters_ordered:
	suite_node.append(node_reporter.to_xml())
	testsuites = ET.Element("testsuites")
	testsuites.append(suite_node)
	logfile.write(ET.tostring(testsuites, encoding="unicode"))
	def pytest_terminal_summary(self, terminalreporter: TerminalReporter) -> None:
	terminalreporter.write_sep("-", f"generated xml file: {self.logfile}")
	def add_global_property(self, name: str, value: object) -> None:
	__tracebackhide__ = True
	_check_record_param_type("name", name)
	self.global_properties.append((name, bin_xml_escape(value)))
	def _get_global_properties_node(self) -> Optional[ET.Element]:
	"""Return a Junit node containing custom properties, if any."""
	if self.global_properties:
	properties = ET.Element("properties")
	for name, value in self.global_properties:
	properties.append(ET.Element("property", name=name, value=value))
	return properties
	return None

is a example of a plugin object, that when registered with pytest makes the junitxml report

[PatrickFlaherty]

Thank you @RonnyPfannschmidt !