qc.csv

CsvTestSuite

CsvTestSuite(data_stream: Csv)

Bases: Suite

Test suite to check if CSV files generated are well formatted.

Provides tests to validate that CSV files conform to expected formatting standards and contain valid data.

Attributes:

Name	Type	Description
data_stream		The CSV data stream to test.

Examples:

from contraqctor.contract.csv import Csv, CsvParams
from contraqctor.qc.csv import CsvTestSuite
from contraqctor.qc.base import Runner

# Create and load a CSV data stream
params = CsvParams(path="data/measurements.csv")
csv_stream = Csv("measurements", reader_params=params).load()

# Create and run the test suite
suite = CsvTestSuite(csv_stream)
runner = Runner().add_suite(suite)
results = runner.run_all_with_progress()

Initialize the CSV test suite.

Parameters:

Name	Type	Description	Default
data_stream	Csv	The CSV data stream to test.	required

Source code in src/contraqctor/qc/csv.py

def __init__(self, data_stream: Csv):
    """Initialize the CSV test suite.

    Args:
        data_stream: The CSV data stream to test.
    """
    self.data_stream = data_stream

description property

description: Optional[str]

Get the description of the test suite from its docstring.

Returns:

Type	Description
Optional[str]	Optional[str]: The docstring of the class, or None if not available.

name property

name: str

Get the name of the test suite.

Returns:

Name	Type	Description
str	str	The name of the test suite class.

test_is_instance_of_pandas_dataframe

test_is_instance_of_pandas_dataframe()

Check if the data stream is a pandas DataFrame.

Source code in src/contraqctor/qc/csv.py

def test_is_instance_of_pandas_dataframe(self):
    """
    Check if the data stream is a pandas DataFrame.
    """
    if not self.data_stream.has_data:
        return self.fail_test(None, "Data stream does not have loaded data")
    if not isinstance(self.data_stream.data, pd.DataFrame):
        return self.fail_test(None, "Data stream is not a pandas DataFrame")
    return self.pass_test(None, "Data stream is a pandas DataFrame")

test_is_not_empty

test_is_not_empty()

Check if the DataFrame is not empty.

Source code in src/contraqctor/qc/csv.py

def test_is_not_empty(self):
    """
    Check if the DataFrame is not empty.
    """
    df = self.data_stream.data
    if df.empty:
        return self.fail_test(None, "Data stream is empty")
    return self.pass_test(None, "Data stream is not empty")

test_infer_missing_headers

test_infer_missing_headers()

Infer if the DataFrame was loaded from a CSV without headers.

Source code in src/contraqctor/qc/csv.py

def test_infer_missing_headers(self):
    """
    Infer if the DataFrame was loaded from a CSV without headers.
    """
    if not self.data_stream.reader_params.strict_header:
        return self.skip_test("CSV was loaded with strict_header=False")

    df = self.data_stream.data
    if df.empty or len(df.columns) == 0:
        return self.fail_test(None, "Data stream is empty or has no columns")

    # Check if column names are default integer indexes (0, 1, 2...)
    if all(isinstance(col, int) or (isinstance(col, str) and col.isdigit()) for col in df.columns):
        return self.fail_test(None, "Data stream has non-integer column names")

    return self.pass_test(None, "DataFramed was likely loaded from a CSV with headers")

get_tests

get_tests() -> Generator[ITest, None, None]

Find all methods starting with 'test'.

Yields:

Name	Type	Description
ITest	ITest	Test methods found in the suite.

Source code in src/contraqctor/qc/base.py

def get_tests(self) -> t.Generator[ITest, None, None]:
    """Find all methods starting with 'test'.

    Yields:
        ITest: Test methods found in the suite.
    """
    for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
        if name.startswith("test"):
            yield method

pass_test

pass_test() -> Result

pass_test(result: Any) -> Result

pass_test(result: Any, message: str) -> Result

pass_test(result: Any, *, context: Any) -> Result

pass_test(
    result: Any, message: str, *, context: Any
) -> Result

pass_test(
    result: Any = None,
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a passing test result.

Parameters:

Name	Type	Description	Default
result	Any	The value to include in the test result.	None
message	Optional[str]	Optional message describing why the test passed.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with PASSED status.

Source code in src/contraqctor/qc/base.py

def pass_test(
    self, result: t.Any = None, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None
) -> Result:
    """Create a passing test result.

    Args:
        result: The value to include in the test result.
        message: Optional message describing why the test passed.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with PASSED status.
    """
    calling_func_name, description = self._get_caller_info()

    return Result(
        status=Status.PASSED,
        result=result,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

warn_test

warn_test() -> Result

warn_test(result: Any) -> Result

warn_test(result: Any, message: str) -> Result

warn_test(result: Any, *, context: Any) -> Result

warn_test(
    result: Any, message: str, *, context: Any
) -> Result

warn_test(
    result: Any = None,
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a warning test result.

Creates a result with WARNING status, or FAILED if warnings are elevated.

Parameters:

Name	Type	Description	Default
result	Any	The value to include in the test result.	None
message	Optional[str]	Optional message describing the warning.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with WARNING or FAILED status.

Source code in src/contraqctor/qc/base.py

def warn_test(
    self, result: t.Any = None, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None
) -> Result:
    """Create a warning test result.

    Creates a result with WARNING status, or FAILED if warnings are elevated.

    Args:
        result: The value to include in the test result.
        message: Optional message describing the warning.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with WARNING or FAILED status.
    """
    calling_func_name, description = self._get_caller_info()

    return Result(
        status=Status.WARNING if not _elevate_warning.get() else Status.FAILED,
        result=result,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

fail_test

fail_test() -> Result

fail_test(result: Any) -> Result

fail_test(result: Any, message: str) -> Result

fail_test(
    result: Any, message: str, *, context: Any
) -> Result

fail_test(
    result: Optional[Any] = None,
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a failing test result.

Parameters:

Name	Type	Description	Default
result	Optional[Any]	The value to include in the test result.	None
message	Optional[str]	Optional message describing why the test failed.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with FAILED status.

Source code in src/contraqctor/qc/base.py

def fail_test(
    self, result: t.Optional[t.Any] = None, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None
) -> Result:
    """Create a failing test result.

    Args:
        result: The value to include in the test result.
        message: Optional message describing why the test failed.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with FAILED status.
    """
    calling_func_name, description = self._get_caller_info()

    return Result(
        status=Status.FAILED,
        result=result,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

skip_test

skip_test() -> Result

skip_test(message: str) -> Result

skip_test(message: str, *, context: Any) -> Result

skip_test(
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a skipped test result.

Creates a result with SKIPPED status, or FAILED if skips are elevated.

Parameters:

Name	Type	Description	Default
message	Optional[str]	Optional message explaining why the test was skipped.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with SKIPPED or FAILED status.

Source code in src/contraqctor/qc/base.py

def skip_test(self, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None) -> Result:
    """Create a skipped test result.

    Creates a result with SKIPPED status, or FAILED if skips are elevated.

    Args:
        message: Optional message explaining why the test was skipped.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with SKIPPED or FAILED status.
    """
    calling_func_name, description = self._get_caller_info()
    return Result(
        status=Status.SKIPPED if not _elevate_skippable.get() else Status.FAILED,
        result=None,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

setup

setup() -> None

Run before each test method.

This method can be overridden by subclasses to implement setup logic that runs before each test.

Source code in src/contraqctor/qc/base.py

def setup(self) -> None:
    """Run before each test method.

    This method can be overridden by subclasses to implement
    setup logic that runs before each test.
    """
    pass

teardown

teardown() -> None

Run after each test method.

This method can be overridden by subclasses to implement teardown logic that runs after each test.

Source code in src/contraqctor/qc/base.py

def teardown(self) -> None:
    """Run after each test method.

    This method can be overridden by subclasses to implement
    teardown logic that runs after each test.
    """
    pass

run_test

run_test(
    test_method: ITest,
) -> Generator[Result, None, None]

Run a single test method and yield its results.

Handles setup, test execution, result processing, and teardown.

Parameters:

Name	Type	Description	Default
test_method	ITest	The test method to run.	required

Yields:

Name	Type	Description
Result	Result	Result objects produced by the test method.

Source code in src/contraqctor/qc/base.py

def run_test(self, test_method: ITest) -> t.Generator[Result, None, None]:
    """Run a single test method and yield its results.

    Handles setup, test execution, result processing, and teardown.

    Args:
        test_method: The test method to run.

    Yields:
        Result: Result objects produced by the test method.
    """
    test_name = test_method.__name__
    suite_name = self.name
    test_description = getattr(test_method, "__doc__", None)

    try:
        self.setup()
        result = test_method()
        if inspect.isgenerator(result):
            for sub_result in result:
                yield self._process_test_result(sub_result, test_method, test_name, test_description)
        else:
            yield self._process_test_result(result, test_method, test_name, test_description)
    except Exception as e:
        tb = traceback.format_exc()
        yield Result(
            status=Status.ERROR,
            result=None,
            test_name=test_name,
            suite_name=suite_name,
            description=test_description,
            message=f"Error during test execution: {str(e)}",
            exception=e,
            traceback=tb,
            test_reference=test_method,
            suite_reference=self,
        )
    finally:
        self.teardown()

run_all

run_all() -> Generator[Result, None, None]

Run all test methods in the suite.

Finds all test methods and runs them in sequence.

Yields:

Name	Type	Description
Result	Result	Result objects produced by all test methods.

Source code in src/contraqctor/qc/base.py

def run_all(self) -> t.Generator[Result, None, None]:
    """Run all test methods in the suite.

    Finds all test methods and runs them in sequence.

    Yields:
        Result: Result objects produced by all test methods.
    """
    for test in self.get_tests():
        yield from self.run_test(test)