Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
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)
# 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)
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)
def get_pclient():
c = peeringdb.PeeringDB(helper.CONFIG)
c._updater._ContextClass = _PartialEnabledContext
return c
def get_client():
return peeringdb.PeeringDB(helper.CONFIG)
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)
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,
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)
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()),
)