end
return DNSAction.None, ""
end
- addLuaAction("added.trailing.tests.powerdns.com.", replaceTrailingData)
+ addAction("added.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData))
function fillBuffer(dq)
local available = dq.size - dq.len
end
return DNSAction.None, ""
end
- addLuaAction("max.trailing.tests.powerdns.com.", fillBuffer)
+ addAction("max.trailing.tests.powerdns.com.", LuaAction(fillBuffer))
function exceedBuffer(dq)
local available = dq.size - dq.len
end
return DNSAction.None, ""
end
- addLuaAction("limited.trailing.tests.powerdns.com.", exceedBuffer)
+ addAction("limited.trailing.tests.powerdns.com.", LuaAction(exceedBuffer))
"""
@classmethod
def startResponders(cls):
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(receivedQuery, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(receivedQuery, receivedResponse) = sender(query, response)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(_, receivedResponse) = sender(query, response)
self.assertTrue(receivedResponse)
self.assertEquals(receivedResponse, expectedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(receivedQuery, receivedResponse) = sender(query, response)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
end
return DNSAction.None, ""
end
- addLuaAction("removed.trailing.tests.powerdns.com.", removeTrailingData)
+ addAction("removed.trailing.tests.powerdns.com.", LuaAction(removeTrailingData))
function reportTrailingData(dq)
local tail = dq:getTrailingData()
return DNSAction.Spoof, "-" .. tail .. ".echoed.trailing.tests.powerdns.com."
end
- addLuaAction("echoed.trailing.tests.powerdns.com.", reportTrailingData)
+ addAction("echoed.trailing.tests.powerdns.com.", LuaAction(reportTrailingData))
function replaceTrailingData(dq)
local success = dq:setTrailingData("ABC")
end
return DNSAction.None, ""
end
- addLuaAction("replaced.trailing.tests.powerdns.com.", replaceTrailingData)
- addLuaAction("replaced.trailing.tests.powerdns.com.", reportTrailingData)
+ addAction("replaced.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData))
+ addAction("replaced.trailing.tests.powerdns.com.", LuaAction(reportTrailingData))
function reportTrailingHex(dq)
local tail = dq:getTrailingData()
end)
return DNSAction.Spoof, "-0x" .. hex .. ".echoed-hex.trailing.tests.powerdns.com."
end
- addLuaAction("echoed-hex.trailing.tests.powerdns.com.", reportTrailingHex)
+ addAction("echoed-hex.trailing.tests.powerdns.com.", LuaAction(reportTrailingHex))
function replaceTrailingData_unsafe(dq)
local success = dq:setTrailingData("\\xB0\\x00\\xDE\\xADB\\xF0\\x9F\\x91\\xBB\\xC3\\xBE")
end
return DNSAction.None, ""
end
- addLuaAction("replaced-unsafe.trailing.tests.powerdns.com.", replaceTrailingData_unsafe)
- addLuaAction("replaced-unsafe.trailing.tests.powerdns.com.", reportTrailingHex)
+ addAction("replaced-unsafe.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData_unsafe))
+ addAction("replaced-unsafe.trailing.tests.powerdns.com.", LuaAction(reportTrailingHex))
"""
def testTrailingDropped(self):
sender = getattr(self, method)
# Verify that queries with no trailing data make it through.
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(receivedQuery, receivedResponse) = sender(query, response)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
self.assertEquals(response, receivedResponse)
# Verify that queries with trailing data don't make it through.
- # (_, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (_, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertEquals(receivedResponse, None)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(receivedQuery, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags