]>
git.ipfire.org Git - thirdparty/pdns.git/blob - regression-tests.dnsdist/test_CacheHitResponses.py
5 from dnsdisttests
import DNSDistTest
7 class TestCacheHitResponses(DNSDistTest
):
10 pc = newPacketCache(100, 86400, 1)
11 getPool(""):setCache(pc)
12 addCacheHitResponseAction(makeRule("dropwhencached.cachehitresponses.tests.powerdns.com."), DropResponseAction())
13 newServer{address="127.0.0.1:%s"}
16 def testDroppedWhenCached(self
):
18 CacheHitResponse: Drop when served from the cache
21 name
= 'dropwhencached.cachehitresponses.tests.powerdns.com.'
22 query
= dns
.message
.make_query(name
, 'AAAA', 'IN')
23 response
= dns
.message
.make_response(query
)
24 rrset
= dns
.rrset
.from_text(name
,
29 response
.answer
.append(rrset
)
31 # first query to fill the cache
32 (receivedQuery
, receivedResponse
) = self
.sendUDPQuery(query
, response
)
33 self
.assertTrue(receivedQuery
)
34 self
.assertTrue(receivedResponse
)
35 receivedQuery
.id = query
.id
36 self
.assertEquals(query
, receivedQuery
)
37 self
.assertEquals(receivedResponse
, response
)
39 # now the result should be cached, and so dropped
40 (_
, receivedResponse
) = self
.sendUDPQuery(query
, response
=None, useQueue
=False)
41 print(receivedResponse
)
42 self
.assertEquals(receivedResponse
, None)
46 # should not be cached anymore and so valid
47 (receivedQuery
, receivedResponse
) = self
.sendUDPQuery(query
, response
)
48 self
.assertTrue(receivedQuery
)
49 self
.assertTrue(receivedResponse
)
50 receivedQuery
.id = query
.id
51 self
.assertEquals(query
, receivedQuery
)
52 self
.assertEquals(receivedResponse
, response
)
55 for key
in self
._responsesCounter
:
56 total
+= self
._responsesCounter
[key
]
57 TestCacheHitResponses
._responsesCounter
[key
] = 0
59 self
.assertEquals(total
, 2)
61 # TCP should not be cached
62 # first query to fill the cache
63 (receivedQuery
, receivedResponse
) = self
.sendTCPQuery(query
, response
)
64 self
.assertTrue(receivedQuery
)
65 self
.assertTrue(receivedResponse
)
66 receivedQuery
.id = query
.id
67 self
.assertEquals(query
, receivedQuery
)
68 self
.assertEquals(receivedResponse
, response
)
70 # now the result should be cached, and so dropped
71 (_
, receivedResponse
) = self
.sendTCPQuery(query
, response
=None, useQueue
=False)
72 self
.assertEquals(receivedResponse
, None)
76 # should not be cached anymore and so valid
77 (receivedQuery
, receivedResponse
) = self
.sendTCPQuery(query
, response
)
78 self
.assertTrue(receivedQuery
)
79 self
.assertTrue(receivedResponse
)
80 receivedQuery
.id = query
.id
81 self
.assertEquals(query
, receivedQuery
)
82 self
.assertEquals(receivedResponse
, response
)
85 for key
in self
._responsesCounter
:
86 total
+= self
._responsesCounter
[key
]
87 TestCacheHitResponses
._responsesCounter
[key
] = 0
89 self
.assertEquals(total
, 2)