]>
git.ipfire.org Git - thirdparty/pdns.git/blob - regression-tests.dnsdist/test_Size.py
3 from dnsdisttests
import DNSDistTest
5 class TestSize(DNSDistTest
):
9 addAction(PayloadSizeRule("smaller", %d), SpoofAction("192.0.2.1"))
10 addAction(PayloadSizeRule("greater", %d), SpoofAction("192.0.2.2"))
11 addAction(PayloadSizeRule("equal", %d), SpoofAction("192.0.2.3"))
12 newServer{address="127.0.0.1:%d"}
14 _config_params
= ['_payloadSize', '_payloadSize', '_payloadSize', '_testServerPort']
16 def testPayloadSize(self
):
18 Size: Check that PayloadSizeRule works
20 name
= 'payload.size.tests.powerdns.com.'
21 query
= dns
.message
.make_query(name
, 'A', 'IN')
22 query
.flags
&= ~dns
.flags
.RD
23 expectedResponse
= dns
.message
.make_response(query
)
24 rrset
= dns
.rrset
.from_text(name
,
29 expectedResponse
.answer
.append(rrset
)
30 self
.assertEqual(len(query
.to_wire()), self
._payloadSize
)
32 for method
in ("sendUDPQuery", "sendTCPQuery"):
33 sender
= getattr(self
, method
)
34 (_
, receivedResponse
) = sender(query
, response
=None, useQueue
=False)
35 self
.assertTrue(receivedResponse
)
36 self
.assertEqual(receivedResponse
, expectedResponse
)
38 class TestTruncateOversizedResponse(DNSDistTest
):
41 _config_template
= """
42 addResponseAction(PayloadSizeRule("greater", %d), TCResponseAction())
43 newServer{address="127.0.0.1:%d"}
45 _config_params
= ['_payloadSize', '_testServerPort']
47 def testTruncateOversizedResponse(self
):
49 Size: Truncate oversized response
51 name
= 'truncate-oversized.size.tests.powerdns.com.'
52 query
= dns
.message
.make_query(name
, 'TXT', 'IN')
53 query
.flags
&= ~dns
.flags
.RD
54 expectedResponse
= dns
.message
.make_response(query
)
55 expectedResponse
.flags |
= dns
.flags
.TC
57 backendResponse
= dns
.message
.make_response(query
)
61 content
= content
+ ' '
62 content
= content
+ 'A' * 255
63 rrset
= dns
.rrset
.from_text(name
,
68 backendResponse
.answer
.append(rrset
)
69 self
.assertGreater(len(backendResponse
.to_wire()), self
._payloadSize
)
71 (receivedQuery
, receivedResponse
) = self
.sendUDPQuery(query
, response
=backendResponse
)
72 self
.assertTrue(receivedQuery
)
73 self
.assertTrue(receivedResponse
)
74 self
.assertEqual(receivedResponse
, expectedResponse
)
76 (receivedQuery
, receivedResponse
) = self
.sendTCPQuery(query
, response
=backendResponse
)
77 self
.assertTrue(receivedQuery
)
78 self
.assertTrue(receivedResponse
)
79 self
.assertEqual(receivedResponse
, backendResponse
)