diff options
Diffstat (limited to 'test/mocks/netconf-pnp-simulator/engine/tests')
4 files changed, 181 insertions, 43 deletions
diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py b/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py index 2f848c361..11ff6ffc4 100644 --- a/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py +++ b/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py @@ -1,11 +1,41 @@ +import logging.config + from ncclient import manager, operations + import settings -import unittest -class NCTestCase(unittest.TestCase): +LOGGER = logging.getLogger(__name__) + + +def check_reply_ok(reply): + assert reply is not None + _log_netconf_msg("Received", reply.xml) + assert reply.ok is True + assert reply.error is None + + +def check_reply_err(reply): + assert reply is not None + _log_netconf_msg("Received", reply.xml) + assert reply.ok is False + assert reply.error is not None + + +def check_reply_data(reply): + check_reply_ok(reply) + + +def _log_netconf_msg(header: str, body: str): + """Log a message using a format inspired by NETCONF 1.1 """ + LOGGER.info("%s:\n\n#%d\n%s\n##", header, len(body), body) + + +class NCTestCase: """ Base class for NETCONF test cases. Provides a NETCONF connection and some helper methods. """ - def setUp(self): + nc: manager.Manager + + def setup(self): self.nc = manager.connect( host=settings.HOST, port=settings.PORT, @@ -16,22 +46,5 @@ class NCTestCase(unittest.TestCase): hostkey_verify=False) self.nc.raise_mode = operations.RaiseMode.NONE - def tearDown(self): + def teardown(self): self.nc.close_session() - - def check_reply_ok(self, reply): - self.assertIsNotNone(reply) - if settings.DEBUG: - print(reply.xml) - self.assertTrue(reply.ok) - self.assertIsNone(reply.error) - - def check_reply_err(self, reply): - self.assertIsNotNone(reply) - if settings.DEBUG: - print(reply.xml) - self.assertFalse(reply.ok) - self.assertIsNotNone(reply.error) - - def check_reply_data(self, reply): - self.check_reply_ok(reply) diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/settings.py b/test/mocks/netconf-pnp-simulator/engine/tests/settings.py index 716fdb7a2..124e333cd 100644 --- a/test/mocks/netconf-pnp-simulator/engine/tests/settings.py +++ b/test/mocks/netconf-pnp-simulator/engine/tests/settings.py @@ -5,5 +5,3 @@ HOST = "127.0.0.1" PORT = int(os.environ["NETCONF_PNP_SIMULATOR_830_TCP_PORT"]) USERNAME = "netconf" KEY_FILENAME = "../config/ssh/id_rsa" - -DEBUG = False diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/test_basic_operations.py b/test/mocks/netconf-pnp-simulator/engine/tests/test_basic_operations.py index 62d41c259..06164e6b5 100644 --- a/test/mocks/netconf-pnp-simulator/engine/tests/test_basic_operations.py +++ b/test/mocks/netconf-pnp-simulator/engine/tests/test_basic_operations.py @@ -1,52 +1,49 @@ -import unittest import nctest + class TestBasicOperations(nctest.NCTestCase): """ Tests basic NETCONF operations with no prerequisites on datastore content. """ def test_capabilities(self): - self.assertTrue(":startup" in self.nc.server_capabilities) - self.assertTrue(":candidate" in self.nc.server_capabilities) - self.assertTrue(":validate" in self.nc.server_capabilities) - self.assertTrue(":xpath" in self.nc.server_capabilities) + assert ":startup" in self.nc.server_capabilities + assert ":candidate" in self.nc.server_capabilities + assert ":validate" in self.nc.server_capabilities + assert ":xpath" in self.nc.server_capabilities def test_get(self): reply = self.nc.get() - self.check_reply_data(reply) + nctest.check_reply_data(reply) def test_get_config_startup(self): reply = self.nc.get_config(source='startup') - self.check_reply_data(reply) + nctest.check_reply_data(reply) def test_get_config_running(self): reply = self.nc.get_config(source='running') - self.check_reply_data(reply) + nctest.check_reply_data(reply) def test_copy_config(self): reply = self.nc.copy_config(source='startup', target='candidate') - self.check_reply_ok(reply) + nctest.check_reply_ok(reply) def test_neg_filter(self): reply = self.nc.get(filter=("xpath", "/non-existing-module:non-existing-data")) - self.check_reply_err(reply) + nctest.check_reply_err(reply) def test_lock(self): reply = self.nc.lock("startup") - self.check_reply_ok(reply) + nctest.check_reply_ok(reply) reply = self.nc.lock("running") - self.check_reply_ok(reply) + nctest.check_reply_ok(reply) reply = self.nc.lock("candidate") - self.check_reply_ok(reply) + nctest.check_reply_ok(reply) reply = self.nc.lock("startup") - self.check_reply_err(reply) + nctest.check_reply_err(reply) reply = self.nc.unlock("startup") - self.check_reply_ok(reply) + nctest.check_reply_ok(reply) reply = self.nc.unlock("running") - self.check_reply_ok(reply) + nctest.check_reply_ok(reply) reply = self.nc.unlock("candidate") - self.check_reply_ok(reply) - -if __name__ == '__main__': - unittest.main() + nctest.check_reply_ok(reply) diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/test_turing_machine.py b/test/mocks/netconf-pnp-simulator/engine/tests/test_turing_machine.py new file mode 100644 index 000000000..8ac38b0f5 --- /dev/null +++ b/test/mocks/netconf-pnp-simulator/engine/tests/test_turing_machine.py @@ -0,0 +1,130 @@ +import nctest + +_NAMESPACES = { + "nc": "urn:ietf:params:xml:ns:netconf:base:1.0", + "tm": "http://example.net/turing-machine" +} + + +def check_labels_only_in_data(data): + children = data.xpath("/nc:rpc-reply/nc:data/*", namespaces=_NAMESPACES) + assert children + for child in children: + assert child.tag.endswith("turing-machine") + children = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/*", namespaces=_NAMESPACES) + assert children + for child in children: + assert child.tag.endswith("transition-function") + children = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/*", namespaces=_NAMESPACES) + assert children + for child in children: + assert child.tag.endswith("delta") + children = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/tm:delta/*", + namespaces=_NAMESPACES) + assert children + for child in children: + assert child.tag.endswith("label") + + +def check_deltas_in_data(data): + deltas = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/*", namespaces=_NAMESPACES) + assert deltas + for d in deltas: + assert d.tag.endswith("delta") + + +class TestTuringMachine(nctest.NCTestCase): + """ Tests basic NETCONF operations on the turing-machine YANG module. """ + + def test_get(self): + reply = self.nc.get() + nctest.check_reply_data(reply) + check_deltas_in_data(reply.data) + + def test_get_config_startup(self): + reply = self.nc.get_config(source="startup") + nctest.check_reply_data(reply) + check_deltas_in_data(reply.data) + + def test_get_config_running(self): + reply = self.nc.get_config(source="running") + nctest.check_reply_data(reply) + check_deltas_in_data(reply.data) + + def test_get_subtree_filter(self): + filter_xml = """<nc:filter xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0"> + <turing-machine xmlns="http://example.net/turing-machine"> + <transition-function> + <delta> + <label /> + </delta> + </transition-function> + </turing-machine> + </nc:filter>""" + reply = self.nc.get_config(source="running", filter=filter_xml) + nctest.check_reply_data(reply) + check_deltas_in_data(reply.data) + check_labels_only_in_data(reply.data) + + def test_get_xpath_filter(self): + # https://github.com/ncclient/ncclient/issues/166 + filter_xml = """<nc:filter type="xpath" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" + xmlns:tm="http://example.net/turing-machine" + select="/tm:turing-machine/transition-function/delta/label" /> + """ + reply = self.nc.get(filter=filter_xml) + nctest.check_reply_data(reply) + check_deltas_in_data(reply.data) + check_labels_only_in_data(reply.data) + + def test_edit_config(self): + config_xml = """<nc:config xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0"> + <turing-machine xmlns="http://example.net/turing-machine"> + <transition-function> + <delta nc:operation="{}"> + <label>test-transition-rule</label> + <input> + <symbol>{}</symbol> + <state>{}</state> + </input> + </delta> + </transition-function> + </turing-machine></nc:config>""" + # merge + reply = self.nc.edit_config(target='running', config=config_xml.format("merge", 9, 99)) + nctest.check_reply_ok(reply) + # get + reply = self.nc.get_config(source="running") + nctest.check_reply_data(reply) + deltas = reply.data.xpath( + "/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/tm:delta[tm:label='test-transition-rule']", + namespaces=_NAMESPACES) + assert len(deltas) == 1 + # create already existing - expect error + reply = self.nc.edit_config(target='running', config=config_xml.format("create", 9, 99)) + nctest.check_reply_err(reply) + # replace + reply = self.nc.edit_config(target='running', config=config_xml.format("replace", 9, 88)) + nctest.check_reply_ok(reply) + # get + reply = self.nc.get_config(source="running") + nctest.check_reply_data(reply) + states = reply.data.xpath( + "/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/tm:delta[tm:label='test-transition-rule']/" + "tm:input/tm:state", + namespaces=_NAMESPACES) + assert len(states) == 1 + assert states[0].text == "88" + # delete + reply = self.nc.edit_config(target='running', config=config_xml.format("delete", 9, 88)) + nctest.check_reply_ok(reply) + # delete non-existing - expect error + reply = self.nc.edit_config(target='running', config=config_xml.format("delete", 9, 88)) + nctest.check_reply_err(reply) + # get - should be empty + reply = self.nc.get_config(source="running") + nctest.check_reply_data(reply) + deltas = reply.data.xpath( + "/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/tm:delta[tm:label='test-transition-rule']", + namespaces=_NAMESPACES) + assert not deltas |