peeringdb-peeringdb/tests/test_ixf_member_views.py

import json
import os
import re
from pprint import pprint
import reversion
import requests
import jsonschema
import time
import io
import datetime

from django.db import transaction
from django.core.cache import cache
from django.test import Client, TestCase, RequestFactory
from django.urls import reverse

from peeringdb_server.models import (
    Organization,
    Network,
    NetworkIXLan,
    IXLan,
    IXLanPrefix,
    InternetExchange,
    IXFMemberData,
    IXLanIXFMemberImportAttempt,
    IXLanIXFMemberImportLog,
    IXLanIXFMemberImportLogEntry,
    User,
    DeskProTicket,
    Group,
)
from peeringdb_server import ixf

import pytest
from .util import override_group_id


@pytest.mark.django_db
def test_reset_ixf_proposals(admin_user, entities, ip_addresses):
    network = entities["network"]
    ixlan = entities["ixlan"][0]

    client = setup_client(admin_user)
    url = reverse("net-reset-ixf-proposals", args=(network.id,))

    create_IXFMemberData(network, ixlan, ip_addresses, True)

    response = client.post(url)
    content = response.content.decode("utf-8")

    assert response.status_code == 200
    assert IXFMemberData.objects.filter(dismissed=True).count() == 0


@pytest.mark.django_db
def test_dismiss_ixf_proposals(admin_user, entities, ip_addresses):

    network = entities["network"]
    ixlan = entities["ixlan"][0]

    ids = create_IXFMemberData(network, ixlan, ip_addresses, False)

    client = setup_client(admin_user)
    url = reverse("net-dismiss-ixf-proposal", args=(network.id, ids[-1]))

    response = client.post(url)
    content = response.content.decode("utf-8")

    assert response.status_code == 200
    assert IXFMemberData.objects.filter(pk=ids[-1]).first().dismissed == True


@pytest.mark.django_db
def test_reset_ixf_proposals_no_perm(regular_user, entities, ip_addresses):
    network = entities["network"]
    ixlan = entities["ixlan"][0]

    client = setup_client(regular_user)
    url = reverse("net-reset-ixf-proposals", args=(network.id,))

    create_IXFMemberData(network, ixlan, ip_addresses, True)

    response = client.post(url)
    content = response.content.decode("utf-8")

    assert response.status_code == 401
    assert "Permission denied" in content


@pytest.mark.django_db
def test_dismiss_ixf_proposals_no_perm(regular_user, entities, ip_addresses):

    network = entities["network"]
    ixlan = entities["ixlan"][0]

    ids = create_IXFMemberData(network, ixlan, ip_addresses, False)

    client = setup_client(regular_user)
    url = reverse("net-dismiss-ixf-proposal", args=(network.id, ids[-1]))

    response = client.post(url)
    content = response.content.decode("utf-8")

    assert response.status_code == 401
    assert "Permission denied" in content


@pytest.mark.django_db
def test_ix_order(admin_user, entities, ip_addresses, ip_addresses_other):
    """
    Test that multiple exchanges proposing changes appear
    sorted by exchange name
    """

    network = entities["network"]
    ixlan_a = entities["ixlan"][0]
    ixlan_b = entities["ixlan"][1]

    create_IXFMemberData(network, ixlan_a, ip_addresses, False)
    create_IXFMemberData(network, ixlan_b, ip_addresses_other, False)

    client = setup_client(admin_user)

    url = reverse("net-view", args=(network.id,))

    with override_group_id():
        response = client.get(url)
    content = response.content.decode("utf-8")

    assert response.status_code == 200

    matches = re.findall('<a class="ix-name">([^<]+)</a>', content)
    assert matches == ["Test Exchange One", "Test Exchange Two"]


@pytest.mark.django_db()
def test_dismissed_note(admin_user, entities, ip_addresses):
    """
    Test that dismissed hints that are no longer relevant (noop)
    don't show the "you have dimissed suggestions" notification (#809)
    """

    network = entities["network"]
    ixlan_a = entities["ixlan"][0]

    create_IXFMemberData(network, ixlan_a, [ip_addresses[0]], True)

    client = setup_client(admin_user)
    url = reverse("net-view", args=(network.id,))

    with override_group_id():
        response = client.get(url)
    content = response.content.decode("utf-8")

    # dismissed suggestion still relevant, confirm note is shown

    assert response.status_code == 200
    assert "You have dismissed some suggestions" in content

    # create netixlan, causing the suggestion to become noop

    NetworkIXLan.objects.create(
        network=network,
        asn=network.asn,
        ixlan=ixlan_a,
        status="ok",
        speed=0,
        ipaddr4=ip_addresses[0][0],
        ipaddr6=ip_addresses[0][1],
    )

    with override_group_id():
        response = client.get(url)

    content = response.content.decode("utf-8")

    # dismissed suggestion no longer relevant, confirm note is gibe

    assert response.status_code == 200
    assert "You have dismissed some suggestions" not in content


@pytest.mark.django_db
def test_check_ixf_proposals(admin_user, ixf_importer_user, entities, ip_addresses):
    network = Network.objects.create(
        name="Network w allow ixp update disabled",
        org=entities["org"][0],
        asn=1001,
        allow_ixp_update=False,
        status="ok",
        info_prefixes4=42,
        info_prefixes6=42,
        website="http://netflix.com/",
        policy_general="Open",
        policy_url="https://www.netflix.com/openconnect/",
        info_unicast=False,
        info_ipv6=False,
    )
    ixlan = entities["ixlan"][0]

    # We create one Netixlan that matches the ASN and ipaddr6 of the import.json
    # Therefore, the hint will suggest we modify this netixlan
    netixlan = NetworkIXLan.objects.create(
        network=network,
        ixlan=ixlan,
        asn=network.asn,
        speed=10000,
        ipaddr4="195.69.147.251",
        ipaddr6="2001:7f8:1::a500:2906:3",
        status="ok",
        is_rs_peer=True,
        operational=True,
    )

    with open(
        os.path.join(
            os.path.dirname(__file__),
            "data",
            "ixf",
            "views",
            "import.json",
        ),
    ) as fh:
        json_data = json.load(fh)

    importer = ixf.Importer()
    importer.update(ixlan, data=json_data)

    client = setup_client(admin_user)
    url = reverse("net-view", args=(network.id,))

    with override_group_id():
        response = client.get(url)
    assert response.status_code == 200
    content = response.content.decode("utf-8")

    # Suggest add
    assert 'data-field="ipaddr4" value="195.69.147.250"' in content
    assert 'data-field="ipaddr6" value="2001:7f8:1::a500:2906:1"' in content

    # Suggest modify
    assert 'data-field="ipaddr4" data-value=""' in content
    assert 'data-field="ipaddr6" data-value="2001:7f8:1::a500:2906:3"' in content


# Functions and fixtures


def setup_client(user):
    client = Client()
    client.force_login(user)
    return client


def create_IXFMemberData(network, ixlan, ip_addresses, dismissed):
    """
    Creates IXFMember data. Returns the ids of the created instances.
    """
    ids = []
    for ip_address in ip_addresses:
        ixfmember = IXFMemberData.instantiate(
            network.asn, ip_address[0], ip_address[1], ixlan, data={"foo": "bar"}
        )
        ixfmember.save()
        ixfmember.dismissed = dismissed
        ixfmember.save()
        ids.append(ixfmember.id)
    return ids


@pytest.fixture
def ip_addresses():
    """
    Returns a list of tuples of ipaddr4 and ipaddr6
    """
    return [
        ("195.69.144.1", "2001:7f8:1::a500:2906:1"),
        ("195.69.144.2", "2001:7f8:1::a500:2906:2"),
        ("195.69.144.3", "2001:7f8:1::a500:2906:3"),
        ("195.69.144.4", "2001:7f8:1::a500:2906:4"),
        ("195.69.144.5", "2001:7f8:1::a500:2906:5"),
    ]


@pytest.fixture
def ip_addresses_other():
    """
    Returns a list of tuples of ipaddr4 and ipaddr6
    """
    return [
        ("195.70.144.1", "2001:7f8:2::a500:2906:1"),
        ("195.70.144.2", "2001:7f8:2::a500:2906:2"),
        ("195.70.144.3", "2001:7f8:2::a500:2906:3"),
        ("195.70.144.4", "2001:7f8:2::a500:2906:4"),
        ("195.70.144.5", "2001:7f8:2::a500:2906:5"),
    ]


@pytest.fixture
def entities():
    entities = {}

    with reversion.create_revision():
        entities["org"] = [Organization.objects.create(name="Netflix", status="ok")]

        # create exchange(s)
        entities["ix"] = [
            InternetExchange.objects.create(
                name="Test Exchange One", org=entities["org"][0], status="ok"
            ),
            InternetExchange.objects.create(
                name="Test Exchange Two", org=entities["org"][0], status="ok"
            ),
        ]

        # create ixlan(s)
        entities["ixlan"] = [ix.ixlan for ix in entities["ix"]]

        # create ixlan prefix(s)
        entities["ixpfx"] = [
            IXLanPrefix.objects.create(
                ixlan=entities["ixlan"][0],
                status="ok",
                prefix="195.69.144.0/22",
                protocol="IPv4",
            ),
            IXLanPrefix.objects.create(
                ixlan=entities["ixlan"][0],
                status="ok",
                prefix="2001:7f8:1::/64",
                protocol="IPv6",
            ),
            IXLanPrefix.objects.create(
                ixlan=entities["ixlan"][1],
                status="ok",
                prefix="195.70.144.0/22",
                protocol="IPv4",
            ),
            IXLanPrefix.objects.create(
                ixlan=entities["ixlan"][1],
                status="ok",
                prefix="2001:7f8:2::/64",
                protocol="IPv6",
            ),
        ]

        # create network(s)
        entities["network"] = Network.objects.create(
            name="Network w allow ixp update enabled",
            org=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",
        )

    return entities


@pytest.fixture
def admin_user():
    from django.conf import settings

    guest_group, _ = Group.objects.get_or_create(name="guest")
    user_group, _ = Group.objects.get_or_create(name="user")

    print("Guest: {} {} ".format(guest_group, guest_group.id))
    print("User: {} {} ".format(user_group, user_group.id))

    admin_user = User.objects.create_user(
        "admin", "admin@localhost", first_name="admin", last_name="admin"
    )
    admin_user.is_superuser = True
    admin_user.is_staff = True
    admin_user.save()
    admin_user.set_password("admin")
    admin_user.save()
    return admin_user


@pytest.fixture
def regular_user():
    user = User.objects.create_user(
        "user", "user@localhost", first_name="user", last_name="user"
    )
    user.set_password("user")
    user.save()
    return user