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_ignore_hidden_keys(self, tmp_path):
"""Check that the config key starting with `_` are ignored and also
don't cause a config merge error"""
_write_yaml(tmp_path / "base" / "catalog1.yml", {"k1": "v1", "_k2": "v2"})
_write_yaml(tmp_path / "base" / "catalog2.yml", {"k3": "v3", "_k2": "v4"})
conf = ConfigLoader(str(tmp_path))
catalog = conf.get("**/catalog*")
assert catalog.keys() == {"k1", "k3"}
_write_yaml(tmp_path / "base" / "catalog3.yml", {"k1": "dup", "_k2": "v5"})
pattern = (
r"^Duplicate keys found in .*catalog3\.yml and\:\n\- .*catalog1\.yml\: k1$"
)
with pytest.raises(ValueError, match=pattern):
conf.get("**/catalog*")
"""Check that same configuration file is not loaded more than once."""
paths = [
str(tmp_path / "base"),
str(tmp_path / "dev"),
str(tmp_path / "dev" / "user1"),
]
_write_yaml(
tmp_path / "base" / "catalog0.yml", {"env": "base", "common": "common"}
)
_write_yaml(
tmp_path / "dev" / "catalog1.yml", {"env": "dev", "dev_specific": "wiz"}
)
_write_yaml(tmp_path / "dev" / "user1" / "catalog2.yml", {"user1_c2": True})
_write_yaml(tmp_path / "dev" / "user1" / "catalog3.yml", {"user1_c3": True})
catalog = ConfigLoader(paths).get("catalog*", "catalog*/**", "user1/catalog2*")
assert catalog == dict(
env="dev", common="common", dev_specific="wiz", user1_c2=True, user1_c3=True
)
log_messages = [record.getMessage() for record in caplog.records]
expected_path = (tmp_path / "dev" / "user1" / "catalog2.yml").resolve()
expected_message = "Config file(s): {} already processed, skipping loading...".format(
str(expected_path)
)
assert expected_message in log_messages
def test_empty_conf_paths(self):
"""Check the error if config paths were not specified or are empty"""
pattern = (
r"`conf_paths` must contain at least one path to load "
r"configuration files from"
)
with pytest.raises(ValueError, match=pattern):
ConfigLoader([])
with pytest.raises(ValueError, match=pattern):
ConfigLoader("")
def test_load_local_config(self, conf_paths):
"""Make sure that configs from `local/` override the ones
from `base/`"""
conf = ConfigLoader(conf_paths)
params = conf.get("parameters*")
db_conf = conf.get("db*")
catalog = conf.get("catalog*")
assert params["param1"] == 1
assert db_conf["prod"]["url"] == "postgresql://user:pass@url_prod/db"
assert catalog["trains"]["type"] == "MemoryDataSet"
assert catalog["cars"]["type"] == "CSVLocalDataSet"
assert catalog["boats"]["type"] == "MemoryDataSet"
assert not catalog["cars"]["save_args"]["index"]
def test_duplicate_paths(self, tmp_path, caplog):
"""Check that trying to load the same environment config multiple times logs a
warning and skips the reload"""
paths = [str(tmp_path / "base"), str(tmp_path / "base")]
_write_yaml(tmp_path / "base" / "catalog.yml", {"env": "base", "a": "a"})
with pytest.warns(UserWarning, match="Duplicate environment detected"):
conf = ConfigLoader(paths)
assert conf.conf_paths == paths[:1]
conf.get("catalog*", "catalog*/**")
log_messages = [record.getMessage() for record in caplog.records]
assert not log_messages
Args:
project_path: The root directory of the Kedro project.
env: The environment used for loading configuration.
kwargs: Ignore any additional arguments added in the future.
Returns:
ConfigLoader which can be queried to access the project config.
"""
project_path_obj = Path(project_path)
env = env or DEFAULT_RUN_ENV
conf_paths = [
str(project_path_obj / CONF_ROOT / "base"),
str(project_path_obj / CONF_ROOT / env),
]
return ConfigLoader(conf_paths)
def _create_config_loader( # pylint: disable=no-self-use
self, conf_paths: Iterable[str]
) -> ConfigLoader:
"""A factory method for the ConfigLoader instantiation.
Returns:
Instance of `ConfigLoader`.
"""
return ConfigLoader(conf_paths)
from kedro.config import ConfigLoader
IDENTIFIER_PATTERN = re.compile(
r"""\$\{
(?P