CernVM-FS  2.9.0
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
test_cvmfs_geo.py
Go to the documentation of this file.
1 from __future__ import print_function
2 import unittest
3 import socket
4 
5 import cvmfs_geo
6 from cvmfs_geo import distance_on_unit_sphere
7 from cvmfs_geo import addr_geoinfo
8 from cvmfs_geo import name_geoinfo
9 from cvmfs_geo import geosort_servers
10 
11 ###
12 # Simulate a small geo IP database, since we can't always
13 # expect a full one to be available. IPv4 addresses are
14 # always preferred, so for those with IPv6 use only IPv6.
15 
16 def getaddrs(name, type):
17  addrs = []
18  for info in socket.getaddrinfo(name,80,0,0,socket.IPPROTO_TCP):
19  if info[0] == type:
20  addrs.append(info[4][0])
21  return addrs
22 
23 CERNgeo = {
24  'latitude': 46.2324,
25  'longitude': 6.0502
26 }
27 CERNname = 'cvmfs-stratum-one.cern.ch'
28 CERNaddrs = getaddrs(CERNname, socket.AF_INET6)
29 if len(CERNaddrs) == 0: # fallback on IPv4-only systems
30  CERNaddrs = getaddrs(CERNname, socket.AF_INET)
31 
32 FNALgeo = {
33  'latitude': 41.7768,
34  'longitude': -88.4604
35 }
36 FNALname = 'cvmfs.fnal.gov'
37 FNALaddrs = getaddrs(FNALname, socket.AF_INET)
38 IHEPgeo = {
39  'latitude': 39.9289,
40  'longitude': 116.3883
41 }
42 IHEPname = 'cvmfs-stratum-one.ihep.ac.cn'
43 IHEPaddrs = getaddrs(IHEPname, socket.AF_INET)
44 RALgeo = {
45  'latitude': 51.75,
46  'longitude': -1.25
47 }
48 RALname = 'cernvmfs.gridpp.rl.ac.uk'
49 RALaddrs = getaddrs(RALname, socket.AF_INET6)
50 if len(RALaddrs) == 0: # fallback on IPv4-only systems
51  RALaddrs = getaddrs(RALname, socket.AF_INET)
52 
53 
54 class giTestDb():
55  def get(self, addr):
56  answer = None
57  if addr in FNALaddrs:
58  answer = FNALgeo
59  elif addr in IHEPaddrs:
60  answer = IHEPgeo
61  elif addr in CERNaddrs:
62  answer = CERNgeo
63  elif addr in RALaddrs:
64  answer = RALgeo
65  else:
66  return None
67 
68  return {'location' : answer}
69 
70 cvmfs_geo.gireader = giTestDb()
71 
72 ####
73 
74 class GeoTest(unittest.TestCase):
75  def setUp(self):
76  pass
77 
78  def tearDown(self):
79  pass
80 
81  def test1Distance(self):
82  self.assertEqual(0.0, distance_on_unit_sphere(0, 0, 0, 0))
83  self.assertAlmostEqual(1.11458455,
84  distance_on_unit_sphere(FNALgeo['latitude'], FNALgeo['longitude'],
85  CERNgeo['latitude'], CERNgeo['longitude']))
86  self.assertAlmostEqual(1.11458455,
87  distance_on_unit_sphere(CERNgeo['latitude'], CERNgeo['longitude'],
88  FNALgeo['latitude'], FNALgeo['longitude']))
89  self.assertAlmostEqual(1.6622382,
90  distance_on_unit_sphere(IHEPgeo['latitude'], IHEPgeo['longitude'],
91  FNALgeo['latitude'], FNALgeo['longitude']))
92  self.assertAlmostEqual(0.1274021,
93  distance_on_unit_sphere(CERNgeo['latitude'], CERNgeo['longitude'],
94  RALgeo['latitude'], RALgeo['longitude']))
95  self.assertAlmostEqual(1.2830254,
96  distance_on_unit_sphere(IHEPgeo['latitude'], IHEPgeo['longitude'],
97  RALgeo['latitude'], RALgeo['longitude']))
98  # surprisingly, CERN is slightly further from IHEP than RAL
99  self.assertAlmostEqual(1.2878979,
100  distance_on_unit_sphere(IHEPgeo['latitude'], IHEPgeo['longitude'],
101  CERNgeo['latitude'], CERNgeo['longitude']))
102 
103  def test2AddrGeoinfo(self):
104  now = 0
105  self.assertEqual(CERNgeo, addr_geoinfo(now, CERNaddrs[0]))
106  self.assertEqual(FNALgeo, addr_geoinfo(now, FNALaddrs[0]))
107  self.assertEqual(IHEPgeo, addr_geoinfo(now, IHEPaddrs[0]))
108  self.assertEqual(RALgeo, addr_geoinfo(now, RALaddrs[0]))
109 
110  def test3NameGeoinfo(self):
111  self.assertEqual(0, len(cvmfs_geo.geo_cache))
112  now = 0
113  self.assertEqual(CERNgeo, name_geoinfo(now, CERNname))
114  self.assertEqual(FNALgeo, name_geoinfo(now, FNALname))
115  self.assertEqual(IHEPgeo, name_geoinfo(now, IHEPname))
116  self.assertEqual(RALgeo, name_geoinfo(now, RALname))
117  self.assertEqual(4, len(cvmfs_geo.geo_cache))
118 
119  # test the caching, when there's no database available
120  savegireader = cvmfs_geo.gireader
121  cvmfs_geo.gireader = None
122  now = 1
123  self.assertEqual(CERNgeo, name_geoinfo(now, CERNname))
124  self.assertEqual(FNALgeo, name_geoinfo(now, FNALname))
125  self.assertEqual(IHEPgeo, name_geoinfo(now, IHEPname))
126  self.assertEqual(RALgeo, name_geoinfo(now, RALname))
127  cvmfs_geo.gireader = savegireader
128 
130  self.assertEqual([True, [3, 0, 1, 2]],
131  geosort_servers(0, RALgeo, [CERNname, FNALname, IHEPname, RALname]))
132  self.assertEqual([True, [0, 3, 2, 1]],
133  geosort_servers(0, RALgeo, [RALname, IHEPname, FNALname, CERNname]))
134  self.assertEqual([True, [1, 0, 3, 2]],
135  geosort_servers(0, IHEPgeo, [RALname, IHEPname, FNALname, CERNname]))
136  self.assertEqual([True, [2, 3, 0, 1]],
137  geosort_servers(0, CERNgeo, [FNALname, IHEPname, CERNname, RALname]))
138  self.assertEqual([True, [3, 2, 1, 0]],
139  geosort_servers(0, FNALgeo, [IHEPname, CERNname, RALname, FNALname]))
140 
141 
142 if __name__ == '__main__':
143  unittest.main()
def geosort_servers
Definition: cvmfs_geo.py:193
def distance_on_unit_sphere
Definition: cvmfs_geo.py:74
def getaddrs
Simulate a small geo IP database, since we can't always expect a full one to be available.
def addr_geoinfo
Definition: cvmfs_geo.py:116
def name_geoinfo
Definition: cvmfs_geo.py:131