[tor-commits] [ooni-probe/develop] Write unittests for geoip web services.

isis at torproject.org isis at torproject.org
Thu Jun 6 16:41:37 UTC 2013


commit e851611afd663f4c58d482d39960b2dae6ac7de6
Author: Arturo Filastò <art at fuffa.org>
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
+





More information about the tor-commits mailing list