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_deserializer_wrong(self):
with pytest.raises(ValidationError):
deserializer.Inventory.deserialize(
**{"hosts": {"wrong": {"host": "should_be_hostname"}}}
)
def test_inventory_pagination(self, requests_mock):
inv = get_inv(requests_mock, "2.3.5", True)
with open("{}/{}/expected.json".format(BASE_PATH, "2.3.5"), "r") as f:
expected = json.load(f)
assert expected == Inventory.serialize(inv).dict()
def test_inventory(self, requests_mock):
inv = get_inv(requests_mock, "2.3.5", False)
with open("{}/{}/expected.json".format(BASE_PATH, "2.3.5"), "r") as f:
expected = json.load(f)
assert expected == Inventory.serialize(inv).dict()
import os
from nornir import InitNornir
from nornir.core.deserializer.inventory import Inventory
dir_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "test_InitNornir")
def transform_func(host):
host["processed_by_transform_function"] = True
class StringInventory(Inventory):
def __init__(self, **kwargs):
inv_dict = {"hosts": {"host1": {}, "host2": {}}, "groups": {}, "defaults": {}}
super().__init__(**inv_dict, **kwargs)
class Test(object):
def test_InitNornir_defaults(self):
os.chdir("tests/inventory_data/")
nr = InitNornir()
os.chdir("../../")
assert not nr.data.dry_run
assert nr.config.core.num_workers == 20
assert len(nr.inventory.hosts)
assert len(nr.inventory.groups)
def test_InitNornir_file(self):
def deserialize(cls, **kwargs: Any) -> configuration.InventoryConfig:
inv = InventoryConfig(**kwargs)
return configuration.InventoryConfig(
plugin=cast(Type[Inventory], _resolve_import_from_string(inv.plugin)),
options=inv.options,
transform_function=_resolve_import_from_string(inv.transform_function),
transform_function_options=inv.transform_function_options,
)
import os
from typing import Any
from nornir.core.deserializer.inventory import Inventory, InventoryElement
import requests
class NSOTInventory(Inventory):
"""
Inventory plugin that uses `nsot `_ as backend.
Note:
An extra attribute ``site`` will be assigned to the host. The value will be
the name of the site the host belongs to.
Environment Variables:
* ``NSOT_URL``: Corresponds to nsot_url argument
* ``NSOT_EMAIL``: Corresponds to nsot_email argument
* ``NSOT_AUTH_HEADER``: Corresponds to nsot_auth_header argument
* ``NSOT_SECRET_KEY``: Corresponds to nsot_secret_key argument
Arguments:
flatten_attributes: Assign host attributes to the root object. Useful
for filtering hosts.
def dict(self) -> Dict:
"""
Return serialized dictionary of inventory
"""
return deserializer.inventory.Inventory.serialize(self).dict()
import os
from typing import Any, Dict, List, Optional, Union
from nornir.core.deserializer.inventory import Inventory, HostsDict
import requests
class NBInventory(Inventory):
def __init__(
self,
nb_url: Optional[str] = None,
nb_token: Optional[str] = None,
use_slugs: bool = True,
ssl_verify: Union[bool, str] = True,
flatten_custom_fields: bool = True,
filter_parameters: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> None:
"""
Netbox plugin
Arguments:
nb_url: Netbox url, defaults to http://localhost:8080.
You can also use env variable NB_URL
def serialize(cls, inv: inventory.Inventory) -> "Inventory":
hosts = {}
for n, h in inv.hosts.items():
hosts[n] = InventoryElement.serialize(h)
groups = {}
for n, g in inv.groups.items():
groups[n] = InventoryElement.serialize(g)
defaults = Defaults.serialize(inv.defaults)
return Inventory(hosts=hosts, groups=groups, defaults=defaults)
import os
from typing import Any, Optional
from nornir.core.deserializer.inventory import (
HostsDict,
GroupsDict,
Inventory,
VarsDict,
)
import ruamel.yaml
logger = logging.getLogger(__name__)
class SimpleInventory(Inventory):
def __init__(
self,
host_file: str = "hosts.yaml",
group_file: str = "groups.yaml",
defaults_file: str = "defaults.yaml",
hosts: Optional[HostsDict] = None,
groups: Optional[GroupsDict] = None,
defaults: Optional[VarsDict] = None,
*args: Any,
**kwargs: Any
) -> None:
yml = ruamel.yaml.YAML(typ="safe")
if hosts is None:
with open(os.path.expanduser(host_file), "r") as f:
hosts = yml.load(f)