import json # import logging import os import tempfile import pytest import jsonschema import inventory_provider # logging.basicConfig(level=logging.DEBUG) DEFAULT_REQUEST_HEADERS = { "Content-type": "application/json", "Accept": ["application/json"] } MODULE_DIR = os.path.realpath(os.path.join( os.path.dirname(__file__), "..", "inventory_provider")) OID_LIST_CONF = """ # # This file is located in dbupdates/conf and is used by scripts under dbupdates/scripts. # It holds OID values for retrieving details of a router. # ## IPv4 v4Address=.1.3.6.1.2.1.4.20.1.1 v4InterfaceOID=.1.3.6.1.2.1.4.20.1.2 v4InterfaceName=.1.3.6.1.2.1.31.1.1.1.1 v4Mask=.1.3.6.1.2.1.4.20.1.3 ## IPv6 v6AddressAndMask=.1.3.6.1.2.1.55.1.8.1.2 v6InterfaceName=.1.3.6.1.2.1.55.1.5.1.2 """ # noqa E501 ROUTERS_COMMUNITY_CONF = """ ###################################################################################################################################### ## ## ## This is a configuration file that stores router names and the SNMP community name in <router>=<community>,<IP address> format. ## ## ## ###################################################################################################################################### mx2.ath.gr.geant.net=0pBiFbD,62.40.114.59 mx1.tal.ee.geant.net=0pBiFbD,62.40.96.1 mx2.tal.ee.geant.net=0pBiFbD,62.40.96.2 mx2.rig.lv.geant.net=0pBiFbD,62.40.96.4 mx1.kau.lt.geant.net=0pBiFbD,62.40.96.6 mx2.kau.lt.geant.net=0pBiFbD,62.40.96.5 mx2.zag.hr.geant.net=0pBiFbD,62.40.96.8 mx2.lju.si.geant.net=0pBiFbD,62.40.96.10 mx1.bud.hu.geant.net=0pBiFbD,62.40.97.1 mx1.pra.cz.geant.net=0pBiFbD,62.40.97.2 mx2.bra.sk.geant.net=0pBiFbD,62.40.97.4 mx1.lon.uk.geant.net=0pBiFbD,62.40.97.5 mx1.vie.at.geant.net=0pBiFbD,62.40.97.7 mx2.bru.be.geant.net=0pBiFbD,62.40.96.20 mx1.poz.pl.geant.net=0pBiFbD,62.40.97.10 mx1.ams.nl.geant.net=0pBiFbD,62.40.97.11 mx1.fra.de.geant.net=0pBiFbD,62.40.97.12 mx1.par.fr.geant.net=0pBiFbD,62.40.97.13 mx1.gen.ch.geant.net=0pBiFbD,62.40.97.14 mx1.mil2.it.geant.net=0pBiFbD,62.40.97.15 mx1.lis.pt.geant.net=0pBiFbD,62.40.96.16 mx2.lis.pt.geant.net=0pBiFbD,62.40.96.17 mx1.mad.es.geant.net=0pBiFbD,62.40.97.16 mx1.sof.bg.geant.net=0pBiFbD,62.40.96.21 mx1.buc.ro.geant.net=0pBiFbD,62.40.96.19 mx1.ham.de.geant.net=0pBiFbD,62.40.96.26 mx1.dub.ie.geant.net=0pBiFbD,62.40.96.3 mx1.dub2.ie.geant.net=0pBiFbD,62.40.96.25 mx1.mar.fr.geant.net=0pBiFbD,62.40.96.12 mx1.lon2.uk.geant.net=0pBiFbD,62.40.96.15 # rt1.clpk.us.geant.net=GEANT_RO,10.200.64.128 # rt1.denv.us.geant.net=GEANT_RO,10.200.67.128 mx1.ath2.gr.geant.net=0pBiFbD,62.40.96.39 # qfx.par.fr.geant.net=0pBiFbD,62.40.117.170 # qfx.fra.de.geant.net=0pBiFbD,62.40.117.162 """ # noqa E501 def data_config_filename(tmp_dir_name): config = { "alarms-db": { "hostname": "xxxxxxx.yyyyy.zzz", "dbname": "xxxxxx", "username": "xxxxxx", "password": "xxxxxxxx" }, "oid_list.conf": os.path.join( tmp_dir_name, "oid_list.conf"), "routers_community.conf": os.path.join( tmp_dir_name, "routers_community.conf"), "ssh": { "private-key": "private-key-filename", "known-hosts": "known-hosts=filename" }, "redis": { "hostname": "xxxxxx", "port": 6379 } } with open(config["oid_list.conf"], "w") as f: f.write(OID_LIST_CONF) with open(config["routers_community.conf"], "w") as f: f.write(ROUTERS_COMMUNITY_CONF) filename = os.path.join(tmp_dir_name, "config.json") with open(filename, "w") as f: f.write(json.dumps(config)) return filename @pytest.fixture def app_config(): with tempfile.TemporaryDirectory() as tmpdir: app_config_filename = os.path.join(tmpdir, "app.config") with open(app_config_filename, "w") as f: f.write("%s = '%s'\n" % ( "INVENTORY_PROVIDER_CONFIG_FILENAME", data_config_filename(tmpdir))) yield app_config_filename @pytest.fixture def client(app_config): os.environ["SETTINGS_FILENAME"] = app_config # with release_webservice.create_app().test_client() as c: # yield c with inventory_provider.create_app().test_client() as c: yield c def test_version_request(client): version_schema = { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "api": { "type": "string", "pattern": r'\d+\.\d+' }, "module": { "type": "string", "pattern": r'\d+\.\d+' } }, "required": ["api", "module"], "additionalProperties": False } rv = client.post( "data/version", headers=DEFAULT_REQUEST_HEADERS) assert rv.status_code == 200 jsonschema.validate( json.loads(rv.data.decode("utf-8")), version_schema)