psdi_data_conversion/testing/conversion_setup.py

Created 2026-10-20 by Bryan Gillis.

This module provides functions and callable classes which can be used as setup callbacks for test specs.

Global variables

var JUNK_FILE_CONTENT

A text string added to junk files created, which can be checked for to see if output files contain this string

Functions

def write_junk_file(dir: str, filename: str)

Expand source code

def write_junk_file(dir: str, filename: str):
    """Write out a junk file at a desired location"""
    open(os.path.join(dir, filename), "w").write(JUNK_FILE_CONTENT)

Write out a junk file at a desired location

Classes

class BlockInputFile (l_filenames: str | list[str])

Expand source code

class BlockInputFile:
    """Callable class which will block a file or files in the input directory with a provided filename
    """

    def __init__(self, l_filenames: str | list[str]):
        if isinstance(l_filenames, str):
            self.l_filenames = [l_filenames]
        else:
            self.l_filenames = l_filenames

    def __call__(self, test_spec: SingleConversionTestSpec, input_dir, **_) -> None:
        for filename in self.l_filenames:
            write_junk_file(input_dir, filename)

Callable class which will block a file or files in the input directory with a provided filename

class BlockOutput

Expand source code

class BlockOutput:
    """Callable class which will block the intended output filename of a conversion with a junk file before the
    conversion is performed
    """

    def __call__(self, test_spec: SingleConversionTestSpec, output_dir, **_) -> None:
        # Get the expected name of the output file
        if test_spec.ex_out_filename:
            ex_out_filename = test_spec.ex_out_filename
        else:
            ex_out_filename = replace_extension(test_spec.filename, get_format_info(test_spec.to_format).name)

        write_junk_file(output_dir, ex_out_filename)

Callable class which will block the intended output filename of a conversion with a junk file before the conversion is performed

class BlockOutputExt (l_ext: str | collections.abc.Iterable[str],
l_ext_free: str | collections.abc.Iterable[str] | None = None)

Expand source code

class BlockOutputExt:
    """Callable class which will block the files in the output directory which have the output filename with its
    extension replaced by any in the provided list
    """

    def __init__(self, l_ext: str | Iterable[str], l_ext_free: str | Iterable[str] | None = None):
        if isinstance(l_ext, str):
            self.l_ext = [l_ext]
        else:
            self.l_ext = l_ext
        if isinstance(l_ext_free, str):
            self.l_ext_free = [l_ext]
        elif l_ext_free is None:
            self.l_ext_free = []
        else:
            self.l_ext_free = l_ext_free

    def __call__(self, test_spec: SingleConversionTestSpec, output_dir, **_) -> None:
        for ext in self.l_ext_free:
            qualified_filename = os.path.join(output_dir, replace_extension(test_spec.filename, ext))
            if os.path.exists(qualified_filename):
                os.remove(qualified_filename)
        for ext in self.l_ext:
            write_junk_file(output_dir, replace_extension(test_spec.filename, ext))

Callable class which will block the files in the output directory which have the output filename with its extension replaced by any in the provided list

class MultiSetup (*l_setups: Callable[typing.Concatenate[SingleConversionTestSpec, ...], None])

Expand source code

class MultiSetup:
    """Callable class which stores a list of other setup tasks to call, and runs them all in sequence. All tasks will
    be run, with no try blocks being applied to stop exceptions from propagating upwards as soon as they happen
    """

    def __init__(self, *l_setups: Callable[Concatenate[SingleConversionTestSpec, ...], None]):

        self.l_setups: Iterable[Callable[Concatenate[SingleConversionTestSpec, ...], None]] = l_setups
        """A list of setup tasks to be called in turn"""

    def __call__(self, test_spec: SingleConversionTestSpec, **kwargs) -> None:
        """When called and passed a test spec, run each stored task on the test spec in turn"""

        for setup in self.l_setups:
            setup(test_spec, **kwargs)

Callable class which stores a list of other setup tasks to call, and runs them all in sequence. All tasks will be run, with no try blocks being applied to stop exceptions from propagating upwards as soon as they happen

Instance variables

var l_setups

A list of setup tasks to be called in turn

class RenameInput (new_filename: str)

Expand source code

class RenameInput:
    """Callable class which renames the input file to a desired filename before the conversion is performed"""

    def __init__(self, new_filename: str):
        self.new_filename = new_filename

    def __call__(self, test_spec: SingleConversionTestSpec, input_dir, **_):
        qualified_old_filename = os.path.join(input_dir, test_spec.filename)
        qualified_new_filename = os.path.join(input_dir, self.new_filename)
        os.rename(qualified_old_filename, qualified_new_filename)
        test_spec.filename = self.new_filename

Callable class which renames the input file to a desired filename before the conversion is performed