1
0
mirror of https://github.com/django/django.git synced 2024-12-22 17:16:24 +00:00

Fixed #34305 -- Changed GeoIP2 tests to use MaxMind test databases.

GEOIP_SETTINGS is removed from the global scope as this prevents
modifications to the settings using @override_settings in tests.

Additional improvements now that we have stable test databases include:

- Made testing more comprehensive and improved coverage
- Patched socket.gethostbyname() for whole test case
- Added testing of non-free GeoIP2 databases

Co-authored-by: Tom Forbes <tom@tomforb.es>
This commit is contained in:
Nick Pope 2021-03-30 11:52:12 +01:00 committed by Mariusz Felisiak
parent ed4f83782d
commit a93375e8ab
9 changed files with 159 additions and 129 deletions

View File

@ -25,19 +25,12 @@ __all__ = ["HAS_GEOIP2"]
try:
import geoip2.database
except ImportError:
except ImportError: # pragma: no cover
HAS_GEOIP2 = False
else:
HAS_GEOIP2 = True
__all__ += ["GeoIP2", "GeoIP2Exception"]
# Creating the settings dictionary with any settings, if needed.
GEOIP_SETTINGS = {
"GEOIP_PATH": getattr(settings, "GEOIP_PATH", None),
"GEOIP_CITY": getattr(settings, "GEOIP_CITY", "GeoLite2-City.mmdb"),
"GEOIP_COUNTRY": getattr(settings, "GEOIP_COUNTRY", "GeoLite2-Country.mmdb"),
}
class GeoIP2Exception(Exception):
pass
@ -95,7 +88,7 @@ class GeoIP2:
raise GeoIP2Exception("Invalid GeoIP caching option: %s" % cache)
# Getting the GeoIP data path.
path = path or GEOIP_SETTINGS["GEOIP_PATH"]
path = path or getattr(settings, "GEOIP_PATH", None)
if not path:
raise GeoIP2Exception(
"GeoIP path must be provided via parameter or the GEOIP_PATH setting."
@ -106,12 +99,16 @@ class GeoIP2:
# Constructing the GeoIP database filenames using the settings
# dictionary. If the database files for the GeoLite country
# and/or city datasets exist, then try to open them.
country_db = path / (country or GEOIP_SETTINGS["GEOIP_COUNTRY"])
country_db = path / (
country or getattr(settings, "GEOIP_COUNTRY", "GeoLite2-Country.mmdb")
)
if country_db.is_file():
self._country = geoip2.database.Reader(str(country_db), mode=cache)
self._country_file = country_db
city_db = path / (city or GEOIP_SETTINGS["GEOIP_CITY"])
city_db = path / (
city or getattr(settings, "GEOIP_CITY", "GeoLite2-City.mmdb")
)
if city_db.is_file():
self._city = geoip2.database.Reader(str(city_db), mode=cache)
self._city_file = city_db

Binary file not shown.

After

Width:  |  Height:  |  Size: 22 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 19 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 12 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 21 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 18 KiB

View File

@ -0,0 +1,4 @@
This work is licensed under the Creative Commons Attribution-ShareAlike 3.0
Unported License. To view a copy of this license, visit
http://creativecommons.org/licenses/by-sa/3.0/ or send a letter to Creative
Commons, 444 Castro Street, Suite 900, Mountain View, California, 94041, USA.

View File

@ -0,0 +1,3 @@
These test databases are taken from the following repository:
https://github.com/maxmind/MaxMind-DB/

View File

@ -1,50 +1,59 @@
import os
import itertools
import pathlib
from unittest import mock, skipUnless
from django.conf import settings
from django.contrib.gis.geoip2 import HAS_GEOIP2
from django.contrib.gis.geos import GEOSGeometry
from django.test import SimpleTestCase
from django.test import SimpleTestCase, override_settings
from django.utils.deprecation import RemovedInDjango60Warning
if HAS_GEOIP2:
import geoip2
from django.contrib.gis.geoip2 import GeoIP2, GeoIP2Exception
# Note: Requires both the GeoIP country and city datasets.
# The GEOIP_DATA path should be the only setting set (the directory
# should contain links or the actual database files 'GeoLite2-City.mmdb' and
# 'GeoLite2-City.mmdb'.
@skipUnless(
HAS_GEOIP2 and getattr(settings, "GEOIP_PATH", None),
"GeoIP is required along with the GEOIP_PATH setting.",
def build_geoip_path(*parts):
return pathlib.Path(__file__).parent.joinpath("data/geoip2", *parts).resolve()
@skipUnless(HAS_GEOIP2, "GeoIP2 is required.")
@override_settings(
GEOIP_CITY="GeoLite2-City-Test.mmdb",
GEOIP_COUNTRY="GeoLite2-Country-Test.mmdb",
)
class GeoIPTest(SimpleTestCase):
addr = "129.237.192.1"
fqdn = "ku.edu"
class GeoLite2Test(SimpleTestCase):
fqdn = "sky.uk"
ipv4 = "2.125.160.216"
ipv6 = "::ffff:027d:a0d8"
def test01_init(self):
"GeoIP initialization."
g1 = GeoIP2() # Everything inferred from GeoIP path
path = settings.GEOIP_PATH
g2 = GeoIP2(path, 0) # Passing in data path explicitly.
# path accepts str and pathlib.Path.
if isinstance(path, str):
g3 = GeoIP2(pathlib.Path(path))
else:
g3 = GeoIP2(str(path))
@classmethod
def setUpClass(cls):
# Avoid referencing __file__ at module level.
cls.enterClassContext(override_settings(GEOIP_PATH=build_geoip_path()))
# Always mock host lookup to avoid test breakage if DNS changes.
cls.enterClassContext(mock.patch("socket.gethostbyname", return_value=cls.ipv4))
super().setUpClass()
def test_init(self):
# Everything inferred from GeoIP path.
g1 = GeoIP2()
# Path passed explicitly.
g2 = GeoIP2(settings.GEOIP_PATH, GeoIP2.MODE_AUTO)
# Path provided as a string.
g3 = GeoIP2(str(settings.GEOIP_PATH))
for g in (g1, g2, g3):
self.assertTrue(g._country)
self.assertTrue(g._city)
# Only passing in the location of one database.
city = os.path.join(path, "GeoLite2-City.mmdb")
cntry = os.path.join(path, "GeoLite2-Country.mmdb")
g4 = GeoIP2(city, country="")
g4 = GeoIP2(settings.GEOIP_PATH / settings.GEOIP_CITY, country="")
self.assertTrue(g4._city)
self.assertIsNone(g4._country)
g5 = GeoIP2(cntry, city="")
g5 = GeoIP2(settings.GEOIP_PATH / settings.GEOIP_COUNTRY, city="")
self.assertTrue(g5._country)
self.assertIsNone(g5._city)
# Improper parameters.
@ -57,99 +66,96 @@ class GeoIPTest(SimpleTestCase):
else:
e = TypeError
with self.assertRaises(e):
GeoIP2(bad, 0)
GeoIP2(bad, GeoIP2.MODE_AUTO)
def test_no_database_file(self):
invalid_path = os.path.join(os.path.dirname(__file__), "data")
msg = "Could not load a database from %s." % invalid_path
invalid_path = pathlib.Path(__file__).parent.joinpath("data/invalid").resolve()
msg = f"Could not load a database from {invalid_path}."
with self.assertRaisesMessage(GeoIP2Exception, msg):
GeoIP2(invalid_path)
def test02_bad_query(self):
"GeoIP query parameter checking."
cntry_g = GeoIP2(city="<foo>")
# No city database available, these calls should fail.
with self.assertRaises(GeoIP2Exception):
cntry_g.city("tmc.edu")
def test_bad_query(self):
g = GeoIP2(city="<invalid>")
# Non-string query should raise TypeError
with self.assertRaises(TypeError):
cntry_g.country_code(17)
with self.assertRaises(TypeError):
cntry_g.country_name(GeoIP2)
functions = (g.city, g.geos, g.lat_lon, g.lon_lat)
msg = "Invalid GeoIP city data file: "
for function in functions:
with self.subTest(function=function.__qualname__):
with self.assertRaisesMessage(GeoIP2Exception, msg):
function("example.com")
@mock.patch("socket.gethostbyname")
def test03_country(self, gethostbyname):
"GeoIP country querying methods."
gethostbyname.return_value = "128.249.1.1"
g = GeoIP2(city="<foo>")
functions += (g.country, g.country_code, g.country_name)
values = (123, 123.45, b"", (), [], {}, set(), frozenset(), GeoIP2)
msg = "GeoIP query must be a string, not type"
for function, value in itertools.product(functions, values):
with self.subTest(function=function.__qualname__, type=type(value)):
with self.assertRaisesMessage(TypeError, msg):
function(value)
for query in (self.fqdn, self.addr):
self.assertEqual(
"US",
g.country_code(query),
"Failed for func country_code and query %s" % query,
)
self.assertEqual(
"United States",
g.country_name(query),
"Failed for func country_name and query %s" % query,
)
self.assertEqual(
{"country_code": "US", "country_name": "United States"},
g.country(query),
)
def test_country(self):
g = GeoIP2(city="<invalid>")
for query in (self.fqdn, self.ipv4, self.ipv6):
with self.subTest(query=query):
self.assertEqual(
g.country(query),
{
"country_code": "GB",
"country_name": "United Kingdom",
},
)
self.assertEqual(g.country_code(query), "GB")
self.assertEqual(g.country_name(query), "United Kingdom")
@mock.patch("socket.gethostbyname")
def test04_city(self, gethostbyname):
"GeoIP city querying methods."
gethostbyname.return_value = "129.237.192.1"
g = GeoIP2(country="<foo>")
def test_city(self):
g = GeoIP2(country="<invalid>")
for query in (self.fqdn, self.ipv4, self.ipv6):
with self.subTest(query=query):
self.assertEqual(
g.city(query),
{
"city": "Boxford",
"continent_code": "EU",
"continent_name": "Europe",
"country_code": "GB",
"country_name": "United Kingdom",
"dma_code": None,
"is_in_european_union": False,
"latitude": 51.75,
"longitude": -1.25,
"postal_code": "OX1",
"region": "ENG",
"time_zone": "Europe/London",
},
)
for query in (self.fqdn, self.addr):
# Country queries should still work.
self.assertEqual(
"US",
g.country_code(query),
"Failed for func country_code and query %s" % query,
)
self.assertEqual(
"United States",
g.country_name(query),
"Failed for func country_name and query %s" % query,
)
self.assertEqual(
{"country_code": "US", "country_name": "United States"},
g.country(query),
)
geom = g.geos(query)
self.assertIsInstance(geom, GEOSGeometry)
self.assertEqual(geom.srid, 4326)
self.assertEqual(geom.tuple, (-1.25, 51.75))
# City information dictionary.
d = g.city(query)
self.assertEqual("NA", d["continent_code"])
self.assertEqual("North America", d["continent_name"])
self.assertEqual("US", d["country_code"])
self.assertEqual("Lawrence", d["city"])
self.assertEqual("KS", d["region"])
self.assertEqual("America/Chicago", d["time_zone"])
self.assertFalse(d["is_in_european_union"])
geom = g.geos(query)
self.assertIsInstance(geom, GEOSGeometry)
self.assertEqual(g.lat_lon(query), (51.75, -1.25))
self.assertEqual(g.lon_lat(query), (-1.25, 51.75))
# Country queries should still work.
self.assertEqual(
g.country(query),
{
"country_code": "GB",
"country_name": "United Kingdom",
},
)
self.assertEqual(g.country_code(query), "GB")
self.assertEqual(g.country_name(query), "United Kingdom")
for e1, e2 in (
geom.tuple,
g.lon_lat(query),
g.lat_lon(query),
):
self.assertIsInstance(e1, float)
self.assertIsInstance(e2, float)
def test06_ipv6_query(self):
"GeoIP can lookup IPv6 addresses."
g = GeoIP2()
d = g.city("2002:81ed:c9a5::81ed:c9a5") # IPv6 address for www.nhm.ku.edu
self.assertEqual("US", d["country_code"])
self.assertEqual("Lawrence", d["city"])
self.assertEqual("KS", d["region"])
def test_not_found(self):
g1 = GeoIP2(city="<invalid>")
g2 = GeoIP2(country="<invalid>")
for function, query in itertools.product(
(g1.country, g2.city), ("127.0.0.1", "::1")
):
with self.subTest(function=function.__qualname__, query=query):
msg = f"The address {query} is not in the database."
with self.assertRaisesMessage(geoip2.errors.AddressNotFoundError, msg):
function(query)
def test_del(self):
g = GeoIP2()
@ -162,8 +168,7 @@ class GeoIPTest(SimpleTestCase):
self.assertIs(country._db_reader.closed, True)
def test_repr(self):
path = settings.GEOIP_PATH
g = GeoIP2(path=path)
g = GeoIP2()
meta = g._reader.metadata()
version = "%s.%s" % (
meta.binary_format_major_version,
@ -181,26 +186,47 @@ class GeoIPTest(SimpleTestCase):
)
self.assertEqual(repr(g), expected)
@mock.patch("socket.gethostbyname", return_value="expected")
def test_check_query(self, gethostbyname):
def test_check_query(self):
g = GeoIP2()
self.assertEqual(g._check_query("127.0.0.1"), "127.0.0.1")
self.assertEqual(
g._check_query("2002:81ed:c9a5::81ed:c9a5"), "2002:81ed:c9a5::81ed:c9a5"
)
self.assertEqual(g._check_query("invalid-ip-address"), "expected")
self.assertEqual(g._check_query(self.ipv4), self.ipv4)
self.assertEqual(g._check_query(self.ipv6), self.ipv6)
self.assertEqual(g._check_query(self.fqdn), self.ipv4)
def test_coords_deprecation_warning(self):
g = GeoIP2()
msg = "GeoIP2.coords() is deprecated. Use GeoIP2.lon_lat() instead."
with self.assertWarnsMessage(RemovedInDjango60Warning, msg):
e1, e2 = g.coords(self.fqdn)
e1, e2 = g.coords(self.ipv4)
self.assertIsInstance(e1, float)
self.assertIsInstance(e2, float)
def test_open_deprecation_warning(self):
msg = "GeoIP2.open() is deprecated. Use GeoIP2() instead."
with self.assertWarnsMessage(RemovedInDjango60Warning, msg):
g = GeoIP2.open(settings.GEOIP_PATH, 0)
g = GeoIP2.open(settings.GEOIP_PATH, GeoIP2.MODE_AUTO)
self.assertTrue(g._country)
self.assertTrue(g._city)
@skipUnless(HAS_GEOIP2, "GeoIP2 is required.")
@override_settings(
GEOIP_CITY="GeoIP2-City-Test.mmdb",
GEOIP_COUNTRY="GeoIP2-Country-Test.mmdb",
)
class GeoIP2Test(GeoLite2Test):
"""Non-free GeoIP2 databases are supported."""
@skipUnless(HAS_GEOIP2, "GeoIP2 is required.")
class ErrorTest(SimpleTestCase):
def test_missing_path(self):
msg = "GeoIP path must be provided via parameter or the GEOIP_PATH setting."
with self.settings(GEOIP_PATH=None):
with self.assertRaisesMessage(GeoIP2Exception, msg):
GeoIP2()
def test_unsupported_database(self):
msg = "Unable to recognize database edition: GeoLite2-ASN"
with self.settings(GEOIP_PATH=build_geoip_path("GeoLite2-ASN-Test.mmdb")):
with self.assertRaisesMessage(GeoIP2Exception, msg):
GeoIP2()