[ENH]: Move parse_yaml to api #499

Merged
synchon merged 2 commits from update/parse-yaml-api into main 2026-07-21 14:03:19 +00:00
7 changed files with 429 additions and 421 deletions

View file

@ -0,0 +1 @@
Move ``parse_yaml`` from ``junifer.cli`` to ``junifer.api`` by `Synchon Mandal`_

View file

@ -1,4 +1,19 @@
__all__ = ["decorators", "collect", "queue", "run", "reset", "list_elements"] __all__ = [
"decorators",
"collect",
"queue",
"run",
"reset",
"list_elements",
"parse_yaml",
]
from . import decorators from . import decorators
from .functions import collect, list_elements, reset, run, queue from .functions import (
collect,
list_elements,
parse_yaml,
reset,
run,
queue,
)

View file

@ -6,8 +6,11 @@
# License: AGPL # License: AGPL
import atexit import atexit
import importlib
import importlib.util
import os import os
import shutil import shutil
import sys
from pathlib import Path from pathlib import Path
import structlog import structlog
@ -32,7 +35,14 @@ from ..typing import (
from ..utils import raise_error, warn_with_log, yaml from ..utils import raise_error, warn_with_log, yaml
__all__ = ["collect", "list_elements", "queue", "reset", "run"] __all__ = [
"collect",
"list_elements",
"parse_yaml",
"queue",
"reset",
"run",
]
_log = structlog.get_logger("junifer") _log = structlog.get_logger("junifer")
logger = _log.bind(pkg="api") logger = _log.bind(pkg="api")
@ -468,3 +478,125 @@ def list_elements(
elements_to_list.append(str_element) elements_to_list.append(str_element)
return "\n".join(elements_to_list) return "\n".join(elements_to_list)
def parse_yaml(filepath: str | Path) -> dict: # noqa: C901
"""Parse YAML.
Parameters
----------
filepath : str or pathlib.Path
The filepath to read from.
Returns
-------
dict
The contents represented as dictionary.
"""
# Convert str to Path
if not isinstance(filepath, Path):
filepath = Path(filepath)
logger.info(f"Parsing yaml file: {filepath.absolute()!s}")
# Filepath existence check
if not filepath.exists():
raise_error(f"File does not exist: {filepath.absolute()!s}")
# Filepath reading
contents = yaml.load(filepath)
if "elements" in contents:
if contents["elements"] is None:
raise_error(
"The elements key was defined but its content is empty. "
"Please define the elements to operate on or remove the key."
)
# load modules
if "with" in contents:
to_load = contents["with"]
# Convert load modules to list
if not isinstance(to_load, list):
to_load = [to_load]
# Initialize list to have absolute paths for custom modules
final_to_load = []
for t_module in to_load:
if t_module.endswith(".py"):
logger.debug(f"Importing file: {t_module}")
# This resolves both absolute and relative paths
file_path = filepath.parent / t_module
if not file_path.exists():
raise_error(
f"File in 'with' section does not exist: {file_path}"
)
# Add the parent directory to the sys.path so that the
# any imports from this module work correctly
t_path = str(file_path.parent)
if t_path not in sys.path:
sys.path.append(str(file_path.parent))
spec = importlib.util.spec_from_file_location(
t_module, file_path
)
module = importlib.util.module_from_spec(spec) # type: ignore
sys.modules[t_module] = module
spec.loader.exec_module(module) # type: ignore
# Add absolute path to final list
final_to_load.append(str(file_path.resolve()))
# Check if the module has junifer_module_deps function
if hasattr(module, "junifer_module_deps"):
logger.debug(
f"Module {t_module} has junifer_module_deps function"
)
# Get the dependencies
deps = module.junifer_module_deps()
# Add the dependencies to the final list
for dep in deps:
if dep not in final_to_load:
final_to_load.append(
str((file_path.parent / dep).resolve())
)
else:
logger.info(f"Importing module: {t_module}")
importlib.import_module(t_module)
# Add module to final list
final_to_load.append(t_module)
# Replace modules to be loaded so that custom modules will take the
# absolute path. This was not the case as found in #224. Similar thing
# is done with the storage URI below.
contents["with"] = final_to_load
# Compute path for the URI parameter in storage files that are relative
# This is a tricky thing that appeared in #127. The problem is that
# the path in the URI parameter is relative to YAML file, not to the
# current working directory. If we leave it as is in the contents
# dict, then it will be used later in the ``build`` function as is,
# which will be computed relative to the current working directory.
# The solution is to compute the absolute path and replace the
# relative path in the contents dict with the absolute path.
# Check if the storage file is defined
if "storage" in contents:
if "uri" in contents["storage"]:
# Check if the storage file is relative
uri_path = Path(contents["storage"]["uri"])
if not uri_path.is_absolute():
# Compute the absolute path
contents["storage"]["uri"] = str(
(filepath.parent / uri_path).resolve()
)
# Allow relative path if queue env kind is venv; same motivation as above
if "queue" in contents:
if "env" in contents["queue"]:
if "venv" == contents["queue"]["env"]["kind"]:
# Check if the env name is relative
venv_path = Path(contents["queue"]["env"]["name"])
if not venv_path.is_absolute():
# Compute the absolute path
contents["queue"]["env"]["name"] = str(
(filepath.parent / venv_path).resolve()
)
return contents

View file

@ -6,6 +6,7 @@
# License: AGPL # License: AGPL
import logging import logging
import sys
from contextlib import AbstractContextManager, nullcontext from contextlib import AbstractContextManager, nullcontext
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@ -15,7 +16,7 @@ from nibabel.filebasedimages import ImageFileError
from ruamel.yaml import YAML from ruamel.yaml import YAML
import junifer.testing.registry # noqa: F401 import junifer.testing.registry # noqa: F401
from junifer.api import collect, list_elements, queue, reset, run from junifer.api import collect, list_elements, parse_yaml, queue, reset, run
from junifer.datagrabber.base import BaseDataGrabber from junifer.datagrabber.base import BaseDataGrabber
from junifer.pipeline import PipelineComponentRegistry from junifer.pipeline import PipelineComponentRegistry
from junifer.typing import Elements from junifer.typing import Elements
@ -756,3 +757,271 @@ def test_list_elements(
""" """
listed_elements = list_elements(datagrabber, elements) listed_elements = list_elements(datagrabber, elements)
assert "sub-01" in listed_elements assert "sub-01" in listed_elements
def test_parse_yaml_failure() -> None:
"""Test YAML parsing failure."""
with pytest.raises(ValueError, match="does not exist"):
parse_yaml("foo.yaml")
def test_parse_yaml_empty_elements_failure(tmp_path: Path) -> None:
"""Test YAML parsing with empty elements failure.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_empty_elements_failure.yaml"
fname.write_text("elements:")
# Check test file
with pytest.raises(ValueError, match="elements key was defined"):
parse_yaml(fname)
def test_parse_yaml_success(tmp_path: Path) -> None:
"""Test YAML parsing success.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_success.yaml"
fname.write_text("foo: bar")
# Check test file
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
def test_parse_yaml_success_with_module_autoload(tmp_path: Path) -> None:
"""Test YAML parsing with single module autoload success.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_with_single_module_autoload.yaml"
fname.write_text("foo: bar\nwith: numpy")
# Check test file
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "with" in contents
assert contents["with"] == ["numpy"]
assert "numpy" in sys.modules
assert "junifer.configs.wrong_config" not in sys.modules
def test_parse_yaml_failure_with_multi_module_autoload(tmp_path: Path) -> None:
"""Test YAML parsing with multi module autoload failure.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_with_multi_module_autoload.yaml"
fname.write_text(
"foo: bar\nwith:\n - numpy\n - junifer.testing.wrong_config"
)
# Check test file
with pytest.raises(ImportError, match="wrong_config"):
parse_yaml(fname)
def test_parse_yaml_with_wrong_path(tmp_path: Path) -> None:
"""Test YAML parsing with wrong paths in with.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_relative_with"
# Write yaml that includes a relative path
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_wrong_path.yaml"
yaml_fname.write_text("foo: bar\nwith:\n - missingt.py\n - scipy\n")
# Check test file
with pytest.raises(ValueError, match="does not exist"):
parse_yaml(yaml_fname)
def test_parse_yaml_relative_path(tmp_path: Path) -> None:
"""Test YAML parsing with relative paths in with.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_relative_with"
# Write .py to include
py_path = t_tmp_path / "external"
py_path.mkdir(exist_ok=True, parents=True)
py_fname = py_path / "first.py"
py_fname.write_text("import numpy as np\n")
# Write yaml that includes a relative path
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_relative_path.yaml"
yaml_fname.write_text(
"foo: bar\nwith:\n - ../external/first.py\n - scipy\n"
)
# Check test file
parse_yaml(yaml_fname)
def test_parse_yaml_absolute_path(tmp_path: Path) -> None:
"""Test YAML parsing with absolute paths in with.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_relative_with"
# Write .py to include
py_path = t_tmp_path / "external"
py_path.mkdir(exist_ok=True, parents=True)
py_fname = py_path / "first.py"
py_fname.write_text("import numpy as np\n")
# Write yaml that includes a relative path
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_relative_path.yaml"
yaml_fname.write_text(
f"foo: bar\nwith:\n - {py_fname.absolute()}\n - scipy\n"
)
# Check test file
parse_yaml(yaml_fname)
def test_parse_yaml_multi_module_deps(tmp_path: Path) -> None:
"""Test YAML parsing with multi-module import with deps.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_with_multi_module"
# Write .py to include
py_path = t_tmp_path / "external"
py_path.mkdir(exist_ok=True, parents=True)
py_fname_1 = py_path / "first.py"
py_fname_1.write_text(
"import numpy as np\nfrom second import hej\n"
"def junifer_module_deps(): return ['second.py']\n"
)
py_fname_2 = py_path / "second.py"
py_fname_2.write_text("def hej(): print('hej')\n")
# Write yaml
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_multi_module.yaml"
yaml_fname.write_text(
"foo: bar\nwith:\n - ../external/first.py\n - scipy\n"
)
# Check test file
parse_yaml(yaml_fname)
def test_parse_storage_uri_relative(tmp_path: Path) -> None:
"""Test YAML parsing with storage and relative URI.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text("foo: bar\nwith: numpy\nstorage:\n uri: test.db\n")
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
assert "uri" in contents["storage"]
assert contents["storage"]["uri"] == str(tmp_path / "test.db")
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text(
"foo: bar\nwith: numpy\nstorage:\n uri: ../another/test.db\n"
)
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
assert "uri" in contents["storage"]
assert contents["storage"]["uri"] == str(
(tmp_path / "../another/test.db").resolve()
)
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text(
"foo: bar\nwith: numpy\nstorage:\n uri: /absolute/test.db\n"
)
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
assert "uri" in contents["storage"]
assert contents["storage"]["uri"] == "/absolute/test.db"
# Just to trick coverage
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text("foo: bar\nwith: numpy\nstorage:\n kind: SomeStorage\n")
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
def test_parse_yaml_queue_venv_relative(tmp_path: Path) -> None:
"""Test YAML parsing with relative venv queue.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
fname = tmp_path / "test_parse_yaml_queue_venv_relative.yaml"
fname.write_text("queue:\n env:\n kind: venv\n name: .venv\n")
_ = parse_yaml(fname)

View file

@ -17,7 +17,7 @@ from ..utils import (
raise_error, raise_error,
yaml, yaml,
) )
from .parser import parse_elements, parse_yaml from .parser import parse_elements
from .utils import ( from .utils import (
_get_dependency_information, _get_dependency_information,
_get_environment_information, _get_environment_information,
@ -162,7 +162,7 @@ def run(
configure_logging(level=verbose, level_datalad=verbose_datalad) configure_logging(level=verbose, level_datalad=verbose_datalad)
# TODO(synchon): add validation # TODO(synchon): add validation
# Parse YAML # Parse YAML
config = parse_yaml(filepath) config = cli_func.parse_yaml(filepath)
# Retrieve working directory # Retrieve working directory
workdir = config["workdir"] workdir = config["workdir"]
# Fetch datagrabber # Fetch datagrabber
@ -232,7 +232,7 @@ def collect(
configure_logging(level=verbose, level_datalad=verbose_datalad) configure_logging(level=verbose, level_datalad=verbose_datalad)
# TODO: add validation # TODO: add validation
# Parse YAML # Parse YAML
config = parse_yaml(filepath) config = cli_func.parse_yaml(filepath)
# Fetch storage # Fetch storage
storage = config["storage"] storage = config["storage"]
# Perform operation # Perform operation
@ -299,7 +299,7 @@ def queue(
configure_logging(level=verbose, level_datalad=verbose_datalad) configure_logging(level=verbose, level_datalad=verbose_datalad)
# TODO: add validation # TODO: add validation
# Parse YAML # Parse YAML
config = parse_yaml(filepath) # type: ignore config = cli_func.parse_yaml(filepath) # type: ignore
# Check queue section # Check queue section
if "queue" not in config: if "queue" not in config:
raise_error(f"No queue configuration found in {filepath}.") raise_error(f"No queue configuration found in {filepath}.")
@ -450,7 +450,7 @@ def reset(
# Setup logging # Setup logging
configure_logging(level=verbose, level_datalad=verbose_datalad) configure_logging(level=verbose, level_datalad=verbose_datalad)
# Parse YAML # Parse YAML
config = parse_yaml(filepath) config = cli_func.parse_yaml(filepath)
# Perform operation # Perform operation
cli_func.reset(config) cli_func.reset(config)
@ -510,7 +510,7 @@ def list_elements(
# Setup logging # Setup logging
configure_logging(level=verbose, level_datalad=verbose_datalad) configure_logging(level=verbose, level_datalad=verbose_datalad)
# Parse YAML # Parse YAML
config = parse_yaml(filepath) config = cli_func.parse_yaml(filepath)
# Fetch datagrabber # Fetch datagrabber
datagrabber = config["datagrabber"] datagrabber = config["datagrabber"]
# Parse elements # Parse elements

View file

@ -4,144 +4,15 @@
# Synchon Mandal <s.mandal@fz-juelich.de> # Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL # License: AGPL
import importlib
import importlib.util
import sys
from pathlib import Path from pathlib import Path
import pandas as pd import pandas as pd
import structlog
from ..typing import Elements from ..typing import Elements
from ..utils import raise_error, warn_with_log, yaml from ..utils import logger, raise_error, warn_with_log
__all__ = ["parse_elements", "parse_yaml"] __all__ = ["parse_elements"]
_log = structlog.get_logger("junifer")
logger = _log.bind(pkg="cli")
def parse_yaml(filepath: str | Path) -> dict: # noqa: C901
"""Parse YAML.
Parameters
----------
filepath : str or pathlib.Path
The filepath to read from.
Returns
-------
dict
The contents represented as dictionary.
"""
# Convert str to Path
if not isinstance(filepath, Path):
filepath = Path(filepath)
logger.info(f"Parsing yaml file: {filepath.absolute()!s}")
# Filepath existence check
if not filepath.exists():
raise_error(f"File does not exist: {filepath.absolute()!s}")
# Filepath reading
contents = yaml.load(filepath)
if "elements" in contents:
if contents["elements"] is None:
raise_error(
"The elements key was defined but its content is empty. "
"Please define the elements to operate on or remove the key."
)
# load modules
if "with" in contents:
to_load = contents["with"]
# Convert load modules to list
if not isinstance(to_load, list):
to_load = [to_load]
# Initialize list to have absolute paths for custom modules
final_to_load = []
for t_module in to_load:
if t_module.endswith(".py"):
logger.debug(f"Importing file: {t_module}")
# This resolves both absolute and relative paths
file_path = filepath.parent / t_module
if not file_path.exists():
raise_error(
f"File in 'with' section does not exist: {file_path}"
)
# Add the parent directory to the sys.path so that the
# any imports from this module work correctly
t_path = str(file_path.parent)
if t_path not in sys.path:
sys.path.append(str(file_path.parent))
spec = importlib.util.spec_from_file_location(
t_module, file_path
)
module = importlib.util.module_from_spec(spec) # type: ignore
sys.modules[t_module] = module
spec.loader.exec_module(module) # type: ignore
# Add absolute path to final list
final_to_load.append(str(file_path.resolve()))
# Check if the module has junifer_module_deps function
if hasattr(module, "junifer_module_deps"):
logger.debug(
f"Module {t_module} has junifer_module_deps function"
)
# Get the dependencies
deps = module.junifer_module_deps()
# Add the dependencies to the final list
for dep in deps:
if dep not in final_to_load:
final_to_load.append(
str((file_path.parent / dep).resolve())
)
else:
logger.info(f"Importing module: {t_module}")
importlib.import_module(t_module)
# Add module to final list
final_to_load.append(t_module)
# Replace modules to be loaded so that custom modules will take the
# absolute path. This was not the case as found in #224. Similar thing
# is done with the storage URI below.
contents["with"] = final_to_load
# Compute path for the URI parameter in storage files that are relative
# This is a tricky thing that appeared in #127. The problem is that
# the path in the URI parameter is relative to YAML file, not to the
# current working directory. If we leave it as is in the contents
# dict, then it will be used later in the ``build`` function as is,
# which will be computed relative to the current working directory.
# The solution is to compute the absolute path and replace the
# relative path in the contents dict with the absolute path.
# Check if the storage file is defined
if "storage" in contents:
if "uri" in contents["storage"]:
# Check if the storage file is relative
uri_path = Path(contents["storage"]["uri"])
if not uri_path.is_absolute():
# Compute the absolute path
contents["storage"]["uri"] = str(
(filepath.parent / uri_path).resolve()
)
# Allow relative path if queue env kind is venv; same motivation as above
if "queue" in contents:
if "env" in contents["queue"]:
if "venv" == contents["queue"]["env"]["kind"]:
# Check if the env name is relative
venv_path = Path(contents["queue"]["env"]["name"])
if not venv_path.is_absolute():
# Compute the absolute path
contents["queue"]["env"]["name"] = str(
(filepath.parent / venv_path).resolve()
)
return contents
def parse_elements(element: tuple[str, ...], config: dict) -> Elements | None: def parse_elements(element: tuple[str, ...], config: dict) -> Elements | None:

View file

@ -1,280 +0,0 @@
"""Provide tests for CLI parser."""
# Authors: Federico Raimondo <f.raimondo@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
import sys
from pathlib import Path
import pytest
from junifer.cli.parser import parse_yaml
def test_parse_yaml_failure() -> None:
"""Test YAML parsing failure."""
with pytest.raises(ValueError, match="does not exist"):
parse_yaml("foo.yaml")
def test_parse_yaml_empty_elements_failure(tmp_path: Path) -> None:
"""Test YAML parsing with empty elements failure.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_empty_elements_failure.yaml"
fname.write_text("elements:")
# Check test file
with pytest.raises(ValueError, match="elements key was defined"):
parse_yaml(fname)
def test_parse_yaml_success(tmp_path: Path) -> None:
"""Test YAML parsing success.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_success.yaml"
fname.write_text("foo: bar")
# Check test file
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
def test_parse_yaml_success_with_module_autoload(tmp_path: Path) -> None:
"""Test YAML parsing with single module autoload success.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_with_single_module_autoload.yaml"
fname.write_text("foo: bar\nwith: numpy")
# Check test file
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "with" in contents
assert contents["with"] == ["numpy"]
assert "numpy" in sys.modules
assert "junifer.configs.wrong_config" not in sys.modules
def test_parse_yaml_failure_with_multi_module_autoload(tmp_path: Path) -> None:
"""Test YAML parsing with multi module autoload failure.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Write test file
fname = tmp_path / "test_parse_yaml_with_multi_module_autoload.yaml"
fname.write_text(
"foo: bar\nwith:\n - numpy\n - junifer.testing.wrong_config"
)
# Check test file
with pytest.raises(ImportError, match="wrong_config"):
parse_yaml(fname)
def test_parse_yaml_with_wrong_path(tmp_path: Path) -> None:
"""Test YAML parsing with wrong paths in with.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_relative_with"
# Write yaml that includes a relative path
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_wrong_path.yaml"
yaml_fname.write_text("foo: bar\nwith:\n - missingt.py\n - scipy\n")
# Check test file
with pytest.raises(ValueError, match="does not exist"):
parse_yaml(yaml_fname)
def test_parse_yaml_relative_path(tmp_path: Path) -> None:
"""Test YAML parsing with relative paths in with.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_relative_with"
# Write .py to include
py_path = t_tmp_path / "external"
py_path.mkdir(exist_ok=True, parents=True)
py_fname = py_path / "first.py"
py_fname.write_text("import numpy as np\n")
# Write yaml that includes a relative path
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_relative_path.yaml"
yaml_fname.write_text(
"foo: bar\nwith:\n - ../external/first.py\n - scipy\n"
)
# Check test file
parse_yaml(yaml_fname)
def test_parse_yaml_absolute_path(tmp_path: Path) -> None:
"""Test YAML parsing with absolute paths in with.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_relative_with"
# Write .py to include
py_path = t_tmp_path / "external"
py_path.mkdir(exist_ok=True, parents=True)
py_fname = py_path / "first.py"
py_fname.write_text("import numpy as np\n")
# Write yaml that includes a relative path
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_relative_path.yaml"
yaml_fname.write_text(
f"foo: bar\nwith:\n - {py_fname.absolute()}\n - scipy\n"
)
# Check test file
parse_yaml(yaml_fname)
def test_parse_yaml_multi_module_deps(tmp_path: Path) -> None:
"""Test YAML parsing with multi-module import with deps.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
t_tmp_path = tmp_path / "test_with_multi_module"
# Write .py to include
py_path = t_tmp_path / "external"
py_path.mkdir(exist_ok=True, parents=True)
py_fname_1 = py_path / "first.py"
py_fname_1.write_text(
"import numpy as np\nfrom second import hej\n"
"def junifer_module_deps(): return ['second.py']\n"
)
py_fname_2 = py_path / "second.py"
py_fname_2.write_text("def hej(): print('hej')\n")
# Write yaml
yaml_path = t_tmp_path / "yamls"
yaml_path.mkdir(exist_ok=True, parents=True)
yaml_fname = yaml_path / "test_parse_yaml_multi_module.yaml"
yaml_fname.write_text(
"foo: bar\nwith:\n - ../external/first.py\n - scipy\n"
)
# Check test file
parse_yaml(yaml_fname)
def test_parse_storage_uri_relative(tmp_path: Path) -> None:
"""Test YAML parsing with storage and relative URI.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text("foo: bar\nwith: numpy\nstorage:\n uri: test.db\n")
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
assert "uri" in contents["storage"]
assert contents["storage"]["uri"] == str(tmp_path / "test.db")
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text(
"foo: bar\nwith: numpy\nstorage:\n uri: ../another/test.db\n"
)
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
assert "uri" in contents["storage"]
assert contents["storage"]["uri"] == str(
(tmp_path / "../another/test.db").resolve()
)
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text(
"foo: bar\nwith: numpy\nstorage:\n uri: /absolute/test.db\n"
)
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
assert "uri" in contents["storage"]
assert contents["storage"]["uri"] == "/absolute/test.db"
# Just to trick coverage
fname = tmp_path / "test_parse_yaml_with_storage_uri.yaml"
fname.write_text("foo: bar\nwith: numpy\nstorage:\n kind: SomeStorage\n")
contents = parse_yaml(fname)
assert "foo" in contents
assert contents["foo"] == "bar"
assert "storage" in contents
def test_parse_yaml_queue_venv_relative(tmp_path: Path) -> None:
"""Test YAML parsing with relative venv queue.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
fname = tmp_path / "test_parse_yaml_queue_venv_relative.yaml"
fname.write_text("queue:\n env:\n kind: venv\n name: .venv\n")
_ = parse_yaml(fname)