twisted/names/test/test_cache.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 import time
   5
   6 from twisted.trial import unittest
   7
   8 from twisted.names import dns, cache
   9 from twisted.internet import task
  10
  11 class Caching(unittest.TestCase):
  12     """
  13     Tests for L{cache.CacheResolver}.
  14     """
  15
  16     def test_lookup(self):
  17         c = cache.CacheResolver({
  18             dns.Query(name='example.com', type=dns.MX, cls=dns.IN):
  19                 (time.time(), ([], [], []))})
  20         return c.lookupMailExchange('example.com').addCallback(
  21             self.assertEqual, ([], [], []))
  22
  23
  24     def test_constructorExpires(self):
  25         """
  26         Cache entries passed into L{cache.CacheResolver.__init__} get
  27         cancelled just like entries added with cacheResult
  28         """
  29
  30         r = ([dns.RRHeader("example.com", dns.A, dns.IN, 60,
  31                            dns.Record_A("127.0.0.1", 60))],
  32              [dns.RRHeader("example.com", dns.A, dns.IN, 50,
  33                            dns.Record_A("127.0.0.1", 50))],
  34              [dns.RRHeader("example.com", dns.A, dns.IN, 40,
  35                            dns.Record_A("127.0.0.1", 40))])
  36
  37         clock = task.Clock()
  38         query = dns.Query(name="example.com", type=dns.A, cls=dns.IN)
  39
  40         c = cache.CacheResolver({ query : (clock.seconds(), r)}, reactor=clock)
  41
  42         # 40 seconds is enough to expire the entry because expiration is based
  43         # on the minimum TTL.
  44         clock.advance(40)
  45
  46         self.assertNotIn(query, c.cache)
  47
  48         return self.assertFailure(
  49             c.lookupAddress("example.com"), dns.DomainError)
  50
  51
  52     def test_normalLookup(self):
  53         """
  54         When a cache lookup finds a cached entry from 1 second ago, it is
  55         returned with a TTL of original TTL minus the elapsed 1 second.
  56         """
  57         r = ([dns.RRHeader("example.com", dns.A, dns.IN, 60,
  58                            dns.Record_A("127.0.0.1", 60))],
  59              [dns.RRHeader("example.com", dns.A, dns.IN, 50,
  60                            dns.Record_A("127.0.0.1", 50))],
  61              [dns.RRHeader("example.com", dns.A, dns.IN, 40,
  62                            dns.Record_A("127.0.0.1", 40))])
  63
  64         clock = task.Clock()
  65
  66         c = cache.CacheResolver(reactor=clock)
  67         c.cacheResult(dns.Query(name="example.com", type=dns.A, cls=dns.IN), r)
  68
  69         clock.advance(1)
  70
  71         def cbLookup(result):
  72             self.assertEquals(result[0][0].ttl, 59)
  73             self.assertEquals(result[1][0].ttl, 49)
  74             self.assertEquals(result[2][0].ttl, 39)
  75             self.assertEquals(result[0][0].name.name, "example.com")
  76
  77         return c.lookupAddress("example.com").addCallback(cbLookup)
  78
  79
  80     def test_negativeTTLLookup(self):
  81         """
  82         When the cache is queried exactly as the cached entry should expire
  83         but before it has actually been cleared, the TTL will be 0, not
  84         negative.
  85         """
  86         r = ([dns.RRHeader("example.com", dns.A, dns.IN, 60,
  87                            dns.Record_A("127.0.0.1", 60))],
  88              [dns.RRHeader("example.com", dns.A, dns.IN, 50,
  89                            dns.Record_A("127.0.0.1", 50))],
  90              [dns.RRHeader("example.com", dns.A, dns.IN, 40,
  91                            dns.Record_A("127.0.0.1", 40))])
  92
  93         clock = task.Clock()
  94         # Make sure timeouts never happen, so entries won't get cleared:
  95         clock.callLater = lambda *args, **kwargs: None
  96
  97         c = cache.CacheResolver({
  98             dns.Query(name="example.com", type=dns.A, cls=dns.IN) :
  99                 (clock.seconds(), r)}, reactor=clock)
 100
 101         clock.advance(60.1)
 102
 103         def cbLookup(result):
 104             self.assertEquals(result[0][0].ttl, 0)
 105             self.assertEquals(result[0][0].ttl, 0)
 106             self.assertEquals(result[0][0].ttl, 0)
 107             self.assertEquals(result[0][0].name.name, "example.com")
 108
 109         return c.lookupAddress("example.com").addCallback(cbLookup)