1
0
mirror of https://github.com/peeringdb/peeringdb.git synced 2024-05-11 05:55:09 +00:00
Files
2024-04-15 10:42:10 -05:00

1076 lines
30 KiB
Python

import ipaddress
import os
from unittest.mock import patch
import pytest
import requests
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.management import call_command
from django.test import RequestFactory, override_settings
from rest_framework.exceptions import ValidationError as RestValidationError
import peeringdb_server.geo as geo
from peeringdb_server.context import current_request
from peeringdb_server.models import (
Facility,
InternetExchange,
IXLan,
IXLanPrefix,
Network,
NetworkContact,
NetworkIXLan,
Organization,
ProtectedAction,
)
from peeringdb_server.validators import (
validate_address_space,
validate_asn_prefix,
validate_distance_geocode,
validate_info_prefixes4,
validate_info_prefixes6,
validate_irr_as_set,
validate_latitude,
validate_longitude,
validate_phonenumber,
validate_prefix_overlap,
validate_social_media,
validate_website_override,
)
from tests.test_ixf_member_import_protocol import setup_test_data
pytestmark = pytest.mark.django_db
INVALID_ADDRESS_SPACES = [
"0.0.0.0/1",
"0.0.0.0/8",
"10.0.0.0/8",
"127.0.0.0/8",
"169.254.0.0/16",
# FIXME: this fails still
#'172.0.0.0/11',
"172.16.0.0/12",
"192.0.2.0/24",
"192.168.0.0/16",
"198.18.0.0/15",
"198.51.100.0/24",
"203.0.113.0/24",
# FIXME: this fails still
#'224.0.0.0/3',
"224.0.0.0/4",
"240.0.0.0/4",
"100.64.0.0/10",
"0000::/8",
"0064:ff9b::/96",
"0100::/8",
"0200::/7",
"0400::/6",
"0800::/5",
"1000::/4",
"2001::/33",
"2001:0:8000::/33",
"2001:0002::/48",
"2001:0003::/32",
"2001:10::/28",
"2001:20::/28",
"2001:db8::/32",
"2002::/16",
"3ffe::/16",
"4000::/2",
"4000::/3",
"5f00::/8",
"6000::/3",
"8000::/3",
"a000::/3",
"c000::/3",
"e000::/4",
"f000::/5",
"f800::/6",
"fc00::/7",
"fe80::/10",
"fec0::/10",
"ff00::/8",
]
@pytest.fixture(params=INVALID_ADDRESS_SPACES)
def prefix(request):
return request.param
# @pytest.mark.django_db
def test_validate_address_space(prefix):
"""
Tests peeringdb_server.validators.validate_address_space
"""
with pytest.raises(ValidationError) as exc:
validate_address_space(ipaddress.ip_network(str(prefix)))
@override_settings(DATA_QUALITY_MAX_PREFIX_V4_LIMIT=500000)
def test_validate_info_prefixes4():
"""
Tests peeringdb_server.validators.validate_info_prefixes4
"""
with pytest.raises(ValidationError):
validate_info_prefixes4(500001)
with pytest.raises(ValidationError):
validate_info_prefixes4(-1)
validate_info_prefixes4(500000)
assert validate_info_prefixes4(None) == 0
assert validate_info_prefixes4("") == 0
@override_settings(DATA_QUALITY_MAX_PREFIX_V6_LIMIT=500000)
def test_validate_info_prefixes6():
"""
Tests peeringdb_server.validators.validate_info_prefixes6
"""
with pytest.raises(ValidationError):
validate_info_prefixes6(500001)
with pytest.raises(ValidationError):
validate_info_prefixes6(-1)
validate_info_prefixes6(500000)
assert validate_info_prefixes6(None) == 0
assert validate_info_prefixes6("") == 0
@override_settings(
DATA_QUALITY_MIN_PREFIXLEN_V4=24,
DATA_QUALITY_MAX_PREFIXLEN_V4=24,
DATA_QUALITY_MIN_PREFIXLEN_V6=48,
DATA_QUALITY_MAX_PREFIXLEN_V6=48,
)
def test_validate_prefixlen():
"""
Tests prefix length limits
"""
with pytest.raises(ValidationError):
validate_address_space("37.77.32.0/20")
with pytest.raises(ValidationError):
validate_address_space("131.72.77.240/28")
with pytest.raises(ValidationError):
validate_address_space("2403:c240::/32")
with pytest.raises(ValidationError):
validate_address_space("2001:504:0:2::/64")
@pytest.mark.django_db
def test_validate_prefix_overlap():
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(name="Text exchange", status="ok", org=org)
ixlan = ix.ixlan
pfx1 = IXLanPrefix.objects.create(
ixlan=ixlan,
protocol="IPv4",
prefix=ipaddress.ip_network("198.32.125.0/24"),
status="ok",
)
with pytest.raises(ValidationError) as exc:
validate_prefix_overlap("198.32.124.0/23")
@pytest.mark.parametrize(
"value,validated",
[
# success validation
("RIPE::AS-FOO", "RIPE::AS-FOO"),
("AS-FOO@RIPE", "AS-FOO@RIPE"),
("AS-FOO-BAR@RIPE", "AS-FOO-BAR@RIPE"),
("ripe::as-foo", "RIPE::AS-FOO"),
("as-foo@ripe", "AS-FOO@RIPE"),
("as-foo@ripe as-bar@ripe", "AS-FOO@RIPE AS-BAR@RIPE"),
("as-foo@ripe,as-bar@ripe", "AS-FOO@RIPE AS-BAR@RIPE"),
("as-foo@ripe, as-bar@ripe", "AS-FOO@RIPE AS-BAR@RIPE"),
(
"RIPE::AS12345:AS-FOO RIPE::AS12345:AS-FOO:AS9876",
"RIPE::AS12345:AS-FOO RIPE::AS12345:AS-FOO:AS9876",
),
("ripe::as-foo:as123:as345", "RIPE::AS-FOO:AS123:AS345"),
("RIPE::AS12345", "RIPE::AS12345"),
("AS12345@RIPE", "AS12345@RIPE"),
("RIPE::AS123456:RS-FOO", "RIPE::AS123456:RS-FOO"),
("as-foo", "AS-FOO"),
("rs-foo", "RS-FOO"),
("as-foo as-bar", "AS-FOO AS-BAR"),
("rs-foo as-bar", "RS-FOO AS-BAR"),
("rs-foo rs-bar", "RS-FOO RS-BAR"),
("AS15562", "AS15562"),
("AS-15562", "AS-15562"),
("AS15562 AS33333", "AS15562 AS33333"),
# hyphenated source validation
# we currently do not have valid hyphentated sources in the IRR_SOURCE
# so this test is commented out
# ("AS-20C@ARIN-NONAUTH", "AS-20C@ARIN-NONAUTH"),
# ("ARIN-NONAUTH::AS-20C", "ARIN-NONAUTH::AS-20C"),
# fail validation
("UNKNOWN::AS-FOO", False),
("AS-FOO@UNKNOWN", False),
("ASFOO@UNKNOWN", False),
("UNKNOWN::ASFOO", False),
("RIPE:AS-FOO", False),
("RIPE::RS15562:RS-FOO", False),
("RIPE::AS123456:RS-FOO:AS-FOO", False),
('!"*([])?.=+/\\', False),
('RIPE::!"*([])?.=+/\\', False),
('!"*([])?.=+/\\@RIPE', False),
# > DATA_QUALITY_MAX_IRR_DEPTH
("ripe::as-foo:as123:as345:as678", False),
],
)
def test_validate_irr_as_set(value, validated):
if not validated:
with pytest.raises(ValidationError):
validate_irr_as_set(value)
else:
assert validate_irr_as_set(value) == validated
@pytest.mark.django_db
def test_validate_phonenumber():
# test standalone validator
validate_phonenumber("+1 206 555 0199")
validate_phonenumber("012065550199", "US")
with pytest.raises(ValidationError):
validate_phonenumber("invalid number")
with pytest.raises(ValidationError):
validate_phonenumber("012065550199")
# test model field validation
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(
name="Text exchange",
status="ok",
org=org,
country="US",
city="Some city",
region_continent="North America",
media="Ethernet",
)
net = Network.objects.create(name="Text network", asn=12345, status="ok", org=org)
poc = NetworkContact.objects.create(network=net, status="ok", role="Abuse")
# test poc phone validation
with pytest.raises(ValidationError):
poc.phone = "invalid"
poc.full_clean()
poc.phone = "+1 206 555 0199"
poc.full_clean()
# test ix phone validation
with pytest.raises(ValidationError):
ix.tech_phone = "invalid"
ix.full_clean()
ix.tech_phone = "+1 206 555 0199"
ix.full_clean()
with pytest.raises(ValidationError):
ix.policy_phone = "invalid"
ix.full_clean()
ix.policy_phone = "+1 206 555 0199"
ix.full_clean()
@pytest.mark.django_db
def test_validate_ixpfx_ixlan_status_match():
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(
name="Text exchange", status="pending", org=org
)
ixlan = ix.ixlan
pfx = IXLanPrefix.objects.create(
ixlan=ixlan,
protocol="IPv4",
prefix=ipaddress.ip_network("198.32.125.0/24"),
status="ok",
)
with pytest.raises(ValidationError) as exc1:
pfx.clean()
assert (
exc1.value.args[0]
== "IXLanPrefix with status 'ok' cannot be linked to a IXLan with status 'pending'."
)
ixlan.status = "deleted"
ixlan.save()
pfx.status = "pending"
pfx.save()
with pytest.raises(ValidationError) as exc2:
pfx.clean()
assert (
exc2.value.args[0]
== "IXLanPrefix with status 'pending' cannot be linked to a IXLan with status 'deleted'."
)
@pytest.mark.django_db
@override_settings(
DATA_QUALITY_MAX_PREFIX_V4_LIMIT=500000,
DATA_QUALITY_MAX_PREFIX_V6_LIMIT=500000,
DATA_QUALITY_MIN_PREFIXLEN_V4=24,
DATA_QUALITY_MAX_PREFIXLEN_V4=24,
DATA_QUALITY_MIN_PREFIXLEN_V6=48,
DATA_QUALITY_MAX_PREFIXLEN_V6=48,
DATA_QUALITY_MAX_IRR_DEPTH=3,
DATA_QUALITY_MIN_SPEED=10,
DATA_QUALITY_MAX_SPEED=100,
)
def test_bypass_validation():
User = get_user_model()
superuser = User.objects.create_user(
username="superuser",
password="superuser",
email="su@localhost",
is_superuser=True,
)
user = User.objects.create_user(
username="user", password="user", email="user@localhost"
)
factory = RequestFactory()
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(
name="Text exchange",
status="ok",
org=org,
country="US",
city="Some city",
region_continent="North America",
media="Ethernet",
)
net = Network.objects.create(name="Text network", asn=12345, status="ok", org=org)
# super user should bypass validation
request = factory.get("/")
request.user = superuser
with current_request(request):
validate_address_space("37.77.32.0/20")
validate_address_space("131.72.77.240/28")
validate_address_space("2403:c240::/32")
validate_address_space("2001:504:0:2::/64")
validate_info_prefixes4(500001)
validate_info_prefixes6(500001)
NetworkIXLan(speed=1, network=net, ixlan=ix.ixlan).clean()
NetworkIXLan(speed=1000, network=net, ixlan=ix.ixlan).clean()
validate_irr_as_set("ripe::as-foo:as123:as345:as678")
# user should NOT bypass validation
request = factory.get("/")
request.user = user
with current_request(request):
with pytest.raises(ValidationError):
validate_address_space("37.77.32.0/20")
with pytest.raises(ValidationError):
validate_address_space("131.72.77.240/28")
with pytest.raises(ValidationError):
validate_address_space("2403:c240::/32")
with pytest.raises(ValidationError):
validate_address_space("2001:504:0:2::/64")
with pytest.raises(ValidationError):
validate_info_prefixes4(500001)
with pytest.raises(ValidationError):
validate_info_prefixes6(500001)
with pytest.raises(ValidationError):
NetworkIXLan(speed=1, network=net, ixlan=ix.ixlan).clean()
with pytest.raises(ValidationError):
NetworkIXLan(speed=1000, network=net, ixlan=ix.ixlan).clean()
with pytest.raises(ValidationError):
validate_irr_as_set("ripe::as-foo:as123:as345:as678")
@pytest.mark.django_db
def test_ghost_peer_vs_real_peer_one_netixlan():
"""
Tests that a real peer can claim the ip addresses of a gohst peer. #983
In this test both ipv4 and ipv6 exist on the same netixlan.
"""
# set up entities
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
network_other = Network.objects.create(
asn=1010, name="AS1010", status="ok", org=org
)
ixlan = ix.ixlan
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
ixlan.save()
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="195.69.144.0/22",
protocol="IPv4",
)
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="2001:7f8:1::/64",
protocol="IPv6",
)
IP4 = "195.69.147.250"
IP6 = "2001:7f8:1::a500:2906:1"
ghost_peer = NetworkIXLan.objects.create(
network=network_other,
ixlan=ixlan,
asn=network_other.asn,
speed=20000,
ipaddr4=IP4,
ipaddr6=IP6,
status="ok",
is_rs_peer=False,
operational=False,
)
# setup IX-F cache
data = setup_test_data("ixf.member.1")
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", data)
ix = ixlan.ix
# real peer should exist in IX-F data
real4, real6 = ix.peer_exists_in_ixf_data(1001, IP4, IP6)
assert real4
assert real6
# ghost peer should NOT exist in IX-F data
ghost4, ghost6 = ix.peer_exists_in_ixf_data(1010, IP4, IP6)
assert not ghost4
assert not ghost6
# create and save a real peer that has the same ip addresses
# as the ghost peer
real_peer = NetworkIXLan(
network=network,
status="ok",
ipaddr4=IP4,
ipaddr6=IP6,
ixlan=ixlan,
speed=1000,
asn=network.asn,
)
# run full validation (this will run `validate_real_vs_ghost_peer`)
real_peer.full_clean()
real_peer.save()
# real peer has been saved and since it claimed both ip4 and ip6, the ghost
# peer is now deleted
ghost_peer.refresh_from_db()
assert ghost_peer.status == "deleted"
@pytest.mark.django_db
def test_ghost_peer_vs_real_peer_two_netixlan():
"""
Tests that a real peer can claim the ip addresses of a gohst peer. #983
In this test both ipv4 and ipv6 exist on separate netixlans.
In this test both conflicting netixlans will have neither ipv4 nor ipv6 set in the end
and will be deleted
"""
# set up entities
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
network_other = Network.objects.create(
asn=1010, name="AS1010", status="ok", org=org
)
ixlan = ix.ixlan
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
ixlan.save()
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="195.69.144.0/22",
protocol="IPv4",
)
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="2001:7f8:1::/64",
protocol="IPv6",
)
IP4 = "195.69.147.250"
IP6 = "2001:7f8:1::a500:2906:1"
ghost_peer_a = NetworkIXLan.objects.create(
network=network_other,
ixlan=ixlan,
asn=network_other.asn,
speed=20000,
ipaddr4=IP4,
ipaddr6=None,
status="ok",
is_rs_peer=False,
operational=False,
)
ghost_peer_b = NetworkIXLan.objects.create(
network=network_other,
ixlan=ixlan,
asn=network_other.asn,
speed=20000,
ipaddr4=None,
ipaddr6=IP6,
status="ok",
is_rs_peer=False,
operational=False,
)
# setup IX-F data
data = setup_test_data("ixf.member.1")
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", data)
ix = ixlan.ix
# real peer should exist in IX-F data
real4, real6 = ix.peer_exists_in_ixf_data(1001, IP4, IP6)
assert real4
assert real6
# ghost peer should NOT exist in IX-F data
ghost4, ghost6 = ix.peer_exists_in_ixf_data(1010, IP4, IP6)
assert not ghost4
assert not ghost6
# create and save a real peer that has the same ip addresses
# as the ghost peer
real_peer = NetworkIXLan(
network=network,
status="ok",
ipaddr4=IP4,
ipaddr6=IP6,
ixlan=ixlan,
speed=1000,
asn=network.asn,
)
# run full validation (this will run `validate_real_vs_ghost_peer`)
real_peer.full_clean()
real_peer.save()
# real peer has been saved and since it claimed both ip4 and ip6, the ghost
# peer is now deleted
ghost_peer_a.refresh_from_db()
ghost_peer_b.refresh_from_db()
assert ghost_peer_a.status == "deleted"
assert ghost_peer_b.status == "deleted"
@pytest.mark.django_db
def test_ghost_peer_vs_real_peer_two_netixlan_partial():
"""
Tests that a real peer can claim the ip addresses of a gohst peer. #983
In this test both ipv4 and ipv6 exist on separate netixlans.
In this test the conflicting netixlans will have the other ip address still set and will not be deleted.
"""
# set up entities
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
network_other = Network.objects.create(
asn=1010, name="AS1010", status="ok", org=org
)
ixlan = ix.ixlan
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
ixlan.save()
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="195.69.144.0/22",
protocol="IPv4",
)
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="2001:7f8:1::/64",
protocol="IPv6",
)
IP4 = "195.69.147.250"
IP6 = "2001:7f8:1::a500:2906:1"
ghost_peer_a = NetworkIXLan.objects.create(
network=network_other,
ixlan=ixlan,
asn=network_other.asn,
speed=20000,
ipaddr4=IP4,
ipaddr6="2001:7f8:1::a500:2906:2",
status="ok",
is_rs_peer=False,
operational=False,
)
ghost_peer_b = NetworkIXLan.objects.create(
network=network_other,
ixlan=ixlan,
asn=network_other.asn,
speed=20000,
ipaddr4="195.69.147.251",
ipaddr6=IP6,
status="ok",
is_rs_peer=False,
operational=False,
)
# setup IX-F data
data = setup_test_data("ixf.member.1")
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", data)
ix = ixlan.ix
# real peer should exist in IX-F data
real4, real6 = ix.peer_exists_in_ixf_data(1001, IP4, IP6)
assert real4
assert real6
# ghost peer should NOT exist in IX-F data
ghost4, ghost6 = ix.peer_exists_in_ixf_data(1010, IP4, IP6)
assert not ghost4
assert not ghost6
# create and save a real peer that has the same ip addresses
# as the ghost peer
real_peer = NetworkIXLan(
network=network,
status="ok",
ipaddr4=IP4,
ipaddr6=IP6,
ixlan=ixlan,
speed=1000,
asn=network.asn,
)
# run full validation (this will run `validate_real_vs_ghost_peer`)
real_peer.full_clean()
real_peer.save()
# real peer has been saved and since it only claimed one ip address
# from either ghost peer, both ghost peers remain
ghost_peer_a.refresh_from_db()
ghost_peer_b.refresh_from_db()
assert ghost_peer_a.status == "ok"
assert ghost_peer_a.ipaddr4 is None
assert ghost_peer_a.ipaddr6 is not None
assert ghost_peer_b.status == "ok"
assert ghost_peer_b.ipaddr4 is not None
assert ghost_peer_b.ipaddr6 is None
@pytest.mark.django_db
def test_ghost_peer_vs_real_peer_invalid_ixf_data():
"""
Tests that a real peer can claim the ip addresses of a gohst peer. #983
Test the handling of invalid IX-F data, in which case the ghost peer vs real peer
logic should be skipped.
"""
# set up entities
org = Organization.objects.create(name="Test org", status="ok")
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
network_other = Network.objects.create(
asn=1010, name="AS1010", status="ok", org=org
)
ixlan = ix.ixlan
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
ixlan.save()
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="195.69.144.0/22",
protocol="IPv4",
)
IXLanPrefix.objects.create(
ixlan=ixlan,
status="ok",
prefix="2001:7f8:1::/64",
protocol="IPv6",
)
IP4 = "195.69.147.250"
IP6 = "2001:7f8:1::a500:2906:1"
ghost_peer = NetworkIXLan.objects.create(
network=network_other,
ixlan=ixlan,
asn=network_other.asn,
speed=20000,
ipaddr4=IP4,
ipaddr6=IP6,
status="ok",
is_rs_peer=False,
operational=False,
)
# setup IX-F data
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", {"invalid": "data"})
ix = ixlan.ix
real_peer = NetworkIXLan(
network=network,
status="ok",
ipaddr4=IP4,
ipaddr6=IP6,
ixlan=ixlan,
speed=1000,
asn=network.asn,
)
# run full validation (this will run `validate_real_vs_ghost_peer`)
with pytest.raises(Exception) as excinfo:
real_peer.full_clean()
assert "IP already exists" in str(excinfo.value)
@pytest.mark.parametrize(
"value,validated",
[
# success validation
(
[
{"service": "website", "identifier": "https://www.example.com"},
{"service": "x", "identifier": "unknown12"},
],
[
{"service": "website", "identifier": "https://www.example.com"},
{"service": "x", "identifier": "unknown12"},
],
),
(
[
{"service": "instagram", "identifier": "john__doe_"},
{"service": "tiktok", "identifier": "unknown_12"},
],
[
{"service": "instagram", "identifier": "john__doe_"},
{"service": "tiktok", "identifier": "unknown_12"},
],
),
(
[
{"service": "instagram", "identifier": "john-doe-"},
{"service": "tiktok", "identifier": "-unknown-12"},
],
[
{"service": "instagram", "identifier": "john-doe-"},
{"service": "tiktok", "identifier": "-unknown-12"},
],
),
# fail validation
(
[
{"service": "website", "identifier": "https://www.example.com"},
{"service": "x", "identifier": "unknown11."},
],
False,
),
([{"service": "instagram", "identifier": "john__doe_*"}], False),
(
[
{"service": "website", "identifier": ""},
{"service": "x", "identifier": "unknown"},
],
False,
),
(
[
{"service": "website", "identifier": "https://www.example.com"},
{"service": "", "identifier": "unknown"},
],
False,
),
(
[
{"service": "website", "identifier": None},
{"service": "x", "identifier": "unknown"},
],
False,
),
(
[
{"service": "website", "identifier": "https://www.example.com"},
{"service": None, "identifier": "unknown"},
],
False,
),
(
[
{"service": "website", "identifier": "https://www.example.com"},
{"service": "website", "identifier": "https://www.unknown.com"},
],
False,
),
(
[
{"service": "website", "identifier": "https://www.example.com"},
{"foo": "bar"},
],
False,
),
(
{
"service": {
"service": "website",
"identifier": "https://www.example.com",
},
},
False,
),
],
)
@pytest.mark.django_db
def test_validate_social_media(value, validated):
if not validated:
with pytest.raises(ValidationError):
validate_social_media(value)
else:
assert validate_social_media(value) == validated
@pytest.mark.parametrize(
"website,org_website,validated",
[
# success validation website not null
(
"https://www.example.com",
"https://www.example1.com",
"https://www.example.com",
),
# success validation website null and overrided by organization website
(None, "https://www.example1.com", "https://www.example1.com"),
# fail validation
(
None,
None,
False,
),
],
)
@pytest.mark.django_db
def test_validate_website_override(website, org_website, validated):
if not validated:
with pytest.raises(ValidationError):
validate_website_override(website, org_website)
else:
assert validate_website_override(website, org_website) == validated
@pytest.mark.django_db
def test_org_create_with_none_social_media():
org = Organization.objects.create(name="Test org", status="ok", social_media=None)
assert org.social_media == {}
@pytest.mark.parametrize(
"value,is_valid,validated",
[
# success validation
("63311", True, "63311"),
("as63311", True, "63311"),
("asn63311", True, "63311"),
("AS63311", True, "63311"),
("ASN63311", True, "63311"),
# fail validation
("AN63311", False, None),
("6as3311", False, None),
("63311asn", False, None),
],
)
@pytest.mark.django_db
def test_validate_asn_prefix(value, is_valid, validated):
print(is_valid)
if not is_valid:
with pytest.raises(RestValidationError):
validate_asn_prefix(value)
else:
assert validate_asn_prefix(value) == validated
@pytest.mark.parametrize(
"value,is_valid,validated",
[
# success validation
(37.7749, True, 37.7749),
(-23.5505, True, -23.5505),
(51.5074, True, 51.5074),
(40.7128, True, 40.7128),
("-33.8688", True, -33.8688),
# fail validation
(95.1234, False, None),
(-120.5678, False, None),
("-122.5678", False, None),
("abcdef", False, None),
],
)
@pytest.mark.django_db
def test_validate_latitude(value, is_valid, validated):
if not is_valid:
with pytest.raises(ValidationError):
validate_latitude(value)
else:
assert validate_latitude(value) == validated
@pytest.mark.parametrize(
"value,is_valid,validated",
[
# success validation
(-122.4194, True, -122.4194),
(-46.6333, True, -46.6333),
(-0.1270, True, -0.1270),
(-74.0060, True, -74.0060),
("151.2093", True, 151.2093),
# fail validation
(190.1234, False, None),
(-250.5678, False, None),
("360.9876", False, None),
("abcdef", False, None),
],
)
@pytest.mark.django_db
def test_validate_longitude(value, is_valid, validated):
if not is_valid:
with pytest.raises(ValidationError):
validate_longitude(value)
else:
assert validate_longitude(value) == validated
def geo_mock_init(self, key, timeout):
pass
def geo_gmaps_mock_geocode_freeform(location):
return {"lat": 40.712776, "lng": -74.005974}
@pytest.mark.parametrize(
"current_geocode,new_geocode,city,is_valid,validated",
[
# test success with exists geocode (max 1km from previous geocode)
(
(50.951533, 1.852570),
(50.951533, 1.851440),
"london",
True,
(50.951533, 1.851440),
),
(
(40.712776, -74.005974),
(40.712790, -74.003974),
"new york",
True,
(40.712790, -74.003974),
),
# # test fail with exists geocode (max 1km from previous geocode)
((40.712776, -74.005974), (40.712790, -73.003974), "new york", False, None),
((50.951533, 1.852570), (51.951533, 0.851440), "london", False, None),
# test success with not exists geocode (max 50km from city)
(
(None, None),
(40.712771, -74.005970),
"new york",
True,
(40.712771, -74.005970),
),
(
(None, None),
(40.716822, -73.991032),
"new york",
True,
(40.716822, -73.991032),
),
# test fail with not exists geocode (max 50km from city)
((None, None), (36.169941, -115.139832), "new york", False, None),
((None, None), (36.201902, -115.328808), "new york", False, None),
],
)
@patch.object(geo.GoogleMaps, "__init__", geo_mock_init)
@patch.object(geo.Melissa, "__init__", geo_mock_init)
@pytest.mark.django_db
def test_validate_distance_geocode(
current_geocode, new_geocode, city, is_valid, validated, settings
):
settings.MELISSA_KEY = ""
settings.GOOGLE_GEOLOC_API_KEY = ""
with patch.object(
geo.GoogleMaps, "geocode_freeform", side_effect=geo_gmaps_mock_geocode_freeform
):
if not is_valid:
with pytest.raises(ValidationError):
validate_distance_geocode(current_geocode, new_geocode, city)
else:
assert (
validate_distance_geocode(current_geocode, new_geocode, city)
== validated
)