Skip to content

blockutils.e2e

Module that includes helper methods to run end to end tests on UP42 block images.

For instance you can setup a simple e2e test as such:

from pathlib import Path

import geojson

from blockutils.e2e import E2ETest

# Disable unused params for assert
# pylint: disable=unused-argument
def asserts(input_dir: Path, output_dir: Path, quicklook_dir: Path, logger):
    # Print out bbox of one tile
    geojson_path = output_dir / "data.json"

    with open(str(geojson_path)) as f:
        feature_collection = geojson.load(f)

    logger.info(feature_collection.features[0].bbox)

    # Check number of files in output_prefix
    output_ndvi = output_dir / Path(
        feature_collection.features[0].properties["up42.data_path"]
    )

    logger.info(output_ndvi)

    assert output_ndvi.exists()


if __name__ == "__main__":
    e2e = E2ETest(image_name="ndvi")
    e2e.add_parameters({"output_original_raster": False})
    # you need to have access to the bucket
    e2e.add_gs_bucket("gs://blocks-e2e-testing/pleiades_rgbnir/input/*")
    e2e.asserts = asserts
    e2e.run()

E2ETest

in_ci: bool property readonly

Is this code being ran in a CI environment?

Returns:

Type Description
bool

bool

__init__(self, image_name, variant='', interactive_mode=False) special

E2ETest main class.

Parameters:

Name Type Description Default
image_name str

The name you assigned to your docker image.

required
variant str

An optional variant of e2e test in case you want to define more than one. Defaults to "".

''
interactive_mode bool

adds an additional -it flag to the docker run command in case debugging is needed while running e2e locally. For running e2e in Circle CI this needs to be False (default).

False
Source code in blockutils/e2e.py
def __init__(
    self, image_name: str, variant: str = "", interactive_mode: bool = False
):
    """E2ETest main class.

    Args:
        image_name (str): The name you assigned to your docker image.
        variant (str, optional): An optional variant of e2e test in
            case you want to define more than one. Defaults to "".
        interactive_mode (bool): adds an additional -it flag to the docker run command in case debugging is needed
            while running e2e locally. For running e2e in Circle CI this needs to be False (default).
    """
    self.image_name = image_name
    self.variant = variant
    self.parameters: dict = {}
    self.additional_env: str = ""
    self.gs_bucket: str = ""
    self.local_files: str = ""
    # Only used when a certain local file needs to be mounted in docker container.
    self.add_files: List[Tuple] = []
    # Only deletes the output if ran in CI
    self.delete_output = self.in_ci
    self.interactive_mode: bool = interactive_mode

    self.test_dir = Path(f"/tmp/e2e_{self.image_name}")
    if self.variant:
        self.test_dir = Path(f"/tmp/e2e_{self.image_name}_{self.variant}")
    self.input_dir, self.output_dir, self.quicklook_dir = setup_test_directories(
        self.test_dir
    )
    self.expected_exit_code = 0

add_additional_env(self, additional_env)

With this method you can pass additional environment variables in your docker run command.

Parameters:

Name Type Description Default
additional_env str

For example "-e A_VAR=5"

required
Source code in blockutils/e2e.py
def add_additional_env(self, additional_env: str):
    """With this method you can pass additional environment variables
    in your docker run command.

    Args:
        additional_env (str): For example "-e `A_VAR=5`"
    """
    self.additional_env = additional_env

add_gs_bucket(self, gs_bucket)

Adds a GCS bucket location to fetch data from and places it in the input folder for the block run.

Parameters:

Name Type Description Default
gs_bucket str

A GCS style location (i.e. gs://some-bucket/my-input/*). You need to have access to this bucket and you need to be authenticated.

required
Source code in blockutils/e2e.py
def add_gs_bucket(self, gs_bucket: str):
    """Adds a GCS bucket location to fetch data from and places it
    in the input folder for the block run.

    Args:
        gs_bucket (str): A GCS style location (i.e. `gs://some-bucket/my-input/*`).
            You need to have access to this bucket and you need to be authenticated.
    """
    self.gs_bucket = gs_bucket

add_local_files(self, local_path)

Adds a local location of file to fetch data from and places it in the input folder for the block run.

Parameters:

Name Type Description Default
gs_bucket str

A GCS style location (i.e. gs://some-bucket/my-input/*). You need to have access to this bucket and you need to be authenticated.

required
Source code in blockutils/e2e.py
def add_local_files(self, local_path: str):
    """Adds a local location of file to fetch data from and places it
    in the input folder for the block run.

    Args:
        gs_bucket (str): A GCS style location (i.e. `gs://some-bucket/my-input/*`).
            You need to have access to this bucket and you need to be authenticated.
    """
    self.local_files = local_path
    assert (
        Path(self.local_files).resolve().is_dir()
    ), "Local path should be a folder!"

add_parameters(self, paramaters)

Add specific job parameters for this e2e test.

Parameters:

Name Type Description Default
paramaters dict

A dictionary with parameters.

required
Source code in blockutils/e2e.py
def add_parameters(self, paramaters: dict):
    """Add specific job parameters for this e2e test.

    Args:
        paramaters (dict): A dictionary with parameters.
    """
    self.parameters = paramaters

asserts(input_dir, output_dir, quicklook_dir, logger=<Logger blockutils.e2e (DEBUG)>) staticmethod

Abstract method to be used to create asserts for an e2e test.

Parameters:

Name Type Description Default
input_dir Path

Input directory.

required
output_dir Path

Output directory.

required
quicklook_dir Path

Quicklook directory.

required
logger Logger

A logger to use during asserts. Defaults to LOGGER.

<Logger blockutils.e2e (DEBUG)>
Source code in blockutils/e2e.py
@staticmethod
def asserts(
    input_dir: Path, output_dir: Path, quicklook_dir: Path, logger: Logger = LOGGER
):
    """Abstract method to be used to create asserts for an e2e test.

    Args:
        input_dir (Path): Input directory.
        output_dir (Path): Output directory.
        quicklook_dir (Path): Quicklook directory.
        logger (Logger, optional): A logger to use during asserts. Defaults to LOGGER.
    """
    pass  # pylint: disable=unnecessary-pass

cleanup_volumes(self)

Cleanup created volumes.

Source code in blockutils/e2e.py
def cleanup_volumes(self):
    """Cleanup created volumes."""
    LOGGER.info("Cleaning up volumes...")
    self.run_subprocess("docker rm -v e2e")
    self.run_subprocess("docker rm -v job")

copy_results(self)

Copy results to output folder in local environment.

Source code in blockutils/e2e.py
def copy_results(self):
    """Copy results to output folder in local environment."""
    LOGGER.info("Copying results...")
    self.run_subprocess(f"docker cp job:/tmp/. {str(self.test_dir)}")

create_volume(self)

Creates a docker volume with mounted folders (and copied folders in CI).

Source code in blockutils/e2e.py
def create_volume(self):
    """Creates a docker volume with mounted folders (and copied folders in CI)."""
    LOGGER.info("Creating a volume...")
    base_cmd = (
        f"docker create -v {str(self.input_dir)}:/tmp/input "
        f"-v {str(self.output_dir)}:/tmp/output "
        f"-v {str(self.quicklook_dir)}:/tmp/quicklooks "
    )

    for in_file, dst_file in self.add_files:
        base_cmd += f"-v {str(in_file)}:{str(dst_file)} "

    create_cmd = base_cmd + f"--name e2e {self.image_name}"
    self.run_subprocess(create_cmd)
    if self.in_ci:
        for in_file, dst_file in self.add_files:
            self.run_subprocess(f"docker cp {str(in_file)} e2e:{str(dst_file)}")

        self.run_subprocess(f"docker cp {str(self.input_dir)}/. e2e:/tmp/input")

download_input(self)

Downloads input and places it in the rigth folder.

Source code in blockutils/e2e.py
def download_input(self):
    """Downloads input and places it in the rigth folder."""
    LOGGER.info("Downloading input...")
    self.run_subprocess(f"gsutil -m cp -r {self.gs_bucket} {self.input_dir}")
    LOGGER.info(f"Contents of input {list(self.input_dir.glob('**/*'))}")

run(self)

Run defined end-to-end test.

Source code in blockutils/e2e.py
def run(self):
    """Run defined end-to-end test."""
    if self.gs_bucket:
        self.download_input()

    if self.local_files:
        self.copy_local_file()

    # Make sure entire docker component passes,
    # Otherwise clean volumes and output
    try:
        self.create_volume()

        self.run_container()

        self.copy_results()
    except AssertionError:
        LOGGER.error(f"Test for {self.image_name} failed!")
        self.cleanup_volumes()
        self._delete_output()
        raise

    self.cleanup_volumes()

    self.run_asserts()

    if self.delete_output:
        self._delete_output()

run_asserts(self)

Run defined asserts.

Source code in blockutils/e2e.py
def run_asserts(self):
    """Run defined asserts."""
    self.asserts(self.input_dir, self.output_dir, self.quicklook_dir, LOGGER)

run_container(self)

Actual run of docker container

Source code in blockutils/e2e.py
def run_container(self):
    """Actual run of docker container"""
    default_run_cmd = f"docker run {'-it' if self.interactive_mode else ''} --volumes-from e2e --name job -e"

    LOGGER.info("Running container...")
    run_cmd = (
        f"{default_run_cmd} 'UP42_TASK_PARAMETERS={json.dumps(self.parameters)}'"
        f" {self.additional_env if self.additional_env else ''} {self.image_name}"
    )

    start = timeit.default_timer()
    self.run_subprocess(run_cmd, exit_code=self.expected_exit_code)
    stop = timeit.default_timer()
    LOGGER.info(f"Time: {stop - start}")

run_subprocess(cmd, assert_return_code=True, exit_code=0) staticmethod

Utility to run arbitrary commands.

Parameters:

Name Type Description Default
cmd str

Some bash style command (i.e. rm file.txt)

required
assert_return_code bool

Make sure return code is exit_code. Defaults to True.

True
exit_code int

Exit code to assert with. Defaults to 0.

0
Source code in blockutils/e2e.py
@staticmethod
def run_subprocess(cmd: str, assert_return_code=True, exit_code=0):
    """Utility to run arbitrary commands.

    Args:
        cmd (str): Some bash style command (i.e. `rm file.txt`)
        assert_return_code (bool, optional): Make sure return code is exit_code. Defaults to True.
        exit_code (int, optional): Exit code to assert with. Defaults to 0.
    """
    sub = subprocess.run(cmd, shell=True)
    return_code = sub.returncode
    if assert_return_code:
        LOGGER.info(f"CMD: {cmd}")
        LOGGER.info(f"EXIT CODE: {return_code}")
        assert return_code == exit_code