python-eventlet/tests/greendns_test.py

   1 # coding: utf-8
   2 """Tests for the eventlet.support.greendns module"""
   3
   4 import os
   5 import socket
   6 import tempfile
   7 import time
   8
   9 import tests
  10 from tests import mock
  11 try:
  12     import dns.rdatatype
  13     import dns.rdtypes.IN.A
  14     import dns.rdtypes.IN.AAAA
  15     import dns.resolver
  16     import dns.reversename
  17     import dns.rrset
  18     from eventlet.support import greendns
  19     greendns_available = True
  20 except ImportError:
  21     greendns_available = False
  22     greendns = mock.Mock()
  23
  24
  25 def greendns_requirement(_f):
  26     """We want to skip tests if greendns is not installed.
  27     """
  28     return greendns_available
  29
  30
  31 class TestHostsResolver(tests.LimitedTestCase):
  32
  33     def _make_host_resolver(self):
  34         """Returns a HostResolver instance
  35
  36         The hosts file will be empty but accessible as a py.path.local
  37         instance using the ``hosts`` attribute.
  38         """
  39         hosts = tempfile.NamedTemporaryFile()
  40         hr = greendns.HostsResolver(fname=hosts.name)
  41         hr.hosts = hosts
  42         hr._last_stat = 0
  43         return hr
  44
  45     @tests.skip_unless(greendns_requirement)
  46     def test_default_fname(self):
  47         hr = greendns.HostsResolver()
  48         assert os.path.exists(hr.fname)
  49
  50     @tests.skip_unless(greendns_requirement)
  51     def test_readlines_lines(self):
  52         hr = self._make_host_resolver()
  53         hr.hosts.write(b'line0\n')
  54         hr.hosts.flush()
  55         assert hr._readlines() == ['line0']
  56         hr._last_stat = 0
  57         hr.hosts.write(b'line1\n')
  58         hr.hosts.flush()
  59         assert hr._readlines() == ['line0', 'line1']
  60         hr._last_stat = 0
  61         hr.hosts.write(b'#comment0\nline0\n #comment1\nline1')
  62         assert hr._readlines() == ['line0', 'line1']
  63
  64     @tests.skip_unless(greendns_requirement)
  65     def test_readlines_missing_file(self):
  66         hr = self._make_host_resolver()
  67         hr.hosts.close()
  68         hr._last_stat = 0
  69         assert hr._readlines() == []
  70
  71     @tests.skip_unless(greendns_requirement)
  72     def test_load_no_contents(self):
  73         hr = self._make_host_resolver()
  74         hr._load()
  75         assert not hr._v4
  76         assert not hr._v6
  77         assert not hr._aliases
  78
  79     @tests.skip_unless(greendns_requirement)
  80     def test_load_v4_v6_cname_aliases(self):
  81         hr = self._make_host_resolver()
  82         hr.hosts.write(b'1.2.3.4 v4.example.com v4\n'
  83                        b'dead:beef::1 v6.example.com v6\n')
  84         hr.hosts.flush()
  85         hr._load()
  86         assert hr._v4 == {'v4.example.com': '1.2.3.4', 'v4': '1.2.3.4'}
  87         assert hr._v6 == {'v6.example.com': 'dead:beef::1',
  88                           'v6': 'dead:beef::1'}
  89         assert hr._aliases == {'v4': 'v4.example.com',
  90                                'v6': 'v6.example.com'}
  91
  92     @tests.skip_unless(greendns_requirement)
  93     def test_load_v6_link_local(self):
  94         hr = self._make_host_resolver()
  95         hr.hosts.write(b'fe80:: foo\n'
  96                        b'fe80:dead:beef::1 bar\n')
  97         hr.hosts.flush()
  98         hr._load()
  99         assert not hr._v4
 100         assert not hr._v6
 101
 102     @tests.skip_unless(greendns_requirement)
 103     def test_query_A(self):
 104         hr = self._make_host_resolver()
 105         hr._v4 = {'v4.example.com': '1.2.3.4'}
 106         ans = hr.query('v4.example.com')
 107         assert ans[0].address == '1.2.3.4'
 108
 109     @tests.skip_unless(greendns_requirement)
 110     def test_query_ans_types(self):
 111         # This assumes test_query_A above succeeds
 112         hr = self._make_host_resolver()
 113         hr._v4 = {'v4.example.com': '1.2.3.4'}
 114         hr._last_stat = time.time()
 115         ans = hr.query('v4.example.com')
 116         assert isinstance(ans, greendns.dns.resolver.Answer)
 117         assert ans.response is None
 118         assert ans.qname == dns.name.from_text('v4.example.com')
 119         assert ans.rdtype == dns.rdatatype.A
 120         assert ans.rdclass == dns.rdataclass.IN
 121         assert ans.canonical_name == dns.name.from_text('v4.example.com')
 122         assert ans.expiration
 123         assert isinstance(ans.rrset, dns.rrset.RRset)
 124         assert ans.rrset.rdtype == dns.rdatatype.A
 125         assert ans.rrset.rdclass == dns.rdataclass.IN
 126         ttl = greendns.HOSTS_TTL
 127         assert ttl - 1 <= ans.rrset.ttl <= ttl + 1
 128         rr = ans.rrset[0]
 129         assert isinstance(rr, greendns.dns.rdtypes.IN.A.A)
 130         assert rr.rdtype == dns.rdatatype.A
 131         assert rr.rdclass == dns.rdataclass.IN
 132         assert rr.address == '1.2.3.4'
 133
 134     @tests.skip_unless(greendns_requirement)
 135     def test_query_AAAA(self):
 136         hr = self._make_host_resolver()
 137         hr._v6 = {'v6.example.com': 'dead:beef::1'}
 138         ans = hr.query('v6.example.com', dns.rdatatype.AAAA)
 139         assert ans[0].address == 'dead:beef::1'
 140
 141     @tests.skip_unless(greendns_requirement)
 142     def test_query_unknown_raises(self):
 143         hr = self._make_host_resolver()
 144         with tests.assert_raises(greendns.dns.resolver.NoAnswer):
 145             hr.query('example.com')
 146
 147     @tests.skip_unless(greendns_requirement)
 148     def test_query_unknown_no_raise(self):
 149         hr = self._make_host_resolver()
 150         ans = hr.query('example.com', raise_on_no_answer=False)
 151         assert isinstance(ans, greendns.dns.resolver.Answer)
 152         assert ans.response is None
 153         assert ans.qname == dns.name.from_text('example.com')
 154         assert ans.rdtype == dns.rdatatype.A
 155         assert ans.rdclass == dns.rdataclass.IN
 156         assert ans.canonical_name == dns.name.from_text('example.com')
 157         assert ans.expiration
 158         assert isinstance(ans.rrset, greendns.dns.rrset.RRset)
 159         assert ans.rrset.rdtype == dns.rdatatype.A
 160         assert ans.rrset.rdclass == dns.rdataclass.IN
 161         assert len(ans.rrset) == 0
 162
 163     @tests.skip_unless(greendns_requirement)
 164     def test_query_CNAME(self):
 165         hr = self._make_host_resolver()
 166         hr._aliases = {'host': 'host.example.com'}
 167         ans = hr.query('host', dns.rdatatype.CNAME)
 168         assert ans[0].target == dns.name.from_text('host.example.com')
 169         assert str(ans[0].target) == 'host.example.com.'
 170
 171     @tests.skip_unless(greendns_requirement)
 172     def test_query_unknown_type(self):
 173         hr = self._make_host_resolver()
 174         with tests.assert_raises(greendns.dns.resolver.NoAnswer):
 175             hr.query('example.com', dns.rdatatype.MX)
 176
 177     @tests.skip_unless(greendns_requirement)
 178     def test_getaliases(self):
 179         hr = self._make_host_resolver()
 180         hr._aliases = {'host': 'host.example.com',
 181                        'localhost': 'host.example.com'}
 182         res = set(hr.getaliases('host'))
 183         assert res == set(['host.example.com', 'localhost'])
 184
 185     @tests.skip_unless(greendns_requirement)
 186     def test_getaliases_unknown(self):
 187         hr = self._make_host_resolver()
 188         assert hr.getaliases('host.example.com') == []
 189
 190     @tests.skip_unless(greendns_requirement)
 191     def test_getaliases_fqdn(self):
 192         hr = self._make_host_resolver()
 193         hr._aliases = {'host': 'host.example.com'}
 194         res = set(hr.getaliases('host.example.com'))
 195         assert res == set(['host'])
 196
 197
 198 def _make_mock_base_resolver():
 199     """A mocked base resolver class"""
 200     class RR(object):
 201         pass
 202
 203     class Resolver(object):
 204         aliases = ['cname.example.com']
 205         raises = None
 206         rr = RR()
 207
 208         def query(self, *args, **kwargs):
 209             self.args = args
 210             self.kwargs = kwargs
 211             if self.raises:
 212                 raise self.raises()
 213             if hasattr(self, 'rrset'):
 214                 rrset = self.rrset
 215             else:
 216                 rrset = [self.rr]
 217             return greendns.HostsAnswer('foo', 1, 1, rrset, False)
 218
 219         def getaliases(self, *args, **kwargs):
 220             return self.aliases
 221
 222     return Resolver
 223
 224
 225 class TestProxyResolver(tests.LimitedTestCase):
 226
 227     @tests.skip_unless(greendns_requirement)
 228     def test_clear(self):
 229         rp = greendns.ResolverProxy()
 230         resolver = rp._resolver
 231         rp.clear()
 232         assert rp._resolver != resolver
 233
 234     @tests.skip_unless(greendns_requirement)
 235     def _make_mock_hostsresolver(self):
 236         """A mocked HostsResolver"""
 237         base_resolver = _make_mock_base_resolver()
 238         base_resolver.rr.address = '1.2.3.4'
 239         return base_resolver()
 240
 241     @tests.skip_unless(greendns_requirement)
 242     def _make_mock_resolver(self):
 243         """A mocked Resolver"""
 244         base_resolver = _make_mock_base_resolver()
 245         base_resolver.rr.address = '5.6.7.8'
 246         return base_resolver()
 247
 248     @tests.skip_unless(greendns_requirement)
 249     def test_hosts(self):
 250         hostsres = self._make_mock_hostsresolver()
 251         rp = greendns.ResolverProxy(hostsres)
 252         ans = rp.query('host.example.com')
 253         assert ans[0].address == '1.2.3.4'
 254
 255     @tests.skip_unless(greendns_requirement)
 256     def test_hosts_noanswer(self):
 257         hostsres = self._make_mock_hostsresolver()
 258         res = self._make_mock_resolver()
 259         rp = greendns.ResolverProxy(hostsres)
 260         rp._resolver = res
 261         hostsres.raises = greendns.dns.resolver.NoAnswer
 262         ans = rp.query('host.example.com')
 263         assert ans[0].address == '5.6.7.8'
 264
 265     @tests.skip_unless(greendns_requirement)
 266     def test_resolver(self):
 267         res = self._make_mock_resolver()
 268         rp = greendns.ResolverProxy()
 269         rp._resolver = res
 270         ans = rp.query('host.example.com')
 271         assert ans[0].address == '5.6.7.8'
 272
 273     @tests.skip_unless(greendns_requirement)
 274     def test_noanswer(self):
 275         res = self._make_mock_resolver()
 276         rp = greendns.ResolverProxy()
 277         rp._resolver = res
 278         res.raises = greendns.dns.resolver.NoAnswer
 279         with tests.assert_raises(greendns.dns.resolver.NoAnswer):
 280             rp.query('host.example.com')
 281
 282     @tests.skip_unless(greendns_requirement)
 283     def test_nxdomain(self):
 284         res = self._make_mock_resolver()
 285         rp = greendns.ResolverProxy()
 286         rp._resolver = res
 287         res.raises = greendns.dns.resolver.NXDOMAIN
 288         with tests.assert_raises(greendns.dns.resolver.NXDOMAIN):
 289             rp.query('host.example.com')
 290
 291     @tests.skip_unless(greendns_requirement)
 292     def test_noanswer_hosts(self):
 293         hostsres = self._make_mock_hostsresolver()
 294         res = self._make_mock_resolver()
 295         rp = greendns.ResolverProxy(hostsres)
 296         rp._resolver = res
 297         hostsres.raises = greendns.dns.resolver.NoAnswer
 298         res.raises = greendns.dns.resolver.NoAnswer
 299         with tests.assert_raises(greendns.dns.resolver.NoAnswer):
 300             rp.query('host.example.com')
 301
 302     def _make_mock_resolver_aliases(self):
 303
 304         class RR(object):
 305             target = 'host.example.com'
 306
 307         class Resolver(object):
 308             call_count = 0
 309             exc_type = greendns.dns.resolver.NoAnswer
 310
 311             def query(self, *args, **kwargs):
 312                 self.args = args
 313                 self.kwargs = kwargs
 314                 self.call_count += 1
 315                 if self.call_count < 2:
 316                     return greendns.HostsAnswer(args[0], 1, 5, [RR()], False)
 317                 else:
 318                     raise self.exc_type()
 319
 320         return Resolver()
 321
 322     @tests.skip_unless(greendns_requirement)
 323     def test_getaliases(self):
 324         aliases_res = self._make_mock_resolver_aliases()
 325         rp = greendns.ResolverProxy()
 326         rp._resolver = aliases_res
 327         aliases = set(rp.getaliases('alias.example.com'))
 328         assert aliases == set(['host.example.com'])
 329
 330     @tests.skip_unless(greendns_requirement)
 331     def test_getaliases_fqdn(self):
 332         aliases_res = self._make_mock_resolver_aliases()
 333         rp = greendns.ResolverProxy()
 334         rp._resolver = aliases_res
 335         rp._resolver.call_count = 1
 336         assert rp.getaliases('host.example.com') == []
 337
 338     @tests.skip_unless(greendns_requirement)
 339     def test_getaliases_nxdomain(self):
 340         aliases_res = self._make_mock_resolver_aliases()
 341         rp = greendns.ResolverProxy()
 342         rp._resolver = aliases_res
 343         rp._resolver.call_count = 1
 344         rp._resolver.exc_type = greendns.dns.resolver.NXDOMAIN
 345         assert rp.getaliases('host.example.com') == []
 346
 347
 348 class TestResolve(tests.LimitedTestCase):
 349
 350     def setUp(self):
 351         base_resolver = _make_mock_base_resolver()
 352         base_resolver.rr.address = '1.2.3.4'
 353         self._old_resolver = greendns.resolver
 354         greendns.resolver = base_resolver()
 355
 356     def tearDown(self):
 357         greendns.resolver = self._old_resolver
 358
 359     @tests.skip_unless(greendns_requirement)
 360     def test_A(self):
 361         ans = greendns.resolve('host.example.com', socket.AF_INET)
 362         assert ans[0].address == '1.2.3.4'
 363         assert greendns.resolver.args == ('host.example.com', dns.rdatatype.A)
 364
 365     @tests.skip_unless(greendns_requirement)
 366     def test_AAAA(self):
 367         greendns.resolver.rr.address = 'dead:beef::1'
 368         ans = greendns.resolve('host.example.com', socket.AF_INET6)
 369         assert ans[0].address == 'dead:beef::1'
 370         assert greendns.resolver.args == ('host.example.com', dns.rdatatype.AAAA)
 371
 372     @tests.skip_unless(greendns_requirement)
 373     def test_unknown_rdtype(self):
 374         with tests.assert_raises(socket.gaierror):
 375             greendns.resolve('host.example.com', socket.AF_INET6 + 1)
 376
 377     @tests.skip_unless(greendns_requirement)
 378     def test_timeout(self):
 379         greendns.resolver.raises = greendns.dns.exception.Timeout
 380         with tests.assert_raises(socket.gaierror):
 381             greendns.resolve('host.example.com')
 382
 383     @tests.skip_unless(greendns_requirement)
 384     def test_exc(self):
 385         greendns.resolver.raises = greendns.dns.exception.DNSException
 386         with tests.assert_raises(socket.gaierror):
 387             greendns.resolve('host.example.com')
 388
 389     @tests.skip_unless(greendns_requirement)
 390     def test_noraise_noanswer(self):
 391         greendns.resolver.rrset = None
 392         ans = greendns.resolve('example.com', raises=False)
 393         assert not ans.rrset
 394
 395     @tests.skip_unless(greendns_requirement)
 396     def test_noraise_nxdomain(self):
 397         greendns.resolver.raises = greendns.dns.resolver.NXDOMAIN
 398         ans = greendns.resolve('example.com', raises=False)
 399         assert not ans.rrset
 400
 401
 402 class TestResolveCname(tests.LimitedTestCase):
 403
 404     def setUp(self):
 405         base_resolver = _make_mock_base_resolver()
 406         base_resolver.rr.target = 'cname.example.com'
 407         self._old_resolver = greendns.resolver
 408         greendns.resolver = base_resolver()
 409
 410     def tearDown(self):
 411         greendns.resolver = self._old_resolver
 412
 413     @tests.skip_unless(greendns_requirement)
 414     def test_success(self):
 415         cname = greendns.resolve_cname('alias.example.com')
 416         assert cname == 'cname.example.com'
 417
 418     @tests.skip_unless(greendns_requirement)
 419     def test_timeout(self):
 420         greendns.resolver.raises = greendns.dns.exception.Timeout
 421         with tests.assert_raises(socket.gaierror):
 422             greendns.resolve_cname('alias.example.com')
 423
 424     @tests.skip_unless(greendns_requirement)
 425     def test_nodata(self):
 426         greendns.resolver.raises = greendns.dns.exception.DNSException
 427         with tests.assert_raises(socket.gaierror):
 428             greendns.resolve_cname('alias.example.com')
 429
 430     @tests.skip_unless(greendns_requirement)
 431     def test_no_answer(self):
 432         greendns.resolver.raises = greendns.dns.resolver.NoAnswer
 433         assert greendns.resolve_cname('host.example.com') == 'host.example.com'
 434
 435
 436 def _make_mock_resolve():
 437     """A stubbed out resolve function
 438
 439     This monkeypatches the greendns.resolve() function with a mock.
 440     You must give it answers by calling .add().
 441     """
 442
 443     class MockAnswer(list):
 444         pass
 445
 446     class MockResolve(object):
 447
 448         def __init__(self):
 449             self.answers = {}
 450
 451         def __call__(self, name, family=socket.AF_INET, raises=True):
 452             qname = dns.name.from_text(name)
 453             try:
 454                 rrset = self.answers[name][family]
 455             except KeyError:
 456                 if raises:
 457                     raise greendns.dns.resolver.NoAnswer()
 458                 rrset = dns.rrset.RRset(qname, 1, 1)
 459             ans = MockAnswer()
 460             ans.qname = qname
 461             ans.rrset = rrset
 462             ans.extend(rrset.items)
 463             return ans
 464
 465         def add(self, name, addr):
 466             """Add an address to a name and family"""
 467             try:
 468                 rdata = dns.rdtypes.IN.A.A(dns.rdataclass.IN,
 469                                            dns.rdatatype.A, addr)
 470                 family = socket.AF_INET
 471             except (socket.error, dns.exception.SyntaxError):
 472                 rdata = dns.rdtypes.IN.AAAA.AAAA(dns.rdataclass.IN,
 473                                                  dns.rdatatype.AAAA, addr)
 474                 family = socket.AF_INET6
 475             family_dict = self.answers.setdefault(name, {})
 476             rrset = family_dict.get(family)
 477             if not rrset:
 478                 family_dict[family] = rrset = dns.rrset.RRset(
 479                     dns.name.from_text(name), rdata.rdclass, rdata.rdtype)
 480             rrset.add(rdata)
 481
 482     resolve = MockResolve()
 483     return resolve
 484
 485
 486 class TestGetaddrinfo(tests.LimitedTestCase):
 487
 488     def _make_mock_resolve_cname(self):
 489         """A stubbed out cname function"""
 490
 491         class ResolveCname(object):
 492             qname = None
 493             cname = 'cname.example.com'
 494
 495             def __call__(self, host):
 496                 self.qname = host
 497                 return self.cname
 498
 499         resolve_cname = ResolveCname()
 500         return resolve_cname
 501
 502     def setUp(self):
 503         self._old_resolve = greendns.resolve
 504         self._old_resolve_cname = greendns.resolve_cname
 505         self._old_orig_getaddrinfo = greendns.socket.getaddrinfo
 506
 507     def tearDown(self):
 508         greendns.resolve = self._old_resolve
 509         greendns.resolve_cname = self._old_resolve_cname
 510         greendns.socket.getaddrinfo = self._old_orig_getaddrinfo
 511
 512     @tests.skip_unless(greendns_requirement)
 513     def test_getaddrinfo(self):
 514         greendns.resolve = _make_mock_resolve()
 515         greendns.resolve.add('example.com', '127.0.0.2')
 516         greendns.resolve.add('example.com', '::1')
 517         res = greendns.getaddrinfo('example.com', 'ssh')
 518         addr = ('127.0.0.2', 22)
 519         tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr)
 520         udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr)
 521         addr = ('::1', 22, 0, 0)
 522         tcp6 = (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr)
 523         udp6 = (socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr)
 524         filt_res = [ai[:3] + (ai[4],) for ai in res]
 525         assert tcp in filt_res
 526         assert udp in filt_res
 527         assert tcp6 in filt_res
 528         assert udp6 in filt_res
 529
 530     @tests.skip_unless(greendns_requirement)
 531     def test_getaddrinfo_idn(self):
 532         greendns.resolve = _make_mock_resolve()
 533         idn_name = u'евентлет.com'
 534         greendns.resolve.add(idn_name.encode('idna').decode('ascii'), '127.0.0.2')
 535         res = greendns.getaddrinfo(idn_name, 'ssh')
 536         addr = ('127.0.0.2', 22)
 537         tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr)
 538         udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr)
 539         filt_res = [ai[:3] + (ai[4],) for ai in res]
 540         assert tcp in filt_res
 541         assert udp in filt_res
 542
 543     @tests.skip_unless(greendns_requirement)
 544     def test_getaddrinfo_inet(self):
 545         greendns.resolve = _make_mock_resolve()
 546         greendns.resolve.add('example.com', '127.0.0.2')
 547         res = greendns.getaddrinfo('example.com', 'ssh', socket.AF_INET)
 548         addr = ('127.0.0.2', 22)
 549         tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr)
 550         udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr)
 551         assert tcp in [ai[:3] + (ai[4],) for ai in res]
 552         assert udp in [ai[:3] + (ai[4],) for ai in res]
 553
 554     @tests.skip_unless(greendns_requirement)
 555     def test_getaddrinfo_inet6(self):
 556         greendns.resolve = _make_mock_resolve()
 557         greendns.resolve.add('example.com', '::1')
 558         res = greendns.getaddrinfo('example.com', 'ssh', socket.AF_INET6)
 559         addr = ('::1', 22, 0, 0)
 560         tcp = (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr)
 561         udp = (socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr)
 562         assert tcp in [ai[:3] + (ai[4],) for ai in res]
 563         assert udp in [ai[:3] + (ai[4],) for ai in res]
 564
 565     @tests.skip_unless(greendns_requirement)
 566     def test_getaddrinfo_only_a_ans(self):
 567         greendns.resolve = _make_mock_resolve()
 568         greendns.resolve.add('example.com', '1.2.3.4')
 569         res = greendns.getaddrinfo('example.com', 0)
 570         addr = [('1.2.3.4', 0)] * len(res)
 571         assert addr == [ai[-1] for ai in res]
 572
 573     @tests.skip_unless(greendns_requirement)
 574     def test_getaddrinfo_only_aaaa_ans(self):
 575         greendns.resolve = _make_mock_resolve()
 576         greendns.resolve.add('example.com', 'dead:beef::1')
 577         res = greendns.getaddrinfo('example.com', 0)
 578         addr = [('dead:beef::1', 0, 0, 0)] * len(res)
 579         assert addr == [ai[-1] for ai in res]
 580
 581     @tests.skip_unless(greendns_requirement)
 582     def test_canonname(self):
 583         greendns.resolve = _make_mock_resolve()
 584         greendns.resolve.add('host.example.com', '1.2.3.4')
 585         greendns.resolve_cname = self._make_mock_resolve_cname()
 586         res = greendns.getaddrinfo('host.example.com', 0,
 587                                    0, 0, 0, socket.AI_CANONNAME)
 588         assert res[0][3] == 'cname.example.com'
 589
 590     @tests.skip_unless(greendns_requirement)
 591     def test_host_none(self):
 592         res = greendns.getaddrinfo(None, 80)
 593         for addr in set(ai[-1] for ai in res):
 594             assert addr in [('127.0.0.1', 80), ('::1', 80, 0, 0)]
 595
 596     @tests.skip_unless(greendns_requirement)
 597     def test_host_none_passive(self):
 598         res = greendns.getaddrinfo(None, 80, 0, 0, 0, socket.AI_PASSIVE)
 599         for addr in set(ai[-1] for ai in res):
 600             assert addr in [('0.0.0.0', 80), ('::', 80, 0, 0)]
 601
 602     @tests.skip_unless(greendns_requirement)
 603     def test_v4mapped(self):
 604         greendns.resolve = _make_mock_resolve()
 605         greendns.resolve.add('example.com', '1.2.3.4')
 606         res = greendns.getaddrinfo('example.com', 80,
 607                                    socket.AF_INET6, 0, 0, socket.AI_V4MAPPED)
 608         addrs = set(ai[-1] for ai in res)
 609         assert addrs == set([('::ffff:1.2.3.4', 80, 0, 0)])
 610
 611     @tests.skip_unless(greendns_requirement)
 612     def test_v4mapped_all(self):
 613         greendns.resolve = _make_mock_resolve()
 614         greendns.resolve.add('example.com', '1.2.3.4')
 615         greendns.resolve.add('example.com', 'dead:beef::1')
 616         res = greendns.getaddrinfo('example.com', 80, socket.AF_INET6, 0, 0,
 617                                    socket.AI_V4MAPPED | socket.AI_ALL)
 618         addrs = set(ai[-1] for ai in res)
 619         for addr in addrs:
 620             assert addr in [('::ffff:1.2.3.4', 80, 0, 0),
 621                             ('dead:beef::1', 80, 0, 0)]
 622
 623     @tests.skip_unless(greendns_requirement)
 624     def test_numericserv(self):
 625         greendns.resolve = _make_mock_resolve()
 626         greendns.resolve.add('example.com', '1.2.3.4')
 627         with tests.assert_raises(socket.gaierror):
 628             greendns.getaddrinfo('example.com', 'www', 0, 0, 0, socket.AI_NUMERICSERV)
 629
 630     @tests.skip_unless(greendns_requirement)
 631     def test_numerichost(self):
 632         greendns.resolve = _make_mock_resolve()
 633         greendns.resolve.add('example.com', '1.2.3.4')
 634         with tests.assert_raises(socket.gaierror):
 635             greendns.getaddrinfo('example.com', 80, 0, 0, 0, socket.AI_NUMERICHOST)
 636
 637     @tests.skip_unless(greendns_requirement)
 638     def test_noport(self):
 639         greendns.resolve = _make_mock_resolve()
 640         greendns.resolve.add('example.com', '1.2.3.4')
 641         ai = greendns.getaddrinfo('example.com', None)
 642         assert ai[0][-1][1] == 0
 643
 644     @tests.skip_unless(greendns_requirement)
 645     def test_AI_ADDRCONFIG(self):
 646         # When the users sets AI_ADDRCONFIG but only has an IPv4
 647         # address configured we will iterate over the results, but the
 648         # call for the IPv6 address will fail rather then return an
 649         # empty list.  In that case we should catch the exception and
 650         # only return the ones which worked.
 651         def getaddrinfo(addr, port, family, socktype, proto, aiflags):
 652             if addr == '127.0.0.1':
 653                 return [(socket.AF_INET, 1, 0, '', ('127.0.0.1', 0))]
 654             elif addr == '::1' and aiflags & socket.AI_ADDRCONFIG:
 655                 raise socket.error(socket.EAI_ADDRFAMILY,
 656                                    'Address family for hostname not supported')
 657             elif addr == '::1' and not aiflags & socket.AI_ADDRCONFIG:
 658                 return [(socket.AF_INET6, 1, 0, '', ('::1', 0, 0, 0))]
 659         greendns.socket.getaddrinfo = getaddrinfo
 660         greendns.resolve = _make_mock_resolve()
 661         greendns.resolve.add('localhost', '127.0.0.1')
 662         greendns.resolve.add('localhost', '::1')
 663         res = greendns.getaddrinfo('localhost', None,
 664                                    0, 0, 0, socket.AI_ADDRCONFIG)
 665         assert res == [(socket.AF_INET, 1, 0, '', ('127.0.0.1', 0))]
 666
 667     @tests.skip_unless(greendns_requirement)
 668     def test_AI_ADDRCONFIG_noaddr(self):
 669         # If AI_ADDRCONFIG is used but there is no address we need to
 670         # get an exception, not an empty list.
 671         def getaddrinfo(addr, port, family, socktype, proto, aiflags):
 672             raise socket.error(socket.EAI_ADDRFAMILY,
 673                                'Address family for hostname not supported')
 674         greendns.socket.getaddrinfo = getaddrinfo
 675         greendns.resolve = _make_mock_resolve()
 676         try:
 677             greendns.getaddrinfo('::1', None, 0, 0, 0, socket.AI_ADDRCONFIG)
 678         except socket.error as e:
 679             assert e.errno == socket.EAI_ADDRFAMILY
 680
 681
 682 class TestIsIpAddr(tests.LimitedTestCase):
 683
 684     @tests.skip_unless(greendns_requirement)
 685     def test_isv4(self):
 686         assert greendns.is_ipv4_addr('1.2.3.4')
 687
 688     @tests.skip_unless(greendns_requirement)
 689     def test_isv4_false(self):
 690         assert not greendns.is_ipv4_addr('260.0.0.0')
 691
 692     @tests.skip_unless(greendns_requirement)
 693     def test_isv6(self):
 694         assert greendns.is_ipv6_addr('dead:beef::1')
 695
 696     @tests.skip_unless(greendns_requirement)
 697     def test_isv6_invalid(self):
 698         assert not greendns.is_ipv6_addr('foobar::1')
 699
 700     @tests.skip_unless(greendns_requirement)
 701     def test_v4(self):
 702         assert greendns.is_ip_addr('1.2.3.4')
 703
 704     @tests.skip_unless(greendns_requirement)
 705     def test_v4_illegal(self):
 706         assert not greendns.is_ip_addr('300.0.0.1')
 707
 708     @tests.skip_unless(greendns_requirement)
 709     def test_v6_addr(self):
 710         assert greendns.is_ip_addr('::1')
 711
 712     @tests.skip_unless(greendns_requirement)
 713     def test_isv4_none(self):
 714         assert not greendns.is_ipv4_addr(None)
 715
 716     @tests.skip_unless(greendns_requirement)
 717     def test_isv6_none(self):
 718         assert not greendns.is_ipv6_addr(None)
 719
 720     @tests.skip_unless(greendns_requirement)
 721     def test_none(self):
 722         assert not greendns.is_ip_addr(None)
 723
 724
 725 class TestGethostbyname(tests.LimitedTestCase):
 726
 727     def setUp(self):
 728         self._old_resolve = greendns.resolve
 729         greendns.resolve = _make_mock_resolve()
 730
 731     def tearDown(self):
 732         greendns.resolve = self._old_resolve
 733
 734     @tests.skip_unless(greendns_requirement)
 735     def test_ipaddr(self):
 736         assert greendns.gethostbyname('1.2.3.4') == '1.2.3.4'
 737
 738     @tests.skip_unless(greendns_requirement)
 739     def test_name(self):
 740         greendns.resolve.add('host.example.com', '1.2.3.4')
 741         assert greendns.gethostbyname('host.example.com') == '1.2.3.4'
 742
 743
 744 class TestGetaliases(tests.LimitedTestCase):
 745
 746     def _make_mock_resolver(self):
 747         base_resolver = _make_mock_base_resolver()
 748         resolver = base_resolver()
 749         resolver.aliases = ['cname.example.com']
 750         return resolver
 751
 752     def setUp(self):
 753         self._old_resolver = greendns.resolver
 754         greendns.resolver = self._make_mock_resolver()
 755
 756     def tearDown(self):
 757         greendns.resolver = self._old_resolver
 758
 759     @tests.skip_unless(greendns_requirement)
 760     def test_getaliases(self):
 761         assert greendns.getaliases('host.example.com') == ['cname.example.com']
 762
 763
 764 class TestGethostbyname_ex(tests.LimitedTestCase):
 765
 766     def _make_mock_getaliases(self):
 767
 768         class GetAliases(object):
 769             aliases = ['cname.example.com']
 770
 771             def __call__(self, *args, **kwargs):
 772                 return self.aliases
 773
 774         getaliases = GetAliases()
 775         return getaliases
 776
 777     def setUp(self):
 778         self._old_resolve = greendns.resolve
 779         greendns.resolve = _make_mock_resolve()
 780         self._old_getaliases = greendns.getaliases
 781
 782     def tearDown(self):
 783         greendns.resolve = self._old_resolve
 784         greendns.getaliases = self._old_getaliases
 785
 786     @tests.skip_unless(greendns_requirement)
 787     def test_ipaddr(self):
 788         res = greendns.gethostbyname_ex('1.2.3.4')
 789         assert res == ('1.2.3.4', [], ['1.2.3.4'])
 790
 791     @tests.skip_unless(greendns_requirement)
 792     def test_name(self):
 793         greendns.resolve.add('host.example.com', '1.2.3.4')
 794         greendns.getaliases = self._make_mock_getaliases()
 795         greendns.getaliases.aliases = []
 796         res = greendns.gethostbyname_ex('host.example.com')
 797         assert res == ('host.example.com', [], ['1.2.3.4'])
 798
 799     @tests.skip_unless(greendns_requirement)
 800     def test_multiple_addrs(self):
 801         greendns.resolve.add('host.example.com', '1.2.3.4')
 802         greendns.resolve.add('host.example.com', '1.2.3.5')
 803         greendns.getaliases = self._make_mock_getaliases()
 804         greendns.getaliases.aliases = []
 805         res = greendns.gethostbyname_ex('host.example.com')
 806         assert res == ('host.example.com', [], ['1.2.3.4', '1.2.3.5'])
 807
 808
 809 def test_reverse_name():
 810     tests.run_isolated('greendns_from_address_203.py')