| 416 | ) |
| 417 | |
| 418 | def test_cache_flush(self): |
| 419 | name1 = dns.name.from_text("name1") |
| 420 | name2 = dns.name.from_text("name2") |
| 421 | name3 = dns.name.from_text("name3") |
| 422 | basic_cache = dns.resolver.Cache() |
| 423 | lru_cache = dns.resolver.LRUCache(100) |
| 424 | for cache in [basic_cache, lru_cache]: |
| 425 | answer1 = FakeAnswer(time.time() + 10) |
| 426 | answer2 = FakeAnswer(time.time() + 10) |
| 427 | cache.put((name1, dns.rdatatype.A, dns.rdataclass.IN), answer1) |
| 428 | cache.put((name2, dns.rdatatype.A, dns.rdataclass.IN), answer2) |
| 429 | canswer = cache.get((name1, dns.rdatatype.A, dns.rdataclass.IN)) |
| 430 | self.assertTrue(canswer is answer1) |
| 431 | canswer = cache.get((name2, dns.rdatatype.A, dns.rdataclass.IN)) |
| 432 | self.assertTrue(canswer is answer2) |
| 433 | # explicit flush of nonexistent key, just to exercise the branch |
| 434 | cache.flush((name3, dns.rdatatype.A, dns.rdataclass.IN)) |
| 435 | canswer = cache.get((name1, dns.rdatatype.A, dns.rdataclass.IN)) |
| 436 | self.assertTrue(canswer is answer1) |
| 437 | canswer = cache.get((name2, dns.rdatatype.A, dns.rdataclass.IN)) |
| 438 | self.assertTrue(canswer is answer2) |
| 439 | # explicit flush |
| 440 | cache.flush((name1, dns.rdatatype.A, dns.rdataclass.IN)) |
| 441 | canswer = cache.get((name1, dns.rdatatype.A, dns.rdataclass.IN)) |
| 442 | self.assertTrue(canswer is None) |
| 443 | canswer = cache.get((name2, dns.rdatatype.A, dns.rdataclass.IN)) |
| 444 | self.assertTrue(canswer is answer2) |
| 445 | # flush all |
| 446 | cache.flush() |
| 447 | canswer = cache.get((name1, dns.rdatatype.A, dns.rdataclass.IN)) |
| 448 | self.assertTrue(canswer is None) |
| 449 | canswer = cache.get((name2, dns.rdatatype.A, dns.rdataclass.IN)) |
| 450 | self.assertTrue(canswer is None) |
| 451 | |
| 452 | def test_LRUCache_set_max_size(self): |
| 453 | cache = dns.resolver.LRUCache(4) |