mir/kea
2
0
Fork 0
mirror of https://gitlab.isc.org/isc-projects/kea synced 2025-08-29 21:18:02 +00:00
Code Issues Releases Wiki Activity

[1866] remove '()' from RCode/Opcode.XXX constants, making all tests pass now

Browse Source
This commit is contained in:
JINMEI Tatuya 2013-01-28 14:59:30 -08:00
parent 4f0716782c
commit d985d0dde4
17 changed files with 308 additions and 308 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/bin/ddns/ddns.py.in
View File

@ -469,7 +469,7 @@ class DDNSServer:
self.__request_msg.clear(Message.PARSE)
# specify PRESERVE_ORDER as we need to handle each RR separately.
self.__request_msg.from_wire(req_data, Message.PRESERVE_ORDER)
if self.__request_msg.get_opcode() != Opcode.UPDATE():
if self.__request_msg.get_opcode() != Opcode.UPDATE:
raise self.InternalError('Update request has unexpected '
'opcode: ' +
str(self.__request_msg.get_opcode()))

22
src/bin/ddns/tests/ddns_test.py
View File

@ -169,9 +169,9 @@ class FakeUpdateSession:
self.__msg.make_response()
self.__msg.clear_section(SECTION_ZONE)
if self.__faked_result == UPDATE_SUCCESS:
self.__msg.set_rcode(Rcode.NOERROR())
self.__msg.set_rcode(Rcode.NOERROR)
else:
self.__msg.set_rcode(Rcode.REFUSED())
self.__msg.set_rcode(Rcode.REFUSED)
return self.__msg
class FakeKeyringModule:
@ -887,12 +887,12 @@ class TestDDNSServer(unittest.TestCase):
self.__select_answer = ([], [10], [])
self.assertRaises(KeyError, self.ddns_server.run)
def create_msg(opcode=Opcode.UPDATE(), zones=[TEST_ZONE_RECORD], prereq=[],
def create_msg(opcode=Opcode.UPDATE, zones=[TEST_ZONE_RECORD], prereq=[],
tsigctx=None):
msg = Message(Message.RENDER)
msg.set_qid(TEST_QID)
msg.set_opcode(opcode)
msg.set_rcode(Rcode.NOERROR())
msg.set_rcode(Rcode.NOERROR)
for z in zones:
msg.add_question(z)
for p in prereq:
@ -936,7 +936,7 @@ class TestDDNSSession(unittest.TestCase):
return FakeUpdateSession(req_message, client_addr, zone_config,
self.__faked_result)
def check_update_response(self, resp_wire, expected_rcode=Rcode.NOERROR(),
def check_update_response(self, resp_wire, expected_rcode=Rcode.NOERROR,
tsig_ctx=None, tcp=False):
'''Check if given wire data are valid form of update response.
@ -963,7 +963,7 @@ class TestDDNSSession(unittest.TestCase):
self.assertNotEqual(None, tsig_record)
self.assertEqual(TSIGError.NOERROR,
tsig_ctx.verify(tsig_record, resp_wire))
self.assertEqual(Opcode.UPDATE(), msg.get_opcode())
self.assertEqual(Opcode.UPDATE, msg.get_opcode())
self.assertEqual(expected_rcode, msg.get_rcode())
self.assertEqual(TEST_QID, msg.get_qid())
for section in [SECTION_ZONE, SECTION_PREREQUISITE, SECTION_UPDATE]:
@ -977,7 +977,7 @@ class TestDDNSSession(unittest.TestCase):
server_addr = TEST_SERVER6 if ipv6 else TEST_SERVER4
client_addr = TEST_CLIENT6 if ipv6 else TEST_CLIENT4
tsig = TSIGContext(tsig_key) if tsig_key is not None else None
rcode = Rcode.NOERROR() if result == UPDATE_SUCCESS else Rcode.REFUSED()
rcode = Rcode.NOERROR if result == UPDATE_SUCCESS else Rcode.REFUSED
has_response = (result != UPDATE_DROP)
self.assertEqual(has_response,
@ -1015,7 +1015,7 @@ class TestDDNSSession(unittest.TestCase):
# Opcode is not UPDATE
self.assertFalse(self.server.handle_request(
(self.__sock, None, None, create_msg(opcode=Opcode.QUERY()))))
(self.__sock, None, None, create_msg(opcode=Opcode.QUERY))))
self.assertEqual((None, None), (s._sent_data, s._sent_addr))
# TSIG verification error. We use UPDATE_DROP to signal check_session
@ -1031,7 +1031,7 @@ class TestDDNSSession(unittest.TestCase):
TEST_CLIENT6,
create_msg())))
# this check ensures sendto() was really attempted.
self.check_update_response(self.__sock._sent_data, Rcode.NOERROR())
self.check_update_response(self.__sock._sent_data, Rcode.NOERROR)
def test_tcp_request(self):
# A simple case using TCP: all resopnse data are sent out at once.
@ -1040,7 +1040,7 @@ class TestDDNSSession(unittest.TestCase):
self.assertTrue(self.server.handle_request((s, TEST_SERVER6,
TEST_CLIENT6,
create_msg())))
self.check_update_response(s._sent_data, Rcode.NOERROR(), tcp=True)
self.check_update_response(s._sent_data, Rcode.NOERROR, tcp=True)
# In the current implementation, the socket should be closed
# immedidately after a successful send.
self.assertEqual(1, s._close_called)
@ -1071,7 +1071,7 @@ class TestDDNSSession(unittest.TestCase):
s.make_send_ready()
self.assertEqual(DNSTCPContext.SEND_DONE,
self.server._tcp_ctxs[s.fileno()][0].send_ready())
self.check_update_response(s._sent_data, Rcode.NOERROR(), tcp=True)
self.check_update_response(s._sent_data, Rcode.NOERROR, tcp=True)
def test_tcp_request_error(self):
# initial send() on the TCP socket will fail. The request handling

20
src/bin/xfrin/tests/xfrin_test.py
View File

@ -317,7 +317,7 @@ class MockXfrinConnection(XfrinConnection):
return len(data)
def create_response_data(self, response=True, auth=True, bad_qid=False,
rcode=Rcode.NOERROR(),
rcode=Rcode.NOERROR,
questions=default_questions,
answers=default_answers,
authorities=[],
@ -327,7 +327,7 @@ class MockXfrinConnection(XfrinConnection):
if bad_qid:
qid += 1
resp.set_qid(qid)
resp.set_opcode(Opcode.QUERY())
resp.set_opcode(Opcode.QUERY)
resp.set_rcode(rcode)
if response:
resp.set_header_flag(Message.HEADERFLAG_QR)
@ -712,7 +712,7 @@ class TestXfrinConnection(unittest.TestCase):
'bad_qid': False,
'response': True,
'auth': True,
'rcode': Rcode.NOERROR(),
'rcode': Rcode.NOERROR,
'answers': default_answers,
'authorities': [],
'tsig': False,
@ -881,8 +881,8 @@ class TestAXFR(TestXfrinConnection):
def test_create_query(self):
def check_query(expected_qtype, expected_auth):
'''Helper method to repeat the same pattern of tests'''
self.assertEqual(Opcode.QUERY(), msg.get_opcode())
self.assertEqual(Rcode.NOERROR(), msg.get_rcode())
self.assertEqual(Opcode.QUERY, msg.get_opcode())
self.assertEqual(Rcode.NOERROR, msg.get_rcode())
self.assertEqual(1, msg.get_rr_count(Message.SECTION_QUESTION))
self.assertEqual(TEST_ZONE_NAME, msg.get_question()[0].get_name())
self.assertEqual(expected_qtype, msg.get_question()[0].get_type())
@ -968,7 +968,7 @@ class TestAXFR(TestXfrinConnection):
# server tsig check fail, return with RCODE 9 (NOTAUTH)
self.conn._send_query(RRType.SOA)
self.conn.reply_data = \
self.conn.create_response_data(rcode=Rcode.NOTAUTH())
self.conn.create_response_data(rcode=Rcode.NOTAUTH)
self.assertRaises(XfrinProtocolError,
self.conn._handle_xfrin_responses)
@ -992,7 +992,7 @@ class TestAXFR(TestXfrinConnection):
lambda key: self.__create_mock_tsig(key, TSIGError.BAD_SIG)
self.conn._send_query(RRType.AXFR)
self.conn.reply_data = self.conn.create_response_data(
rcode=Rcode.SERVFAIL())
rcode=Rcode.SERVFAIL)
# xfrin should check TSIG before other part of incoming message
# validate log message for XfrinException
self.__match_exception(XfrinProtocolError,
@ -1019,7 +1019,7 @@ class TestAXFR(TestXfrinConnection):
def test_response_error_code(self):
self.conn._send_query(RRType.AXFR)
self.conn.reply_data = self.conn.create_response_data(
rcode=Rcode.SERVFAIL())
rcode=Rcode.SERVFAIL)
self.assertRaises(XfrinProtocolError,
self.conn._handle_xfrin_responses)
@ -1069,7 +1069,7 @@ class TestAXFR(TestXfrinConnection):
self.assertRaises(XfrinProtocolError, self.conn._check_soa_serial)
def test_soacheck_error_code(self):
self.soa_response_params['rcode'] = Rcode.SERVFAIL()
self.soa_response_params['rcode'] = Rcode.SERVFAIL
self.conn.response_generator = self._create_soa_response_data
self.assertRaises(XfrinProtocolError, self.conn._check_soa_serial)
@ -1191,7 +1191,7 @@ class TestAXFR(TestXfrinConnection):
self.conn._tsig_key = TSIG_KEY
self.conn._tsig_ctx_creator = \
lambda key: self.__create_mock_tsig(key, TSIGError.BAD_SIG)
self.soa_response_params['rcode'] = Rcode.NOTAUTH()
self.soa_response_params['rcode'] = Rcode.NOTAUTH
self.conn.response_generator = self._create_soa_response_data
self.assertRaises(XfrinProtocolError, self.conn._check_soa_serial)

6
src/bin/xfrin/xfrin.py.in
View File

@ -715,8 +715,8 @@ class XfrinConnection(asyncore.dispatcher):
query_id = random.randint(0, 0xFFFF)
self._query_id = query_id
msg.set_qid(query_id)
msg.set_opcode(Opcode.QUERY())
msg.set_rcode(Rcode.NOERROR())
msg.set_opcode(Opcode.QUERY)
msg.set_rcode(Rcode.NOERROR)
msg.add_question(Question(self._zone_name, self._rrclass, query_type))
# Remember our serial, if known
@ -992,7 +992,7 @@ class XfrinConnection(asyncore.dispatcher):
# cause interoperability trouble with stricter checks.
msg_rcode = msg.get_rcode()
if msg_rcode != Rcode.NOERROR():
if msg_rcode != Rcode.NOERROR:
raise XfrinProtocolError('error response: %s' %
msg_rcode.to_text())

56
src/bin/xfrout/tests/xfrout_test.py.in
View File

@ -238,8 +238,8 @@ class TestXfroutSessionBase(unittest.TestCase):
msg = Message(Message.RENDER)
query_id = 0x1035
msg.set_qid(query_id)
msg.set_opcode(Opcode.QUERY())
msg.set_rcode(Rcode.NOERROR())
msg.set_opcode(Opcode.QUERY)
msg.set_rcode(Rcode.NOERROR)
req_type = RRType.AXFR if ixfr is None else RRType.IXFR
if with_question:
msg.add_question(Question(zone_name, RRClass.IN,
@ -342,7 +342,7 @@ class TestXfroutSession(TestXfroutSessionBase):
self.xfrsess._request_data = self.mdata
self.xfrsess._server.increase_transfers_counter = lambda : False
XfroutSession._handle(self.xfrsess)
self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.REFUSED())
self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.REFUSED)
def test_quota_ok(self):
'''The default case in terms of the xfrout quota.
@ -355,7 +355,7 @@ class TestXfroutSession(TestXfroutSessionBase):
# Replace the data source client to avoid datasrc related exceptions
self.xfrsess.ClientClass = MockDataSrcClient
XfroutSession._handle(self.xfrsess)
self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.FORMERR())
self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.FORMERR)
def test_exception_from_session(self):
'''Test the case where the main processing raises an exception.
@ -379,7 +379,7 @@ class TestXfroutSession(TestXfroutSessionBase):
request_data = self.create_request_data(ixfr=2011111801)
rcode, msg = self.xfrsess._parse_query_message(request_data)
self.assertEqual(RRType.IXFR, self.xfrsess._request_type)
self.assertEqual(Rcode.NOERROR(), rcode)
self.assertEqual(Rcode.NOERROR, rcode)
# Broken request: no question
self.assertRaises(RuntimeError, self.xfrsess._parse_query_message,
@ -759,19 +759,19 @@ class TestXfroutSession(TestXfroutSessionBase):
self.xfrsess.ClientClass = MockDataSrcClient
# Successful case. A zone iterator should be set up.
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR)
self.assertNotEqual(None, self.xfrsess._iterator)
# Failure cases
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), Name('notauth.example.com'), TEST_RRCLASS),
Rcode.NOTAUTH())
Rcode.NOTAUTH)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), Name('nosoa.example.com'), TEST_RRCLASS),
Rcode.SERVFAIL())
Rcode.SERVFAIL)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), Name('multisoa.example.com'), TEST_RRCLASS),
Rcode.SERVFAIL())
Rcode.SERVFAIL)
def test_xfrout_ixfr_setup(self):
self.xfrsess.ClientClass = MockDataSrcClient
@ -781,14 +781,14 @@ class TestXfroutSession(TestXfroutSessionBase):
# up.
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR)
self.assertNotEqual(None, self.xfrsess._jnl_reader)
# Successful case, but as a result of falling back to AXFR-style
# IXFR. A zone iterator should be set up instead of a journal reader.
self.mdata = self.create_request_data(ixfr=IXFR_NG_VERSION)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR)
self.assertNotEqual(None, self.xfrsess._iterator)
self.assertEqual(None, self.xfrsess._jnl_reader)
@ -797,7 +797,7 @@ class TestXfroutSession(TestXfroutSessionBase):
# indicating that the response will contain just one SOA.
self.mdata = self.create_request_data(ixfr=SOA_CURRENT_VERSION+1)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR)
self.assertEqual(None, self.xfrsess._iterator)
self.assertEqual(None, self.xfrsess._jnl_reader)
@ -805,7 +805,7 @@ class TestXfroutSession(TestXfroutSessionBase):
# the local SOA.
self.mdata = self.create_request_data(ixfr=SOA_CURRENT_VERSION)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), TEST_ZONE_NAME, TEST_RRCLASS), Rcode.NOERROR)
self.assertEqual(None, self.xfrsess._iterator)
self.assertEqual(None, self.xfrsess._jnl_reader)
@ -814,7 +814,7 @@ class TestXfroutSession(TestXfroutSessionBase):
zone_name = Name('maxserial.example.com') # whose SOA is 0xffffffff
self.mdata = self.create_request_data(ixfr=1, zone_name=zone_name)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOERROR)
self.assertEqual(None, self.xfrsess._iterator)
self.assertEqual(None, self.xfrsess._jnl_reader)
@ -823,7 +823,7 @@ class TestXfroutSession(TestXfroutSessionBase):
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
zone_name=zone_name)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOERROR())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOERROR)
self.assertNotEqual(None, self.xfrsess._iterator)
# Failure cases
@ -831,42 +831,42 @@ class TestXfroutSession(TestXfroutSessionBase):
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
zone_name=zone_name)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOTAUTH())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOTAUTH)
# this is a strange case: zone's SOA will be found but the journal
# reader won't be created due to 'no such zone'.
zone_name = Name('notauth2.example.com')
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
zone_name=zone_name)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOTAUTH())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.NOTAUTH)
zone_name = Name('nosoa.example.com')
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
zone_name=zone_name)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.SERVFAIL())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.SERVFAIL)
zone_name = Name('multisoa.example.com')
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
zone_name=zone_name)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.SERVFAIL())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.SERVFAIL)
# query name doesn't match the SOA's owner
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.FORMERR())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.FORMERR)
# query's RR class doesn't match the SOA's class
zone_name = TEST_ZONE_NAME # make sure the name matches this time
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
soa_class=RRClass.CH)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.FORMERR())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.FORMERR)
# multiple SOA RRs
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION,
num_soa=2)
self.assertEqual(self.xfrsess._xfrout_setup(
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.FORMERR())
self.getmsg(), zone_name, TEST_RRCLASS), Rcode.FORMERR)
def test_dns_xfrout_start_formerror(self):
# formerror
@ -876,7 +876,7 @@ class TestXfroutSession(TestXfroutSessionBase):
def test_dns_xfrout_start_notauth(self):
def notauth(msg, name, rrclass):
return Rcode.NOTAUTH()
return Rcode.NOTAUTH
self.xfrsess._xfrout_setup = notauth
self.xfrsess.dns_xfrout_start(self.sock, self.mdata)
get_msg = self.sock.read_msg()
@ -887,11 +887,11 @@ class TestXfroutSession(TestXfroutSessionBase):
raise isc.datasrc.Error('exception for the sake of test')
self.xfrsess.ClientClass = internal_raise
self.xfrsess.dns_xfrout_start(self.sock, self.mdata)
self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.SERVFAIL())
self.assertEqual(self.sock.read_msg().get_rcode(), Rcode.SERVFAIL)
def test_dns_xfrout_start_noerror(self):
def noerror(msg, name, rrclass):
return Rcode.NOERROR()
return Rcode.NOERROR
self.xfrsess._xfrout_setup = noerror
def myreply(msg, sock):
@ -905,7 +905,7 @@ class TestXfroutSession(TestXfroutSessionBase):
def test_dns_xfrout_start_with_notcallable_xfrreqdone(self):
def noerror(msg, name, rrclass):
return Rcode.NOERROR()
return Rcode.NOERROR
self.xfrsess._xfrout_setup = noerror
def myreply(msg, sock):
@ -1177,7 +1177,7 @@ class TestXfroutSessionWithSQLite3(TestXfroutSessionBase):
self.assertEqual(self.get_counter('ixfr_ended'), 0)
XfroutSession._handle(self.xfrsess)
response = self.sock.read_msg(Message.PRESERVE_ORDER);
self.assertEqual(Rcode.NOERROR(), response.get_rcode())
self.assertEqual(Rcode.NOERROR, response.get_rcode())
self.check_axfr_stream(response)
self.assertEqual(self.xfrsess._request_type, RRType.AXFR)
self.assertNotEqual(self.xfrsess._request_type, RRType.IXFR)
@ -1191,7 +1191,7 @@ class TestXfroutSessionWithSQLite3(TestXfroutSessionBase):
self.create_request_data(ixfr=IXFR_NG_VERSION)
XfroutSession._handle(self.xfrsess)
response = self.sock.read_msg(Message.PRESERVE_ORDER);
self.assertEqual(Rcode.NOERROR(), response.get_rcode())
self.assertEqual(Rcode.NOERROR, response.get_rcode())
# This is an AXFR-style IXFR. So the question section should indicate
# that it's an IXFR resposne.
self.assertEqual(RRType.IXFR, response.get_question()[0].get_type())

48
src/bin/xfrout/xfrout.py.in
View File

@ -227,9 +227,9 @@ class XfroutSession():
self._tsig_key_ring)
tsig_error = self._tsig_ctx.verify(tsig_record, request_data)
if tsig_error != TSIGError.NOERROR:
return Rcode.NOTAUTH()
return Rcode.NOTAUTH
return Rcode.NOERROR()
return Rcode.NOERROR
def _parse_query_message(self, mdata):
''' parse query message to [socket,message]'''
@ -239,11 +239,11 @@ class XfroutSession():
Message.from_wire(msg, mdata)
except Exception as err: # Exception is too broad
logger.error(XFROUT_PARSE_QUERY_ERROR, err)
return Rcode.FORMERR(), None
return Rcode.FORMERR, None
# TSIG related checks
rcode = self._check_request_tsig(msg, mdata)
if rcode != Rcode.NOERROR():
if rcode != Rcode.NOERROR:
return rcode, msg
# Make sure the question is valid. This should be ensured by
@ -283,7 +283,7 @@ class XfroutSession():
logger.debug(DBG_XFROUT_TRACE, XFROUT_QUERY_REJECTED,
self._request_type, format_addrinfo(self._remote),
format_zone_str(zone_name, zone_class))
return Rcode.REFUSED(), msg
return Rcode.REFUSED, msg
return rcode, msg
@ -351,16 +351,16 @@ class XfroutSession():
'''
result, finder = self._datasrc_client.find_zone(zone_name)
if result != DataSourceClient.SUCCESS:
return (Rcode.NOTAUTH(), None)
return (Rcode.NOTAUTH, None)
result, soa_rrset, _ = finder.find(zone_name, RRType.SOA)
if result != ZoneFinder.SUCCESS:
return (Rcode.SERVFAIL(), None)
return (Rcode.SERVFAIL, None)
# Especially for database-based zones, a working zone may be in
# a broken state where it has more than one SOA RR. We proactively
# check the condition and abort the xfr attempt if we identify it.
if soa_rrset.get_rdata_count() != 1:
return (Rcode.SERVFAIL(), None)
return (Rcode.NOERROR(), soa_rrset)
return (Rcode.SERVFAIL, None)
return (Rcode.NOERROR, soa_rrset)
def __axfr_setup(self, zone_name):
'''Setup a zone iterator for AXFR or AXFR-style IXFR.
@ -379,16 +379,16 @@ class XfroutSession():
# update get_iterator() API so that we can distinguish "no such
# zone" and other cases (#1373). For now we consider all these
# cases as NOTAUTH.
return Rcode.NOTAUTH()
return Rcode.NOTAUTH
# If we are an authoritative name server for the zone, but fail
# to find the zone's SOA record in datasource, xfrout can't
# provide zone transfer for it.
self._soa = self._iterator.get_soa()
if self._soa is None or self._soa.get_rdata_count() != 1:
return Rcode.SERVFAIL()
return Rcode.SERVFAIL
return Rcode.NOERROR()
return Rcode.NOERROR
def __ixfr_setup(self, request_msg, zone_name, zone_class):
'''Setup a zone journal reader for IXFR.
@ -411,15 +411,15 @@ class XfroutSession():
if auth_rrset.get_rdata_count() != 1:
logger.info(XFROUT_IXFR_MULTIPLE_SOA,
format_addrinfo(self._remote))
return Rcode.FORMERR()
return Rcode.FORMERR
remote_soa = auth_rrset
if remote_soa is None:
logger.info(XFROUT_IXFR_NO_SOA, format_addrinfo(self._remote))
return Rcode.FORMERR()
return Rcode.FORMERR
# Retrieve the local SOA
rcode, self._soa = self._get_zone_soa(zone_name)
if rcode != Rcode.NOERROR():
if rcode != Rcode.NOERROR:
return rcode
# RFC1995 says "If an IXFR query with the same or newer version
@ -437,7 +437,7 @@ class XfroutSession():
logger.info(XFROUT_IXFR_UPTODATE, format_addrinfo(self._remote),
format_zone_str(zone_name, zone_class),
begin_serial, end_serial)
return Rcode.NOERROR()
return Rcode.NOERROR
# Set up the journal reader or fall back to AXFR-style IXFR
try:
@ -462,12 +462,12 @@ class XfroutSession():
# between these two operations. We treat it as NOTAUTH.
logger.warn(XFROUT_IXFR_NO_ZONE, format_addrinfo(self._remote),
format_zone_str(zone_name, zone_class))
return Rcode.NOTAUTH()
return Rcode.NOTAUTH
# Use the reader as the iterator to generate the response.
self._iterator = self._jnl_reader
return Rcode.NOERROR()
return Rcode.NOERROR
def _xfrout_setup(self, request_msg, zone_name, zone_class):
'''Setup a context for xfr responses according to the request type.
@ -500,17 +500,17 @@ class XfroutSession():
#TODO. create query message and parse header
if rcode_ is None: # Dropped by ACL
return
elif rcode_ == Rcode.NOTAUTH() or rcode_ == Rcode.REFUSED():
elif rcode_ == Rcode.NOTAUTH or rcode_ == Rcode.REFUSED:
return self._reply_query_with_error_rcode(msg, sock_fd, rcode_)
elif rcode_ != Rcode.NOERROR():
elif rcode_ != Rcode.NOERROR:
return self._reply_query_with_error_rcode(msg, sock_fd,
Rcode.FORMERR())
Rcode.FORMERR)
elif not quota_ok:
logger.warn(XFROUT_QUERY_QUOTA_EXCCEEDED, self._request_typestr,
format_addrinfo(self._remote),
self._server._max_transfers_out)
return self._reply_query_with_error_rcode(msg, sock_fd,
Rcode.REFUSED())
Rcode.REFUSED)
question = msg.get_question()[0]
zone_name = question.get_name()
@ -522,8 +522,8 @@ class XfroutSession():
except Exception as ex:
logger.error(XFROUT_XFR_TRANSFER_CHECK_ERROR, self._request_typestr,
format_addrinfo(self._remote), zone_str, ex)
rcode_ = Rcode.SERVFAIL()
if rcode_ != Rcode.NOERROR():
rcode_ = Rcode.SERVFAIL
if rcode_ != Rcode.NOERROR:
logger.info(XFROUT_XFR_TRANSFER_FAILED, self._request_typestr,
format_addrinfo(self._remote), zone_str, rcode_)
return self._reply_query_with_error_rcode(msg, sock_fd, rcode_)

6
src/lib/dns/python/tests/edns_python_test.py
View File

@ -108,8 +108,8 @@ class EDNSTest(unittest.TestCase):
def test_towire_renderer(self):
renderer = MessageRenderer()
extrcode_noerror = Rcode.NOERROR().get_extended_code()
extrcode_badvers = Rcode.BADVERS().get_extended_code()
extrcode_noerror = Rcode.NOERROR.get_extended_code()
extrcode_badvers = Rcode.BADVERS.get_extended_code()
self.assertEqual(1, self.edns_base.to_wire(renderer, extrcode_noerror))
wiredata = read_wire_data("edns_toWire1.wire")
@ -148,7 +148,7 @@ class EDNSTest(unittest.TestCase):
self.assertEqual(0, renderer.get_length())
def test_towire_buffer(self):
extrcode_noerror = Rcode.NOERROR().get_extended_code()
extrcode_noerror = Rcode.NOERROR.get_extended_code()
obuffer = bytes()
obuffer = self.edns_base.to_wire(obuffer, extrcode_noerror)

30
src/lib/dns/python/tests/message_python_test.py
View File

@ -59,8 +59,8 @@ LONG_TXT4 = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef012
def create_message():
message_render = Message(Message.RENDER)
message_render.set_qid(0x1035)
message_render.set_opcode(Opcode.QUERY())
message_render.set_rcode(Rcode.NOERROR())
message_render.set_opcode(Opcode.QUERY)
message_render.set_rcode(Rcode.NOERROR)
message_render.set_header_flag(Message.HEADERFLAG_QR)
message_render.set_header_flag(Message.HEADERFLAG_RD)
message_render.set_header_flag(Message.HEADERFLAG_AA)
@ -161,7 +161,7 @@ class MessageTest(unittest.TestCase):
def test_set_rcode(self):
self.assertRaises(TypeError, self.r.set_rcode, "wrong")
rcode = Rcode.BADVERS()
rcode = Rcode.BADVERS
self.r.set_rcode(rcode)
self.assertEqual(rcode, self.r.get_rcode())
@ -173,7 +173,7 @@ class MessageTest(unittest.TestCase):
def test_set_opcode(self):
self.assertRaises(TypeError, self.r.set_opcode, "wrong")
opcode = Opcode.IQUERY()
opcode = Opcode.IQUERY
self.r.set_opcode(opcode)
self.assertEqual(opcode, self.r.get_opcode())
@ -336,19 +336,19 @@ class MessageTest(unittest.TestCase):
renderer.get_data())
def test_to_wire_without_opcode(self):
self.r.set_rcode(Rcode.NOERROR())
self.r.set_rcode(Rcode.NOERROR)
self.assertRaises(InvalidMessageOperation, self.r.to_wire,
MessageRenderer())
def test_to_wire_without_rcode(self):
self.r.set_opcode(Opcode.QUERY())
self.r.set_opcode(Opcode.QUERY)
self.assertRaises(InvalidMessageOperation, self.r.to_wire,
MessageRenderer())
def __common_tsigmessage_setup(self, flags=[Message.HEADERFLAG_RD],
rrtype=RRType("A"), answer_data=None):
self.r.set_opcode(Opcode.QUERY())
self.r.set_rcode(Rcode.NOERROR())
self.r.set_opcode(Opcode.QUERY)
self.r.set_rcode(Rcode.NOERROR)
for flag in flags:
self.r.set_header_flag(flag)
if answer_data is not None:
@ -407,8 +407,8 @@ class MessageTest(unittest.TestCase):
self.__common_tsig_checks("message_toWire4.wire")
def test_to_wire_tsig_truncation3(self):
self.r.set_opcode(Opcode.QUERY())
self.r.set_rcode(Rcode.NOERROR())
self.r.set_opcode(Opcode.QUERY)
self.r.set_rcode(Rcode.NOERROR)
for i in range(1, 68):
self.r.add_question(Question(Name("www.example.com"),
RRClass("IN"), RRType(i)))
@ -469,11 +469,11 @@ test.example.com. 3600 IN A 192.0.2.2
self.assertEqual(msg_str, str(message_render))
def test_to_text_without_opcode(self):
self.r.set_rcode(Rcode.NOERROR())
self.r.set_rcode(Rcode.NOERROR)
self.assertRaises(InvalidMessageOperation, self.r.to_text)
def test_to_text_without_rcode(self):
self.r.set_opcode(Opcode.QUERY())
self.r.set_opcode(Opcode.QUERY)
self.assertRaises(InvalidMessageOperation, self.r.to_text)
def test_from_wire(self):
@ -488,8 +488,8 @@ test.example.com. 3600 IN A 192.0.2.2
message_parse = Message(0)
factoryFromFile(message_parse, "message_fromWire1")
self.assertEqual(0x1035, message_parse.get_qid())
self.assertEqual(Opcode.QUERY(), message_parse.get_opcode())
self.assertEqual(Rcode.NOERROR(), message_parse.get_rcode())
self.assertEqual(Opcode.QUERY, message_parse.get_opcode())
self.assertEqual(Rcode.NOERROR, message_parse.get_rcode())
self.assertTrue(message_parse.get_header_flag(Message.HEADERFLAG_QR))
self.assertTrue(message_parse.get_header_flag(Message.HEADERFLAG_RD))
self.assertTrue(message_parse.get_header_flag(Message.HEADERFLAG_AA))
@ -568,7 +568,7 @@ test.example.com. 3600 IN A 192.0.2.2
# Extended Rcode = BADVERS
message_parse = Message(Message.PARSE)
factoryFromFile(message_parse, "message_fromWire10.wire")
self.assertEqual(Rcode.BADVERS(), message_parse.get_rcode())
self.assertEqual(Rcode.BADVERS, message_parse.get_rcode())
# Maximum extended Rcode
message_parse.clear(Message.PARSE)

8
src/lib/dns/python/tests/messagerenderer_python_test.py
View File

@ -31,8 +31,8 @@ class MessageRendererTest(unittest.TestCase):
message = Message(Message.RENDER)
message.set_qid(123)
message.set_opcode(Opcode.QUERY())
message.set_rcode(Rcode.NOERROR())
message.set_opcode(Opcode.QUERY)
message.set_rcode(Rcode.NOERROR)
message.add_question(Question(name, c, t))
self.message1 = message
@ -40,8 +40,8 @@ class MessageRendererTest(unittest.TestCase):
message.set_qid(123)
message.set_header_flag(Message.HEADERFLAG_AA, True)
message.set_header_flag(Message.HEADERFLAG_QR, True)
message.set_opcode(Opcode.QUERY())
message.set_rcode(Rcode.NOERROR())
message.set_opcode(Opcode.QUERY)
message.set_rcode(Rcode.NOERROR)
message.add_question(Question(name, c, t))
rrset = RRset(name, c, t, ttl)
rrset.add_rdata(Rdata(t, c, "192.0.2.98"))

82
src/lib/dns/python/tests/opcode_python_test.py
View File

@ -34,53 +34,53 @@ class OpcodeTest(unittest.TestCase):
self.assertEqual(Opcode.UPDATE_CODE, Opcode(5).get_code())
self.assertEqual(Opcode.RESERVED15_CODE, Opcode(15).get_code())
self.assertEqual(Opcode.QUERY_CODE, Opcode.QUERY().get_code())
self.assertEqual(Opcode.IQUERY_CODE, Opcode.IQUERY().get_code())
self.assertEqual(Opcode.NOTIFY_CODE, Opcode.NOTIFY().get_code())
self.assertEqual(Opcode.UPDATE_CODE, Opcode.UPDATE().get_code())
self.assertEqual(Opcode.RESERVED15_CODE, Opcode.RESERVED15().get_code())
self.assertEqual(Opcode.QUERY_CODE, Opcode.QUERY.get_code())
self.assertEqual(Opcode.IQUERY_CODE, Opcode.IQUERY.get_code())
self.assertEqual(Opcode.NOTIFY_CODE, Opcode.NOTIFY.get_code())
self.assertEqual(Opcode.UPDATE_CODE, Opcode.UPDATE.get_code())
self.assertEqual(Opcode.RESERVED15_CODE, Opcode.RESERVED15.get_code())
def test_get_code(self):
self.assertEqual(0, Opcode.QUERY().get_code())
self.assertEqual(1, Opcode.IQUERY().get_code())
self.assertEqual(2, Opcode.STATUS().get_code())
self.assertEqual(3, Opcode.RESERVED3().get_code())
self.assertEqual(4, Opcode.NOTIFY().get_code())
self.assertEqual(5, Opcode.UPDATE().get_code())
self.assertEqual(6, Opcode.RESERVED6().get_code())
self.assertEqual(7, Opcode.RESERVED7().get_code())
self.assertEqual(8, Opcode.RESERVED8().get_code())
self.assertEqual(9, Opcode.RESERVED9().get_code())
self.assertEqual(10, Opcode.RESERVED10().get_code())
self.assertEqual(11, Opcode.RESERVED11().get_code())
self.assertEqual(12, Opcode.RESERVED12().get_code())
self.assertEqual(13, Opcode.RESERVED13().get_code())
self.assertEqual(14, Opcode.RESERVED14().get_code())
self.assertEqual(15, Opcode.RESERVED15().get_code())
self.assertEqual(0, Opcode.QUERY.get_code())
self.assertEqual(1, Opcode.IQUERY.get_code())
self.assertEqual(2, Opcode.STATUS.get_code())
self.assertEqual(3, Opcode.RESERVED3.get_code())
self.assertEqual(4, Opcode.NOTIFY.get_code())
self.assertEqual(5, Opcode.UPDATE.get_code())
self.assertEqual(6, Opcode.RESERVED6.get_code())
self.assertEqual(7, Opcode.RESERVED7.get_code())
self.assertEqual(8, Opcode.RESERVED8.get_code())
self.assertEqual(9, Opcode.RESERVED9.get_code())
self.assertEqual(10, Opcode.RESERVED10.get_code())
self.assertEqual(11, Opcode.RESERVED11.get_code())
self.assertEqual(12, Opcode.RESERVED12.get_code())
self.assertEqual(13, Opcode.RESERVED13.get_code())
self.assertEqual(14, Opcode.RESERVED14.get_code())
self.assertEqual(15, Opcode.RESERVED15.get_code())
def test_to_text(self):
self.assertEqual("QUERY", Opcode.QUERY().to_text())
self.assertEqual("QUERY", str(Opcode.QUERY()))
self.assertEqual("IQUERY", Opcode.IQUERY().to_text())
self.assertEqual("STATUS", Opcode.STATUS().to_text())
self.assertEqual("RESERVED3", Opcode.RESERVED3().to_text())
self.assertEqual("NOTIFY", Opcode.NOTIFY().to_text())
self.assertEqual("UPDATE", Opcode.UPDATE().to_text())
self.assertEqual("RESERVED6", Opcode.RESERVED6().to_text())
self.assertEqual("RESERVED7", Opcode.RESERVED7().to_text())
self.assertEqual("RESERVED8", Opcode.RESERVED8().to_text())
self.assertEqual("RESERVED9", Opcode.RESERVED9().to_text())
self.assertEqual("RESERVED10", Opcode.RESERVED10().to_text())
self.assertEqual("RESERVED11", Opcode.RESERVED11().to_text())
self.assertEqual("RESERVED12", Opcode.RESERVED12().to_text())
self.assertEqual("RESERVED13", Opcode.RESERVED13().to_text())
self.assertEqual("RESERVED14", Opcode.RESERVED14().to_text())
self.assertEqual("RESERVED15", Opcode.RESERVED15().to_text())
self.assertEqual("QUERY", Opcode.QUERY.to_text())
self.assertEqual("QUERY", str(Opcode.QUERY))
self.assertEqual("IQUERY", Opcode.IQUERY.to_text())
self.assertEqual("STATUS", Opcode.STATUS.to_text())
self.assertEqual("RESERVED3", Opcode.RESERVED3.to_text())
self.assertEqual("NOTIFY", Opcode.NOTIFY.to_text())
self.assertEqual("UPDATE", Opcode.UPDATE.to_text())
self.assertEqual("RESERVED6", Opcode.RESERVED6.to_text())
self.assertEqual("RESERVED7", Opcode.RESERVED7.to_text())
self.assertEqual("RESERVED8", Opcode.RESERVED8.to_text())
self.assertEqual("RESERVED9", Opcode.RESERVED9.to_text())
self.assertEqual("RESERVED10", Opcode.RESERVED10.to_text())
self.assertEqual("RESERVED11", Opcode.RESERVED11.to_text())
self.assertEqual("RESERVED12", Opcode.RESERVED12.to_text())
self.assertEqual("RESERVED13", Opcode.RESERVED13.to_text())
self.assertEqual("RESERVED14", Opcode.RESERVED14.to_text())
self.assertEqual("RESERVED15", Opcode.RESERVED15.to_text())
def test_richcmp(self):
o1 = Opcode.QUERY()
o2 = Opcode.NOTIFY()
o3 = Opcode.NOTIFY()
o1 = Opcode.QUERY
o2 = Opcode.NOTIFY
o3 = Opcode.NOTIFY
self.assertTrue(o2 == o3)
self.assertFalse(o2 != o3)
self.assertTrue(o1 != o2)

60
src/lib/dns/python/tests/rcode_python_test.py
View File

@ -54,36 +54,36 @@ class RcodeTest(unittest.TestCase):
self.assertEqual(Rcode.RESERVED15_CODE, Rcode(15).get_code())
self.assertEqual(Rcode.BADVERS_CODE, Rcode(16).get_code())
self.assertEqual(Rcode.NOERROR_CODE, Rcode.NOERROR().get_code())
self.assertEqual(Rcode.FORMERR_CODE, Rcode.FORMERR().get_code())
self.assertEqual(Rcode.NOTIMP_CODE, Rcode.NOTIMP().get_code())
self.assertEqual(Rcode.REFUSED_CODE, Rcode.REFUSED().get_code())
self.assertEqual(Rcode.RESERVED15_CODE, Rcode.RESERVED15().get_code())
self.assertEqual(Rcode.BADVERS_CODE, Rcode.BADVERS().get_code())
self.assertEqual(Rcode.NOERROR_CODE, Rcode.NOERROR.get_code())
self.assertEqual(Rcode.FORMERR_CODE, Rcode.FORMERR.get_code())
self.assertEqual(Rcode.NOTIMP_CODE, Rcode.NOTIMP.get_code())
self.assertEqual(Rcode.REFUSED_CODE, Rcode.REFUSED.get_code())
self.assertEqual(Rcode.RESERVED15_CODE, Rcode.RESERVED15.get_code())
self.assertEqual(Rcode.BADVERS_CODE, Rcode.BADVERS.get_code())
def test_get_code(self):
self.assertEqual(0, Rcode.NOERROR().get_code())
self.assertEqual(1, Rcode.FORMERR().get_code())
self.assertEqual(2, Rcode.SERVFAIL().get_code())
self.assertEqual(3, Rcode.NXDOMAIN().get_code())
self.assertEqual(4, Rcode.NOTIMP().get_code())
self.assertEqual(5, Rcode.REFUSED().get_code())
self.assertEqual(6, Rcode.YXDOMAIN().get_code())
self.assertEqual(7, Rcode.YXRRSET().get_code())
self.assertEqual(8, Rcode.NXRRSET().get_code())
self.assertEqual(9, Rcode.NOTAUTH().get_code())
self.assertEqual(10, Rcode.NOTZONE().get_code())
self.assertEqual(11, Rcode.RESERVED11().get_code())
self.assertEqual(12, Rcode.RESERVED12().get_code())
self.assertEqual(13, Rcode.RESERVED13().get_code())
self.assertEqual(14, Rcode.RESERVED14().get_code())
self.assertEqual(15, Rcode.RESERVED15().get_code())
self.assertEqual(16, Rcode.BADVERS().get_code())
self.assertEqual(0, Rcode.NOERROR.get_code())
self.assertEqual(1, Rcode.FORMERR.get_code())
self.assertEqual(2, Rcode.SERVFAIL.get_code())
self.assertEqual(3, Rcode.NXDOMAIN.get_code())
self.assertEqual(4, Rcode.NOTIMP.get_code())
self.assertEqual(5, Rcode.REFUSED.get_code())
self.assertEqual(6, Rcode.YXDOMAIN.get_code())
self.assertEqual(7, Rcode.YXRRSET.get_code())
self.assertEqual(8, Rcode.NXRRSET.get_code())
self.assertEqual(9, Rcode.NOTAUTH.get_code())
self.assertEqual(10, Rcode.NOTZONE.get_code())
self.assertEqual(11, Rcode.RESERVED11.get_code())
self.assertEqual(12, Rcode.RESERVED12.get_code())
self.assertEqual(13, Rcode.RESERVED13.get_code())
self.assertEqual(14, Rcode.RESERVED14.get_code())
self.assertEqual(15, Rcode.RESERVED15.get_code())
self.assertEqual(16, Rcode.BADVERS.get_code())
def test_get_extended_code(self):
self.assertEqual(0, Rcode.NOERROR().get_extended_code())
self.assertEqual(0, Rcode.YXRRSET().get_extended_code())
self.assertEqual(1, Rcode.BADVERS().get_extended_code())
self.assertEqual(0, Rcode.NOERROR.get_extended_code())
self.assertEqual(0, Rcode.YXRRSET.get_extended_code())
self.assertEqual(1, Rcode.BADVERS.get_extended_code())
self.assertEqual(0xab, Rcode(0xabf).get_extended_code())
self.assertEqual(0xff, Rcode(0xfff).get_extended_code())
@ -107,13 +107,13 @@ class RcodeTest(unittest.TestCase):
self.assertEqual("RESERVED15", Rcode(15).to_text())
self.assertEqual("BADVERS", Rcode(16).to_text())
self.assertEqual("17", Rcode(Rcode.BADVERS().get_code() + 1).to_text())
self.assertEqual("17", Rcode(Rcode.BADVERS.get_code() + 1).to_text())
self.assertEqual("4095", Rcode(0xfff).to_text())
def test_richcmp(self):
r1 = Rcode.NOERROR()
r2 = Rcode.FORMERR()
r3 = Rcode.FORMERR()
r1 = Rcode.NOERROR
r2 = Rcode.FORMERR
r3 = Rcode.FORMERR
self.assertTrue(r2 == r3)
self.assertTrue(r1 != r2)
self.assertFalse(r1 == r2)

8
src/lib/dns/python/tests/tsig_python_test.py
View File

@ -61,10 +61,10 @@ class TSIGContextTest(unittest.TestCase):
def createMessageAndSign(self, id, qname, ctx, message_flags=RD_FLAG,
qtype=RRType.A, answer_data=None,
answer_type=None, add_question=True,
rcode=Rcode.NOERROR()):
rcode=Rcode.NOERROR):
self.message.clear(Message.RENDER)
self.message.set_qid(id)
self.message.set_opcode(Opcode.QUERY())
self.message.set_opcode(Opcode.QUERY)
self.message.set_rcode(rcode)
if (message_flags & QR_FLAG) != 0:
self.message.set_header_flag(Message.HEADERFLAG_QR)
@ -120,7 +120,7 @@ class TSIGContextTest(unittest.TestCase):
self.assertEqual(TSIGContext.STATE_INIT, self.tsig_ctx.get_state())
# And there should be no error code.
self.assertEqual(TSIGError(Rcode.NOERROR()), self.tsig_ctx.get_error())
self.assertEqual(TSIGError(Rcode.NOERROR), self.tsig_ctx.get_error())
# No message signed yet
self.assertRaises(TSIGContextError, self.tsig_ctx.last_had_signature)
@ -329,7 +329,7 @@ class TSIGContextTest(unittest.TestCase):
tsig = self.createMessageAndSign(test_qid, self.test_name,
self.tsig_verify_ctx,
QR_FLAG, RRType.SOA, None, None,
True, Rcode.NOTAUTH())
True, Rcode.NOTAUTH)
expected_otherdata = b"\x00\x00\x4d\xa8\xbe\x86"
expected_mac = b"\xd4\xb0\x43\xf6\xf4\x44\x95\xec\x8a\x01\x26" +\

26
src/lib/dns/python/tests/tsigerror_python_test.py
View File

@ -28,7 +28,7 @@ class TSIGErrorTest(unittest.TestCase):
def test_from_rcode(self):
# We use RCODE for code values from 0-15.
self.assertEqual(0, TSIGError(Rcode.NOERROR()).get_code())
self.assertEqual(0, TSIGError(Rcode.NOERROR).get_code())
self.assertEqual(15, TSIGError(Rcode(15)).get_code())
# From error code 16 TSIG errors define a separate space, so passing
@ -50,19 +50,19 @@ class TSIGErrorTest(unittest.TestCase):
self.assertEqual(TSIGError.BAD_TIME_CODE, TSIGError.BAD_TIME.get_code())
def test_equal(self):
self.assertTrue(TSIGError.NOERROR == TSIGError(Rcode.NOERROR()))
self.assertTrue(TSIGError(Rcode.NOERROR()) == TSIGError.NOERROR)
self.assertTrue(TSIGError.NOERROR == TSIGError(Rcode.NOERROR))
self.assertTrue(TSIGError(Rcode.NOERROR) == TSIGError.NOERROR)
self.assertTrue(TSIGError.BAD_SIG == TSIGError(16))
self.assertTrue(TSIGError(16) == TSIGError.BAD_SIG)
def test_nequal(self):
self.assertTrue(TSIGError.BAD_KEY != TSIGError(Rcode.NOERROR()))
self.assertTrue(TSIGError(Rcode.NOERROR()) != TSIGError.BAD_KEY)
self.assertTrue(TSIGError.BAD_KEY != TSIGError(Rcode.NOERROR))
self.assertTrue(TSIGError(Rcode.NOERROR) != TSIGError.BAD_KEY)
def test_to_text(self):
# TSIGError derived from the standard Rcode
self.assertEqual("NOERROR", TSIGError(Rcode.NOERROR()).to_text())
self.assertEqual("NOERROR", TSIGError(Rcode.NOERROR).to_text())
# Well known TSIG errors
self.assertEqual("BADSIG", TSIGError.BAD_SIG.to_text())
@ -74,21 +74,21 @@ class TSIGErrorTest(unittest.TestCase):
self.assertEqual("65535", TSIGError(65535).to_text());
# also check str() works same way
self.assertEqual("NOERROR", str(TSIGError(Rcode.NOERROR())))
self.assertEqual("NOERROR", str(TSIGError(Rcode.NOERROR)))
self.assertEqual("BADSIG", str(TSIGError.BAD_SIG))
def test_to_rcode(self):
# TSIGError derived from the standard Rcode
self.assertEqual(Rcode.NOERROR(), TSIGError(Rcode.NOERROR()).to_rcode())
self.assertEqual(Rcode.NOERROR, TSIGError(Rcode.NOERROR).to_rcode())
# Well known TSIG errors
self.assertEqual(Rcode.NOTAUTH(), TSIGError.BAD_SIG.to_rcode())
self.assertEqual(Rcode.NOTAUTH(), TSIGError.BAD_KEY.to_rcode())
self.assertEqual(Rcode.NOTAUTH(), TSIGError.BAD_TIME.to_rcode())
self.assertEqual(Rcode.NOTAUTH, TSIGError.BAD_SIG.to_rcode())
self.assertEqual(Rcode.NOTAUTH, TSIGError.BAD_KEY.to_rcode())
self.assertEqual(Rcode.NOTAUTH, TSIGError.BAD_TIME.to_rcode())
# Unknown (or not yet supported) codes are treated as SERVFAIL.
self.assertEqual(Rcode.SERVFAIL(), TSIGError(19).to_rcode())
self.assertEqual(Rcode.SERVFAIL(), TSIGError(65535).to_rcode())
self.assertEqual(Rcode.SERVFAIL, TSIGError(19).to_rcode())
self.assertEqual(Rcode.SERVFAIL, TSIGError(65535).to_rcode())
# Check there's no redundant refcount (which would cause leak)
self.assertEqual(1, sys.getrefcount(TSIGError.BAD_SIG.to_rcode()))

70
src/lib/python/isc/ddns/session.py
View File

@ -248,14 +248,14 @@ class UpdateSession:
self.__check_update_acl(self.__zname, self.__zclass)
self._create_diff()
prereq_result = self.__check_prerequisites()
if prereq_result != Rcode.NOERROR():
if prereq_result != Rcode.NOERROR:
self.__make_response(prereq_result)
return UPDATE_ERROR, self.__zname, self.__zclass
update_result = self.__do_update()
if update_result != Rcode.NOERROR():
if update_result != Rcode.NOERROR:
self.__make_response(update_result)
return UPDATE_ERROR, self.__zname, self.__zclass
self.__make_response(Rcode.NOERROR())
self.__make_response(Rcode.NOERROR)
return UPDATE_SUCCESS, self.__zname, self.__zclass
except UpdateError as e:
if not e.nolog:
@ -272,7 +272,7 @@ class UpdateSession:
except isc.datasrc.Error as e:
logger.error(LIBDDNS_DATASRC_ERROR,
ClientFormatter(self.__client_addr, self.__tsig), e)
self.__make_response(Rcode.SERVFAIL())
self.__make_response(Rcode.SERVFAIL)
return UPDATE_ERROR, None, None
def _get_update_zone(self):
@ -295,11 +295,11 @@ class UpdateSession:
n_zones = self.__message.get_rr_count(SECTION_ZONE)
if n_zones != 1:
raise UpdateError('Invalid number of records in zone section: ' +
str(n_zones), None, None, Rcode.FORMERR())
str(n_zones), None, None, Rcode.FORMERR)
zrecord = self.__message.get_question()[0]
if zrecord.get_type() != RRType.SOA:
raise UpdateError('update zone section contains non-SOA',
None, None, Rcode.FORMERR())
None, None, Rcode.FORMERR)
# See if we're serving a primary zone specified in the zone section.
zname = zrecord.get_name()
@ -316,12 +316,12 @@ class UpdateSession:
logger.debug(DBGLVL_TRACE_BASIC, LIBDDNS_UPDATE_FORWARD_FAIL,
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('forward', zname, zclass, Rcode.NOTIMP(), True)
raise UpdateError('forward', zname, zclass, Rcode.NOTIMP, True)
# zone wasn't found
logger.debug(DBGLVL_TRACE_BASIC, LIBDDNS_UPDATE_NOTAUTH,
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('notauth', zname, zclass, Rcode.NOTAUTH(), True)
raise UpdateError('notauth', zname, zclass, Rcode.NOTAUTH, True)
def _create_diff(self):
'''
@ -352,7 +352,7 @@ class UpdateSession:
logger.info(LIBDDNS_UPDATE_DENIED,
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('rejected', zname, zclass, Rcode.REFUSED(), True)
raise UpdateError('rejected', zname, zclass, Rcode.REFUSED, True)
if action == DROP:
logger.info(LIBDDNS_UPDATE_DROPPED,
ClientFormatter(self.__client_addr, self.__tsig),
@ -459,7 +459,7 @@ class UpdateSession:
def __check_prerequisites(self):
'''Check the prerequisites section of the UPDATE Message.
RFC2136 Section 2.4.
Returns a dns Rcode signaling either no error (Rcode.NOERROR())
Returns a dns Rcode signaling either no error (Rcode.NOERROR)
or that one of the prerequisites failed (any other Rcode).
'''
@ -473,7 +473,7 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.NOTZONE()
return Rcode.NOTZONE
# Algorithm taken from RFC2136 Section 3.2
if rrset.get_class() == RRClass.ANY:
@ -483,10 +483,10 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
elif rrset.get_type() == RRType.ANY:
if not self.__prereq_name_in_use(rrset):
rcode = Rcode.NXDOMAIN()
rcode = Rcode.NXDOMAIN
logger.info(LIBDDNS_PREREQ_NAME_IN_USE_FAILED,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
@ -494,7 +494,7 @@ class UpdateSession:
return rcode
else:
if not self.__prereq_rrset_exists(rrset):
rcode = Rcode.NXRRSET()
rcode = Rcode.NXRRSET
logger.info(LIBDDNS_PREREQ_RRSET_EXISTS_FAILED,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
@ -507,10 +507,10 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
elif rrset.get_type() == RRType.ANY:
if not self.__prereq_name_not_in_use(rrset):
rcode = Rcode.YXDOMAIN()
rcode = Rcode.YXDOMAIN
logger.info(LIBDDNS_PREREQ_NAME_NOT_IN_USE_FAILED,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
@ -518,7 +518,7 @@ class UpdateSession:
return rcode
else:
if not self.__prereq_rrset_does_not_exist(rrset):
rcode = Rcode.YXRRSET()
rcode = Rcode.YXRRSET
logger.info(LIBDDNS_PREREQ_RRSET_DOES_NOT_EXIST_FAILED,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
@ -530,7 +530,7 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
else:
collect_rrsets(exact_match_rrsets, rrset)
else:
@ -538,11 +538,11 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
for collected_rrset in exact_match_rrsets:
if not self.__prereq_rrset_exists_value(collected_rrset):
rcode = Rcode.NXRRSET()
rcode = Rcode.NXRRSET
logger.info(LIBDDNS_PREREQ_RRSET_EXISTS_VAL_FAILED,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
@ -550,7 +550,7 @@ class UpdateSession:
return rcode
# All prerequisites are satisfied
return Rcode.NOERROR()
return Rcode.NOERROR
def __set_soa_rrset(self, rrset):
'''Sets the given rrset to the member __added_soa (which
@ -570,7 +570,7 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.NOTZONE()
return Rcode.NOTZONE
if rrset.get_class() == self.__zclass:
# In fact, all metatypes are in a specific range,
# so one check can test TKEY to ANY
@ -581,7 +581,7 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
if rrset.get_type() == RRType.SOA:
# In case there's multiple soa records in the update
# somehow, just take the last
@ -593,40 +593,40 @@ class UpdateSession:
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
if rrset.get_rdata_count() > 0:
logger.info(LIBDDNS_UPDATE_DELETE_RRSET_NOT_EMPTY,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
if rrset.get_type().get_code() >= 249 and\
rrset.get_type().get_code() <= 254:
logger.info(LIBDDNS_UPDATE_DELETE_BAD_TYPE,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
elif rrset.get_class() == RRClass.NONE:
if rrset.get_ttl().get_value() != 0:
logger.info(LIBDDNS_UPDATE_DELETE_RR_NONZERO_TTL,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
if rrset.get_type().get_code() >= 249:
logger.info(LIBDDNS_UPDATE_DELETE_RR_BAD_TYPE,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.FORMERR
else:
logger.info(LIBDDNS_UPDATE_BAD_CLASS,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
RRsetFormatter(rrset))
return Rcode.FORMERR()
return Rcode.NOERROR()
return Rcode.FORMERR
return Rcode.NOERROR
def __do_update_add_single_rr(self, rr, existing_rrset):
'''Helper for __do_update_add_rrs_to_rrset: only add the
@ -800,7 +800,7 @@ class UpdateSession:
# for now servfail on such a broken state
if result != ZoneFinder.SUCCESS:
raise UpdateError("Error finding SOA record in datasource.",
self.__zname, self.__zclass, Rcode.SERVFAIL())
self.__zname, self.__zclass, Rcode.SERVFAIL)
serial_operation = DDNS_SOA()
if self.__added_soa is not None and\
serial_operation.soa_update_check(old_soa, self.__added_soa):
@ -820,7 +820,7 @@ class UpdateSession:
'''
# prescan
prescan_result = self.__do_prescan()
if prescan_result != Rcode.NOERROR():
if prescan_result != Rcode.NOERROR:
return prescan_result
# update
@ -850,13 +850,13 @@ class UpdateSession:
self.__do_update_delete_rrs_from_rrset(rrset)
self.__diff.commit()
return Rcode.NOERROR()
return Rcode.NOERROR
except isc.datasrc.Error as dse:
logger.info(LIBDDNS_UPDATE_DATASRC_ERROR, dse)
return Rcode.SERVFAIL()
return Rcode.SERVFAIL
except Exception as uce:
logger.error(LIBDDNS_UPDATE_UNCAUGHT_EXCEPTION,
ClientFormatter(self.__client_addr),
ZoneFormatter(self.__zname, self.__zclass),
uce)
return Rcode.SERVFAIL()
return Rcode.SERVFAIL

164
src/lib/python/isc/ddns/tests/session_tests.py
View File

@ -42,8 +42,8 @@ def create_update_msg(zones=[TEST_ZONE_RECORD], prerequisites=[],
updates=[], tsig_key=None):
msg = Message(Message.RENDER)
msg.set_qid(5353) # arbitrary chosen
msg.set_opcode(Opcode.UPDATE())
msg.set_rcode(Rcode.NOERROR())
msg.set_opcode(Opcode.UPDATE)
msg.set_rcode(Rcode.NOERROR)
for z in zones:
msg.add_question(z)
for p in prerequisites:
@ -216,7 +216,7 @@ class SessionTestBase(unittest.TestCase):
'''Perform common checks on update resposne message.'''
self.assertTrue(msg.get_header_flag(Message.HEADERFLAG_QR))
# note: we convert opcode to text it'd be more helpful on failure.
self.assertEqual(Opcode.UPDATE().to_text(), msg.get_opcode().to_text())
self.assertEqual(Opcode.UPDATE.to_text(), msg.get_opcode().to_text())
self.assertEqual(expected_rcode.to_text(), msg.get_rcode().to_text())
# All sections should be cleared
self.assertEqual(0, msg.get_rr_count(SECTION_ZONE))
@ -305,20 +305,20 @@ class SessionTest(SessionTestBase):
self.assertEqual(UPDATE_ERROR, result)
self.assertEqual(None, zname)
self.assertEqual(None, zclass)
self.check_response(session.get_message(), Rcode.FORMERR())
self.check_response(session.get_message(), Rcode.FORMERR)
# Zone section contains multiple records
msg = create_update_msg(zones=[TEST_ZONE_RECORD, TEST_ZONE_RECORD])
session = UpdateSession(msg, TEST_CLIENT4, None)
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.FORMERR())
self.check_response(session.get_message(), Rcode.FORMERR)
# Zone section's type is not SOA
msg = create_update_msg(zones=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
RRType.A)])
session = UpdateSession(msg, TEST_CLIENT4, None)
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.FORMERR())
self.check_response(session.get_message(), Rcode.FORMERR)
def test_update_secondary(self):
# specified zone is configured as a secondary. Since this
@ -330,7 +330,7 @@ class SessionTest(SessionTestBase):
ZoneConfig({(TEST_ZONE_NAME, TEST_RRCLASS)},
TEST_RRCLASS, self._datasrc_client))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.NOTIMP())
self.check_response(session.get_message(), Rcode.NOTIMP)
def check_notauth(self, zname, zclass=TEST_RRCLASS):
'''Common test sequence for the 'notauth' test'''
@ -339,7 +339,7 @@ class SessionTest(SessionTestBase):
ZoneConfig({(TEST_ZONE_NAME, TEST_RRCLASS)},
TEST_RRCLASS, self._datasrc_client))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.NOTAUTH())
self.check_response(session.get_message(), Rcode.NOTAUTH)
def test_update_notauth(self):
'''Update attempt for non authoritative zones'''
@ -364,7 +364,7 @@ class SessionTest(SessionTestBase):
TEST_RRCLASS,
BadDataSourceClient()))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.SERVFAIL())
self.check_response(session.get_message(), Rcode.SERVFAIL)
def test_foreach_rr_in_rrset(self):
rrset = create_rrset("www.example.org", TEST_RRCLASS,
@ -632,7 +632,7 @@ class SessionTest(SessionTestBase):
self.assertEqual(expected.to_text(),
session._UpdateSession__message.get_rcode().to_text())
# And that the result looks right
if expected == Rcode.NOERROR():
if expected == Rcode.NOERROR:
self.assertEqual(UPDATE_SUCCESS, result)
else:
self.assertEqual(UPDATE_ERROR, result)
@ -672,7 +672,7 @@ class SessionTest(SessionTestBase):
self.assertEqual(expected.to_text(),
session._UpdateSession__message.get_rcode().to_text())
# And that the result looks right
if expected == Rcode.NOERROR():
if expected == Rcode.NOERROR:
self.assertEqual(UPDATE_SUCCESS, result)
else:
self.assertEqual(UPDATE_ERROR, result)
@ -727,36 +727,36 @@ class SessionTest(SessionTestBase):
name_not_in_use_no = create_rrset("www.example.org", RRClass.NONE,
RRType.ANY, 0)
# check 'no' result codes
self.check_prerequisite_result(Rcode.NXRRSET(),
self.check_prerequisite_result(Rcode.NXRRSET,
[ rrset_exists_no ])
self.check_prerequisite_result(Rcode.NXRRSET(),
self.check_prerequisite_result(Rcode.NXRRSET,
[ rrset_exists_value_no ])
self.check_prerequisite_result(Rcode.YXRRSET(),
self.check_prerequisite_result(Rcode.YXRRSET,
[ rrset_does_not_exist_no ])
self.check_prerequisite_result(Rcode.NXDOMAIN(),
self.check_prerequisite_result(Rcode.NXDOMAIN,
[ name_in_use_no ])
self.check_prerequisite_result(Rcode.YXDOMAIN(),
self.check_prerequisite_result(Rcode.YXDOMAIN,
[ name_not_in_use_no ])
# the 'yes' codes should result in ok
# individually
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ rrset_exists_yes ] )
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ rrset_exists_value_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ rrset_does_not_exist_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ name_in_use_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ name_not_in_use_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ rrset_exists_value_1,
rrset_exists_value_2,
rrset_exists_value_3])
# and together
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ rrset_exists_yes,
rrset_exists_value_yes,
rrset_does_not_exist_yes,
@ -768,7 +768,7 @@ class SessionTest(SessionTestBase):
# try out a permutation, note that one rrset is split up,
# and the order of the RRs should not matter
self.check_prerequisite_result(Rcode.NOERROR(),
self.check_prerequisite_result(Rcode.NOERROR,
[ rrset_exists_value_3,
rrset_exists_yes,
rrset_exists_value_2,
@ -777,7 +777,7 @@ class SessionTest(SessionTestBase):
# Should fail on the first error, even if most of the
# prerequisites are ok
self.check_prerequisite_result(Rcode.NXDOMAIN(),
self.check_prerequisite_result(Rcode.NXDOMAIN,
[ rrset_exists_value_3,
rrset_exists_yes,
rrset_exists_value_2,
@ -787,38 +787,38 @@ class SessionTest(SessionTestBase):
def test_prerequisite_notzone(self):
rrset = create_rrset("some.other.zone.", RRClass.ANY, RRType.SOA, 0)
self.check_prerequisite_result(Rcode.NOTZONE(), [ rrset ])
self.check_prerequisite_result(Rcode.NOTZONE, [ rrset ])
def test_prerequisites_formerr(self):
# test for form errors in the prerequisite section
# Class ANY, non-zero TTL
rrset = create_rrset("example.org", RRClass.ANY, RRType.SOA, 1)
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
self.check_prerequisite_result(Rcode.FORMERR, [ rrset ])
# Class ANY, but with rdata
rrset = create_rrset("example.org", RRClass.ANY, RRType.A, 0,
[ b'\x00\x00\x00\x00' ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
self.check_prerequisite_result(Rcode.FORMERR, [ rrset ])
# Class NONE, non-zero TTL
rrset = create_rrset("example.org", RRClass.NONE, RRType.SOA, 1)
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
self.check_prerequisite_result(Rcode.FORMERR, [ rrset ])
# Class NONE, but with rdata
rrset = create_rrset("example.org", RRClass.NONE, RRType.A, 0,
[ b'\x00\x00\x00\x00' ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
self.check_prerequisite_result(Rcode.FORMERR, [ rrset ])
# Matching class and type, but non-zero TTL
rrset = create_rrset("www.example.org", RRClass.IN, RRType.A, 1,
[ "192.0.2.1" ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
self.check_prerequisite_result(Rcode.FORMERR, [ rrset ])
# Completely different class
rrset = create_rrset("example.org", RRClass.CH, RRType.TXT, 0,
[ "foo" ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
self.check_prerequisite_result(Rcode.FORMERR, [ rrset ])
def __prereq_helper(self, method, expected, rrset):
'''Calls the given method with self._datasrc_client
@ -907,7 +907,7 @@ class SessionTest(SessionTestBase):
RRType.ANY, 0)
# Test a prerequisite that would fail
self.check_full_handle_result(Rcode.NXDOMAIN(), [], [ name_in_use_no ])
self.check_full_handle_result(Rcode.NXDOMAIN, [], [ name_in_use_no ])
# Change ACL so that it would be denied
self._acl_map = {(TEST_ZONE_NAME, TEST_RRCLASS):
@ -915,7 +915,7 @@ class SessionTest(SessionTestBase):
# The prerequisite should now not be reached; it should fail on the
# ACL
self.check_full_handle_result(Rcode.REFUSED(), [], [ name_in_use_no ])
self.check_full_handle_result(Rcode.REFUSED, [], [ name_in_use_no ])
def test_prescan(self):
'''Test whether the prescan succeeds on data that is ok, and whether
@ -923,29 +923,29 @@ class SessionTest(SessionTestBase):
# prepare a set of correct update statements
self.__initialize_update_rrsets()
self.check_prescan_result(Rcode.NOERROR(), [ self.rrset_update_a ])
self.check_prescan_result(Rcode.NOERROR, [ self.rrset_update_a ])
# check if soa is noticed
self.check_prescan_result(Rcode.NOERROR(), [ self.rrset_update_soa ],
self.check_prescan_result(Rcode.NOERROR, [ self.rrset_update_soa ],
self.rrset_update_soa)
# Other types of succesful prechecks
self.check_prescan_result(Rcode.NOERROR(), [ self.rrset_update_soa2 ],
self.check_prescan_result(Rcode.NOERROR, [ self.rrset_update_soa2 ],
self.rrset_update_soa2)
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_del_name ])
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_del_name_apex ])
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset ])
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_del_mx_apex ])
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset_part ])
# and check a few permutations of the above
# all of them (with one of the soas)
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[
self.rrset_update_a,
self.rrset_update_soa,
@ -960,16 +960,16 @@ class SessionTest(SessionTestBase):
# Two soas. Should we reject or simply use the last?
# (RFC is not really explicit on this, but between the lines I read
# use the last)
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_soa,
self.rrset_update_soa2 ],
self.rrset_update_soa2)
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[ self.rrset_update_soa2,
self.rrset_update_soa ],
self.rrset_update_soa)
self.check_prescan_result(Rcode.NOERROR(),
self.check_prescan_result(Rcode.NOERROR,
[
self.rrset_update_del_mx_apex,
self.rrset_update_del_name,
@ -985,35 +985,35 @@ class SessionTest(SessionTestBase):
'''Test whether prescan fails on bad data'''
# out of zone data
rrset = create_rrset("different.zone", RRClass.ANY, RRType.TXT, 0)
self.check_prescan_result(Rcode.NOTZONE(), [ rrset ])
self.check_prescan_result(Rcode.NOTZONE, [ rrset ])
# forbidden type, zone class
rrset = create_rrset(TEST_ZONE_NAME, TEST_RRCLASS, RRType.ANY, 0,
[ b'\x00' ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
self.check_prescan_result(Rcode.FORMERR, [ rrset ])
# non-zero TTL, class ANY
rrset = create_rrset(TEST_ZONE_NAME, RRClass.ANY, RRType.TXT, 1)
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
self.check_prescan_result(Rcode.FORMERR, [ rrset ])
# non-zero Rdata, class ANY
rrset = create_rrset(TEST_ZONE_NAME, RRClass.ANY, RRType.TXT, 0,
[ "foo" ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
self.check_prescan_result(Rcode.FORMERR, [ rrset ])
# forbidden type, class ANY
rrset = create_rrset(TEST_ZONE_NAME, RRClass.ANY, RRType.AXFR, 0,
[ b'\x00' ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
self.check_prescan_result(Rcode.FORMERR, [ rrset ])
# non-zero TTL, class NONE
rrset = create_rrset(TEST_ZONE_NAME, RRClass.NONE, RRType.TXT, 1)
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
self.check_prescan_result(Rcode.FORMERR, [ rrset ])
# forbidden type, class NONE
rrset = create_rrset(TEST_ZONE_NAME, RRClass.NONE, RRType.AXFR, 0,
[ b'\x00' ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
self.check_prescan_result(Rcode.FORMERR, [ rrset ])
def __check_inzone_data(self, expected_result, name, rrtype,
expected_rrset = None):
@ -1066,7 +1066,7 @@ class SessionTest(SessionTestBase):
self.orig_a_rrset)
# Add two rrs
self.check_full_handle_result(Rcode.NOERROR(), [ self.rrset_update_a ])
self.check_full_handle_result(Rcode.NOERROR, [ self.rrset_update_a ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
@ -1074,7 +1074,7 @@ class SessionTest(SessionTestBase):
extended_a_rrset)
# Adding the same RRsets should not make a difference.
self.check_full_handle_result(Rcode.NOERROR(), [ self.rrset_update_a ])
self.check_full_handle_result(Rcode.NOERROR, [ self.rrset_update_a ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
@ -1082,7 +1082,7 @@ class SessionTest(SessionTestBase):
extended_a_rrset)
# Now delete those two, and we should end up with the original RRset
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset_part ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
@ -1090,7 +1090,7 @@ class SessionTest(SessionTestBase):
self.orig_a_rrset)
# 'Deleting' them again should make no difference
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset_part ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
@ -1099,14 +1099,14 @@ class SessionTest(SessionTestBase):
# But deleting the entire rrset, independent of its contents, should
# work
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("www.example.org"),
RRType.A)
# Check that if we update the SOA, it is updated to our value
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_soa2 ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
@ -1126,7 +1126,7 @@ class SessionTest(SessionTestBase):
RRClass.ANY,
RRType.A,
0)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ rrset_delete_glue ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("sub.example.org."),
@ -1141,7 +1141,7 @@ class SessionTest(SessionTestBase):
RRClass.ANY,
RRType.A,
0)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ rrset_delete_nonexistent_glue ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("sub.example.org."),
@ -1157,7 +1157,7 @@ class SessionTest(SessionTestBase):
RRType.A)
rrset = create_rrset("new.example.org", TEST_RRCLASS, RRType.A,
3600, [ "192.0.2.1", "192.0.2.2" ])
self.check_full_handle_result(Rcode.NOERROR(), [ rrset ])
self.check_full_handle_result(Rcode.NOERROR, [ rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("new.example.org"),
RRType.A,
@ -1170,7 +1170,7 @@ class SessionTest(SessionTestBase):
RRType.TXT)
rrset = create_rrset("new.example.org", TEST_RRCLASS, RRType.TXT,
3600, [ "foo" ])
self.check_full_handle_result(Rcode.NOERROR(), [ rrset ])
self.check_full_handle_result(Rcode.NOERROR, [ rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("new.example.org"),
RRType.TXT,
@ -1200,7 +1200,7 @@ class SessionTest(SessionTestBase):
rrset3 = create_rrset("new_a.example.org", TEST_RRCLASS, RRType.A,
3600, [ "192.0.2.2" ])
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ rrset1, rrset2, rrset3 ])
# The update should have merged rrset1 and rrset3
@ -1230,14 +1230,14 @@ class SessionTest(SessionTestBase):
RRType.A)
# Delete the entire name
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_name ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("www.example.org"),
RRType.A)
# Should still be gone after pointless second delete
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_name ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("www.example.org"),
@ -1285,7 +1285,7 @@ class SessionTest(SessionTestBase):
# Check that we cannot delete the SOA record by direct deletion
# both by name+type and by full rrset
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_soa_apex,
self.rrset_update_soa_del ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
@ -1295,7 +1295,7 @@ class SessionTest(SessionTestBase):
# If we delete everything at the apex, the SOA and NS rrsets should be
# untouched (but serial will be incremented)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_name_apex ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
@ -1312,7 +1312,7 @@ class SessionTest(SessionTestBase):
# Deleting the NS rrset by name and type only, it should also be left
# untouched
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_ns_apex ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
@ -1327,7 +1327,7 @@ class SessionTest(SessionTestBase):
short_ns_rrset = create_rrset("example.org", TEST_RRCLASS,
RRType.NS, 3600,
[ "ns3.example.org." ])
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset_ns ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
@ -1341,7 +1341,7 @@ class SessionTest(SessionTestBase):
new_ns = create_rrset("example.org", TEST_RRCLASS, RRType.NS, 3600,
[ "newns1.example.org", "newns2.example.org" ])
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ new_ns,
self.rrset_update_del_rrset_ns ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
@ -1359,7 +1359,7 @@ class SessionTest(SessionTestBase):
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.MX)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset_mx ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXRRSET,
isc.dns.Name("example.org"),
@ -1372,7 +1372,7 @@ class SessionTest(SessionTestBase):
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_a,
self.rrset_update_del_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
@ -1386,7 +1386,7 @@ class SessionTest(SessionTestBase):
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_a,
self.rrset_update_del_name ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
@ -1400,7 +1400,7 @@ class SessionTest(SessionTestBase):
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_rrset,
self.rrset_update_a ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
@ -1415,7 +1415,7 @@ class SessionTest(SessionTestBase):
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A)
self.check_full_handle_result(Rcode.NOERROR(),
self.check_full_handle_result(Rcode.NOERROR,
[ self.rrset_update_del_name,
self.rrset_update_a ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
@ -1439,7 +1439,7 @@ class SessionTest(SessionTestBase):
rrset = create_rrset("cname.example.org", TEST_RRCLASS, RRType.A,
3600, [ "192.0.2.1" ])
self.check_full_handle_result(Rcode.NOERROR(), [ rrset ])
self.check_full_handle_result(Rcode.NOERROR, [ rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.CNAME,
isc.dns.Name("cname.example.org"),
RRType.A,
@ -1449,7 +1449,7 @@ class SessionTest(SessionTestBase):
new_cname_rrset = create_rrset("cname.example.org", TEST_RRCLASS,
RRType.CNAME, 3600,
[ "mail.example.org." ])
self.check_full_handle_result(Rcode.NOERROR(), [ new_cname_rrset ])
self.check_full_handle_result(Rcode.NOERROR, [ new_cname_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.CNAME,
isc.dns.Name("cname.example.org"),
RRType.A,
@ -1466,7 +1466,7 @@ class SessionTest(SessionTestBase):
new_cname_rrset = create_rrset("www.example.org", TEST_RRCLASS,
RRType.CNAME, 3600,
[ "mail.example.org." ])
self.check_full_handle_result(Rcode.NOERROR(), [ new_cname_rrset ])
self.check_full_handle_result(Rcode.NOERROR, [ new_cname_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A,
@ -1475,13 +1475,13 @@ class SessionTest(SessionTestBase):
def test_update_bad_class(self):
rrset = create_rrset("example.org.", RRClass.CH, RRType.TXT, 0,
[ "foo" ])
self.check_full_handle_result(Rcode.FORMERR(), [ rrset ])
self.check_full_handle_result(Rcode.FORMERR, [ rrset ])
def test_uncaught_exception(self):
def my_exc():
raise Exception("foo")
self._session._UpdateSession__update_soa = my_exc
self.assertEqual(Rcode.SERVFAIL().to_text(),
self.assertEqual(Rcode.SERVFAIL.to_text(),
self._session._UpdateSession__do_update().to_text())
class SessionACLTest(SessionTestBase):
@ -1527,7 +1527,7 @@ class SessionACLTest(SessionTestBase):
self._datasrc_client,
acl_map))
self.assertEqual((UPDATE_ERROR, None, None), session.handle())
self.check_response(session.get_message(), Rcode.REFUSED())
self.check_response(session.get_message(), Rcode.REFUSED)
# If the message contains TSIG, it should match the ACCEPT
# ACL entry, and the request should be granted.

6
src/lib/python/isc/notify/notify_out.py
View File

@ -509,8 +509,8 @@ class NotifyOut:
msg = Message(Message.RENDER)
qid = random.randint(0, 0xFFFF)
msg.set_qid(qid)
msg.set_opcode(Opcode.NOTIFY())
msg.set_rcode(Rcode.NOERROR())
msg.set_opcode(Opcode.NOTIFY)
msg.set_rcode(Rcode.NOERROR)
msg.set_header_flag(Message.HEADERFLAG_AA)
msg.add_question(Question(zone_name, zone_class, RRType.SOA))
msg.add_rrset(Message.SECTION_ANSWER, self._get_zone_soa(zone_name,
@ -566,7 +566,7 @@ class NotifyOut:
Name(zone_notify_info.zone_name).to_text())
return _BAD_QUERY_NAME
if msg.get_opcode() != Opcode.NOTIFY():
if msg.get_opcode() != Opcode.NOTIFY:
logger.warn(NOTIFY_OUT_REPLY_BAD_OPCODE, from_addr[0],
from_addr[1], msg.get_opcode().to_text())
return _BAD_OPCODE

2
tools/query_cmp/src/lib/handledns.py
View File

@ -187,7 +187,7 @@ def send_req(query, server, port=53, timeout=5):
msg = Message(Message.RENDER)
msg.set_qid(int(qheader['id']))
msg.set_opcode(Opcode.QUERY())
msg.set_opcode(Opcode.QUERY)
msg.set_rcode(Rcode(int(qheader['rcode'])))
if qheader['qr'] == 1: