Switch to pygeoip for log processing to allow running it via PyPy

sitescripts/stats/test/logprocessor.py

 # coding: utf-8
 
 # This file is part of the Adblock Plus web scripts,
 # Copyright (C) 2006-2013 Eyeo GmbH
 #
 # Adblock Plus is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 3 as
 # published by the Free Software Foundation.
 #
 # Adblock Plus is distributed in the hope that it will be useful,
@@ skipping 48 matching lines @@
 
   def test_ipprocessing(self):
     country = None
     class FakeGeo(object):
       ip_checked = None
       def country_code_by_addr(self, ip):
         self.ip_checked = ip
         return country
 
     tests = [
-      ("1.2.3.4", "xy", "1.2.3.4", "xy"),
-      ("::ffff:1.2.3.4", "xy", "1.2.3.4", "xy"),
-      ("1.2.3.4", "--", "1.2.3.4", "unknown"),
-      ("1.2.3.4", "", "1.2.3.4", "unknown"),
-      ("::ffff:1.2.3.4", None, "1.2.3.4", "unknown"),
+      ("1.2.3.4", "xy", "1.2.3.4", "v4", "xy"),
+      ("::ffff:1.2.3.4", "xy", "1.2.3.4", "v4", "xy"),
+      ("1.2.3.4", "--", "1.2.3.4", "v4", "unknown"),
+      ("1.2.3.4", "", "1.2.3.4", "v4", "unknown"),
+      ("::ffff:1.2.3.4", None, "1.2.3.4", "v4", "unknown"),
+      ("::1", "xy", "::1", "v6", "xy"),
+      ("FE80:0000:0000:0000:0202:B3FF:FE1E:8329", None, "FE80:0000:0000:0000:0202:B3FF:FE1E:8329", "v6", "unknown"),
     ]
-    for ip, country, expected_ip, expected_country in tests:
+    for ip, country, expected_ip, expected_type, expected_country in tests:
       fake_geo = FakeGeo()
-      self.assertEqual(logprocessor.process_ip(ip, fake_geo), (expected_ip, expected_country), "Processing IP '%s'" % ip)
-      self.assertEqual(fake_geo.ip_checked, expected_ip, "GeoIP check for IP '%s'" % ip)
+      fake_geov6 = FakeGeo()
+      self.assertEqual(logprocessor.process_ip(ip, fake_geo, fake_geov6), (expected_ip, expected_country), "Processing IP '%s'" % ip)
+      if expected_type == "v4":
+        self.assertEqual(fake_geo.ip_checked, expected_ip, "GeoIP check for IP '%s'" % ip)
+        self.assertEqual(fake_geov6.ip_checked, None, "GeoIPv6 check for IP '%s'" % ip)
+      else:
+        self.assertEqual(fake_geo.ip_checked, None, "GeoIP check for IP '%s'" % ip)
+        self.assertEqual(fake_geov6.ip_checked, expected_ip, "GeoIPv6 check for IP '%s'" % ip)
 
   def test_timeparsing(self):
     tests = [
       ("31/Jul/2013:12:03:37", 0, 0, datetime(2013, 07, 31, 12, 03, 37), "201307"),
       ("31/Jul/2013:12:03:37", 5, 0, datetime(2013, 07, 31, 7, 03, 37), "201307"),
       ("31/Jul/2013:12:03:37", -5, 0, datetime(2013, 07, 31, 17, 03, 37), "201307"),
       ("31/Jul/2013:12:03:37", 5, 30, datetime(2013, 07, 31, 6, 33, 37), "201307"),
       ("31/Jul/2013:12:03:37", -5, 30, datetime(2013, 07, 31, 17, 33, 37), "201307"),
     ]
     for timestr, tz_hours, tz_minutes, expected_time, expected_month in tests:
@@ skipping 481 matching lines @@
           "platform": "chromium",
           "platformVersion": "28.0",
           "fullPlatform": "chromium 28.0",
           "downloadInterval": "1 day(s)",
           "firstInDay": True,
         }
       ),
     ]
     for line, expected_record in tests:
       logprocessor.mirror_name = "mirror123"
-      self.assertEqual(logprocessor.parse_record(line, set(), FakeGeo()), expected_record, "Parsing log line '%s'" % line)
+      self.assertEqual(logprocessor.parse_record(line, set(), FakeGeo(), FakeGeo()), expected_record, "Parsing log line '%s'" % line)
 
   def test_record_adding(self):
     tests = [
       (
         {"size": 200},
         {},
         (),
         {"hits": 1, "bandwidth": 200},
       ),
       (
@@ skipping 40 matching lines @@
           "addonName": {"bar": {"hits": 1, "bandwidth": 200}}
         },
       ),
     ]
     for info, section, ignored_fields, expected_result in tests:
       logprocessor.add_record(info, section, ignored_fields)
       self.assertEqual(section, expected_result)
 
 if __name__ == '__main__':
   unittest.main()