How to use peeringdb - 10 common examples

To help you get started, we’ve selected a few peeringdb examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github peeringdb / peeringdb-py / tests / test_sync.py View on Github external
client = client_dup
    # sanity check - do we actually have a duplicate
    swapdup = client.get(Network, 9)
    d = client._fetcher.fetch_latest(Network, FIRST_NET, 0, since=0)
    assert d[0][0]['name'] == swapdup.name

    # obj that doesn't exist remotely
    assert client.get(Network, 12)

    rs = all_resources()
    client.update_all(rs, since=0)

    assert client.get(Network, FIRST_NET)

    # remotely deleted dup should be gone
    B = peeringdb.get_backend()
    with pytest.raises(B.object_missing_error()):
        client.get(Network, 12)
github respawner / peering-manager / peeringdb / tests.py View on Github external
asn = 15169

        # Must not exist
        self.assertIsNone(api.get_autonomous_system(64500))

        # Using an API call (no cached data)
        autonomous_system = api.get_autonomous_system(asn)
        self.assertEqual(autonomous_system.asn, asn)

        # Save the data inside the cache
        details = {
            "id": autonomous_system.id,
            "asn": autonomous_system.asn,
            "name": autonomous_system.name,
        }
        network = Network(**details)
        network.save()

        # Using no API calls (cached data)
        autonomous_system = api.get_autonomous_system(asn)
        self.assertEqual(autonomous_system.asn, asn)
github respawner / peering-manager / peeringdb / tests.py View on Github external
# Must not exist
        self.assertIsNone(api.get_ix_network(0))

        # Using an API call (no cached data)
        ix_network = api.get_ix_network(ix_network_id)
        self.assertEqual(ix_network.id, ix_network_id)

        # Save the data inside the cache
        details = {
            "id": ix_network.id,
            "asn": ix_network.asn,
            "name": ix_network.name,
            "ix_id": ix_network.ix_id,
            "ixlan_id": ix_network.ixlan_id,
        }
        network_ixlan = NetworkIXLAN(**details)
        network_ixlan.save()

        # Using no API calls (cached data)
        ix_network = api.get_ix_network(ix_network_id)
        self.assertEqual(ix_network.id, ix_network_id)
github respawner / peering-manager / peeringdb / tests.py View on Github external
ix_network = api.get_ix_network_by_ip_address(
            ipv6_address=ipv6_address, ipv4_address=ipv4_address
        )
        self.assertEqual(ix_network.id, ix_network_id)

        # Save the data inside the cache
        details = {
            "id": ix_network.id,
            "asn": ix_network.asn,
            "name": ix_network.name,
            "ipaddr6": ipv6_address,
            "ipaddr4": ipv4_address,
            "ix_id": ix_network.ix_id,
            "ixlan_id": ix_network.ixlan_id,
        }
        network_ixlan = NetworkIXLAN(**details)
        network_ixlan.save()

        # Using no API calls (cached data)
        ix_network = api.get_ix_network_by_ip_address(ipv6_address=ipv6_address)
        self.assertEqual(ix_network.id, ix_network_id)
        ix_network = api.get_ix_network_by_ip_address(ipv4_address=ipv4_address)
        self.assertEqual(ix_network.id, ix_network_id)
        ix_network = api.get_ix_network_by_ip_address(
            ipv6_address=ipv6_address, ipv4_address=ipv4_address
        )
        self.assertEqual(ix_network.id, ix_network_id)
github peeringdb / peeringdb-py / tests / test_sync.py View on Github external
def get_pclient():
    c = peeringdb.PeeringDB(helper.CONFIG)
    c._updater._ContextClass = _PartialEnabledContext
    return c
github peeringdb / peeringdb-py / tests / test_sync.py View on Github external
def get_client():
    return peeringdb.PeeringDB(helper.CONFIG)
github peeringdb / peeringdb-py / tests / test_sync.py View on Github external
def test_dry_run(client_empty):
    client = peeringdb.PeeringDB(helper.CONFIG, dry_run=True)
    rs = all_resources()
    client.update_all(rs)
    # still empty?
    with pytest.raises(peeringdb.get_backend().object_missing_error()):
        client.get(Network, FIRST_NET)
github respawner / peering-manager / peeringdb / tests.py View on Github external
def test_get_ix_networks_for_asn(self):
        api = PeeringDB()
        asn = 29467

        # Must not exist
        self.assertIsNone(api.get_ix_networks_for_asn(64500))

        known_ix_networks = [
            29146,
            15321,
            24292,
            15210,
            16774,
            14657,
            23162,
            14659,
            17707,
            27863,
github respawner / peering-manager / peeringdb / tests.py View on Github external
def test_get_autonomous_system(self):
        api = PeeringDB()
        asn = 15169

        # Must not exist
        self.assertIsNone(api.get_autonomous_system(64500))

        # Using an API call (no cached data)
        autonomous_system = api.get_autonomous_system(asn)
        self.assertEqual(autonomous_system.asn, asn)

        # Save the data inside the cache
        details = {
            "id": autonomous_system.id,
            "asn": autonomous_system.asn,
            "name": autonomous_system.name,
        }
        network = Network(**details)
github respawner / peering-manager / peeringdb / tests.py View on Github external
def test_get_last_synchronization(self):
        api = PeeringDB()

        # Test when no sync has been done
        self.assertIsNone(api.get_last_synchronization())

        # Test of sync record with no objects
        time_of_sync = timezone.now()
        api.record_last_sync(time_of_sync, {"added": 0, "updated": 0, "deleted": 0})
        self.assertEqual(api.get_last_sync_time(), 0)

        # Test of sync record with one object
        time_of_sync = timezone.now()
        api.record_last_sync(time_of_sync, {"added": 1, "updated": 0, "deleted": 0})
        self.assertEqual(
            int(api.get_last_synchronization().time.timestamp()),
            int(time_of_sync.timestamp()),
        )