Function assert_load_info

tests/pipeline/utils.py:159–179

Assert that all load packages completed successfully. Args: info (LoadInfo): The load information instance returned by DLT after executing ``pipeline.run()`` or ``pipeline.load()``. expected_load_packages (int, optional): How many load packages should be

(info: LoadInfo, expected_load_packages: int = 1)

def assert_load_info(info: LoadInfo, expected_load_packages: int = 1) -> None:
    """Assert that all load packages completed successfully.

    Args:
        info (LoadInfo): The load information instance returned by DLT after
            executing ``pipeline.run()`` or ``pipeline.load()``.
        expected_load_packages (int, optional): How many load packages should be
            present in *info*. Defaults to 1.

    Raises:
        AssertionError: If the amount of load packages or any job status does
            not match the expectations.
    """
    # make sure we can serialize
    info.asstr(verbosity=2)
    info.asdict()
    assert len(info.loads_ids) == expected_load_packages
    # all packages loaded
    assert all(p.completed_at is not None for p in info.load_packages) is True
    # no failed jobs in any of the packages
    assert all(len(p.jobs["failed_jobs"]) == 0 for p in info.load_packages) is True


#