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 |
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. |
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. |
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. |
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