GaudiTesting.GaudiExeTest.GaudiExeTest Class Reference

Inheritance diagram for GaudiTesting.GaudiExeTest.GaudiExeTest:

Collaboration diagram for GaudiTesting.GaudiExeTest.GaudiExeTest:

Public Member Functions
None	validate_with_reference (cls, bytes data, str key, dict reference, Callable[[str, str], None] record_property)
	validate_json_with_reference (cls, str output_file, str reference_file, detailed=True)
	find_reference_block (cls, str reference_block, Callable preprocessor=None, str signature=None, int signature_offset=0)
	test_count_messages (self, reference, stdout, record_property)
None	test_stdout (self, bytes stdout, Callable record_property, dict reference)
None	test_ttrees (self, bytes stdout, Callable record_property, dict reference)
None	test_histos (self, bytes stdout, Callable record_property, dict reference)
None	test_stderr (self, bytes stderr, Callable record_property, dict reference)
	test_record_options (self, Callable record_property)
Public Member Functions inherited from GaudiTesting.SubprocessBaseTest.SubprocessBaseTest
Path|None	cwd (self)
str	resolve_path (cls, Path|str path)
None	update_env (cls, dict[str, str] env)
FixtureResult	run_program (cls, tmp_path=None)
	run_program_for_dbg (cls, tmp_path)
None	test_fixture_setup (self, Callable[[str, str], None] record_property, FixtureResult fixture_result, Path|None reference_path)
None	test_returncode (self, int returncode)

Static Public Attributes
Callable	options = None
	options_code = None
Callable	preprocessor = normalizeTestSuite
Static Public Attributes inherited from GaudiTesting.SubprocessBaseTest.SubprocessBaseTest
list	command = None
str	reference = None
list	environment = None
int	timeout = 600
int	returncode = 0
dict	popen_kwargs = {}
list	env_to_record

Protected Member Functions
list[str]	_prepare_command (cls, tmp_path=Path())
Protected Member Functions inherited from GaudiTesting.SubprocessBaseTest.SubprocessBaseTest
dict[str, str]	_prepare_environment (cls)
str	_determine_program (cls, str prog)
str	_handle_timeout (cls, subprocess.Popen proc)
	_prepare_execution (cls, tmp_path=None)

Static Protected Member Functions
str	_output_diff (dict reference_data, str output, Callable[[str], str] preprocessor=lambda x:x)
Static Protected Member Functions inherited from GaudiTesting.SubprocessBaseTest.SubprocessBaseTest
str	_collect_stack_trace (subprocess.Popen proc)
None	_terminate_process (subprocess.Popen proc)

Additional Inherited Members
Static Public Member Functions inherited from GaudiTesting.SubprocessBaseTest.SubprocessBaseTest
str	expand_vars_from (str value, dict[str, str] env)
None	unset_vars (dict[str, str] env, list[str] vars_to_unset)

Detailed Description

An extension of SubprocessBaseTest tailored to the Gaudi/LHCb workflow.
It includes additional functionalities for handling options,
preprocessing output, and validating against platform-specific reference files.

Definition at line 35 of file GaudiExeTest.py.

Member Function Documentation

_output_diff()

str GaudiTesting.GaudiExeTest.GaudiExeTest._output_diff ( dict reference_data, str output, Callable[[str], str] preprocessor = lambda x: x )

staticprotected

Compute the difference between the reference data and the current output.

Definition at line 95 of file GaudiExeTest.py.

    ) -> str:
        """
        Compute the difference between the reference data and the current output.
        """
        expected_output = (
            reference_data.splitlines()
            if hasattr(reference_data, "splitlines")
            else reference_data
        )
        actual_output = preprocessor(output).splitlines()
        return "\n".join(
            difflib.unified_diff(
                expected_output,
                actual_output,
                fromfile="expected",
                tofile="actual",
                lineterm="",
            )
        )
 

_prepare_command()

list[str] GaudiTesting.GaudiExeTest.GaudiExeTest._prepare_command ( cls, tmp_path = Path() )

protected

Override the base class to include options.

Reimplemented from GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.

Definition at line 47 of file GaudiExeTest.py.

    def _prepare_command(cls, tmp_path=Path()) -> list[str]:
        """
        Override the base class to include options.
        """
        command = super()._prepare_command(tmp_path=tmp_path)
 
        def generate_unique_options_filename(extension, directory):
            timestamp = time.strftime("%Y%m%d_%H%M%S")
            return directory / f"options_{timestamp}.{extension}"
 
        if hasattr(cls, "options") and cls.options is not None:
            options = cls.options
            filename = None
 
            # Check if options is a function
            if callable(options):
                source_lines = inspect.getsource(options).splitlines()
                clean_source = dedent(
                    "\n".join(source_lines[1:])
                )  # Skip the first line (def options():)
                filename = generate_unique_options_filename("py", tmp_path)
                with open(filename, "w") as file:
                    file.write(clean_source)
                    cls.options_code = CodeWrapper(clean_source, "python")
 
            # Check if options is a dictionary
            elif isinstance(options, dict):
                filename = generate_unique_options_filename("json", tmp_path)
                with open(filename, "w") as file:
                    json.dump(options, file, indent=4)
                    cls.options_code = CodeWrapper(options, "json")
 
            # Check if options is a string
            elif isinstance(options, str):
                options = dedent(options)
                filename = generate_unique_options_filename("opts", tmp_path)
                with open(filename, "w") as file:
                    file.write(options)
                    cls.options_code = CodeWrapper(options, "cpp")
 
            else:
                raise ValueError(f"invalid options type '{type(options).__name__}'")
 
            if filename:
                command.append(str(filename))
        return command
 

find_reference_block()

GaudiTesting.GaudiExeTest.GaudiExeTest.find_reference_block	(		cls,
		str	reference_block,
		Callable	preprocessor = None,
		str	signature = None,
		int	signature_offset = 0 )

Definition at line 194 of file GaudiExeTest.py.

    ):
        def assert_function(
            cls,
            stdout,
            record_property,
            preprocessor=preprocessor,
            signature=signature,
            signature_offset=signature_offset,
        ):
            processed_stdout = (
                preprocessor(stdout.decode("utf-8"))
                if preprocessor
                else stdout.decode("utf-8")
            )
            stdout_lines = processed_stdout.strip().split("\n")
            reference_lines = dedent(reference_block).strip().split("\n")
 
            if signature is None and signature_offset is not None:
                if signature_offset < 0:
                    signature_offset = len(reference_lines) + signature_offset
                signature = reference_lines[signature_offset]
 
            try:
                start_index = stdout_lines.index(signature)
                end_index = start_index + len(reference_lines)
                observed_block = stdout_lines[start_index:end_index]
 
                if observed_block != reference_lines:
                    diff = list(
                        difflib.unified_diff(
                            reference_lines,
                            observed_block,
                            fromfile="expected",
                            tofile="actual",
                        )
                    )
                    diff_text = "\n".join(diff)
                    record_property("block_diff", CodeWrapper(diff_text, "diff"))
                    raise AssertionError(
                        "The observed block does not match the reference."
                    )
            except ValueError:
                raise AssertionError(
                    f"Signature '{signature}' not found in the output."
                )
 
        return assert_function
 

test_count_messages()

GaudiTesting.GaudiExeTest.GaudiExeTest.test_count_messages	(		self,
			reference,
			stdout,
			record_property )

Test the count of error messages in the stdout against expected values.

Definition at line 249 of file GaudiExeTest.py.

    def test_count_messages(self, reference, stdout, record_property):
        """
        Test the count of error messages in the stdout against expected values.
        """
        expected_messages = (
            None if reference is None else reference.get("messages_count")
        )
        if expected_messages is None:
            pytest.skip()
 
        if not isinstance(expected_messages, dict):
            raise ValueError("reference['messages_count'] must be a dict")
        if not expected_messages:
            # an empty dict doesn't make sense, let's assume we are bootstrapping the reference
            # and start from the default
            expected_messages = NO_ERROR_MESSAGES
            reference["messages_count"] = expected_messages
 
        outlines = self.preprocessor(
            stdout.decode("utf-8", errors="backslashreplace")
        ).splitlines()
 
        messages = {key: [] for key in expected_messages}
        for n, line in enumerate(outlines, 1):
            words = line.split()
            if len(words) >= 2 and words[1] in messages:
                messages[words[1]].append((n, line.rstrip()))
 
        messages_count = {key: len(value) for key, value in messages.items()}
        try:
            assert messages_count == expected_messages
        except AssertionError:
            reference["messages_count"] = messages_count
            record_property("unexpected_messages_count", messages)
            raise
 

test_histos()

None GaudiTesting.GaudiExeTest.GaudiExeTest.test_histos	(		self,
		bytes	stdout,
		Callable	record_property,
		dict	reference )

Test the histogram summaries against the reference.

Definition at line 321 of file GaudiExeTest.py.

    ) -> None:
        """
        Test the histogram summaries against the reference.
        """
        if not self.reference or "histos" not in reference:
            pytest.skip()
 
        histos = filter_dict(
            find_histos_summaries(stdout.decode()),
            re.compile(r"Basket|.*size|Compression"),
        )
        try:
            assert histos == reference["histos"]
        except AssertionError:
            reference["histos"] = histos
            if os.environ.get("GAUDI_TEST_IGNORE_STDOUT_VALIDATION") == "1":
                pytest.xfail("Ignoring stdout validation")
            raise
 

test_record_options()

GaudiTesting.GaudiExeTest.GaudiExeTest.test_record_options	(		self,
		Callable	record_property )

Definition at line 357 of file GaudiExeTest.py.

    def test_record_options(self, record_property: Callable):
        if self.options_code:
            record_property("options", self.options_code)

test_stderr()

None GaudiTesting.GaudiExeTest.GaudiExeTest.test_stderr	(		self,
		bytes	stderr,
		Callable	record_property,
		dict	reference )

Test the standard error output against the reference.

Definition at line 343 of file GaudiExeTest.py.

    ) -> None:
        """
        Test the standard error output against the reference.
        """
        err = self.preprocessor(stderr.decode("utf-8", errors="backslashreplace"))
        if self.reference and "stderr" in reference:
            self.validate_with_reference(err, "stderr", reference, record_property)
 
        else:
            assert not err.strip(), "Expected no standard error output, but got some."
 

test_stdout()

None GaudiTesting.GaudiExeTest.GaudiExeTest.test_stdout	(		self,
		bytes	stdout,
		Callable	record_property,
		dict	reference )

Test the standard output against the reference.

Definition at line 286 of file GaudiExeTest.py.

    ) -> None:
        """
        Test the standard output against the reference.
        """
        if not self.reference or "stdout" not in reference:
            pytest.skip("No stdout reference")
 
        out = self.preprocessor(stdout.decode("utf-8", errors="backslashreplace"))
        self.validate_with_reference(out, "stdout", reference, record_property)
 

test_ttrees()

None GaudiTesting.GaudiExeTest.GaudiExeTest.test_ttrees	(		self,
		bytes	stdout,
		Callable	record_property,
		dict	reference )

Test the TTree summaries against the reference.

Definition at line 299 of file GaudiExeTest.py.

    ) -> None:
        """
        Test the TTree summaries against the reference.
        """
        if not self.reference or "ttrees" not in reference:
            pytest.skip()
 
        ttrees = filter_dict(
            find_ttree_summaries(stdout.decode()),
            re.compile(r"Basket|.*size|Compression"),
        )
        try:
            assert ttrees == reference["ttrees"]
        except AssertionError:
            reference["ttrees"] = ttrees
            if os.environ.get("GAUDI_TEST_IGNORE_STDOUT_VALIDATION") == "1":
                pytest.xfail("Ignoring stdout validation")
            raise
 

validate_json_with_reference()

GaudiTesting.GaudiExeTest.GaudiExeTest.validate_json_with_reference	(		cls,
		str	output_file,
		str	reference_file,
			detailed = True )

Validate the JSON output against a reference JSON file.

Definition at line 162 of file GaudiExeTest.py.

    ):
        """
        Validate the JSON output against a reference JSON file.
        """
        assert os.path.isfile(output_file)
 
        try:
            with open(output_file) as f:
                output = json.load(f)
        except json.JSONDecodeError as err:
            pytest.fail(f"json parser error in {output_file}: {err}")
 
        lreference = cls.resolve_path(reference_file)
        assert lreference, "reference file not set"
        assert os.path.isfile(lreference)
 
        try:
            with open(lreference) as f:
                expected = json.load(f)
        except json.JSONDecodeError as err:
            pytest.fail(f"JSON parser error in {lreference}: {err}")
 
        if not detailed:
            assert output == expected
 
        expected = sorted(expected, key=lambda item: (item["component"], item["name"]))
        output = sorted(output, key=lambda item: (item["component"], item["name"]))
        assert output == expected
 

validate_with_reference()

None GaudiTesting.GaudiExeTest.GaudiExeTest.validate_with_reference	(		cls,
		bytes	data,
		str	key,
		dict	reference,
		Callable[[str, str], None]	record_property )

Validate the given data against a reference file for the specified key.

Definition at line 120 of file GaudiExeTest.py.

    ) -> None:
        """
        Validate the given data against a reference file for the specified key.
        """
        if cls.reference:
            try:
                if key in reference:
                    # If the data is less than 100k, we can compare it directly
                    # and let pytest generate a nice diff, otherwise we hide the values
                    # to pytest to avoid that the test (practically) hangs.
                    #
                    # See
                    # - https://gitlab.cern.ch/lhcb/LHCb/-/issues/252
                    # - https://gitlab.cern.ch/gaudi/Gaudi/-/merge_requests/1375
                    # - https://gitlab.cern.ch/lhcb/LHCb/-/merge_requests/4773#note_9075367
                    if len(data) < 100000:
                        assert data == reference[key]
                    else:
                        same_as_reference = data == reference[key]
                        assert same_as_reference, "data is different from reference"
            except AssertionError:
                record_property(
                    f"{key}_diff",
                    CodeWrapper(
                        cls._output_diff(reference[key] or "", data, cls.preprocessor),
                        "diff",
                    ),
                )
                reference[key] = data
                if os.environ.get("GAUDI_TEST_IGNORE_STDOUT_VALIDATION") == "1":
                    pytest.xfail("Ignoring stdout validation")
                raise
        else:
            pytest.skip("No reference file provided")
 

Member Data Documentation

options

Callable GaudiTesting.GaudiExeTest.GaudiExeTest.options = None

static

Definition at line 42 of file GaudiExeTest.py.

options_code

GaudiTesting.GaudiExeTest.GaudiExeTest.options_code = None

static

Definition at line 43 of file GaudiExeTest.py.

preprocessor

GaudiTesting.GaudiExeTest.GaudiExeTest.preprocessor = normalizeTestSuite

static

Definition at line 44 of file GaudiExeTest.py.

---

The documentation for this class was generated from the following file:

• GaudiPolicy/python/GaudiTesting/GaudiExeTest.py