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_from_sane_config_default(sane_config, dummy_dataframe, tmpdir):
catalog = DataCatalog.from_config(
sane_config["catalog"], sane_config["credentials"]
)
catalog_with_default = DataCatalogWithDefault.from_data_catalog(
catalog, default_csv
)
path = str(tmpdir.mkdir("sub").join("missing.csv"))
catalog_with_default.save(path, dummy_dataframe)
reloaded_df = catalog_with_default.load(path)
assert dummy_dataframe.equals(reloaded_df)
def test_from_sane_config_load_versions_warn(self, sane_config):
sane_config["catalog"]["boats"]["versioned"] = True
version = generate_timestamp()
load_version = {"non-boart": version}
pattern = r"\`load_versions\` keys \[non-boart\] are not found in the catalog\."
with pytest.warns(UserWarning, match=pattern):
DataCatalog.from_config(**sane_config, load_versions=load_version)
def test_config_invalid_arguments(self, sane_config):
"""Check the error if the data set config contains invalid arguments"""
sane_config["catalog"]["boats"]["save_and_load_args"] = False
pattern = (
r"DataSet 'boats' must only contain arguments valid for "
r"the constructor of `.*CSVLocalDataSet`"
)
with pytest.raises(DataSetError, match=pattern):
DataCatalog.from_config(**sane_config)
def test_config_missing_class(self, sane_config):
"""Check the error if the type points to nonexistent class"""
sane_config["catalog"]["boats"]["type"] = "kedro.io.CSVLocalDataSetInvalid"
pattern = (
"An exception occurred when parsing config for DataSet `boats`:\n"
"Class `kedro.io.CSVLocalDataSetInvalid` not found"
)
with pytest.raises(DataSetError, match=re.escape(pattern)):
DataCatalog.from_config(**sane_config)
def test_idempotent_catalog(self, sane_config):
"""Test that data catalog instantiations are idempotent"""
_ = DataCatalog.from_config(**sane_config) # NOQA
catalog = DataCatalog.from_config(**sane_config)
assert catalog
def test_idempotent_catalog(self, sane_config):
"""Test that data catalog instantiations are idempotent"""
_ = DataCatalog.from_config(**sane_config) # NOQA
catalog = DataCatalog.from_config(**sane_config)
assert catalog
def test_link_credentials(self, sane_config, mocker):
"""Test credentials being linked to the relevant data set"""
mock_client = mocker.patch("kedro.io.csv_s3.S3FileSystem")
DataCatalog.from_config(**sane_config)
expected_client_kwargs = {
"aws_access_key_id": sane_config["credentials"]["s3_credentials"][
"aws_access_key_id"
],
"aws_secret_access_key": sane_config["credentials"]["s3_credentials"][
"aws_secret_access_key"
],
}
mock_client.assert_called_once_with(client_kwargs=expected_client_kwargs)
DataCatalog defined in `catalog.yml`.
"""
conf_logging = config.get("logging*", "logging*/**")
logging.config.dictConfig(conf_logging)
conf_catalog = config.get("catalog*", "catalog*/**")
try:
conf_creds = config.get("credentials*", "credentials*/**")
except MissingConfigException:
warn("Your Kedro project is missing a credentials file!")
conf_creds = None
conf_params = config.get("parameters*", "parameters*/**")
logging.config.dictConfig(conf_logging)
catalog = DataCatalog.from_config(conf_catalog, conf_creds)
catalog.add_feed_dict({"parameters": conf_params})
return catalog
def _create_catalog( # pylint: disable=no-self-use,too-many-arguments
self,
conf_catalog: Dict[str, Any],
conf_creds: Dict[str, Any],
save_version: str = None,
journal: Journal = None,
load_versions: Dict[str, str] = None,
) -> DataCatalog:
"""A factory method for the DataCatalog instantiation.
Returns:
DataCatalog defined in `catalog.yml`.
"""
return DataCatalog.from_config(
conf_catalog,
conf_creds,
save_version=save_version,
journal=journal,
load_versions=load_versions,
)