Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_config_with_defaults(self, mock_request):
session = ncclient.transport.SSHSession(self.device_handler)
session._server_capabilities = Capabilities([
"urn:ietf:params:netconf:capability:with-defaults:1.0"
"?basic-mode=explicit"
])
obj = GetConfig(session, self.device_handler, raise_mode=RaiseMode.ALL)
source = 'candidate'
obj.request(source, with_defaults='explicit')
expected_xml = etree.fromstring(
''
''
'explicit'
''.format(
base=BASE_NS_1_0,
defaults=NETCONF_WITH_DEFAULTS_NS
)
)
def test_rpc_async(self, mock_thread, mock_send):
device_handler = manager.make_device_handler({'name': 'junos'})
capabilities = Capabilities(device_handler.get_capabilities())
session = ncclient.transport.Session(capabilities)
obj = RPC(
session,
device_handler,
raise_mode=RaiseMode.ALL,
timeout=0,
async_mode=True)
reply = RPCReply(xml1)
obj._reply = reply
node = new_ele("commit")
result = obj._request(node)
self.assertEqual(result, obj)
def test_get_with_defaults_not_supported(self, mock_request):
session = ncclient.transport.SSHSession(self.device_handler)
session._server_capabilities = Capabilities([
"urn:ietf:params:netconf:capability:with-defaults:1.0"
"?basic-mode=explicit"
"&also-supported=report-all,trim"
])
obj = Get(session, self.device_handler, raise_mode=RaiseMode.ALL)
expected_error = (
"Invalid 'with-defaults' mode 'report-all-tagged'; the server "
"only supports the following: explicit, report-all, trim"
)
self.assertRaisesRegexp(
WithDefaultsError,
expected_error,
obj.request,
with_defaults='report-all-tagged'
)
def test_rpc_timeout_error(self, mock_thread, mock_send):
device_handler = manager.make_device_handler({'name': 'junos'})
capabilities = Capabilities(device_handler.get_capabilities())
session = ncclient.transport.Session(capabilities)
obj = RPC(session, device_handler, raise_mode=RaiseMode.ALL, timeout=0)
reply = RPCReply(xml1)
obj.deliver_reply(reply)
node = new_ele("commit")
sub_ele(node, "confirmed")
mock_thread.return_value = False
self.assertRaises(TimeoutExpiredError, obj._request, node)
def test_get_with_defaults_missing_capability(self, mock_request):
session = ncclient.transport.SSHSession(self.device_handler)
session._server_capabilities = Capabilities([])
obj = Get(session, self.device_handler, raise_mode=RaiseMode.ALL)
self.assertRaises(
MissingCapabilityError,
obj.request,
with_defaults='report-all'
)
def test_get_with_defaults_also_supported(self, mock_request):
session = ncclient.transport.SSHSession(self.device_handler)
session._server_capabilities = Capabilities([
"urn:ietf:params:netconf:capability:with-defaults:1.0"
"?basic-mode=explicit"
"&also-supported=report-all,trim"
])
obj = Get(session, self.device_handler, raise_mode=RaiseMode.ALL)
obj.request(with_defaults='report-all')
expected_xml = etree.fromstring(
''
'report-all'
''.format(
base=BASE_NS_1_0,
defaults=NETCONF_WITH_DEFAULTS_NS
)
)
def test_get_with_defaults_basic_mode(self, mock_request):
session = ncclient.transport.SSHSession(self.device_handler)
session._server_capabilities = Capabilities([
"urn:ietf:params:netconf:capability:with-defaults:1.0"
"?basic-mode=explicit"
])
obj = Get(session, self.device_handler, raise_mode=RaiseMode.ALL)
obj.request(with_defaults='explicit')
expected_xml = etree.fromstring(
''
'explicit'
''.format(
base=BASE_NS_1_0,
defaults=NETCONF_WITH_DEFAULTS_NS
)
)
call = mock_request.call_args_list[0][0][0]
def parse(raw):
"Returns tuple of (session-id (str), capabilities (Capabilities)"
sid, capabilities = 0, []
root = to_ele(raw)
for child in root.getchildren():
if child.tag == qualify("session-id") or child.tag == "session-id":
sid = child.text
elif child.tag == qualify("capabilities") or child.tag == "capabilities" :
for cap in child.getchildren():
if cap.tag == qualify("capability") or cap.tag == "capability":
capabilities.append(cap.text)
return sid, Capabilities(capabilities)
def parse(raw):
"Returns tuple of (session-id (str), capabilities (Capabilities)"
sid, capabilities = 0, []
root = to_ele(raw)
for child in root.getchildren():
if child.tag == qualify("session-id") or child.tag == "session-id":
sid = child.text
elif child.tag == qualify("capabilities") or child.tag == "capabilities" :
for cap in child.getchildren():
if cap.tag == qualify("capability") or cap.tag == "capability":
capabilities.append(cap.text)
return sid, Capabilities(capabilities)