import os import json import reversion import requests import jsonschema import time import StringIO from django.db import transaction from django.test import TestCase, Client, RequestFactory from django.core.management import call_command from peeringdb_server.models import ( Organization, Network, NetworkIXLan, IXLan, IXLanPrefix, InternetExchange, IXLanIXFMemberImportAttempt, IXLanIXFMemberImportLog, IXLanIXFMemberImportLogEntry, User) from peeringdb_server.import_views import ( view_import_ixlan_ixf_preview, ) from peeringdb_server import ixf from util import ClientCase class JsonMembersListTestCase(TestCase): # test this version of the json schema; requires the file # to exist at data/json_members_list/members..json version = "0.6" schema_url = "https://raw.githubusercontent.com/euro-ix/json-schemas/v0.6/ixp-member-list.schema.json" # will be loaded with the json data to be tested against during # setUpTestData json_data = {} entities = {} @classmethod def setUpTestData(cls): # load json members list data to test against with open( os.path.join( os.path.dirname(__file__), "data", "json_members_list", "members.{}.json".format(cls.version)), "r") as fh: cls.json_data = json.load(fh) with reversion.create_revision(): # create organization(s) cls.entities["org"] = [ Organization.objects.create(name="Netflix", status="ok") ] # create exchange(s) cls.entities["ix"] = [ InternetExchange.objects.create(name="Test Exchange", org=cls.entities["org"][0], status="ok") ] # create ixlan(s) cls.entities["ixlan"] = [ IXLan.objects.create(ix=cls.entities["ix"][0], status="ok"), IXLan.objects.create(ix=cls.entities["ix"][0], status="ok"), IXLan.objects.create(ix=cls.entities["ix"][0], status="ok") ] # create ixlan prefix(s) cls.entities["ixpfx"] = [ IXLanPrefix.objects.create( ixlan=cls.entities["ixlan"][0], status="ok", prefix="195.69.144.0/22", protocol="IPv4"), IXLanPrefix.objects.create( ixlan=cls.entities["ixlan"][0], status="ok", prefix="2001:7f8:1::/64", protocol="IPv6"), IXLanPrefix.objects.create( ixlan=cls.entities["ixlan"][1], status="ok", prefix="195.66.224.0/22", protocol="IPv4"), IXLanPrefix.objects.create( ixlan=cls.entities["ixlan"][1], status="ok", prefix="2001:7f8:4::/64", protocol="IPv6") ] # create network(s) cls.entities["net"] = [ Network.objects.create( name="Netflix", org=cls.entities["org"][0], asn=2906, info_prefixes4=42, info_prefixes6=42, website="http://netflix.com/", policy_general="Open", policy_url="https://www.netflix.com/openconnect/", allow_ixp_update=True, status="ok", irr_as_set="AS-NFLX"), Network.objects.create(name="Network with deleted netixlans", org=cls.entities["org"][0], asn=1001, allow_ixp_update=True, status="ok"), Network.objects.create( name="Network with allow ixp update off", org=cls.entities["org"][0], asn=1002, status="ok") ] # create netixlans cls.entities["netixlan"] = [ NetworkIXLan.objects.create( network=cls.entities["net"][1], ixlan=cls.entities["ixlan"][1], asn=1001, speed=10000, ipaddr4="195.69.146.250", ipaddr6=None, status="deleted"), NetworkIXLan.objects.create( network=cls.entities["net"][1], ixlan=cls.entities["ixlan"][1], asn=1001, speed=10000, ipaddr4=None, ipaddr6="2001:7f8:1::a500:2906:1", status="deleted"), NetworkIXLan.objects.create( network=cls.entities["net"][0], ixlan=cls.entities["ixlan"][0], asn=2906, speed=10000, ipaddr4="195.69.146.249", ipaddr6=None, status="ok"), NetworkIXLan.objects.create( network=cls.entities["net"][0], ixlan=cls.entities["ixlan"][0], asn=2906, speed=10000, ipaddr4="195.69.146.251", ipaddr6=None, status="ok"), NetworkIXLan.objects.create( network=cls.entities["net"][0], ixlan=cls.entities["ixlan"][0], asn=2906, speed=20000, is_rs_peer=False, ipaddr4="195.69.147.251", ipaddr6=None, status="ok"), NetworkIXLan.objects.create( network=cls.entities["net"][0], ixlan=cls.entities["ixlan"][0], asn=1002, speed=10000, ipaddr4="195.69.147.252", ipaddr6=None, status="ok"), ] def setUp(self): self.ixf_importer = ixf.Importer() def assertLog(self, log, expected): path = os.path.join(os.path.dirname(__file__), "data", "ixf", "logs", "{}.json".format(expected)) with open(path, "r") as fh: self.assertEqual(log, json.load(fh)) def test_update_from_ixf_ixp_member_list(self): ixlan = self.entities["ixlan"][0] n_deleted = self.entities["netixlan"][0] n_deleted2 = self.entities["netixlan"][1] self.assertEqual(unicode(n_deleted.ipaddr4), u'195.69.146.250') self.assertEqual( unicode(n_deleted2.ipaddr6), u'2001:7f8:1::a500:2906:1') self.assertEqual(ixlan.netixlan_set_active.count(), 4) r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) self.assertLog(log, "update_01") self.assertEqual(len(netixlans), 3) self.assertEqual(len(netixlans_deleted), 2) n = netixlans[0] self.assertEqual(unicode(n.ipaddr4), u"195.69.146.250") self.assertEqual(unicode(n.ipaddr6), u"2001:7f8:1::a500:2906:2") self.assertEqual(n.speed, 10000) self.assertEqual(n.status, "ok") self.assertEqual(n.ixlan, ixlan) self.assertEqual(n.asn, 2906) n2 = netixlans[1] self.assertEqual(unicode(n2.ipaddr4), u"195.69.147.250") self.assertEqual(unicode(n2.ipaddr6), u"2001:7f8:1::a500:2906:1") self.assertEqual(n2.speed, 10000) self.assertEqual(n2.status, "ok") self.assertEqual(n2.ixlan, ixlan) self.assertEqual(n.asn, 2906) # test that inactive connections had no effect self.assertEqual(NetworkIXLan.objects.filter(ipaddr4="195.69.146.251", speed=10000, status="ok").count(), 1) self.assertEqual(NetworkIXLan.objects.filter(ipaddr4="195.69.146.252").count(), 0) #self.assertEqual(IXLan.objects.get(id=ixlan.id).netixlan_set_active.count(), 2) #FIXME: this is not practical until #https://github.com/peeringdb/peeringdb/issues/90 is resolved #so skipping those tests right now #n_deleted.refresh_from_db() #n_deleted2.refresh_from_db() #self.assertEqual(n_deleted.ipaddr4, None) #self.assertEqual(n_deleted2.ipaddr6, None) def test_update_from_ixf_ixp_member_list_skip_prefix_mismatch(self): """ Here we test that entries with ipaddresses that cannot be validated against any of the prefixes that exist on the ixlan get skipped """ ixlan = self.entities["ixlan"][1] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) self.assertLog(log, "skip_prefix_mismatch") self.assertEqual(len(netixlans), 0) def test_update_from_ixf_ixp_member_list_skip_missing_prefixes(self): """ Here we test that nothing is done at all if the importer is run on an ixlan that does not have any prefixes """ ixlan = self.entities["ixlan"][2] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) self.assertEqual(len(netixlans), 0) self.assertEqual(len(netixlans_deleted), 0) self.assertEqual(log["errors"], [u'No prefixes defined on ixlan']) def test_update_from_ixf_ixp_member_list_skip_disabled_networks(self): """ Here we test that networks with allow_ixp_update set to False will not be processed """ ixlan = self.entities["ixlan"][0] network = self.entities["net"][0] network.allow_ixp_update = False network.save() r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) self.assertLog(log, "skip_disabled_networks") self.assertEqual(len(netixlans), 0) for netixlan in network.netixlan_set_active.all(): netixlan.refresh_from_db() self.assertEqual(netixlan.status, "ok") def test_update_from_ixf_ixp_member_list_logs(self): ixlan = self.entities["ixlan"][0] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) attempt_dt_1 = ixlan.ixf_import_attempt.updated for netixlan in netixlans: log_entry = ixlan.ixf_import_log_set.last().entries.get( netixlan=netixlan) if netixlan.id == self.entities["netixlan"][4].id: # netixlan was modified self.assertEqual( log_entry.version_before, reversion.models.Version.objects.get_for_object(netixlan)[1]) else: # netixlan was added self.assertEqual( log_entry.version_before, None) self.assertEqual( log_entry.version_after, reversion.models.Version.objects.get_for_object(netixlan)[0]) for netixlan in netixlans_deleted: log_entry = ixlan.ixf_import_log_set.last().entries.get( netixlan=netixlan) self.assertEqual( log_entry.version_before, reversion.models.Version.objects.get_for_object(netixlan)[1]) self.assertEqual( log_entry.version_after, reversion.models.Version.objects.get_for_object(netixlan)[0]) with reversion.create_revision(): netixlans[0].speed = 10 netixlans[0].save() time.sleep(0.1) r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) ixlan.ixf_import_attempt.refresh_from_db() attempt_dt_2 = ixlan.ixf_import_attempt.updated self.assertNotEqual(attempt_dt_1, attempt_dt_2) self.assertEqual(ixlan.ixf_import_log_set.count(), 2) self.assertEqual(len(netixlans), 1) for netixlan in netixlans: log_entry = ixlan.ixf_import_log_set.last().entries.get( netixlan=netixlan) self.assertEqual( log_entry.version_before, reversion.models.Version.objects.get_for_object(netixlan)[1]) self.assertEqual( log_entry.version_after, reversion.models.Version.objects.get_for_object(netixlan)[0]) def test_rollback(self): ixlan = self.entities["ixlan"][0] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) for entry in ixlan.ixf_import_log_set.last().entries.all(): self.assertEqual(entry.rollback_status(), 0) ixlan.ixf_import_log_set.last().rollback() netixlans[0].refresh_from_db() netixlans[1].refresh_from_db() self.assertEqual(netixlans[0].status, "deleted") self.assertEqual(netixlans[1].status, "deleted") ixlan.ixf_import_log_set.last().refresh_from_db() for entry in ixlan.ixf_import_log_set.last().entries.all(): self.assertEqual(entry.rollback_status(), 1) def test_rollback_avoid_ipaddress_conflict(self): ixlan = self.entities["ixlan"][0] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) self.assertEqual(len(netixlans_deleted), 2) netixlan = netixlans_deleted[0] other = NetworkIXLan.objects.create( network=netixlan.network, ixlan=netixlan.ixlan, speed=1000, status="ok", asn=netixlan.asn + 1, ipaddr4=netixlan.ipaddr4) for entry in ixlan.ixf_import_log_set.last().entries.all(): if entry.netixlan == netixlan: self.assertEqual(entry.rollback_status(), 2) ixlan.ixf_import_log_set.last().rollback() netixlan.refresh_from_db() self.assertEqual(netixlan.status, "deleted") other.delete(hard=True) def test_export_view_ixlan(self): """ Test that the /export/ixlan//ixp-member-list endpoint generates the expected result after importing a test data set """ # we only export 0.6 version of the schema, so can skip this test # for the other versions if self.version != "0.6": return # import the data ixlan = self.entities["ixlan"][0] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) # request the view and compare it agaisnt expected data c = Client() resp = c.get("/export/ixlan/{}/ixp-member-list".format(ixlan.id)) self.assertEqual(resp.status_code, 200) data = json.loads(resp.content) with open( os.path.join( os.path.dirname(__file__), "data", "json_members_list", "export.json"), "r") as fh: expected = json.load(fh) data["timestamp"] = expected["timestamp"] self.assertEqual(data, expected) schema = requests.get(self.schema_url).json() jsonschema.validate(data, schema) def test_export_view_ix(self): """ Test that the /export/ix//ixp-member-list endpoint generates the expected result after importing a test data set """ # we only export 0.6 version of the schema, so can skip this test # for the other versions if self.version != "0.6": return # import the data ixlan = self.entities["ixlan"][0] r, netixlans, netixlans_deleted, log = self.ixf_importer.update(ixlan, data=self.json_data) # request the view and compare it agaisnt expected data c = Client() resp = c.get("/export/ix/{}/ixp-member-list".format(ixlan.ix.id)) self.assertEqual(resp.status_code, 200) data = json.loads(resp.content) with open( os.path.join( os.path.dirname(__file__), "data", "json_members_list", "export.json"), "r") as fh: other = json.load(fh) data["timestamp"] = other["timestamp"] self.assertEqual(data, other) schema = requests.get(self.schema_url).json() jsonschema.validate(data, schema) def test_ixp_allow_update_default(self): self.assertEqual(self.entities["net"][2].allow_ixp_update, False) def test_command(self): """ Test the ixf_ixp_member_import command """ ixlan = self.entities["ixlan"][0] ixlan.ixf_ixp_member_list_url = "http://localhost:12345/ixf/member/import" ixlan.ixf_ixp_import_enabled = True ixlan.save() stdout = StringIO.StringIO() stderr = StringIO.StringIO() r = call_command("pdb_ixf_ixp_member_import", only=ixlan.id, commit=True, stdout=stdout, stderr=stderr) self.assertEqual(stdout.getvalue().find("Fetching data for -ixlan1 from"), 0) # importer should skip ixlans where ixf_ixp_import_enabled is # turned off ixlan.ixf_ixp_import_enabled = False ixlan.save() stdout = StringIO.StringIO() stderr = StringIO.StringIO() r = call_command("pdb_ixf_ixp_member_import", only=ixlan.id, commit=True, stdout=stdout, stderr=stderr) self.assertEqual(stdout.getvalue().find("Fetching data for -ixlan1 from"), -1) class JsonMembersListTestCase_V05(JsonMembersListTestCase): version = "0.5" class JsonMembersListTestCase_V04(JsonMembersListTestCase): version = "0.4" class TestImportPreview(ClientCase): """ Test the ixf import preview """ @classmethod def setUpTestData(cls): super(TestImportPreview, cls).setUpTestData() cls.org = Organization.objects.create(name="Test Org", status="ok") cls.ix = InternetExchange.objects.create(name="Test IX", status="ok", org=cls.org) cls.ixlan = IXLan.objects.create(status="ok", ix=cls.ix) IXLanPrefix.objects.create(ixlan=cls.ixlan, status="ok", prefix="195.69.144.0/22", protocol="IPv4") IXLanPrefix.objects.create(ixlan=cls.ixlan, status="ok", prefix="2001:7f8:1::/64", protocol="IPv6") cls.net = Network.objects.create(org=cls.org, status="ok", asn=1000, name="net01") cls.net_2 = Network.objects.create(org=cls.org, status="ok", asn=1001, name="net02") cls.admin_user = User.objects.create_user("admin","admin@localhost","admin") cls.org.admin_usergroup.user_set.add(cls.admin_user) def test_import_preview(self): request = RequestFactory().get("/import/ixlan/{}/ixf/preview/".format(self.ixlan.id)) request.user = self.admin_user response = view_import_ixlan_ixf_preview(request, self.ixlan.id) assert response.status_code == 200 assert json.loads(response.content)["errors"] == ["IXF import url not specified"] def test_import_preview_fail_ratelimit(self): request = RequestFactory().get("/import/ixlan/{}/ixf/preview/".format(self.ixlan.id)) request.user = self.admin_user response = view_import_ixlan_ixf_preview(request, self.ixlan.id) assert response.status_code == 200 response = view_import_ixlan_ixf_preview(request, self.ixlan.id) assert response.status_code == 400 def test_import_preview_fail_permission(self): request = RequestFactory().get("/import/ixlan/{}/ixf/preview/".format(self.ixlan.id)) request.user = self.guest_user response = view_import_ixlan_ixf_preview(request, self.ixlan.id) assert response.status_code == 403 def test_netixlan_diff(self): netix1 = NetworkIXLan.objects.create( network=self.net, ixlan=self.ixlan, status="ok", ipaddr4="195.69.146.250", ipaddr6="2001:7f8:1::a500:2906:1", asn=self.net.asn, speed=1000, is_rs_peer=True) netix2 = NetworkIXLan( network=self.net_2, status="ok", ipaddr4="195.69.146.250", ipaddr6="2001:7f8:1::a500:2906:2", asn=self.net_2.asn, speed=10000, is_rs_peer=False) result = self.ixlan.add_netixlan(netix2, save=False, save_others=False) self.assertEqual(sorted(result["changed"]), ['asn', 'ipaddr6', 'is_rs_peer', 'network_id', 'speed']) netix2.ipaddr4 = "195.69.146.251" netix2.ipaddr6 = netix1.ipaddr6 result = self.ixlan.add_netixlan(netix2, save=False, save_others=False) self.assertEqual(sorted(result["changed"]), ['asn', 'ipaddr4', 'is_rs_peer', 'network_id', 'speed'])