[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[tor-commits] [ooni-probe/master] Write unittests for geoip web services.
commit e851611afd663f4c58d482d39960b2dae6ac7de6
Author: Arturo Filastò <art@xxxxxxxxx>
Date: Fri Mar 15 15:02:04 2013 +0100
Write unittests for geoip web services.
Fix some bugs found in unittesting.
---
ooni/geoip.py | 26 ++++++++----
tests/test_geoip.py | 109 +++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 126 insertions(+), 9 deletions(-)
diff --git a/ooni/geoip.py b/ooni/geoip.py
index 0fe8a6b..55c4fa4 100644
--- a/ooni/geoip.py
+++ b/ooni/geoip.py
@@ -1,5 +1,7 @@
+import xml.etree.ElementTree as ET
import re
import os
+import random
from twisted.web import client, http_headers
from ooni.utils.net import userAgents, BodyReceiver
@@ -42,6 +44,11 @@ def IPToLocation(ipaddr):
class HTTPGeoIPLookupper(object):
url = None
+ _agent = client.Agent
+
+ def __init__(self):
+ self.agent = self._agent(reactor)
+
def _response(self, response):
content_length = response.headers.getRawHeaders('content-length')
@@ -59,16 +66,15 @@ class HTTPGeoIPLookupper(object):
pass
def failed(self, failure):
- log.err("Failed to lookup via %s" % url)
+ log.err("Failed to lookup via %s" % self.url)
log.exception(failure)
return failure
def lookup(self):
- agent = client.Agent(reactor)
headers = {}
headers['User-Agent'] = [random.choice(userAgents)]
- d = agent.request("GET", self.url, http_headers.Headers(headers))
+ d = self.agent.request("GET", self.url, http_headers.Headers(headers))
d.addCallback(self._response)
d.addErrback(self.failed)
return d
@@ -101,7 +107,7 @@ class ProbeIP(object):
strategy = None
geoIPServices = {'ubuntu': UbuntuGeoIP,
'torproject': TorProjectGeoIP,
- 'maximind': MaxMindGeoIP
+ 'maxmind': MaxMindGeoIP
}
address = None
@@ -126,20 +132,22 @@ class ProbeIP(object):
try:
yield self.askGeoIPService()
defer.returnValue(self.address)
- except Exception, e:
- print e
+ except:
log.msg("Unable to lookup the probe IP via GeoIPService")
@defer.inlineCallbacks
def askGeoIPService(self):
- for service_name, service in self.geoIPServices.items():
- s = TorProjectGeoIP()
+ # Shuffle the order in which we test the geoip services.
+ services = self.geoIPServices.items()
+ random.shuffle(services)
+ for service_name, service in services:
+ s = service()
log.msg("Looking up your IP address via %s" % service_name)
try:
self.address = yield s.lookup()
self.strategy = 'geo_ip_service-' + service_name
break
- except:
+ except Exception, e:
log.msg("Failed to lookup your IP via %s" % service_name)
def askTraceroute(self):
diff --git a/tests/test_geoip.py b/tests/test_geoip.py
new file mode 100644
index 0000000..a4ad767
--- /dev/null
+++ b/tests/test_geoip.py
@@ -0,0 +1,109 @@
+from twisted.web import server, static, resource
+from twisted.internet import reactor
+from twisted.trial import unittest
+from twisted.python.filepath import FilePath
+from twisted.protocols.policies import WrappingFactory
+
+from ooni.geoip import ProbeIP, MaxMindGeoIP, TorProjectGeoIP
+from ooni.geoip import UbuntuGeoIP, HTTPGeoIPLookupper, IPToLocation
+
+class UbuntuGeoIPResource(resource.Resource):
+ def render(self, request):
+ return """
+ <Response>
+ <Ip>127.0.0.1</Ip>
+ <Spam>Ham</Spam>
+ </Response>
+ """
+
+class MaxMindGeoIPResource(resource.Resource):
+ def render(self, request):
+ return """
+ <span id="my-ip-address">127.0.0.1</span>
+ """
+
+class TorProjectGeoIPResource(resource.Resource):
+ def render(self, request):
+
+ return """
+ Your IP address appears to be: <b>127.0.0.1</b>
+ """
+
+class GeoIPBaseTest(unittest.TestCase):
+ services = {'ubuntu': UbuntuGeoIPResource,
+ 'maxmind': MaxMindGeoIPResource,
+ 'torproject': TorProjectGeoIPResource
+ }
+ def _listen(self, site):
+ return reactor.listenTCP(0, site, interface="127.0.0.1")
+
+ def setUp(self):
+ r = resource.Resource()
+ for name, service in self.services.items():
+ r.putChild(name, service())
+ self.site = server.Site(r, timeout=None)
+
+ self.wrapper = WrappingFactory(self.site)
+ self.port = self._listen(self.wrapper)
+ self.portno = self.port.getHost().port
+
+ def getUrl(self, service_name):
+ return "http://%s:%s/%s"; % ('127.0.0.1', self.portno, service_name)
+
+ def tearDown(self):
+ return self.port.stopListening()
+
+class TestGeoIPServices(GeoIPBaseTest):
+ def test_torproject_geoip(self):
+ gip = TorProjectGeoIP()
+ gip.url = self.getUrl('torproject')
+ d = gip.lookup()
+ @d.addCallback
+ def cb(res):
+ self.assertEqual(res, '127.0.0.1')
+ return d
+
+ def test_ubuntu_geoip(self):
+ gip = UbuntuGeoIP()
+ gip.url = self.getUrl('ubuntu')
+ d = gip.lookup()
+ @d.addCallback
+ def cb(res):
+ self.assertEqual(res, '127.0.0.1')
+ return d
+
+ def test_maxmind_geoip(self):
+ gip = MaxMindGeoIP()
+ gip.url = self.getUrl('maxmind')
+ d = gip.lookup()
+ @d.addCallback
+ def cb(res):
+ self.assertEqual(res, '127.0.0.1')
+ return d
+
+class TestProbeIP(GeoIPBaseTest):
+ def setUp(self):
+ GeoIPBaseTest.setUp(self)
+
+ # Override the service addresses with those of the fake localhost
+ # resource.
+ self.probe_ip = ProbeIP()
+ for name in self.probe_ip.geoIPServices.keys():
+ self.probe_ip.geoIPServices[name].url = self.getUrl(name)
+
+ def test_ask_geoip_service(self):
+ d = self.probe_ip.askGeoIPService()
+ @d.addCallback
+ def cb(res):
+ self.assertEqual(self.probe_ip.address, '127.0.0.1')
+ return d
+
+ def test_ask_traceroute_service(self):
+ pass
+
+ def test_ask_tor(self):
+ pass
+
+class TestIPToLocation(unittest.TestCase):
+ pass
+
_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits