mirror/peeringdb-peeringdb
1
0
Fork 0
mirror of https://github.com/peeringdb/peeringdb.git synced 2024-05-11 05:55:09 +00:00
Code Releases Activity

Support 202103 (#966)

Browse Source 
* Add check for existing pending affiliations to org

* Fix message

* 883 consolidation for deskpro tickets

* add views tests

* ipaddress normalization during search (#913)

* remove unused imports

* Update lat and long admin fields to not required

* black formatting

* relock pipfile

* fix conditions for failing asn affil test

* Update affiliation logic and add tests

* make name search case insensitive

* add asn to org_id lookup

* black format

* skip sync test while test.peeringdb.com is down

Co-authored-by: Elliot Frank <elliot@20c.com>
Co-authored-by: Stefan Pratter <stefan@20c.com>
This commit is contained in:
Matt Griswold
2021-04-13 08:59:23 -05:00
committed by GitHub
parent 744450e310
commit 2162329ea1
13 changed files with 755 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
tests/data/json_members_list/ixf.member.6.json Normal file
View File

@@ -0,0 +1,180 @@
{
"timestamp": "2020-07-13T09:23:47Z",
"version": "1.0",
"ixp_list": [
{
"shortname": "Test Exchange",
"ixp_id": 1,
"ixf_id": 1
}
],
"member_list": [
{
"asnum": 1001,
"member_type": "peering",
"name": "Netflix",
"url": "http://netflix.com/",
"contact_email": [
"peering@netflix.com",
"mrpeering@netflix.com"
],
"contact_phone": [
"+1 1234 5678"
],
"contact_hours": "8/5",
"peering_policy": "open",
"peering_policy_url": "https://www.netflix.com/openconnect/",
"member_since": "2009-02-04T00:00:00Z",
"connection_list": [
{
"ixp_id": 42,
"connected_since": "2009-02-04T00:00:00Z",
"state": "connected",
"if_list": [
{
"switch_id": 1,
"if_speed": 20000,
"if_type": "LR4"
}
],
"vlan_list": [
{
"vlan_id": 0,
"ipv4": {
"address": "195.69.147.250",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V4",
"mac_address" : [
"00:0a:95:9d:68:16"
]
},
"ipv6": {
"address": "2001:7f8:1::a500:2906:1",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V6",
"mac_address" : [
"00:0a:95:9d:68:16"
]
}
}
]
},
{
"ixp_id": 43,
"connected_since": "2009-02-04T00:00:00Z",
"state": "connected",
"if_list": [
{
"switch_id": 1,
"if_speed": 10000,
"if_type": "LR4"
}
],
"vlan_list": [
{
"vlan_id": 0,
"ipv4": {
"address": "195.69.147.251",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V4",
"mac_address" : [
"00:0a:95:9d:68:16"
]
},
"ipv6": {
"address": "2001:7f8:1::a500:2906:4",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V6",
"mac_address" : [
"00:0a:95:9d:68:16"
]
}
}
]
},
{
"ixp_id": 44,
"connected_since": "2009-02-04T00:00:00Z",
"state": "connected",
"if_list": [
{
"switch_id": 1,
"if_speed": 15000,
"if_type": "LR4"
}
],
"vlan_list": [
{
"vlan_id": 0,
"ipv4": {
"address": "195.69.147.249",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V4",
"mac_address" : [
"00:0a:95:9d:68:16"
]
},
"ipv6": {
"address": "2001:7f8:1::a500:2907:1",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V6",
"mac_address" : [
"00:0a:95:9d:68:16"
]
}
}
]
}
]
},
{
"asnum": 1101,
"member_type": "peering",
"name": "Other Netflix",
"url": "http://netflix2.com/",
"connection_list": [
{
"ixp_id": 42,
"connected_since": "2009-02-04T00:00:00Z",
"state": "connected",
"if_list": [
{
"switch_id": 1,
"if_speed": 30000,
"if_type": "LR4"
}
],
"vlan_list": [
{
"vlan_id": 0,
"ipv4": {
"address": "195.69.147.123",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V4",
"mac_address" : [
"00:0a:95:9d:68:16"
]
},
"ipv6": {
"address": "2001:7f8:1::a500:2212:1",
"routeserver": true,
"max_prefix": 42,
"as_macro": "AS-NFLX-V6",
"mac_address" : [
"00:0a:95:9d:68:16"
]
}
}
]
}
]
}
]
}

6
tests/test_asn_automation.py
View File

@@ -235,13 +235,19 @@ class AsnAutomationTestCase(TestCase):
for i in range(0, settings.MAX_USER_AFFILIATION_REQUESTS + 1):
# For this test we need the orgs to actually exist
models.Organization.objects.create(name=f"AFFILORG{i}", status="ok")
request = self.factory.post(
"/affiliate-to-org", data={"org": f"AFFILORG{i}"}
)
request.user = self.user_b
request._dont_enforce_csrf_checks = True
print("\n")
print(i)
response = pdbviews.view_affiliate_to_org(request)
print(response.content)
if i < settings.MAX_USER_AFFILIATION_REQUESTS:
assert response.status_code == 200
else:

41
tests/test_inet.py
View File

@@ -1,8 +1,9 @@
import pytest
import pytest_filedata
import ipaddress
from peeringdb_server.inet import RdapLookup, RdapNotFoundError, renumber_ipaddress
import pytest
import pytest_filedata
from peeringdb_server.inet import RdapNotFoundError, renumber_ipaddress
def test_rdap_asn_lookup(rdap):
@@ -79,3 +80,37 @@ def test_renumber_ipaddress():
ipaddress.ip_network("206.41.0.0/21"),
ipaddress.ip_network("206.41.111.0/24"),
)
@pytest.mark.parametrize(
"input_str,compressed",
[
("2001:0db8::0001", "2001:db8::1"),
("2001:db8:0:0:0:0:2:1", "2001:db8::2:1"),
("2001:db8::0:1", "2001:db8::1"),
("2001:7f8:f2:e1::af21:3376:1", "2001:7f8:f2:e1:0:af21:3376:1"),
("2001:db8::1:1:1:1:1", "2001:db8:0:1:1:1:1:1"),
("2001:db8:0:0:1:0:0:1", "2001:db8::1:0:0:1"),
("2001:7F8:F2:E1:0:AF21:3376:1", "2001:7f8:f2:e1:0:af21:3376:1"),
],
ids=[
"4.1 handling leading zeros",
"4.2.1 shorten as much as possible (1)",
"4.2.1 shorten as much as possible (2)",
"4.2.2 handling one 16-bit 0 field (1)",
"4.2.2 handling one 16-bit 0 field (2)",
"4.2.3 choice in placement of ::",
"4.3 lowercase",
],
)
def test_ipaddress6_compression(input_str, compressed):
"""
Testing if the ipaddress string formatting
is compliant with RFC 5952
https://tools.ietf.org/html/rfc5952#section-4
Ids of parameters denote which rule in the spec
the test is demonstrating
"""
ipv6 = ipaddress.ip_address(input_str)
assert str(ipv6) == compressed

63
tests/test_ixf_member_import_protocol.py
View File

@@ -137,6 +137,8 @@ def test_resolve_local_ixf(entities, use_ip, save):
# We do not email upon resolve
assert_no_emails(network, ixlan.ix)
importer.update(ixlan, data=data)
importer.notify_proposals()
# Test idempotent
assert_idempotent(importer, ixlan, data)
@@ -2365,18 +2367,19 @@ def test_create_deskpro_tickets_after_x_days(entities):
@pytest.mark.django_db
def test_create_deskpro_tickets_no_contacts(entities):
data = setup_test_data("ixf.member.2")
"""
For issue 883, we want to test that two consolidated emails
are sent if we have two networks missing contacts.
"""
data = setup_test_data("ixf.member.6")
network = entities["net"]["UPDATE_DISABLED"]
network2 = entities["net"]["UPDATE_DISABLED_2"]
ixlan = entities["ixlan"][0]
ix = ixlan.ix
# Delete contacts
for netcontact in entities["netcontact"]:
netcontact.delete()
ix.tech_email = ""
ix.save()
entities["netixlan"].append(
NetworkIXLan.objects.create(
network=network,
@@ -2403,15 +2406,31 @@ def test_create_deskpro_tickets_no_contacts(entities):
operational=True,
)
)
entities["netixlan"].append(
NetworkIXLan.objects.create(
network=network2,
ixlan=ixlan,
asn=network2.asn,
speed=10000,
ipaddr4="195.69.147.239",
ipaddr6="2001:7f8:1::a500:2100:1",
status="ok",
is_rs_peer=True,
operational=True,
)
)
importer = ixf.Importer()
importer.update(ixlan, data=data)
importer.notify_proposals()
# Assert Tickets are created immediately
if network.ipv6_support:
assert DeskProTicket.objects.count() == 4
else:
assert DeskProTicket.objects.count() == 3
# Issue 883: Assert a single consolidated ticket is created
assert DeskProTicket.objects.count() == 2
for ticket in DeskProTicket.objects.all():
assert ticket.cc_set.count() == 0
assert DeskProTicket.objects.filter(subject__contains=str(network.asn)).exists()
assert DeskProTicket.objects.filter(subject__contains=str(network2.asn)).exists()
@pytest.mark.django_db
@@ -2419,7 +2438,6 @@ def test_email_with_partial_contacts(entities):
data = setup_test_data("ixf.member.2")
network = entities["net"]["UPDATE_DISABLED"]
ixlan = entities["ixlan"][0]
ix = ixlan.ix
# Delete network contact but keep ix contact
for netcontact in entities["netcontact"]:
@@ -2455,15 +2473,10 @@ def test_email_with_partial_contacts(entities):
importer.update(ixlan, data=data)
importer.notify_proposals()
# Assert Tickets are created immediately
if network.ipv6_support:
assert DeskProTicket.objects.count() == 4
for ticket in DeskProTicket.objects.all():
assert ticket.cc_set.count() == 1
else:
assert DeskProTicket.objects.count() == 3
for ticket in DeskProTicket.objects.all():
assert ticket.cc_set.count() == 1
# Issue 883: Assert a single consolidated ticket is created
assert DeskProTicket.objects.count() == 1
for ticket in DeskProTicket.objects.all():
assert ticket.cc_set.count() == 0
@pytest.mark.django_db
@@ -2478,7 +2491,6 @@ def test_no_email_if_deskpro_fails(entities, use_ip, save):
data = setup_test_data("ixf.member.2")
network = entities["net"]["UPDATE_DISABLED"]
ixlan = entities["ixlan"][0]
ix = ixlan.ix
# Delete network contacts
for netcontact in entities["netcontact"]:
@@ -2520,11 +2532,10 @@ def test_no_email_if_deskpro_fails(entities, use_ip, save):
importer.update(ixlan, data=data)
importer.notify_proposals()
# Assert Tickets are created immediately
if network.ipv6_support:
assert DeskProTicket.objects.count() == 4
else:
assert DeskProTicket.objects.count() == 3
# Issue 883: Assert a single consolidated ticket is created
assert DeskProTicket.objects.count() == 1
for ticket in DeskProTicket.objects.all():
assert ticket.cc_set.count() == 0
# This is the single consolidated email
assert IXFImportEmail.objects.count() == 1

115
tests/test_views.py Normal file
View File

@@ -0,0 +1,115 @@
import pytest
from rest_framework.test import APIClient
from peeringdb_server.models import (
Network,
Organization,
User,
UserOrgAffiliationRequest,
)
URL = "/affiliate-to-org"
@pytest.fixture
def client():
user = User.objects.create(
username="test",
email="test@localhost",
)
user.set_password("test1234")
user.save()
client = APIClient()
client.login(username="test", password="test1234")
return client
@pytest.fixture
def org():
org = Organization.objects.create(
name="Test Org",
)
return org
@pytest.fixture
def network(org):
net = Network.objects.create(name="test network", org=org, asn=123, status="ok")
return net
def assert_passing_affiliation_request(data, client):
response = client.post(URL, data)
assert response.status_code == 200
assert UserOrgAffiliationRequest.objects.count() == 1
def assert_failing_affiliation_request(data, client):
response = client.post(URL, data)
assert response.status_code == 400
assert "You already requested affiliation to this ASN/org" in str(response.content)
assert UserOrgAffiliationRequest.objects.count() == 1
"""
The following tests are for issue 931:
Limit the number of requests
for affiliation to an ASN/org to 1
"""
@pytest.mark.django_db
def test_affiliate_to_org_multiple(client, org):
assert_passing_affiliation_request({"org": org.id}, client)
assert_failing_affiliation_request({"org": org.id}, client)
@pytest.mark.django_db
def test_affiliate_to_asn_multiple(client, network):
assert_passing_affiliation_request({"asn": 123}, client)
assert_failing_affiliation_request({"asn": 123}, client)
@pytest.mark.django_db
def test_affiliate_to_org_then_asn(client, network, org):
assert_passing_affiliation_request({"org": org.id}, client)
assert_failing_affiliation_request({"asn": 123}, client)
@pytest.mark.django_db
def test_affiliate_to_asn_then_org(client, network, org):
assert_passing_affiliation_request({"org": org.id}, client)
assert_failing_affiliation_request({"asn": 123}, client)
@pytest.mark.django_db
def test_affiliate_to_org_id_takes_precedence_over_asn(client, org):
assert_passing_affiliation_request({"org": org.id, "asn": 2020}, client)
assert_failing_affiliation_request({"org": org.id, "asn": 2111}, client)
@pytest.mark.django_db
def test_affiliate_to_asn_takes_precendence_over_org_name(client, network, org):
assert_passing_affiliation_request({"org": "test name", "asn": 123}, client)
assert_failing_affiliation_request({"org": "different", "asn": 123}, client)
@pytest.mark.django_db
def test_affiliate_to_nonexisting_org_multiple(client):
"""
Multiple affiliations to nonexisting orgs should still get
caught if the provided org name is repetitive
"""
data = {
"org": "Nonexistent org",
}
assert_passing_affiliation_request(data, client)
assert_failing_affiliation_request(data, client)
# If we change the org name we can affiliate to that one as well
other_data = {
"org": "Second nonexistent org",
}
response = client.post(URL, other_data)
assert response.status_code == 200
assert UserOrgAffiliationRequest.objects.count() == 2

4
tests/test_wipe.py
View File

@@ -1,3 +1,5 @@
import pytest
from django.core.management import call_command
from django.contrib.auth import get_user_model
from django.conf import settings
@@ -43,6 +45,8 @@ class TestWipe(ClientCase):
assert get_user_model().objects.all().count() == 1
assert get_user_model().objects.first().is_superuser == True
# FIXME: uncomment when test.peeringdb.com is back up
@pytest.mark.skip(reason="test.peeringdb.com is currently unavailable")
def test_run_with_sync(self):
"""
Test running `pdb_wipe` and sync data from