Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_pipeline(self):
key = "create_pipeline"
pattern = self._deprecation_msg(key)
with warns(DeprecationWarning, match=pattern):
pipeline = get_project_context(key)
assert pipeline() == "pipeline"
def test_get_config(self, tmp_path):
key = "get_config"
pattern = self._deprecation_msg(key)
with warns(DeprecationWarning, match=pattern):
config_loader = get_project_context(key)
assert config_loader(tmp_path) == "config_loader"
def test_template_version(self):
key = "template_version"
pattern = self._deprecation_msg(key)
with warns(DeprecationWarning, match=pattern):
assert get_project_context(key) == "dummy_version"
def test_project_name(self):
key = "project_name"
pattern = self._deprecation_msg(key)
with warns(DeprecationWarning, match=pattern):
assert get_project_context(key) == "dummy_name"
def test_context(self):
dummy_context = get_project_context("context")
assert isinstance(dummy_context, DummyContext)
def _image_callback(ctx, param, value): # pylint: disable=unused-argument
image = value or str(get_project_context("project_path").name)
check_docker_image_exists(image)
return image
def _get_dag_filename():
project_path = get_project_context("project_path")
project_name = get_project_context("project_name")
dest_dir = project_path / "airflow_dags"
return dest_dir / (slugify(project_name, separator="_") + "_dag.py")
def docker_build(uid, gid, image, docker_args):
"""Build a Docker image for the project."""
uid, gid = get_uid_gid(uid, gid)
project_path = get_project_context("project_path")
image = image or str(project_path.name)
template_path = Path(__file__).parent / "template"
verbose = get_project_context("verbose")
copy_template_files(
project_path,
template_path,
["Dockerfile", ".dockerignore", ".dive-ci"],
verbose,
)
combined_args = compose_docker_run_args(
required_args=[
("--build-arg", "KEDRO_UID={0}".format(uid)),
("--build-arg", "KEDRO_GID={0}".format(gid)),
],
# add image tag if only it is not already supplied by the user
optional_args=[("-t", image)],
user_args=docker_args,
)
def _get_dag_filename():
project_path = get_project_context("project_path")
project_name = get_project_context("project_name")
dest_dir = project_path / "airflow_dags"
return dest_dir / (slugify(project_name, separator="_") + "_dag.py")
def _mount_info() -> Dict[str, Union[str, Tuple]]:
project_path = get_project_context("project_path")
res = dict(
host_root=str(project_path),
container_root="/home/kedro",
mount_volumes=DOCKER_DEFAULT_VOLUMES,
)
return res