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_inventory(self):
g1 = inventory.Group(name="g1")
g2 = inventory.Group(name="g2", groups=inventory.ParentGroups(["g1"]))
h1 = inventory.Host(name="h1", groups=inventory.ParentGroups(["g1", "g2"]))
h2 = inventory.Host(name="h2")
hosts = {"h1": h1, "h2": h2}
groups = {"g1": g1, "g2": g2}
inv = inventory.Inventory(hosts=hosts, groups=groups)
assert "h1" in inv.hosts
assert "h2" in inv.hosts
assert "g1" in inv.groups
assert "g2" in inv.groups
assert inv.groups["g1"] in inv.hosts["h1"].groups
assert inv.groups["g1"] in inv.groups["g2"].groups
def test_host(self):
h = inventory.Host(name="host1", hostname="host1")
assert h.hostname == "host1"
assert h.port is None
assert h.username is None
assert h.password is None
assert h.platform is None
assert h.data == {}
data = {"asn": 65100, "router_id": "1.1.1.1"}
h = inventory.Host(
name="host2",
hostname="host2",
username="user",
port=123,
password="",
platform="fake",
data=data,
)
assert h.hostname == "host2"
assert h.port == 123
assert h.username == "user"
assert h.password == ""
assert h.platform == "fake"
assert h.data == data
def test_class1_ex1():
base_path = "../class1/exercises/exercise1/"
nornir_inventory = gen_inventory_dict(base_path)
nr = InitNornir(inventory=nornir_inventory, logging=NORNIR_LOGGING)
assert isinstance(nr, nornir.core.Nornir)
assert isinstance(nr.inventory.hosts, nornir.core.inventory.Hosts)
assert isinstance(nr.inventory.hosts["my_host"], nornir.core.inventory.Host)
assert nr.inventory.hosts["my_host"].hostname == "localhost"
def test_add_group(self):
connection_options = {"username": "test_user", "password": "test_pass"}
data = {"test_var": "test_value"}
defaults = inventory.Defaults(data=data, connection_options=connection_options)
g1 = inventory.Group(name="g1")
g2 = inventory.Group(name="g2", groups=inventory.ParentGroups(["g1"]))
h1 = inventory.Host(name="h1", groups=inventory.ParentGroups(["g1", "g2"]))
h2 = inventory.Host(name="h2")
hosts = {"h1": h1, "h2": h2}
groups = {"g1": g1, "g2": g2}
inv = inventory.Inventory(hosts=hosts, groups=groups, defaults=defaults)
g3_connection_options = {"netmiko": {"extras": {"device_type": "cisco_ios"}}}
inv.add_group(
name="g3", username="test_user", connection_options=g3_connection_options
)
assert "g3" in inv.groups
assert (
inv.groups["g3"].defaults.connection_options.get("username") == "test_user"
)
assert (
inv.groups["g3"].defaults.connection_options.get("password") == "test_pass"
)
assert "test_var" in inv.groups["g3"].defaults.data.keys()
def test_add_host(self):
data = {"test_var": "test_value"}
defaults = inventory.Defaults(data=data)
g1 = inventory.Group(name="g1")
g2 = inventory.Group(name="g2", groups=inventory.ParentGroups(["g1"]))
h1 = inventory.Host(name="h1", groups=inventory.ParentGroups(["g1", "g2"]))
h2 = inventory.Host(name="h2")
hosts = {"h1": h1, "h2": h2}
groups = {"g1": g1, "g2": g2}
inv = inventory.Inventory(hosts=hosts, groups=groups, defaults=defaults)
h3_connection_options = {"netmiko": {"extras": {"device_type": "cisco_ios"}}}
inv.add_host(
name="h3",
groups=["g1"],
platform="TestPlatform",
connection_options=h3_connection_options,
)
assert "h3" in inv.hosts
assert "g1" in inv.hosts["h3"].groups
assert "test_var" in inv.hosts["h3"].defaults.data.keys()
assert inv.hosts["h3"].defaults.data.get("test_var") == "test_value"
assert inv.hosts["h3"].platform == "TestPlatform"
def test_add_host(self):
data = {"test_var": "test_value"}
defaults = inventory.Defaults(data=data)
g1 = inventory.Group(name="g1")
g2 = inventory.Group(name="g2", groups=inventory.ParentGroups(["g1"]))
h1 = inventory.Host(name="h1", groups=inventory.ParentGroups(["g1", "g2"]))
h2 = inventory.Host(name="h2")
hosts = {"h1": h1, "h2": h2}
groups = {"g1": g1, "g2": g2}
inv = inventory.Inventory(hosts=hosts, groups=groups, defaults=defaults)
h3_connection_options = {"netmiko": {"extras": {"device_type": "cisco_ios"}}}
inv.add_host(
name="h3",
groups=["g1"],
platform="TestPlatform",
connection_options=h3_connection_options,
)
assert "h3" in inv.hosts
assert "g1" in inv.hosts["h3"].groups
assert "test_var" in inv.hosts["h3"].defaults.data.keys()
assert inv.hosts["h3"].defaults.data.get("test_var") == "test_value"
assert inv.hosts["h3"].platform == "TestPlatform"
assert (
def test_inventory(self):
g1 = inventory.Group(name="g1")
g2 = inventory.Group(name="g2", groups=inventory.ParentGroups(["g1"]))
h1 = inventory.Host(name="h1", groups=inventory.ParentGroups(["g1", "g2"]))
h2 = inventory.Host(name="h2")
hosts = {"h1": h1, "h2": h2}
groups = {"g1": g1, "g2": g2}
inv = inventory.Inventory(hosts=hosts, groups=groups)
assert "h1" in inv.hosts
assert "h2" in inv.hosts
assert "g1" in inv.groups
assert "g2" in inv.groups
assert inv.groups["g1"] in inv.hosts["h1"].groups
assert inv.groups["g1"] in inv.groups["g2"].groups
def deserialize_host(cls, **kwargs: Any) -> inventory.Host:
return inventory.Host(**cls.deserialize(**kwargs))
raise AttributeError(
"not sure how to establish a connection for {}".format(connection)
)
# We use `filter(name=self.name)` to call the connection task for only
# the given host. We also have to set `num_workers=1` because chances are
# we are already inside a thread
# Task should establish a connection and populate self.connection[connection]
r = self.nornir.filter(name=self.name).run(conn_task, num_workers=1)
if r[self.name].exception:
raise r[self.name].exception
return self.connections[connection]
class Group(Host):
"""Same as :obj:`Host`"""
def children(self):
return {
n: h
for n, h in self.nornir.inventory.hosts.items()
if h.has_parent_group(self)
}
class Inventory(object):
"""
An inventory contains information about hosts and groups.
Arguments:
hosts (dict): keys are hostnames and values are either :obj:`Host` or a dict
self.defaults = defaults or {}
self.groups = groups or {}
for n, g in self.groups.items():
if isinstance(g, dict):
g = Group(name=n, nornir=nornir, **g)
self.groups[n] = g
for group in self.groups.values():
group.groups = self._resolve_groups(group.groups)
self.hosts = {}
for n, h in hosts.items():
if isinstance(h, dict):
h = Host(name=n, nornir=nornir, defaults=self.defaults, **h)
if transform_function:
transform_function(h)
h.groups = self._resolve_groups(h.groups)
self.hosts[n] = h