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()))
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:
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:
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.
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]:
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,
# 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
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.
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)
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
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=[],
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)
'bad_qid': False,
'response': True,
'auth': True,
- 'rcode': Rcode.NOERROR(),
+ 'rcode': Rcode.NOERROR,
'answers': default_answers,
'authorities': [],
'tsig': False,
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())
# 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)
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,
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)
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)
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)
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
# 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())
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,
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.
# 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.
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,
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
# 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)
# 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)
# 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)
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)
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
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
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()
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):
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):
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)
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())
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]'''
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
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
'''
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.
# 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.
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
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:
# 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.
#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()
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_)
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")
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)
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)
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())
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())
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:
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)))
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):
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))
# 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)
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
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"))
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)
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())
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)
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)
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)
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" +\
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
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())
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()))
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:
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):
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()
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):
'''
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),
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).
'''
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:
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),
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),
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),
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),
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:
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),
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
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
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
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
# 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):
'''
# prescan
prescan_result = self.__do_prescan()
- if prescan_result != Rcode.NOERROR():
+ if prescan_result != Rcode.NOERROR:
return prescan_result
# update
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
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:
'''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))
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
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'''
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'''
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,
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)
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)
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,
# 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,
# 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,
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
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):
# 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
# 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,
# 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,
'''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):
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"),
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"),
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"),
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"),
# 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"),
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."),
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."),
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,
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,
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
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"),
# 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,
# 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"),
# 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"),
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"),
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,
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"),
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,
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,
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,
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,
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,
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,
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,
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):
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.
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,
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
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:
projects / thirdparty / kea.git / commitdiff
