Skip to content

Commit

Permalink
Removing Netbox v3 client
Browse files Browse the repository at this point in the history
  • Loading branch information
@johannwagner
johannwagner committed Nov 19, 2024
1 parent 8cd6089 commit 6b18b63
Show file tree
Hide file tree
Showing 4 changed files with 62 additions and 215 deletions.
162 changes: 3 additions & 159 deletions cosmo/netboxclient.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,9 @@ def __init__(self, url, token):
self.url = url
self.token = token

v = self.query_version()
self.version = self.query_version()

if v.startswith("wc_3."):
print("[INFO] Using version 3.x strategy...")
self.child_client = NetboxV3Strategy(url, token)
elif v.startswith("4."):
if self.version.startswith("4."):
print("[INFO] Using version 4.x strategy...")
self.child_client = NetboxV4Strategy(url, token)
else:
Expand All @@ -35,6 +32,7 @@ def query_version(self):

json = r.json()
return json['netbox-version']

def get_data(self, device_config):
return self.child_client.get_data(device_config)

Expand Down Expand Up @@ -86,159 +84,6 @@ def query_rest(self, path, queries):
return return_array


class NetboxV3Strategy(NetboxStrategy):

def get_data(self, device_config):
query_template = Template(
"""
{
device_list(
name: $device_array,
) {
id
name
serial
device_type {
slug
}
platform {
manufacturer {
slug
}
slug
}
primary_ip4 {
address
}
interfaces {
id
name
enabled
type
mode
mtu
mac_address
description
vrf {
id
}
lag {
id
}
ip_addresses {
address
}
untagged_vlan {
id
name
vid
}
tagged_vlans {
id
name
vid
}
tags {
name
slug
}
parent {
id
}
connected_endpoints {
... on InterfaceType {
name
device {
primary_ip4 {
address
}
interfaces {
ip_addresses {
address
}
}
}
}
}
custom_fields
}
staticroute_set {
interface {
name
}
vrf {
name
}
prefix {
prefix
family {
value
}
}
next_hop {
address
}
metric
}
}
vrf_list {
id
name
description
rd
export_targets {
name
}
import_targets {
name
}
}
l2vpn_list {
id
name
type
identifier
terminations {
id
assigned_object {
__typename
... on VLANType {
id
}
... on InterfaceType {
id
device {
name
interfaces (type: "virtual") {
ip_addresses {
address
}
parent {
name
type
}
vrf {
id
}
}
}
}
}
}
}
}"""
)

query = query_template.substitute(
device_array=json.dumps(device_config['router'] + device_config['switch'])
)

r = self.query(query)

return r['data']


class NetboxV4Strategy(NetboxStrategy):

def get_data(self, device_config):
Expand Down Expand Up @@ -378,7 +223,6 @@ def get_data(self, device_config):
static_routes = self.query_rest("api/plugins/routing/staticroutes/", {"device": device_list})

for d in r['data']['device_list']:

device_static_routes = list(filter(lambda sr: str(sr['device']['id']) == d['id'], static_routes))
d['staticroute_set'] = device_static_routes

Expand Down
16 changes: 8 additions & 8 deletions cosmo/tests/integration_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,37 +23,37 @@ def test_missing_netbox_api_token(mocker):

def test_limit_argument_with_commas(mocker):
utils.CommonSetup(mocker, args=[utils.CommonSetup.PROGNAME, '--limit', 'router1,router2'])
utils.RequestResponseMock.patchTool(mocker)
utils.RequestResponseMock.patchNetboxClient(mocker)
assert cosmoMain() == 0

def test_limit_arguments_with_repeat(mocker):
utils.CommonSetup(mocker, args=[utils.CommonSetup.PROGNAME, '--limit', 'router1', '--limit', 'router2'])
utils.RequestResponseMock.patchTool(mocker)
utils.RequestResponseMock.patchNetboxClient(mocker)
assert cosmoMain() == 0

def test_device_generation_ansible(mocker):
testEnv = utils.CommonSetup(mocker, cfgFile='cosmo/tests/cosmo.devgen_ansible.yml')
with open(f"cosmo/tests/test_case_l3vpn.yml") as f:
utils.RequestResponseMock.patchTool(
mocker,{'status_code': 200, 'text': '{"data": ' + json.dumps(yaml.safe_load(f)) + '}'})
test_data = yaml.safe_load(f)
utils.RequestResponseMock.patchNetboxClient(mocker, **test_data)
assert cosmoMain() == 0
testEnv.stop()
assert os.path.isfile('host_vars/test0001/generated-cosmo.yml')

def test_device_generation_nix(mocker):
testEnv = utils.CommonSetup(mocker, cfgFile='cosmo/tests/cosmo.devgen_nix.yml')
with open(f"cosmo/tests/test_case_l3vpn.yml") as f:
utils.RequestResponseMock.patchTool(
mocker,{'status_code': 200, 'text': '{"data": ' + json.dumps(yaml.safe_load(f)) + '}'})
test_data = yaml.safe_load(f)
utils.RequestResponseMock.patchNetboxClient(mocker, **test_data)
assert cosmoMain() == 0
testEnv.stop()
assert os.path.isfile('machines/test0001/generated-cosmo.json')

def test_device_processing_error(mocker):
testEnv = utils.CommonSetup(mocker, cfgFile='cosmo/tests/cosmo.devgen_nix.yml')
with open(f"cosmo/tests/test_case_vendor_unknown.yaml") as f:
utils.RequestResponseMock.patchTool(
mocker,{'status_code': 200, 'text': '{"data": ' + json.dumps(yaml.safe_load(f)) + '}'})
test_data = yaml.safe_load(f)
utils.RequestResponseMock.patchNetboxClient(mocker, **test_data)
with pytest.warns(UserWarning, match="unsupported platform vendor"):
assert cosmoMain() == 0
testEnv.stop()
52 changes: 17 additions & 35 deletions cosmo/tests/test_netboxclient.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import pytest

import cosmo.tests.utils as utils
from cosmo.netboxclient import NetboxClient, NetboxV3Strategy, NetboxV4Strategy
from cosmo.netboxclient import NetboxClient

TEST_URL = 'https://netbox.example.com'
TEST_TOKEN = 'token123'
Expand All @@ -16,46 +16,28 @@
]}


def test_case_query_v3_ok(mocker):
utils.RequestResponseMock.patchTool(mocker)
nc = NetboxV3Strategy(TEST_URL, TEST_TOKEN)
nc.query('check')


def test_case_query_v3_nok(mocker):
with pytest.raises(Exception):
utils.RequestResponseMock.patchTool(
mocker, graphqlData={'status_code': 403, 'text': 'unauthorized'})
nc = NetboxV3Strategy(TEST_URL, TEST_TOKEN)
nc.query('check')


def test_case_query_v4_ok(mocker):
utils.RequestResponseMock.patchTool(mocker)
nc = NetboxV4Strategy(TEST_URL, TEST_TOKEN)
nc.query('check')

def test_case_get_data(mocker):
mockAnswer = {
"device_list": [],
"l2vpn_list": [],
"vrf_list": [],
}
[getMock, postMock] = utils.RequestResponseMock.patchNetboxClient(mocker)

def test_case_query_v4_nok(mocker):
with pytest.raises(Exception):
utils.RequestResponseMock.patchTool(
mocker, graphqlData={'status_code': 403, 'text': 'unauthorized'})
nc = NetboxV4Strategy(TEST_URL, TEST_TOKEN)
nc.query('check')
nc = NetboxClient(TEST_URL, TEST_TOKEN)
assert nc.version == "4.1.2"

getMock.assert_called_once()

def test_case_get_data(mocker):
mockAnswer = []
[versionDetectMock, dataMock] = utils.RequestResponseMock.patchTool(
mocker, graphqlData={'status_code': 200, 'text': '{"data":' + str(mockAnswer) + '}'})
nc = NetboxClient(TEST_URL, TEST_TOKEN)
responseData = nc.get_data(TEST_DEVICE_CFG)
assert responseData == mockAnswer
versionDetectMock.assert_called_once()
dataMock.assert_called_once()
kwargs = dataMock.call_args.kwargs

assert getMock.call_count == 2
assert postMock.call_count == 1

kwargs = postMock.call_args.kwargs
assert 'json' in kwargs
assert 'query' in kwargs['json']
ncQueryStr = kwargs['json']['query']
for device in [*TEST_DEVICE_CFG['router'], *TEST_DEVICE_CFG['switch']]:
assert device in ncQueryStr
assert device in ncQueryStr
47 changes: 34 additions & 13 deletions cosmo/tests/utils.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,25 +35,46 @@ def stop(self):


class RequestResponseMock:
def __init__(self, **kwargs):
self.status_code = kwargs['status_code']
self.text = kwargs['text']

@staticmethod
def patchTool(mocker, graphqlData=None, versionData=None):

if graphqlData is None:
graphqlData = {'status_code': 200, 'text': json.dumps({"data": {"vrf_list": [], "device_list": []}})}
def patchNetboxClient(mocker, **patchKwArgs):
def patchGetFunc(url, **kwargs):
if "/api/status" in url:
return ResponseMock(200, {"netbox-version": "4.1.2"})

return ResponseMock(200, {
"next": None,
"results": [],
})

def patchPostFunc(url, json, **kwargs):
q = json.get("query")
request_lists = [
"device_list",
"vrf_list",
"l2vpn_list"
]
retVal = dict()

for rl in request_lists:
if rl in q:
retVal[rl] = patchKwArgs.get(rl, [])

return ResponseMock(200, {"data": retVal})

postMock1 = mocker.patch('requests.get', side_effect=patchGetFunc)
postMock2 = mocker.patch('requests.post', side_effect=patchPostFunc)
return [postMock1, postMock2]

if versionData is None:
versionData = {'status_code': 200, 'text': json.dumps({"netbox-version": "wc_3.7.5-0.7.0"})}

postMock1 = mocker.patch('requests.get', return_value=RequestResponseMock(**versionData))
postMock2 = mocker.patch('requests.post', return_value=RequestResponseMock(**graphqlData))
return [postMock1, postMock2]
class ResponseMock:
def __init__(self, status_code, obj):
self.status_code = status_code
self.text = json.dumps(obj)
self.obj = obj

def json(self):
return json.loads(self.text)
return self.obj


# it has to be stateful - so I'm making an object
Expand Down

0 comments on commit 6b18b63

Please sign in to comment.