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_assign_to_reftype_plugin(
ref_type: Any, values: List[Any], assign: Any, expectation: Any
) -> None:
for value in values:
cfg = OmegaConf.create({"foo": DictConfig(ref_type=ref_type, content=value)})
with expectation():
assert _utils.get_ref_type(cfg, "foo") == Optional[ref_type]
cfg.foo = assign
assert cfg.foo == assign
# validate assignment does not change ref type.
assert _utils.get_ref_type(cfg, "foo") == Optional[ref_type]
if value is not None:
cfg = OmegaConf.create(
{"foo": DictConfig(ref_type=ref_type, content=value)}
)
with expectation():
cfg2 = OmegaConf.merge(cfg, {"foo": assign})
assert isinstance(cfg2, DictConfig)
assert cfg2.foo == assign
{"foo": DictConfig(content="${bar}"), "bar": 10},
{"foo": 10, "bar": 10},
id="dictconfig_inter",
),
pytest.param(
{"foo": ListConfig(content="${bar}"), "bar": 10},
{"foo": 10, "bar": 10},
id="listconfig_inter",
),
# None containers
pytest.param(
{"foo": DictConfig(content=None)}, {"foo": None}, id="dictconfig_none",
),
pytest.param(
{"foo": ListConfig(content=None)}, {"foo": None}, id="listconfig_none",
),
# Missing containers
def test_save_load_file_deprecated() -> None:
cfg = OmegaConf.create(dict(a=10))
assert isinstance(cfg, DictConfig)
save_load_file_deprecated(cfg)
[{"user": DictConfig(ref_type=User, content=User)}, {"user": Group}],
pytest.raises(ValidationError),
),
([Plugin, ConcretePlugin], ConcretePlugin),
pytest.param(
[{"user": "???"}, {"user": Group}],
{"user": Group},
id="merge_into_missing_node",
),
# missing DictConfig
pytest.param(
[{"dict": DictConfig(content="???")}, {"dict": {"foo": "bar"}}],
{"dict": {"foo": "bar"}},
id="merge_into_missing_DictConfig",
),
# missing Dict[str, str]
pytest.param(
pytest.param(DictConfig(content={}), Optional[Dict[Any, Any]], id="DictConfig"),
pytest.param(
DictConfig(key_type=int, element_type=Color, content={}),
Optional[Dict[int, Color]],
id="DictConfig[int,Color]",
),
pytest.param(
DictConfig(key_type=Color, element_type=int, content={}),
Optional[Dict[Color, int]],
id="DictConfig[Color,int]",
),
pytest.param(
DictConfig(ref_type=Any, content=ConcretePlugin),
Any,
id="DictConfig[ConcretePlugin]_Any_reftype",
),
pytest.param(
pytest.param(Any, {}, DictConfig(content={}), id="any_as_dict"),
pytest.param(Any, [], ListConfig(content=[]), id="any_as_list"),
# int
pytest.param(int, "foo", ValidationError, id="int"),
pytest.param(int, True, ValidationError, id="int"),
pytest.param(int, 1, IntegerNode(1), id="int"),
pytest.param(int, 1.0, ValidationError, id="int"),
pytest.param(int, Color.RED, ValidationError, id="int"),
# float
pytest.param(float, "foo", ValidationError, id="float"),
pytest.param(float, True, ValidationError, id="float"),
pytest.param(float, 1, FloatNode(1), id="float"),
pytest.param(float, 1.0, FloatNode(1.0), id="float"),
pytest.param(float, Color.RED, ValidationError, id="float"),
# # bool
pytest.param(bool, "foo", ValidationError, id="bool"),
pytest.param(bool, True, BooleanNode(True), id="bool"),
is_optional=is_optional,
key_type=key_type,
element_type=element_type,
)
elif is_list:
element_type = get_list_element_type(type_)
node = ListConfig(
content=value,
key=key,
parent=parent,
is_optional=is_optional,
element_type=element_type,
)
elif is_structured_config(type_) or is_structured_config(value):
key_type, element_type = get_dict_key_value_types(type_)
node = DictConfig(
ref_type=type_,
is_optional=is_optional,
content=value,
key=key,
parent=parent,
key_type=key_type,
element_type=element_type,
)
elif type_ == Any or type_ is None:
node = AnyNode(value=value, key=key, parent=parent, is_optional=is_optional)
elif issubclass(type_, Enum):
node = EnumNode(
enum_type=type_,
value=value,
key=key,
parent=parent,
dest_node = dest._get_node(key, validate_access=False)
if isinstance(dest_node, Container) and OmegaConf.is_none(dest, key):
if not OmegaConf.is_none(src_value):
expand(dest_node)
if dest_node is not None:
if dest_node._is_interpolation():
target_node = dest_node._dereference_node(
throw_on_resolution_failure=False
)
if isinstance(target_node, Container):
dest[key] = target_node
dest_node = dest._get_node(key)
if is_structured_config(dest._metadata.element_type):
dest[key] = DictConfig(content=dest._metadata.element_type, parent=dest)
dest_node = dest._get_node(key)
if dest_node is not None:
if isinstance(dest_node, BaseContainer):
if isinstance(src_value, BaseContainer):
dest._validate_merge(key=key, value=src_value)
dest_node._merge_with(src_value)
else:
dest.__setitem__(key, src_value)
else:
if isinstance(src_value, BaseContainer):
dest.__setitem__(key, src_value)
else:
assert isinstance(dest_node, ValueNode)
try:
dest_node._set_value(src_value)
else:
key_matches = []
matches.extend(["{}{}".format(word, match) for match in key_matches])
else:
last_dot = word.rfind(".")
if last_dot != -1:
base_key = word[0:last_dot]
partial_key = word[last_dot + 1 :]
conf_node = config.select(base_key)
key_matches = CompletionPlugin._get_matches(conf_node, partial_key)
matches.extend(
["{}.{}".format(base_key, match) for match in key_matches]
)
else:
if isinstance(config, DictConfig):
for key, value in config.items(resolve=False):
if key.startswith(word):
matches.append(str_rep(key, value))
elif isinstance(config, ListConfig):
for idx, value in enumerate(config):
if str(idx).startswith(word):
matches.append(str_rep(idx, value))
else:
assert False, "Object is not an instance of config : {}".format(
type(config)
)
return matches
def _map_merge(dest: "BaseContainer", src: "BaseContainer") -> None:
"""merge src into dest and return a new copy, does not modified input"""
from omegaconf import DictConfig, OmegaConf, ValueNode
assert isinstance(dest, DictConfig)
assert isinstance(src, DictConfig)
src_type = src._metadata.object_type
# if source DictConfig is missing set the DictConfig one to be missing too.
if src._is_missing():
dest._set_value("???")
return
dest._validate_set_merge_impl(key=None, value=src, is_assign=False)
def expand(node: Container) -> None:
type_ = get_ref_type(node)
if type_ is not None:
_is_optional, type_ = _resolve_optional(type_)
if is_dict_annotation(type_):
node._set_value({})
else:
node._set_value(type_)