Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
Sebastian Schrader
Kea
Commits
1af57091
Commit
1af57091
authored
Nov 18, 2011
by
JINMEI Tatuya
Browse files
[1372] refactoring: renamed _check_xfrout_available to _xfrout_setup
as it's not really about availability anymore.
parent
256c0a08
Changes
2
Hide whitespace changes
Inline
Side-by-side
src/bin/xfrout/tests/xfrout_test.py.in
View file @
1af57091
...
...
@@ -673,46 +673,46 @@ class TestXfroutSession(TestXfroutSessionBase):
def test_get_rrset_len(self):
self.assertEqual(82, get_rrset_len(self.soa_rrset))
def test_
check_
xfrout_axfr_
available
(self):
def test_xfrout_axfr_
setup
(self):
self.xfrsess.ClientClass = MockDataSrcClient
# Successful case. A zone iterator should be set up.
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('example.com')), Rcode.NOERROR())
self.assertNotEqual(None, self.xfrsess._iterator)
# Failure cases
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('notauth.example.com')), Rcode.NOTAUTH())
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('nosoa.example.com')), Rcode.SERVFAIL())
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('multisoa.example.com')), Rcode.SERVFAIL())
def test_
check_
xfrout_ixfr_
available
(self):
def test_xfrout_ixfr_
setup
(self):
self.xfrsess.ClientClass = MockDataSrcClient
self.set_request_type(RRType.IXFR())
# Successful case of pure IXFR. A zone journal reader should be set
# up.
self.mdata = self.create_request_data(ixfr=IXFR_OK_VERSION)
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), TEST_ZONE_NAME), 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._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), TEST_ZONE_NAME), Rcode.NOERROR())
self.assertNotEqual(None, self.xfrsess._iterator)
self.assertEqual(None, self.xfrsess._jnl_reader)
# Failure cases
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('notauth.example.com')), Rcode.NOTAUTH())
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('nosoa.example.com')), Rcode.SERVFAIL())
self.assertEqual(self.xfrsess._
check_
xfrout_
available
(
self.assertEqual(self.xfrsess._xfrout_
setup
(
self.getmsg(), Name('multisoa.example.com')), Rcode.SERVFAIL())
def test_dns_xfrout_start_formerror(self):
...
...
@@ -727,7 +727,7 @@ class TestXfroutSession(TestXfroutSessionBase):
def test_dns_xfrout_start_notauth(self):
def notauth(msg, name):
return Rcode.NOTAUTH()
self.xfrsess._
check_
xfrout_
available
= notauth
self.xfrsess._xfrout_
setup
= notauth
self.xfrsess.dns_xfrout_start(self.sock, self.mdata)
get_msg = self.sock.read_msg()
self.assertEqual(get_msg.get_rcode().to_text(), "NOTAUTH")
...
...
@@ -742,7 +742,7 @@ class TestXfroutSession(TestXfroutSessionBase):
def test_dns_xfrout_start_noerror(self):
def noerror(msg, name):
return Rcode.NOERROR()
self.xfrsess._
check_
xfrout_
available
= noerror
self.xfrsess._xfrout_
setup
= noerror
def myreply(msg, sock):
self.sock.send(b"success")
...
...
src/bin/xfrout/xfrout.py.in
View file @
1af57091
...
...
@@ -153,7 +153,7 @@ class XfroutSession():
self._acl = default_acl
self._zone_config = zone_config
self.ClientClass = client_class # parameterize this for testing
self._soa = None # will be set in
_check
_xfrout_
available
or in tests
self._soa = None # will be set in _xfrout_
setup
or in tests
self._handle()
def create_tsig_ctx(self, tsig_record, tsig_key_ring):
...
...
@@ -391,10 +391,13 @@ class XfroutSession():
return Rcode.NOERROR()
def _check_xfrout_available(self, request_msg, zone_name):
'''Check if xfr request can be responsed.
TODO, Get zone's configuration from cfgmgr or some other place
eg. check allow_transfer setting,
def _xfrout_setup(self, request_msg, zone_name):
'''Setup a context for xfr responses according to the request type.
This method identifies the most appropriate data source for the
request and set up a zone iterator or journal reader depending on
whether the request is AXFR or IXFR. If it identifies any protocol
level error it returns an RCODE other than NOERROR.
'''
...
...
@@ -438,7 +441,7 @@ class XfroutSession():
# TODO: we should also include class in the check
try:
rcode_ = self._
check_
xfrout_
available
(msg, zone_name)
rcode_ = self._xfrout_
setup
(msg, zone_name)
except Exception as ex:
logger.error(XFROUT_XFR_TRANSFER_CHECK_ERROR, self._request_typestr,
format_addrinfo(self._remote), zone_str, ex)
...
...
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment