82 lines
2.3 KiB
Python
Executable File
82 lines
2.3 KiB
Python
Executable File
#!/usr/bin/env python
|
|
# -*- coding: utf-8
|
|
|
|
import os
|
|
import sys
|
|
import socket
|
|
from iptogeo import IPToGeo, IPToGeoException
|
|
|
|
TIMEOUT = None
|
|
# TIMEOUT = 5.0
|
|
|
|
iptogeo = IPToGeo(timeout=TIMEOUT, family=socket.AF_INET6, remote_addr='::1')
|
|
|
|
def get_random_ip_v4():
|
|
ip = '%d.%d.%d.%d' % \
|
|
(ord(os.urandom(1)), ord(os.urandom(1)), ord(os.urandom(1)), ord(os.urandom(1)))
|
|
return ip
|
|
|
|
def test_ip(ip, verbose=True, proxy=iptogeo, result=None):
|
|
try:
|
|
(ipres, country_code) = proxy.ip_to_geo(ip)
|
|
except IPToGeoException, e:
|
|
if verbose:
|
|
print 'Error with IP %s' % (ip)
|
|
print e
|
|
return
|
|
|
|
if verbose:
|
|
if not country_code:
|
|
print 'Country code for %s (%s) not found' % (ip, ipres)
|
|
else:
|
|
print 'Country code for %s (%s) is %s' % (ip, ipres, country_code)
|
|
|
|
if not result is None:
|
|
if not country_code: country_code = ''
|
|
if result != country_code:
|
|
raise Exception('Bad result %s != %s' % (country_code, result))
|
|
|
|
print '#### Coherence test'
|
|
test_ip('1.5.7.3', result='jp')
|
|
test_ip('1.5.255.4', result='jp')
|
|
test_ip('1.6.255.4', result='in')
|
|
test_ip('2.0.0.0', result='fr')
|
|
test_ip('127.0.0.1', result='')
|
|
test_ip('1.55.3.12', result='vn')
|
|
test_ip('1.57.0.0', result='cn')
|
|
test_ip('192.168.2.2', result='')
|
|
|
|
test_ip('2001:600::', result='eu')
|
|
test_ip('2001:608::00', result='de')
|
|
test_ip('2001:0660:0000:0000:0000:0000:0000:0000', result='fr')
|
|
test_ip('2001:670:1234:5678:9abc:def0:1234:5678', result='fi')
|
|
test_ip('8001::', result='')
|
|
test_ip('fe80::1a03:73ff:fee9:8705', result='')
|
|
|
|
print '#### 5 sockets test'
|
|
for i in range(0, 6):
|
|
print 'Test %d' % (i)
|
|
test_ip(get_random_ip_v4(), verbose=False, proxy=IPToGeo(timeout=TIMEOUT))
|
|
|
|
print '#### 35 sockets test'
|
|
for i in range(0, 36):
|
|
print 'Test %d' % (i)
|
|
test_ip(get_random_ip_v4(), verbose=False, proxy=IPToGeo(timeout=TIMEOUT))
|
|
|
|
print '#### 1 socket heavy test'
|
|
for i in range(0, 1000):
|
|
print 'Test %d' % (i)
|
|
test_ip(get_random_ip_v4(), verbose=False)
|
|
|
|
print '#### 35 sockets heavy test'
|
|
geo_proxy = []
|
|
for i in range(0, 36):
|
|
geo_proxy.append(IPToGeo(timeout=TIMEOUT))
|
|
|
|
for i in range(0, 1000):
|
|
print 'Test %d' % (i)
|
|
for proxy in geo_proxy:
|
|
test_ip(get_random_ip_v4(), verbose=False, proxy=proxy)
|
|
|
|
print 'All is OK for me'
|