Skip to content

P
pygeoip
Commit dde3969f by William Tisäter
Options

Merge branch 'master' into sphinx-docs

parents 22be6554 07890956
Showing with 328 additions and 426 deletions
CHANGELOG.md
# ChangeLog for pygeoip
### Release 0.3.1
* New: MaxMind Netspeed database support
* Fix: Release thread lock on exceptions
Not yet uploaded to PyPi
### Release 0.3.0
* New: 100% test coverage
......@@ -15,7 +22,7 @@
* Fix: Aquire lock on instance creation (Thanks to Jiří Techet)
* Fix: Fixed alpha-3 codes ordering, replaced TLS,TKM,TUN,TON with TKM,TUN,TON,TLS (Thanks to Marc Sherry)
[William Tisäter](mailto:william@defunct.cc) -- TBA
Uploaded to PyPi on 2013-11-13
### Release 0.2.7
......@@ -23,7 +30,7 @@
* Fix: Sync timezones from latest ISO-8601 edition
* Fix: `record_by_addr()` will now return `None` when missing data
[William Tisäter](mailto:william@defunct.cc) -- 2013-06-04 15:55:01
Uploaded to PyPi on 2013-07-15
### Release 0.2.6
......@@ -47,5 +54,4 @@
* Fix: Remove dependency of ez_setup.py
* Fix: Add Python 3.2 and 3.3 to tox tests
[William Tisäter](mailto:william@defunct.cc) -- 2013-02-22 02:51:36
Uploaded to PyPi on 2013-02-23
DEVELOPER.md
......@@ -12,7 +12,7 @@ Running the tests requires a couple of sample databases found on the
link below.
Maxmind sample databases for testing can be downloaded here:
https://www.defunct.cc/maxmind-geoip-samples.tar.gz (58 MB)
https://www.defunct.cc/maxmind-geoip-samples.tar.gz (17 MB)
Extract the tarball in the tests directory and run `tox` from the root directory.
......
README.md
# Pure Python GeoIP API #
# Pure Python GeoIP API
This library is based on [Maxmind's GeoIP C API](https://github.com/maxmind/geoip-api-c).
Tested with Python version 2.5, 2.6, 2.7, 3.2 and 3.3.
Tested with Python version 2.6, 2.7, 3.2 and 3.3.
[![Build Status](https://travis-ci.org/appliedsec/pygeoip.png)](https://travis-ci.org/appliedsec/pygeoip) [![Coverage Status](https://coveralls.io/repos/appliedsec/pygeoip/badge.png)](https://coveralls.io/r/appliedsec/pygeoip) [![Downloads](https://pypip.in/d/pygeoip/badge.png)](https://crate.io/packages/pygeoip)
[![Build Status](https://api.travis-ci.org/appliedsec/pygeoip.png?branch=master)](https://travis-ci.org/appliedsec/pygeoip) [![Coverage Status](https://coveralls.io/repos/appliedsec/pygeoip/badge.png)](https://coveralls.io/r/appliedsec/pygeoip) [![Downloads](https://pypip.in/d/pygeoip/badge.png)](https://crate.io/packages/pygeoip)
## Installation ##
## Installation
You can easily install pygeoip from PyPi.
pip install pygeoip
```bash
pip install pygeoip
```
## Supported Databases ##
* COUNTRY_EDITION
* COUNTRY_EDITION_V6
* REGION_EDITION_REV0
* REGION_EDITION_REV1
* CITY_EDITION_REV0
* CITY_EDITION_REV1
* CITY_EDITION_REV1_V6
* ORG_EDITION
* ISP_EDITION
* ASNUM_EDITION
* ASNUM_EDITION_V6
## Issues and Contribution ##
## Issues and Contribution
Bug reports are done by [creating an issue on Github](https://github.com/appliedsec/pygeoip/issues). If you want to contribute you can always [create a pull request](https://github.com/appliedsec/pygeoip/pulls) for discussion and code submission.
## Getting Started ##
## Getting Started
Create your GeoIP instance with appropriate access flag. `STANDARD` reads data from disk when needed, `MEMORY_CACHE` loads database into memory on instantiation and `MMAP_CACHE` loads database into memory using mmap.
import pygeoip
gi4 = pygeoip.GeoIP('/path/to/GeoIP.dat', pygeoip.MEMORY_CACHE)
gi6 = pygeoip.GeoIP('/path/to/GeoIPv6.dat', pygeoip.MEMORY_CACHE)
### Country Lookup ###
>>> gi4.country_code_by_name('google.com')
'US'
>>> gi4.country_code_by_addr('64.233.161.99')
'US'
>>> gi4.country_name_by_addr('64.233.161.99')
'United States'
>>> gi6.country_code_by_name('google.com')
'IE'
>>> gi6.country_code_by_addr('2001:7fd::1')
'EU'
>>> gi6.country_name_by_addr('2001:7fd::1')
'Europe'
### Region Lookup ###
>>> gi = pygeoip.GeoIP('/path/to/GeoIPRegion.dat')
>>> gi.region_by_name('apple.com')
{'region_code': 'CA', 'country_code': 'US'}
```python
>>> import pygeoip
>>> gi = pygeoip.GeoIP('/path/to/GeoIP.dat')
>>> gi.country_name_by_addr('64.233.161.99')
'United States'
```
### Country Lookup
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIP.dat')
>>> gi.country_code_by_name('google.com')
'US'
>>> gi.country_code_by_addr('64.233.161.99')
'US'
>>> gi.country_name_by_addr('64.233.161.99')
'United States'
```
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIPv6.dat')
>>> gi.country_code_by_name('google.com')
'IE'
>>> gi.country_code_by_addr('2001:7fd::1')
'EU'
>>> gi.country_name_by_addr('2001:7fd::1')
'Europe'
```
### Region Lookup
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIPRegion.dat')
>>> gi.region_by_name('apple.com')
{'region_code': 'CA', 'country_code': 'US'}
```
### City Lookup ###
>>> gi = pygeoip.GeoIP('/path/to/GeoIPCity.dat')
>>> gi.record_by_addr('64.233.161.99')
{
'city': u'Mountain View',
'region_code': u'CA',
'area_code': 650,
'time_zone': 'America/Los_Angeles',
'dma_code': 807,
'metro_code': 'San Francisco, CA',
'country_code3': 'USA',
'latitude': 37.41919999999999,
'postal_code': u'94043',
'longitude': -122.0574,
'country_code': 'US',
'country_name': 'United States',
'continent': 'NA'
}
>>> gi.time_zone_by_addr('64.233.161.99')
'America/Los_Angeles'
### Organization Lookup ###
>>> gi = pygeoip.GeoIP('/path/to/GeoIPOrg.dat')
>>> gi.org_by_name('dell.com')
'Dell Computer Corporation'
### ISP Lookup ###
>>> gi = pygeoip.GeoIP('/path/to/GeoIPISP.dat')
>>> gi.org_by_name('cnn.com')
'Turner Broadcasting System'
### ASN Lookup ###
>>> gi = pygeoip.GeoIP('/path/to/GeoIPASNum.dat')
>>> gi.org_by_name('cnn.com')
'AS5662 Turner Broadcasting'
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIPCity.dat')
>>> gi.record_by_addr('64.233.161.99')
{
'city': u'Mountain View',
'region_code': u'CA',
'area_code': 650,
'time_zone': 'America/Los_Angeles',
'dma_code': 807,
'metro_code': 'San Francisco, CA',
'country_code3': 'USA',
'latitude': 37.41919999999999,
'postal_code': u'94043',
'longitude': -122.0574,
'country_code': 'US',
'country_name': 'United States',
'continent': 'NA'
}
>>> gi.time_zone_by_addr('64.233.161.99')
'America/Los_Angeles'
```
### Organization Lookup
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIPOrg.dat')
>>> gi.org_by_name('dell.com')
'Dell Computer Corporation'
```
### ISP Lookup
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIPISP.dat')
>>> gi.isp_by_name('cnn.com')
'Turner Broadcasting System'
```
### ASN Lookup
```python
>>> gi = pygeoip.GeoIP('/path/to/GeoIPASNum.dat')
>>> gi.asn_by_name('cnn.com')
'AS5662 Turner Broadcasting'
```
For more information, [check out the full API documentation](http://packages.python.org/pygeoip).
## Supported Databases
| Type | IPv4 | IPv6 | Details |
| -------------- |:----:|:----:| ------- |
| Country | ✓ | ✓ | [MaxMind Country product page](http://www.maxmind.com/en/country) |
| City | ✓ | ✓ | [MaxMind City product page](http://www.maxmind.com/en/city) |
| Organization | ✓ | | [MaxMind Organization product page](http://www.maxmind.com/en/organization) |
| ISP | ✓ | | [MaxMind ISP product page](http://www.maxmind.com/en/isp) |
| Region | ✓ | | [MaxMind Region product page](http://www.maxmind.com/en/geolocation_landing) |
| ASN | ✓ | ✓ | [MaxMind ASN product page](http://dev.maxmind.com/geoip/legacy/geolite) |
| Netspeed | ✓ | | [MaxMind Netspeed product page](http://www.maxmind.com/en/netspeed) |
pygeoip/__init__.py
......@@ -20,6 +20,7 @@ GNU General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/lgpl.txt>.
"""
__version__ = "0.3.1"
import os
import socket
......@@ -32,16 +33,11 @@ try:
except ImportError: # pragma: no cover
mmap = None
try:
from StringIO import StringIO
range = xrange # Use xrange for Python 2
except ImportError:
from io import StringIO, BytesIO
from pygeoip import util, const
from pygeoip.const import PY2, PY3
from pygeoip.timezone import time_zone_by_country_and_region
range = xrange if PY2 else range
STANDARD = const.STANDARD
MMAP_CACHE = const.MMAP_CACHE
......@@ -72,10 +68,12 @@ class _GeoIPMetaclass(type):
if not kwargs.get('cache', True):
return super(_GeoIPMetaclass, cls).__call__(*args, **kwargs)
cls._instance_lock.acquire()
if filename not in cls._instances:
cls._instances[filename] = super(_GeoIPMetaclass, cls).__call__(*args, **kwargs)
cls._instance_lock.release()
try:
cls._instance_lock.acquire()
if filename not in cls._instances:
cls._instances[filename] = super(_GeoIPMetaclass, cls).__call__(*args, **kwargs)
finally:
cls._instance_lock.release()
return cls._instances[filename]
......@@ -97,6 +95,7 @@ class GeoIP(object):
@param cache: Used in tests to skip instance caching
@type cache: bool
"""
self._lock = Lock()
self._flags = flags
self._netmask = None
......@@ -114,27 +113,18 @@ class GeoIP(object):
elif self._flags & const.MEMORY_CACHE:
f = codecs.open(filename, 'rb', ENCODING)
self._memory = f.read()
self._fp = self._str_to_fp(self._memory)
self._fp = util.str2fp(self._memory)
self._type = 'MEMORY_CACHE'
f.close()
else:
self._fp = codecs.open(filename, 'rb', ENCODING)
self._type = 'STANDARD'
self._lock = Lock()
self._setup_segments()
@classmethod
def _str_to_fp(cls, data):
"""
Convert bytes data to file handle object
@param data: string data
@type data: str
@return: file handle object
@rtype: StringIO or BytesIO
"""
return BytesIO(bytearray(data, ENCODING)) if PY3 else StringIO(data)
try:
self._lock.acquire()
self._setup_segments()
finally:
self._lock.release()
def _setup_segments(self):
"""
......@@ -155,13 +145,13 @@ class GeoIP(object):
* ISP_EDITION
* ASNUM_EDITION
* ASNUM_EDITION_V6
* NETSPEED_EDITION
"""
self._databaseType = const.COUNTRY_EDITION
self._recordLength = const.STANDARD_RECORD_LENGTH
self._databaseSegments = const.COUNTRY_BEGIN
self._lock.acquire()
filepos = self._fp.tell()
self._fp.seek(-3, os.SEEK_END)
......@@ -215,7 +205,6 @@ class GeoIP(object):
self._fp.seek(-4, os.SEEK_CUR)
self._fp.seek(filepos, os.SEEK_SET)
self._lock.release()
def _seek_country(self, ipnum):
"""
......@@ -239,10 +228,12 @@ class GeoIP(object):
else:
startIndex = 2 * self._recordLength * offset
readLength = 2 * self._recordLength
self._lock.acquire()
self._fp.seek(startIndex, os.SEEK_SET)
buf = self._fp.read(readLength)
self._lock.release()
try:
self._lock.acquire()
self._fp.seek(startIndex, os.SEEK_SET)
buf = self._fp.read(readLength)
finally:
self._lock.release()
if PY3 and type(buf) is bytes:
buf = buf.decode(ENCODING)
......@@ -280,10 +271,12 @@ class GeoIP(object):
return None
read_length = (2 * self._recordLength - 1) * self._databaseSegments
self._lock.acquire()
self._fp.seek(seek_org + read_length, os.SEEK_SET)
buf = self._fp.read(const.MAX_ORG_RECORD_LENGTH)
self._lock.release()
try:
self._lock.acquire()
self._fp.seek(seek_org + read_length, os.SEEK_SET)
buf = self._fp.read(const.MAX_ORG_RECORD_LENGTH)
finally:
self._lock.release()
if PY3 and type(buf) is bytes:
buf = buf.decode(ENCODING)
......@@ -352,10 +345,12 @@ class GeoIP(object):
return {}
read_length = (2 * self._recordLength - 1) * self._databaseSegments
self._lock.acquire()
self._fp.seek(seek_country + read_length, os.SEEK_SET)
buf = self._fp.read(const.FULL_RECORD_LENGTH)
self._lock.release()
try:
self._lock.acquire()
self._fp.seek(seek_country + read_length, os.SEEK_SET)
buf = self._fp.read(const.FULL_RECORD_LENGTH)
finally:
self._lock.release()
if PY3 and type(buf) is bytes:
buf = buf.decode(ENCODING)
......@@ -420,10 +415,23 @@ class GeoIP(object):
else:
return socket.gethostbyname(hostname)
def _id_by_addr(self, addr):
def id_by_name(self, hostname):
"""
Returns the database id for specified hostname.
The id might be useful as array index. 0 is unknown.
@param hostname: Hostname
@type hostname: str
@return: network byte order 32-bit integer
@rtype: int
"""
addr = self._gethostbyname(hostname)
return self.id_by_addr(addr)
def id_by_addr(self, addr):
"""
Looks up the index for the country which is the key for the
code and name.
Returns the database id for specified address.
The id might be useful as array index. 0 is unknown.
@param addr: IPv4 or IPv6 address
@type addr: str
......@@ -431,7 +439,7 @@ class GeoIP(object):
@rtype: int
"""
ipv = 6 if addr.find(':') >= 0 else 4
if ipv == 4 and self._databaseType != const.COUNTRY_EDITION:
if ipv == 4 and self._databaseType not in (const.COUNTRY_EDITION, const.NETSPEED_EDITION):
raise GeoIPError('Invalid database type; expected IPv6 address')
if ipv == 6 and self._databaseType != const.COUNTRY_EDITION_V6:
raise GeoIPError('Invalid database type; expected IPv4 address')
......@@ -460,7 +468,7 @@ class GeoIP(object):
"""
VALID_EDITIONS = (const.COUNTRY_EDITION, const.COUNTRY_EDITION_V6)
if self._databaseType in VALID_EDITIONS:
country_id = self._id_by_addr(addr)
country_id = self.id_by_addr(addr)
return const.COUNTRY_CODES[country_id]
elif self._databaseType in const.REGION_CITY_EDITIONS:
return self.region_by_addr(addr).get('country_code')
......@@ -480,6 +488,29 @@ class GeoIP(object):
addr = self._gethostbyname(hostname)
return self.country_code_by_addr(addr)
def netspeed_by_addr(self, addr):
"""
Returns NetSpeed name from address.
@param hostname: IP address
@type hostname: str
@return: netspeed name
@rtype: str
"""
return const.NETSPEED_NAMES[self.id_by_addr(addr)]
def netspeed_by_name(self, hostname):
"""
Returns NetSpeed name from hostname.
@param hostname: Hostname
@type hostname: str
@return: netspeed name
@rtype: str
"""
addr = self._gethostbyname(hostname)
return self.netspeed_by_addr(addr)
def country_name_by_addr(self, addr):
"""
Returns full country name for specified IP address.
......@@ -492,7 +523,7 @@ class GeoIP(object):
"""
VALID_EDITIONS = (const.COUNTRY_EDITION, const.COUNTRY_EDITION_V6)
if self._databaseType in VALID_EDITIONS:
country_id = self._id_by_addr(addr)
country_id = self.id_by_addr(addr)
return const.COUNTRY_NAMES[country_id]
elif self._databaseType in const.CITY_EDITIONS:
return self.record_by_addr(addr).get('country_name')
......@@ -544,6 +575,11 @@ class GeoIP(object):
addr = self._gethostbyname(hostname)
return self.org_by_addr(addr)
isp_by_addr = org_by_addr
isp_by_name = org_by_name
asn_by_addr = org_by_addr
asn_by_name = org_by_name
def record_by_addr(self, addr):
"""
Look up the record for a given IP address.
......
pygeoip/const.py
......@@ -383,6 +383,10 @@ CONTINENT_NAMES = (
'NA', 'NA', 'AF'
)
NETSPEED_NAMES = (
'Unknown', 'Dial-up', 'Cable', 'Corporate'
)
# storage / caching flags
STANDARD = 0
MEMORY_CACHE = 1
......@@ -408,9 +412,11 @@ ORG_EDITION = 5
ISP_EDITION = 4
ASNUM_EDITION = 9
ASNUM_EDITION_V6 = 21
NETSPEED_EDITION = 10
# Not yet supported databases
PROXY_EDITION = 8
NETSPEED_EDITION = 11
NETSPEED_EDITION_REV1 = 32
NETSPEED_EDITION_REV1_V6 = 33
# Collection of databases
IPV6_EDITIONS = (COUNTRY_EDITION_V6, ASNUM_EDITION_V6, CITY_EDITION_REV1_V6)
......
pygeoip/util.py
......@@ -24,6 +24,13 @@ along with this program. If not, see <http://www.gnu.org/licenses/lgpl.txt>.
import socket
import binascii
try:
from StringIO import StringIO
except ImportError:
from io import StringIO, BytesIO
from pygeoip import const
def ip2long(ip):
"""
......@@ -35,3 +42,15 @@ def ip2long(ip):
return int(binascii.hexlify(socket.inet_aton(ip)), 16)
except socket.error:
return int(binascii.hexlify(socket.inet_pton(socket.AF_INET6, ip)), 16)
def str2fp(data):
"""
Convert bytes data to file handle object
@param data: string data
@type data: str
@return: file handle object
@rtype: StringIO or BytesIO
"""
return BytesIO(bytearray(data, const.ENCODING)) if const.PY3 else StringIO(data)
setup.cfg 0 → 100644
[wheel]
universal = 1
setup.py
......@@ -27,9 +27,10 @@ try:
except ImportError:
from distutils.core import setup
from pygeoip import __version__
setup(name='pygeoip',
version='0.3.0',
version=__version__,
description='Pure Python GeoIP API',
author='Jennifer Ennis',
author_email='zaylea@gmail.com',
......@@ -37,7 +38,6 @@ setup(name='pygeoip',
maintainer_email='william@defunct.cc',
url='https://github.com/appliedsec/pygeoip',
classifiers=['Programming Language :: Python',
'Programming Language :: Python :: 2.5',
'Programming Language :: Python :: 2.6',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
......
tests/config.py
......@@ -9,4 +9,6 @@ CITY_DB_PATH = path.join(DATA_DIR, 'GeoLiteCity.dat')
ORG_DB_PATH = path.join(DATA_DIR, 'GeoIPOrg.dat')
ASNUM_DB_PATH = path.join(DATA_DIR, 'GeoIPASNum.dat')
ISP_DB_PATH = path.join(DATA_DIR, 'GeoIPISP.dat')
NETSPEED_DB_PATH = path.join(DATA_DIR, 'GeoIPNetSpeed.dat')
NETSPEEDCELL_DB_PATH = path.join(DATA_DIR, 'GeoIPNetSpeedCell.dat')
CORRUPT_DB_PATH = path.join(DATA_DIR, '../config.py')
tests/test_asnum.py
......@@ -8,34 +8,16 @@ from tests.config import ASNUM_DB_PATH
class TestGeoIPASNumFunctions(unittest.TestCase):
def setUp(self):
self.us_as = 'AS15169 Google Inc.'
self.us_ip = '64.233.161.99'
self.us_hostname = 'google.com'
self.gb_as = 'AS2818 BBC Internet Services, UK'
self.gb_ip = '212.58.253.68'
self.gb_hostname = 'bbc.com'
self.gia = pygeoip.GeoIP(ASNUM_DB_PATH)
self.gi = pygeoip.GeoIP(ASNUM_DB_PATH)
def testOrgByAddr(self):
gb_as = self.gia.org_by_addr(self.gb_ip)
us_as = self.gia.org_by_addr(self.us_ip)
self.assertEqual(gb_as, self.gb_as)
self.assertEqual(us_as, self.us_as)
def testOrgByName(self):
gb_as = self.gia.org_by_name(self.gb_hostname)
us_as = self.gia.org_by_name(self.us_hostname)
self.assertEqual(gb_as, self.gb_as)
self.assertEqual(us_as, self.us_as)
asn = self.gi.asn_by_addr('64.17.254.216')
self.assertEqual(asn, 'AS33224')
@raises(pygeoip.GeoIPError)
def testCodeByAddr(self):
self.gia.country_code_by_addr(self.us_ip)
self.gi.country_code_by_addr('64.17.254.216')
@raises(pygeoip.GeoIPError)
def testNameByAddr(self):
self.gia.country_name_by_addr(self.us_ip)
self.gi.country_name_by_addr('64.17.254.216')
tests/test_cache.py
......@@ -7,18 +7,15 @@ from tests.config import COUNTRY_DB_PATH
class TestGeoIPCacheMethods(unittest.TestCase):
def setUp(self):
self.us_ip = '64.233.161.99'
self.us_code = 'US'
self.gi_mmap = pygeoip.GeoIP(COUNTRY_DB_PATH, MMAP_CACHE)
self.gi_memory = pygeoip.GeoIP(COUNTRY_DB_PATH, MEMORY_CACHE)
self.gi_standard = pygeoip.GeoIP(COUNTRY_DB_PATH, STANDARD)
def testCountryCodeByAddr(self):
code_mmap = self.gi_mmap.country_code_by_addr(self.us_ip)
code_memory = self.gi_memory.country_code_by_addr(self.us_ip)
code_standard = self.gi_standard.country_code_by_addr(self.us_ip)
code_mmap = self.gi_mmap.country_code_by_addr('64.17.254.216')
code_memory = self.gi_memory.country_code_by_addr('64.17.254.216')
code_standard = self.gi_standard.country_code_by_addr('64.17.254.216')
self.assertEqual(code_mmap, self.us_code)
self.assertEqual(code_memory, self.us_code)
self.assertEqual(code_standard, self.us_code)
self.assertEqual(code_mmap, 'US')
self.assertEqual(code_memory, 'US')
self.assertEqual(code_standard, 'US')
tests/test_city.py
......@@ -7,18 +7,6 @@ from tests.config import CITY_DB_PATH
class TestGeoIPCityFunctions(unittest.TestCase):
def setUp(self):
self.us_hostname = 'google.com'
self.us_ip = '64.233.161.99'
self.gb_hostname = 'bbc.com'
self.gb_ip = '212.58.253.68'
self.us_code = 'US'
self.gb_code = 'GB'
self.us_name = 'United States'
self.gb_name = 'United Kingdom'
self.us_record_data = {
'city': 'Mountain View',
'region_code': 'CA',
......@@ -51,58 +39,28 @@ class TestGeoIPCityFunctions(unittest.TestCase):
'time_zone': 'Europe/London'
}
self.us_region_data = {'region_code': 'CA', 'country_code': 'US'}
self.gb_region_data = {'region_code': 'N7', 'country_code': 'GB'}
self.gic = pygeoip.GeoIP(CITY_DB_PATH)
self.gic_mem = pygeoip.GeoIP(CITY_DB_PATH, pygeoip.MEMORY_CACHE)
self.gic_mmap = pygeoip.GeoIP(CITY_DB_PATH, pygeoip.MMAP_CACHE)
def testCountryCodeByName(self):
us_code = self.gic.country_code_by_name(self.us_hostname)
gb_code = self.gic.country_code_by_name(self.gb_hostname)
self.assertEqual(us_code, self.us_code)
self.assertEqual(gb_code, self.gb_code)
def testCountryCodeByAddr(self):
us_code = self.gic.country_code_by_addr(self.us_ip)
gb_code = self.gic.country_code_by_addr(self.gb_ip)
self.assertEqual(us_code, self.us_code)
self.assertEqual(gb_code, self.gb_code)
def testCountryNameByName(self):
us_name = self.gic.country_name_by_name(self.us_hostname)
gb_name = self.gic.country_name_by_name(self.gb_hostname)
self.assertEqual(us_name, self.us_name)
self.assertEqual(gb_name, self.gb_name)
code = self.gic.country_code_by_addr('66.92.181.240')
self.assertEqual(code, 'US')
def testCountryNameByAddr(self):
us_name = self.gic.country_name_by_addr(self.us_ip)
gb_name = self.gic.country_name_by_addr(self.gb_ip)
self.assertEqual(us_name, self.us_name)
self.assertEqual(gb_name, self.gb_name)
us_name = self.gic.country_name_by_addr('64.233.161.99')
gb_name = self.gic.country_name_by_addr('212.58.253.68')
def testRegionByName(self):
us_region_data = self.gic.region_by_name(self.us_hostname)
gb_region_data = self.gic.region_by_name(self.gb_hostname)
self.assertEqual(us_region_data, self.us_region_data)
self.assertEqual(gb_region_data, self.gb_region_data)
self.assertEqual(us_name, 'United States')
self.assertEqual(gb_name, 'United Kingdom')
def testRegionByAddr(self):
us_region = self.gic.region_by_addr(self.us_ip)
gb_region = self.gic.region_by_addr(self.gb_ip)
self.assertEqual(us_region, self.us_region_data)
self.assertEqual(gb_region, self.gb_region_data)
region = self.gic.region_by_addr('66.92.181.240')
self.assertEqual(region, {'region_code': 'CA', 'country_code': 'US'})
def testCacheMethods(self):
mem_record = self.gic_mem.record_by_addr(self.us_ip)
mmap_record = self.gic_mmap.record_by_addr(self.us_ip)
mem_record = self.gic_mem.record_by_addr('64.233.161.99')
mmap_record = self.gic_mmap.record_by_addr('64.233.161.99')
self.assertEqual(mem_record['city'], self.us_record_data['city'])
self.assertEqual(mmap_record['city'], self.us_record_data['city'])
......@@ -113,31 +71,7 @@ class TestGeoIPCityFunctions(unittest.TestCase):
'country_name', 'time_zone')
almost_equal_keys = ('longitude', 'latitude')
us_record = self.gic.record_by_addr(self.us_ip)
for key, value in us_record.items():
if key in equal_keys:
test_value = self.us_record_data[key]
self.assertEqual(value, test_value, 'Key: %s' % key)
elif key in almost_equal_keys:
test_value = self.us_record_data[key]
self.assertAlmostEqual(value, test_value, 3, 'Key: %s' % key)
gb_record = self.gic.record_by_addr(self.gb_ip)
for key, value in gb_record.items():
if key in equal_keys:
test_value = self.gb_record_data[key]
self.assertEqual(value, test_value, 'Key: %s' % key)
elif key in almost_equal_keys:
test_value = self.gb_record_data[key]
self.assertAlmostEqual(value, test_value, 3, 'Key: %s' % key)
def testRecordByName(self):
equal_keys = ('city', 'region_name', 'area_code', 'country_code3',
'postal_code', 'dma_code', 'country_code',
'country_name', 'time_zone')
almost_equal_keys = ('longitude', 'latitude')
us_record = self.gic.record_by_name(self.us_hostname)
us_record = self.gic.record_by_addr('64.233.161.99')
for key, value in us_record.items():
if key in equal_keys:
test_value = self.us_record_data[key]
......@@ -146,7 +80,7 @@ class TestGeoIPCityFunctions(unittest.TestCase):
test_value = self.us_record_data[key]
self.assertAlmostEqual(value, test_value, 3, 'Key: %s' % key)
gb_record = self.gic.record_by_name(self.gb_hostname)
gb_record = self.gic.record_by_addr('212.58.253.68')
for key, value in gb_record.items():
if key in equal_keys:
test_value = self.gb_record_data[key]
......
tests/test_country.py
# -*- coding: utf-8 -*-
import socket
import unittest
from nose.tools import raises
......@@ -9,86 +8,49 @@ from tests.config import COUNTRY_DB_PATH, COUNTRY_V6_DB_PATH
class TestGeoIPCountryFunctions(unittest.TestCase):
def setUp(self):
self.us_hostname = 'google.com'
self.gb_hostname = 'bbc.com'
self.us_ip = '64.233.161.99'
self.gb_ip = '212.58.253.68'
self.ie6_hostname = 'ipv6.loopia.se'
self.ie6_ip = '2a00:1450:400f:800::1002'
self.ie_code = 'IE'
self.us_code = 'US'
self.gb_code = 'GB'
self.se_code = 'SE'
self.ie_name = 'Ireland'
self.us_name = 'United States'
self.gb_name = 'United Kingdom'
self.se_name = 'Sweden'
self.gi = pygeoip.GeoIP(COUNTRY_DB_PATH)
self.gi6 = pygeoip.GeoIP(COUNTRY_V6_DB_PATH)
def testCountryCodeByName(self):
us_code = self.gi.country_code_by_name(self.us_hostname)
gb_code = self.gi.country_code_by_name(self.gb_hostname)
ie6_code = self.gi6.country_code_by_name(self.ie6_hostname)
self.assertEqual(us_code, self.us_code)
self.assertEqual(gb_code, self.gb_code)
self.assertEqual(ie6_code, self.se_code)
def testCountryCodeByAddr(self):
us_code = self.gi.country_code_by_addr(self.us_ip)
gb_code = self.gi.country_code_by_addr(self.gb_ip)
ie6_code = self.gi6.country_code_by_addr(self.ie6_ip)
self.assertEqual(us_code, self.us_code)
self.assertEqual(gb_code, self.gb_code)
self.assertEqual(ie6_code, self.ie_code)
def testCountryNameByName(self):
us_name = self.gi.country_name_by_name(self.us_hostname)
gb_name = self.gi.country_name_by_name(self.gb_hostname)
ie6_name = self.gi6.country_name_by_name(self.ie6_hostname)
us_code = self.gi.country_code_by_addr('64.17.254.216')
it_code = self.gi.country_code_by_addr('78.26.70.208')
jp6_code = self.gi6.country_code_by_addr('2001:200::')
self.assertEqual(us_name, self.us_name)
self.assertEqual(gb_name, self.gb_name)
self.assertEqual(ie6_name, self.se_name)
self.assertEqual(us_code, 'US')
self.assertEqual(it_code, 'IT')
self.assertEqual(jp6_code, 'JP')
def testCountryNameByAddr(self):
us_name = self.gi.country_name_by_addr(self.us_ip)
gb_name = self.gi.country_name_by_addr(self.gb_ip)
ie6_name = self.gi6.country_name_by_addr(self.ie6_ip)
us_name = self.gi.country_name_by_addr('64.17.254.216')
it_name = self.gi.country_name_by_addr('78.26.70.208')
jp6_name = self.gi6.country_name_by_addr('2001:200::')
self.assertEqual(us_name, self.us_name)
self.assertEqual(gb_name, self.gb_name)
self.assertEqual(ie6_name, self.ie_name)
self.assertEqual(us_name, 'United States')
self.assertEqual(it_name, 'Italy')
self.assertEqual(jp6_name, 'Japan')
@raises(pygeoip.GeoIPError)
def testOpen4With6(self):
data = self.gi.country_code_by_addr(self.ie6_ip)
data = self.gi.country_code_by_addr('2001:200::')
raise ValueError(data)
@raises(pygeoip.GeoIPError)
def testOpen6With4(self):
data = self.gi6.country_code_by_addr(self.gb_ip)
data = self.gi6.country_code_by_addr('78.26.70.208')
raise ValueError(data)
@raises(pygeoip.GeoIPError)
def testOrgByAddr(self):
self.gi.org_by_addr(self.us_ip)
self.gi.org_by_addr('78.26.70.208')
@raises(pygeoip.GeoIPError)
def testRecordByAddr(self):
self.gi.record_by_addr(self.us_ip)
self.gi.record_by_addr('78.26.70.208')
@raises(pygeoip.GeoIPError)
def testRegionByAddr(self):
self.gi.region_by_addr(self.us_ip)
self.gi.region_by_addr('78.26.70.208')
@raises(pygeoip.GeoIPError)
def testTimeZoneByAddr(self):
self.gi.time_zone_by_addr(self.us_ip)
self.gi.time_zone_by_addr('78.26.70.208')
tests/test_isp.py
......@@ -7,26 +7,8 @@ from tests.config import ISP_DB_PATH
class TestGeOIPISPFunctions(unittest.TestCase):
def setUp(self):
self.us_isp = 'Turner Broadcasting System'
self.us_ip = '157.166.255.18'
self.us_hostname = 'cnn.com'
self.se_isp = 'IP-Only Tele Communication AB'
self.se_ip = '213.132.112.97'
self.se_hostname = 'blocket.se'
self.gi = pygeoip.GeoIP(ISP_DB_PATH)
def testISPByAddr(self):
se_isp = self.gi.org_by_addr(self.se_ip)
us_isp = self.gi.org_by_addr(self.us_ip)
self.assertEqual(se_isp, self.se_isp)
self.assertEqual(us_isp, self.us_isp)
def testISPByName(self):
se_isp = self.gi.org_by_name(self.se_hostname)
us_isp = self.gi.org_by_name(self.us_hostname)
self.assertEqual(se_isp, self.se_isp)
self.assertEqual(us_isp, self.us_isp)
isp = self.gi.isp_by_addr('70.46.123.145')
self.assertEqual(isp, 'FDN Communications')
tests/test_memory.py
......@@ -11,5 +11,5 @@ class TestMemoryCache(unittest.TestCase):
gi = pygeoip.GeoIP(COUNTRY_DB_PATH, flags=const.MEMORY_CACHE, cache=False)
self.assertEqual(gi._type, 'MEMORY_CACHE')
code = gi.country_code_by_name('dn.se')
self.assertEqual(code, 'SE')
code = gi.country_code_by_addr('64.17.254.216')
self.assertEqual(code, 'US')
tests/test_netspeed.py 0 → 100644
# -*- coding: utf-8 -*-
import unittest
import pygeoip
from tests.config import NETSPEED_DB_PATH, NETSPEEDCELL_DB_PATH
class TestGeoIPNetspeedFunctions(unittest.TestCase):
def setUp(self):
self.gi = pygeoip.GeoIP(NETSPEED_DB_PATH)
def testNetSpeedByAddr(self):
netspeed = self.gi.id_by_addr('17.172.224.47')
self.assertEqual(netspeed, 3)
def testNetSpeedByAddrWrapper(self):
netspeed = self.gi.netspeed_by_addr('17.172.224.47')
self.assertEqual(netspeed, 'Corporate')
tests/test_org.py
......@@ -7,26 +7,8 @@ from tests.config import ORG_DB_PATH
class TestGeoIPOrgFunctions(unittest.TestCase):
def setUp(self):
self.us_org = 'APPLE COMPUTER'
self.us_ip = '17.172.224.47'
self.us_hostname = 'apple.com'
self.gb_org = 'BBC'
self.gb_ip = '212.58.253.68'
self.gb_hostname = 'bbc.com'
self.gio = pygeoip.GeoIP(ORG_DB_PATH)
self.gi = pygeoip.GeoIP(ORG_DB_PATH)
def testOrgByAddr(self):
gb_org = self.gio.org_by_addr(self.gb_ip)
us_org = self.gio.org_by_addr(self.us_ip)
self.assertEqual(gb_org, self.gb_org)
self.assertEqual(us_org, self.us_org)
def testOrgByName(self):
gb_org = self.gio.org_by_name(self.gb_hostname)
us_org = self.gio.org_by_name(self.us_hostname)
self.assertEqual(gb_org, self.gb_org)
self.assertEqual(us_org, self.us_org)
org = self.gi.org_by_addr('70.46.123.145')
self.assertEqual(org, 'DSLAM WAN Allocation')
tests/test_region.py
......@@ -7,48 +7,15 @@ from tests.config import REGION_DB_PATH
class TestGeoIPRegionFunctions(unittest.TestCase):
def setUp(self):
self.se_hostname = 'dn.se'
self.se_region_data = {
'region_code': None,
'country_code': None
}
self.gi = pygeoip.GeoIP(REGION_DB_PATH)
self.ca_hostname = 'cbc.ca'
self.ca_region_data = {
'region_code': 'ON',
'country_code': 'CA'
}
self.us_code = 'US'
self.us_hostname = 'apple.com'
self.us_ip = '17.172.224.47'
self.us_region_data = {
def testRegionByAddr(self):
region_name = self.gi.region_by_addr('17.172.224.47')
self.assertEqual(region_name, {
'region_code': 'CA',
'country_code': 'US'
}
self.gir = pygeoip.GeoIP(REGION_DB_PATH)
def testRegionByName(self):
region_name = self.gir.region_by_name(self.us_hostname)
self.assertEqual(region_name, self.us_region_data)
def testRegionByAddr(self):
region_name = self.gir.region_by_addr(self.us_ip)
self.assertEqual(region_name, self.us_region_data)
def testCountryCodeByName(self):
us_code = self.gir.country_code_by_name(self.us_hostname)
self.assertEqual(us_code, self.us_code)
})
def testCountryCodeByAddr(self):
us_code = self.gir.country_code_by_addr(self.us_ip)
self.assertEqual(us_code, self.us_code)
def testCanadaRegion(self):
region_name = self.gir.region_by_name(self.ca_hostname)
self.assertEqual(region_name, self.ca_region_data)
def testSwedenRegion(self):
region_name = self.gir.region_by_name(self.se_hostname)
self.assertEqual(region_name, self.se_region_data)
us_code = self.gi.country_code_by_addr('17.172.224.47')
self.assertEqual(us_code, 'US')
tests/test_thread.py
# -*- coding: utf-8 -*-
import sys
import unittest
import threading
......@@ -8,32 +9,36 @@ from tests.config import COUNTRY_DB_PATH
class TestGeoIPThreading(unittest.TestCase):
def setUp(self):
self.us_ip = '64.233.161.99'
self.gb_ip = '212.58.253.68'
self.us_code = 'US'
self.gb_code = 'GB'
self.gi = pygeoip.GeoIP(COUNTRY_DB_PATH)
def testCountryDatabase(self):
us_thread = TestThread('us', self.gi, self.us_ip, self.us_code, self.assertEqual)
gb_thread = TestThread('gb', self.gi, self.us_ip, self.us_code, self.assertEqual)
us_thread.start()
gb_thread.start()
us_thread.join()
gb_thread.join()
t1 = TestThread('us', '64.17.254.216', 'US', self)
t2 = TestThread('it', '78.26.70.208', 'IT', self)
t1.start()
t2.start()
t1.join()
t2.join()
if t1.exc_info:
raise t1.exc_info[1]
if t2.exc_info:
raise t2.exc_info[1]
class TestThread(threading.Thread):
def __init__(self, name, gi, ip, code, assertEqual):
def __init__(self, name, ip, code, test):
threading.Thread.__init__(self, name=name)
self.ip = ip
self.gi = gi
self.code = code
self.assertEqual = assertEqual
self.test = test
self.exc_info = None
def run(self):
for i in range(1000):
code = self.gi.country_code_by_addr(self.ip)
self.assertEqual(code, self.code)
try:
for i in range(1000):
code = self.test.gi.country_code_by_addr(self.ip)
self.test.assertEqual(code, self.code)
except AssertionError:
self.exc_info = sys.exc_info()
tests/test_tz.py
......@@ -8,13 +8,7 @@ from tests.config import CITY_DB_PATH
class TestGeoIPTimeZoneFunctions(unittest.TestCase):
def setUp(self):
self.us_hostname = 'google.com'
self.us_ip = '64.233.161.99'
self.gb_hostname = 'bbc.com'
self.gb_ip = '212.58.253.68'
self.gic = pygeoip.GeoIP(CITY_DB_PATH)
self.gi = pygeoip.GeoIP(CITY_DB_PATH)
def testTimeZoneInternals(self):
tz = pygeoip.timezone.time_zone_by_country_and_region('XX')
......@@ -27,15 +21,5 @@ class TestGeoIPTimeZoneFunctions(unittest.TestCase):
self.assertEquals(tz, None)
def testTimeZoneByAddr(self):
us_time_zone = self.gic.time_zone_by_addr(self.us_ip)
gb_time_zone = self.gic.time_zone_by_addr(self.gb_ip)
self.assertEquals(us_time_zone, 'America/Los_Angeles')
self.assertEquals(gb_time_zone, 'Europe/London')
def testTimeZoneByName(self):
us_time_zone = self.gic.time_zone_by_name(self.us_hostname)
gb_time_zone = self.gic.time_zone_by_name(self.gb_hostname)
self.assertEquals(us_time_zone, 'America/Los_Angeles')
self.assertEquals(gb_time_zone, 'Europe/London')
tz = self.gi.time_zone_by_addr('64.17.254.216')
self.assertEquals(tz, 'America/Los_Angeles')
tests/test_unicode.py
......@@ -2,7 +2,7 @@
import unittest
import pygeoip
from pygeoip.const import PY2, PY3
from pygeoip.const import PY2
from tests.config import CITY_DB_PATH
......
tox.ini
[tox]
envlist = py25,py26,py27,py32,py33
[testenv:py25]
setenv=PIP_INSECURE=1
envlist = py26,py27,py32,py33
[testenv]
deps = nose
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment