-
Bjarke Madsen authoredBjarke Madsen authored
test_update.py 15.11 KiB
import responses
import json
import re
from brian_dashboard_manager.grafana.utils.request import TokenRequest
from brian_dashboard_manager.templating.nren_access import get_nrens
from brian_dashboard_manager.grafana.provision import provision_folder, \
generate_all_nrens, provision
from brian_dashboard_manager.grafana.provision import is_re_customer, \
is_cls, is_ias_customer, is_ias_private, is_ias_public, is_ias_upstream, \
is_lag_backbone, is_phy_upstream, is_re_peer, is_gcs, \
is_geantopen, is_l2circuit, is_lhcone_peer, is_lhcone_customer, is_mdvpn,\
parse_backbone_name, parse_phy_upstream_name
DEFAULT_REQUEST_HEADERS = {
"Content-type": "application/json",
"Accept": ["application/json"]
}
TEST_INTERFACES = [
{
"router": "mx1.ath2.gr.geant.net",
"name": "xe-1/0/1",
"bundle": [],
"bundle-parents": [],
"snmp-index": 569,
"description": "PHY RESERVED | New OTEGLOBE ATH2-VIE 10Gb LS",
"circuits": []
},
{
"router": "mx1.ath2.gr.geant.net",
"name": "ge-1/3/7",
"bundle": [],
"bundle-parents": [],
"snmp-index": 543,
"description": "PHY SPARE",
"circuits": []
},
{
"router": "mx1.ham.de.geant.net",
"name": "xe-2/2/0.13",
"bundle": [],
"bundle-parents": [],
"snmp-index": 721,
"description": "SRV_L2CIRCUIT CUSTOMER WP6T3 WP6T3 #ham_lon2-WP6-GTS_20063 |", # noqa: E501
"circuits": [
{
"id": 52382,
"name": "ham_lon2-WP6-GTS_20063_L2c",
"type": "",
"status": "operational"
}
]
},
{
"router": "mx1.fra.de.geant.net",
"name": "ae27",
"bundle": [],
"bundle-parents": [
"xe-10/0/2",
"xe-10/3/2",
"xe-10/3/3"
],
"snmp-index": 760,
"description": "LAG CUSTOMER ULAKBIM SRF9940983 |",
"circuits": [
{
"id": 40983,
"name": "ULAKBIM AP2 LAG",
"type": "",
"status": "operational"
}
]
},
{
"router": "mx2.zag.hr.geant.net",
"name": "xe-2/1/0",
"bundle": [],
"bundle-parents": [],
"snmp-index": 739,
"description": "PHY SPARE",
"circuits": []
},
{
"router": "rt1.rig.lv.geant.net",
"name": "xe-0/1/5",
"bundle": [],
"bundle-parents": [],
"snmp-index": 539,
"description": "PHY SPARE",
"circuits": []
},
{
"router": "srx1.ch.office.geant.net",
"name": "ge-0/0/0",
"bundle": [],
"bundle-parents": [],
"snmp-index": 513,
"description": "Reserved for GEANT OC to test Virgin Media link",
"circuits": []
},
{
"router": "mx1.par.fr.geant.net",
"name": "xe-4/1/4.1",
"bundle": [],
"bundle-parents": [],
"snmp-index": 1516,
"description": "SRV_L2CIRCUIT INFRASTRUCTURE JRA1 JRA1 | #SDX-L2_PILOT-Br52 OF-P3_par ", # noqa: E501
"circuits": []
},
{
"router": "mx1.lon.uk.geant.net",
"name": "lt-1/3/0.61",
"bundle": [],
"bundle-parents": [],
"snmp-index": 1229,
"description": "SRV_IAS INFRASTRUCTURE ACCESS GLOBAL #LON-IAS-RE-Peering | BGP Peering - IAS Side", # noqa: E501
"circuits": []
},
{
"router": "mx1.sof.bg.geant.net",
"name": "xe-2/0/5",
"bundle": [],
"bundle-parents": [],
"snmp-index": 694,
"description": "PHY RESERVED | Prime Telecom Sofia-Bucharest 3_4",
"circuits": []
},
{
"router": "mx1.sof.bg.geant.net",
"name": "xe-2/0/5",
"bundle": [],
"bundle-parents": [],
"snmp-index": 694,
"description": "SRV_GLOBAL CUSTOMER HEANET TESTDESCRIPTION |",
"circuits": []
}
]
def generate_folder(data):
return {
"id": 555,
"uid": data['uid'],
"title": data['title'],
"url": f"/dashboards/f/{data['uid']}/{data['title'].lower()}",
"hasAcl": False,
"canSave": True,
"canEdit": True,
"canAdmin": True,
"createdBy": "Anonymous",
"created": "2021-02-23T15:33:46Z",
"updatedBy": "Anonymous",
"updated": "2021-02-23T15:33:46Z",
"version": 1
}
@responses.activate
def test_provision_folder(data_config, mocker):
dashboards = {
'CLS TESTDASHBOARD': {
'predicate': is_cls,
'tag': 'CLS'
},
'RE PEER TESTDASHBOARD': {
'predicate': is_re_peer,
'tag': 'RE_PEER'
},
'RE CUST TESTDASHBOARD': {
'predicate': is_re_customer,
'tag': 'RE_CUST'
},
'GEANTOPEN TESTDASHBOARD': {
'predicate': is_geantopen,
'tag': 'GEANTOPEN'
},
'GCS TESTDASHBOARD': {
'predicate': is_gcs,
'tag': 'AUTOMATED_L2_CIRCUITS'
},
'L2 CIRCUIT TESTDASHBOARD': {
'predicate': is_l2circuit,
'tag': 'L2_CIRCUITS'
},
'LHCONE PEER TESTDASHBOARD': {
'predicate': is_lhcone_peer,
'tag': 'LHCONE_PEER'
},
'LHCONE CUST TESTDASHBOARD': {
'predicate': is_lhcone_customer,
'tag': 'LHCONE_CUST'
},
'MDVPN Customers TESTDASHBOARD': {
'predicate': is_mdvpn,
'tag': 'MDVPN'
},
'Infrastructure Backbone TESTDASHBOARD': {
'predicate': is_lag_backbone,
'tag': 'BACKBONE',
'errors': True,
'parse_func': parse_backbone_name
},
'IAS PRIVATE TESTDASHBOARD': {
'predicate': is_ias_private,
'tag': 'IAS_PRIVATE'
},
'IAS PUBLIC TESTDASHBOARD': {
'predicate': is_ias_public,
'tag': 'IAS_PUBLIC'
},
'IAS CUSTOMER TESTDASHBOARD': {
'predicate': is_ias_customer,
'tag': 'IAS_CUSTOMER'
},
'IAS UPSTREAM TESTDASHBOARD': {
'predicate': is_ias_upstream,
'tag': 'IAS_UPSTREAM'
},
'GWS PHY Upstream TESTDASHBOARD': {
'predicate': is_phy_upstream,
'tag': 'GWS_UPSTREAM',
'errors': True,
'parse_func': parse_phy_upstream_name
}
}
# just return a generated folder
_mocked_find_folder = mocker.patch(
'brian_dashboard_manager.grafana.provision.find_folder')
_mocked_find_folder.return_value = generate_folder(
{'uid': 'testfolderuid', 'title': 'testfolder'})
# we don't care about testing create_dashboard
_mocked_find_folder = mocker.patch(
'brian_dashboard_manager.grafana.provision.create_dashboard')
_mocked_find_folder.return_value = None
for dashboard in dashboards:
provision_folder(None, 'testfolder', dashboards[dashboard],
TEST_INTERFACES,
'testdatasource', ['CLS TESTDASHBOARD'])
@responses.activate
def test_provision_nrens(data_config, mocker):
NREN_INTERFACES = [
# physical
{
"router": "mx1.dub2.ie.geant.net",
"name": "xe-0/0/0",
"bundle": ["ae10"],
"bundle-parents": [],
"snmp-index": 554,
"description": "PHY CUSTOMER HEANET P_AE10 SRF9948758 | HEANET AP2-3 LL", # noqa: E501
"circuits": []
},
# aggregate
{
"router": "mx1.dub2.ie.geant.net",
"name": "ae10",
"bundle": [],
"bundle-parents": ["xe-0/0/0", "xe-1/0/1", "xe-1/1/0"],
"snmp-index": 596,
"description": "LAG CUSTOMER HEANET SRF9925909 |",
"circuits": [
{
"id": 25909,
"name": "HEANET AP2 LAG",
"type": "",
"status": "operational"
}
]
},
# logical
{
"router": "mx1.dub2.ie.geant.net",
"name": "ae10.12",
"bundle": [],
"bundle-parents": [
"xe-0/0/0",
"xe-1/0/1",
"xe-1/1/0"
],
"snmp-index": 713,
"description": "SRV_GLOBAL CUSTOMER HEANET #HEANET_AP2 | ASN1213 | ", # noqa: E501
"circuits": [
{
"id": 48776,
"name": "HEANET AP2",
"type": "ip access",
"status": "operational"
},
{
"id": 31347,
"name": "HEANET AP2 L2c",
"type": "",
"status": "operational"
}
]
}
]
UID = '1337'
def get_callback(request):
query = request.params.get('query')
return 200, {}, json.dumps({'uid': UID, 'title': query})
responses.add_callback(
method=responses.GET,
url=re.compile(f"http://{data_config['hostname']}/api/search"),
callback=get_callback)
def post_callback(request):
dashboard = json.loads(request.body).get('dashboard')
title = dashboard.get('title')
return 200, {}, json.dumps({'uid': UID, 'title': title})
responses.add_callback(
method=responses.POST,
url=re.compile(f"http://{data_config['hostname']}/api/dashboards/db"),
callback=post_callback)
nrens = get_nrens(NREN_INTERFACES)
assert len(nrens) == 1 and nrens.get('HEANET') is not None
assert len(nrens.get('HEANET').get('AGGREGATES')) == 1
assert len(nrens.get('HEANET').get('SERVICES')) == 1
assert len(nrens.get('HEANET').get('PHYSICAL')) == 2
token_request = TokenRequest(token='testtoken', **data_config)
generate_all_nrens(token_request, NREN_INTERFACES, 1, 'testdatasource')
@responses.activate
def test_provision(data_config, mocker, client):
def get_callback(request):
return 200, {}, json.dumps(TEST_INTERFACES)
responses.add_callback(
method=responses.GET,
url=f"{data_config['inventory_provider']}/poller/interfaces",
callback=get_callback)
responses.add_callback(
method=responses.GET,
url=f"{data_config['inventory_provider']}/data/interfaces",
callback=get_callback)
def folder_get(request):
return 200, {}, json.dumps([])
responses.add_callback(
method=responses.GET,
url=f"http://{data_config['hostname']}/api/folders",
callback=folder_get)
def folder_delete(request):
return 200, {}, json.dumps({"message": "Deleted folder"})
responses.add_callback(
method=responses.DELETE,
url=re.compile(f"http://{data_config['hostname']}/api/folders"),
callback=folder_delete, )
def folder_post(request):
data = json.loads(request.body)
return 200, {}, json.dumps(generate_folder(data))
responses.add_callback(
method=responses.POST,
url=f"http://{data_config['hostname']}/api/folders",
callback=folder_post)
def home_dashboard(request):
return 200, {}, json.dumps([])
responses.add_callback(
method=responses.GET,
url=f"http://{data_config['hostname']}/api/search?query=Home",
callback=home_dashboard)
TEST_DATASOURCE = [{
"name": "brian-influx-datasource",
"type": "influxdb",
"access": "proxy",
"url": "http://test-brian-datasource.geant.org:8086",
"database": "test-db",
"basicAuth": False,
"isDefault": True,
"readOnly": False
}]
def datasources(request):
return 200, {}, json.dumps(TEST_DATASOURCE)
responses.add_callback(
method=responses.GET,
url=f"http://{data_config['hostname']}/api/datasources",
callback=datasources)
def createdashboard(request):
return 200, {}, json.dumps({'id': 666})
responses.add_callback(
method=responses.POST,
url=f"http://{data_config['hostname']}/api/dashboards/db",
callback=createdashboard)
def preferences(request):
return 200, {}, json.dumps({'message': 'Preferences updated'})
responses.add_callback(
method=responses.PUT,
url=f"http://{data_config['hostname']}/api/org/preferences",
callback=preferences)
def homedashboard(request):
return 404, {}, ''
responses.add_callback(
method=responses.GET,
url=f"http://{data_config['hostname']}/api/dashboards/uid/home",
callback=homedashboard)
PROVISIONED_ORGANIZATION = {
'name': data_config['organizations'][0],
'id': 0
}
EXISTING_ORGS = [{**org, 'id': i + 1}
for i, org in enumerate(data_config['organizations'][1:])]
_mocked_get_organizations = mocker.patch(
'brian_dashboard_manager.grafana.provision.get_organizations')
# all organizations are provisioned except the first one.
_mocked_get_organizations.return_value = EXISTING_ORGS.copy()
_mocked_create_organization = mocker.patch(
'brian_dashboard_manager.grafana.provision.create_organization')
# spoof creating first organization
_mocked_create_organization.return_value = PROVISIONED_ORGANIZATION
_mocked_delete_expired_api_tokens = mocker.patch(
'brian_dashboard_manager.grafana.provision.delete_expired_api_tokens')
# we dont care about this, , tested separately
_mocked_delete_expired_api_tokens.return_value = None
_mocked_create_api_token = mocker.patch(
'brian_dashboard_manager.grafana.provision.create_api_token')
_mocked_create_api_token.return_value = {
'key': 'testtoken', 'id': 0} # api token
_mocked_create_datasource = mocker.patch(
'brian_dashboard_manager.grafana.provision.create_datasource')
# we dont care about this, just mark it created
_mocked_create_datasource.return_value = True
_mocked_get_dashboard_definitions = mocker.patch(
'brian_dashboard_manager.grafana.provision.get_dashboard_definitions')
_mocked_gws = mocker.patch(
'brian_dashboard_manager.grafana.provision.get_gws_direct')
_mocked_gws.return_value = []
_mocked_gws_indirect = mocker.patch(
'brian_dashboard_manager.grafana.provision.get_gws_indirect')
_mocked_gws_indirect.return_value = []
UID = 1
ID = 1
VERSION = 1
TITLE = 'testdashboard'
dashboard = {'id': ID, 'uid': UID, 'title': TITLE, 'version': VERSION}
_mocked_get_dashboard_definitions.return_value = [
dashboard # test dashboard
]
_mocked_create_dashboard = mocker.patch(
'brian_dashboard_manager.grafana.provision.create_dashboard')
# we dont care about this, just mark it created
# we dont care about this, tested separately
_mocked_create_dashboard.return_value = {'id': 666}
_mocked_delete_api_token = mocker.patch(
'brian_dashboard_manager.grafana.provision.delete_api_token')
# we dont care about this, tested separately
_mocked_delete_api_token.return_value = None
provision(data_config)