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_api_version(self, *_):
api = pynetbox.api(
host,
)
self.assertEqual(api.version, "1.999")
def common_arguments(self, kwargs, arg, expect, *_):
'''
Ensures the api and endpoint instances have ssl_verify set
as expected
'''
api = pynetbox.api(
host,
**kwargs
)
self.assertIs(getattr(api, arg, "fail"), expect)
for app, endpoint in endpoints.items():
ep = getattr(getattr(api, app), endpoint)
self.assertIs(getattr(ep, arg), expect)
def query_netbox(self):
nb = netbox_api(self.netbox_address, self.netbox_token)
for device in nb.dcim.devices.all():
device_ip = device.primary_ip4 or device.primary_ip6
db.factory(
"device",
**{
"name": device.name,
"ip_address": str(device_ip).split("/")[0],
"subtype": str(device.device_role),
"model": str(device.device_type),
"location": str(device.site),
"vendor": str(device.device_type.manufacturer),
"operating_system": str(device.platform),
"longitude": str(nb.dcim.sites.get(name=device.site).longitude),
"latitude": str(nb.dcim.sites.get(name=device.site).latitude),
},
# Fail if device name is not given
if not module.params["data"].get("name"):
module.fail_json(msg="missing device name")
# Assign variables to be used with module
app = 'dcim'
endpoint = 'devices'
url = module.params["netbox_url"]
token = module.params["netbox_token"]
data = module.params["data"]
state = module.params["state"]
validate_certs = module.params["validate_certs"]
# Attempt to create Netbox API object
try:
nb = pynetbox.api(url, token=token, ssl_verify=validate_certs)
except Exception:
module.fail_json(msg="Failed to establish connection to Netbox API")
try:
nb_app = getattr(nb, app)
except AttributeError:
module.fail_json(msg="Incorrect application specified: %s" % (app))
nb_endpoint = getattr(nb_app, endpoint)
norm_data = normalize_data(data)
try:
if 'present' in state:
result = ensure_device_present(nb, nb_endpoint, norm_data)
else:
result = ensure_device_absent(nb_endpoint, norm_data)
return module.exit_json(**result)
except pynetbox.RequestError as e:
netbox_api_token = kwargs.get("token")
netbox_api_endpoint = kwargs.get("api_endpoint")
netbox_ssl_verify = kwargs.get("validate_certs", True)
netbox_private_key_file = kwargs.get("key_file")
netbox_api_filter = kwargs.get("api_filter")
netbox_raw_return = kwargs.get("raw_data")
if not isinstance(terms, list):
terms = [terms]
try:
session = requests.Session()
session.verify = netbox_ssl_verify
netbox = pynetbox.api(
netbox_api_endpoint,
token=netbox_api_token if netbox_api_token else None,
private_key_file=netbox_private_key_file,
)
netbox.http_session = session
except FileNotFoundError:
raise AnsibleError(
"%s cannot be found. Please make sure file exists."
% netbox_private_key_file
)
results = []
for term in terms:
try:
endpoint = get_endpoint(netbox, term)
def _connect_netbox_api(self, url, token, ssl_verify):
try:
session = requests.Session()
session.verify = ssl_verify
nb = pynetbox.api(url, token=token)
nb.http_session = session
try:
self.version = float(nb.version)
except AttributeError:
self.module.fail_json(msg="Must have pynetbox >=4.1.0")
except Exception:
self.module.fail_json(
msg="Failed to establish connection to Netbox API"
)
return nb
except Exception:
self.module.fail_json(msg="Failed to establish connection to Netbox API")
if not HAS_PYNETBOX:
module.fail_json(msg=missing_required_lib('pynetbox'), exception=PYNETBOX_IMP_ERR)
# Assign variables to be used with module
changed = False
app = 'ipam'
endpoint = 'ip_addresses'
url = module.params["netbox_url"]
token = module.params["netbox_token"]
data = module.params["data"]
state = module.params["state"]
validate_certs = module.params["validate_certs"]
# Attempt to create Netbox API object
try:
nb = pynetbox.api(url, token=token, ssl_verify=validate_certs)
except Exception:
module.fail_json(msg="Failed to establish connection to Netbox API")
try:
nb_app = getattr(nb, app)
except AttributeError:
module.fail_json(msg="Incorrect application specified: %s" % (app))
nb_endpoint = getattr(nb_app, endpoint)
norm_data = normalize_data(data)
try:
norm_data = _check_and_adapt_data(nb, norm_data)
if state in ("new", "present"):
return _handle_state_new_present(
module, state, nb_app, nb_endpoint, norm_data
)
elif state == "absent":
def _nb_obj(auth_required=False):
pynb_kwargs = {}
pynb_kwargs['token'] = _config().get('token')
if auth_required:
pynb_kwargs['private_key_file'] = _config().get('keyfile')
return pynetbox.api(_config().get('url'), **pynb_kwargs)
supports_check_mode=True)
# Fail module if pynetbox is not installed
if not HAS_PYNETBOX:
module.fail_json(msg=missing_required_lib('pynetbox'), exception=PYNETBOX_IMP_ERR)
# Assign variables to be used with module
app = "dcim"
endpoint = "interfaces"
url = module.params["netbox_url"]
token = module.params["netbox_token"]
data = module.params["data"]
state = module.params["state"]
validate_certs = module.params["validate_certs"]
# Attempt to create Netbox API object
try:
nb = pynetbox.api(url, token=token, ssl_verify=validate_certs)
except Exception:
module.fail_json(msg="Failed to establish connection to Netbox API")
try:
nb_app = getattr(nb, app)
except AttributeError:
module.fail_json(msg="Incorrect application specified: %s" % (app))
nb_endpoint = getattr(nb_app, endpoint)
norm_data = normalize_data(data)
try:
norm_data = _check_and_adapt_data(nb, norm_data)
if "present" in state:
return module.exit_json(
**ensure_interface_present(nb, nb_endpoint, norm_data)
)
else: