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_deepcopy_struct(self, src: Any) -> None:
c1 = OmegaConf.create(src)
OmegaConf.set_struct(c1, True)
c2 = copy.deepcopy(c1)
assert c1 == c2
if isinstance(c2, ListConfig):
c2.append(1000)
elif isinstance(c2, DictConfig):
with raises(AttributeError):
c2.foo = 42
def test_create_from_listconfig_preserves_metadata() -> None:
cfg1 = ListConfig(element_type=int, is_optional=False, content=[1, 2, 3])
OmegaConf.set_struct(cfg1, True)
OmegaConf.set_readonly(cfg1, True)
cfg2 = OmegaConf.create(cfg1)
assert cfg1 == cfg2
assert cfg1._metadata == cfg2._metadata
def test_deepcopy_and_merge_and_flags() -> None:
c1 = OmegaConf.create(
{"dataset": {"name": "imagenet", "path": "/datasets/imagenet"}, "defaults": []}
)
OmegaConf.set_struct(c1, True)
c2 = copy.deepcopy(c1)
with raises(ConfigKeyError):
OmegaConf.merge(c2, OmegaConf.from_dotlist(["dataset.bad_key=yes"]))
def validate(cfg: DictConfig) -> None:
assert not OmegaConf.is_struct(cfg)
with pytest.raises(AttributeError):
# noinspection PyStatementEffect
cfg.foo
cfg.dict1.foo = 10
assert cfg.dict1.foo == 10
# setting struct False on a specific typed node opens it up even though it's
# still typed
OmegaConf.set_struct(cfg, False)
cfg.foo = 20
assert cfg.foo == 20
def test_merge_config_with_struct(
in_base: Dict[str, Any], in_merged: Dict[str, Any]
) -> None:
base = OmegaConf.create(in_base)
merged = OmegaConf.create(in_merged)
OmegaConf.set_struct(base, True)
with pytest.raises(ConfigKeyError):
OmegaConf.merge(base, merged)
def test_struct_set_on_nested_dict() -> None:
c = OmegaConf.create(dict(a=dict(b=10)))
OmegaConf.set_struct(c, True)
with pytest.raises(AttributeError):
# noinspection PyStatementEffect
c.foo
assert "a" in c
assert c.a.b == 10
with pytest.raises(AttributeError, match=re.escape("a.foo")):
# noinspection PyStatementEffect
c.a.foo
def test_create_from_dictconfig_preserves_metadata() -> None:
cfg1 = DictConfig(ref_type=Plugin, is_optional=False, content=ConcretePlugin)
OmegaConf.set_struct(cfg1, True)
OmegaConf.set_readonly(cfg1, True)
cfg2 = OmegaConf.create(cfg1)
assert cfg1 == cfg2
assert cfg1._metadata == cfg2._metadata
defaults = hydra_cfg.defaults or []
if config_file is not None:
defaults.append(config_file)
split_at = len(defaults)
ConfigLoader._merge_default_lists(defaults, job_cfg.defaults or [])
consumed = self._apply_defaults_overrides(overrides, defaults)
consumed_free_job_defaults = self._apply_free_defaults(defaults, overrides)
ConfigLoader._validate_defaults(defaults)
# Load and defaults and merge them into cfg
cfg = self._merge_defaults(hydra_cfg, defaults, split_at)
OmegaConf.set_struct(cfg.hydra, True)
OmegaConf.set_struct(cfg, strict)
# Merge all command line overrides after enabling strict flag
all_consumed = consumed + consumed_free_job_defaults
remaining_overrides = [x for x in overrides if x not in all_consumed]
cfg.merge_with_dotlist(remaining_overrides)
remaining = consumed + consumed_free_job_defaults + remaining_overrides
def is_hydra(x):
return x.startswith("hydra.") or x.startswith("hydra/")
cfg.hydra.overrides.task = [x for x in remaining if not is_hydra(x)]
cfg.hydra.overrides.hydra = [x for x in remaining if is_hydra(x)]
with open_dict(cfg.hydra.job):
if "name" not in cfg.hydra.job:
job_cfg = self._create_cfg(cfg_filename=config_file, record_load=False)
defaults = hydra_cfg.defaults or []
if config_file is not None:
defaults.append(config_file)
split_at = len(defaults)
ConfigLoader._merge_default_lists(defaults, job_cfg.defaults or [])
consumed = self._apply_defaults_overrides(overrides, defaults)
consumed_free_job_defaults = self._apply_free_defaults(defaults, overrides)
ConfigLoader._validate_defaults(defaults)
# Load and defaults and merge them into cfg
cfg = self._merge_defaults(hydra_cfg, defaults, split_at)
OmegaConf.set_struct(cfg.hydra, True)
OmegaConf.set_struct(cfg, strict)
# Merge all command line overrides after enabling strict flag
all_consumed = consumed + consumed_free_job_defaults
remaining_overrides = [x for x in overrides if x not in all_consumed]
cfg.merge_with_dotlist(remaining_overrides)
remaining = consumed + consumed_free_job_defaults + remaining_overrides
def is_hydra(x):
return x.startswith("hydra.") or x.startswith("hydra/")
cfg.hydra.overrides.task = [x for x in remaining if not is_hydra(x)]
cfg.hydra.overrides.hydra = [x for x in remaining if is_hydra(x)]
with open_dict(cfg.hydra.job):