aboutsummaryrefslogtreecommitdiffstats
path: root/test/mocks/netconf-pnp-simulator/engine/tests
diff options
context:
space:
mode:
authorebo <eliezio.oliveira@est.tech>2020-03-09 15:42:08 +0000
committerMarco Platania <platania@research.att.com>2020-03-11 21:06:13 +0000
commit769a79125d17ded007894464a7151eafbd65f355 (patch)
treec5a40275f7b7eba8f3bd1e9665ac0ba461209912 /test/mocks/netconf-pnp-simulator/engine/tests
parentc27713e8c8b5467e8e900f7020227de366c67014 (diff)
netconf-pnp-simulator: make PYTHONPATH always globally defined
Add IT using ncclient and tox Issue-ID: INT-1124 Change-Id: I560d4fd2468ac93f8ead36062b2e316821af8d07 Signed-off-by: ebo <eliezio.oliveira@est.tech>
Diffstat (limited to 'test/mocks/netconf-pnp-simulator/engine/tests')
-rw-r--r--test/mocks/netconf-pnp-simulator/engine/tests/README2
-rw-r--r--test/mocks/netconf-pnp-simulator/engine/tests/nctest.py37
-rw-r--r--test/mocks/netconf-pnp-simulator/engine/tests/settings.py11
-rw-r--r--test/mocks/netconf-pnp-simulator/engine/tests/test_basic_operations.py52
-rw-r--r--test/mocks/netconf-pnp-simulator/engine/tests/test_ietf_interfaces.py93
-rw-r--r--test/mocks/netconf-pnp-simulator/engine/tests/test_turing_machine.py124
6 files changed, 319 insertions, 0 deletions
diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/README b/test/mocks/netconf-pnp-simulator/engine/tests/README
new file mode 100644
index 000000000..295585dc2
--- /dev/null
+++ b/test/mocks/netconf-pnp-simulator/engine/tests/README
@@ -0,0 +1,2 @@
+Borrowed from https://github.com/sysrepo/sysrepo-netopeer2-smoketests
+with some minor fixes
diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py b/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py
new file mode 100644
index 000000000..2f848c361
--- /dev/null
+++ b/test/mocks/netconf-pnp-simulator/engine/tests/nctest.py
@@ -0,0 +1,37 @@
+from ncclient import manager, operations
+import settings
+import unittest
+
+class NCTestCase(unittest.TestCase):
+ """ Base class for NETCONF test cases. Provides a NETCONF connection and some helper methods. """
+
+ def setUp(self):
+ self.nc = manager.connect(
+ host=settings.HOST,
+ port=settings.PORT,
+ username=settings.USERNAME,
+ key_filename=settings.KEY_FILENAME,
+ allow_agent=False,
+ look_for_keys=False,
+ hostkey_verify=False)
+ self.nc.raise_mode = operations.RaiseMode.NONE
+
+ 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
new file mode 100644
index 000000000..749eb4cfd
--- /dev/null
+++ b/test/mocks/netconf-pnp-simulator/engine/tests/settings.py
@@ -0,0 +1,11 @@
+import os
+
+HOST = "127.0.0.1"
+# Set by tox-docker
+# Unexpectedly, tox-docker uses the repository prefix instead of the image name to define the
+# variable prefix.
+PORT = int(os.environ["LOCALHOST_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
new file mode 100644
index 000000000..62d41c259
--- /dev/null
+++ b/test/mocks/netconf-pnp-simulator/engine/tests/test_basic_operations.py
@@ -0,0 +1,52 @@
+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)
+
+ def test_get(self):
+ reply = self.nc.get()
+ self.check_reply_data(reply)
+
+ def test_get_config_startup(self):
+ reply = self.nc.get_config(source='startup')
+ self.check_reply_data(reply)
+
+ def test_get_config_running(self):
+ reply = self.nc.get_config(source='running')
+ self.check_reply_data(reply)
+
+ def test_copy_config(self):
+ reply = self.nc.copy_config(source='startup', target='candidate')
+ self.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)
+
+ def test_lock(self):
+ reply = self.nc.lock("startup")
+ self.check_reply_ok(reply)
+ reply = self.nc.lock("running")
+ self.check_reply_ok(reply)
+ reply = self.nc.lock("candidate")
+ self.check_reply_ok(reply)
+
+ reply = self.nc.lock("startup")
+ self.check_reply_err(reply)
+
+ reply = self.nc.unlock("startup")
+ self.check_reply_ok(reply)
+ reply = self.nc.unlock("running")
+ self.check_reply_ok(reply)
+ reply = self.nc.unlock("candidate")
+ self.check_reply_ok(reply)
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test/mocks/netconf-pnp-simulator/engine/tests/test_ietf_interfaces.py b/test/mocks/netconf-pnp-simulator/engine/tests/test_ietf_interfaces.py
new file mode 100644
index 000000000..87733ac37
--- /dev/null
+++ b/test/mocks/netconf-pnp-simulator/engine/tests/test_ietf_interfaces.py
@@ -0,0 +1,93 @@
+import unittest
+import nctest
+import os
+
+class TestIETFInterfaces(nctest.NCTestCase):
+ """ Tests basic NETCONF operations on the turing-machine YANG module. """
+
+ def __init__(self, *args, **kwargs):
+ super(TestIETFInterfaces, self).__init__(*args, **kwargs)
+ self.ns = {"nc": "urn:ietf:params:xml:ns:netconf:base:1.0", "if": "urn:ietf:params:xml:ns:yang:ietf-interfaces"}
+
+ def test_edit_config(self):
+ config_xml = """<nc:config xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0">
+ <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
+ <interface nc:operation="{}">
+ <name>TestInterface</name>
+ <description>Interface under test</description>
+ <type xmlns:ianaift="urn:ietf:params:xml:ns:yang:iana-if-type">ianaift:ethernetCsmacd</type>
+ <ipv4 xmlns="urn:ietf:params:xml:ns:yang:ietf-ip">
+ <mtu>1500</mtu>
+ <address>
+ <ip>192.168.2.100</ip>
+ <prefix-length>24</prefix-length>
+ </address>
+ </ipv4>
+ <ipv6 xmlns="urn:ietf:params:xml:ns:yang:ietf-ip">
+ <address>
+ <ip>2001:db8::10</ip>
+ <prefix-length>32</prefix-length>
+ </address>
+ </ipv6>
+ </interface>
+ </interfaces>
+ </nc:config>"""
+
+ filter_xml = """<nc:filter xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0">
+ <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces" />
+ </nc:filter>"""
+
+ with_default_report_all = """report-all"""
+
+ # get from running - should be empty
+ reply = self.nc.get_config(source="running", filter=filter_xml)
+ self.check_reply_data(reply)
+ deltas = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']", namespaces=self.ns)
+ self.assertEqual(len(deltas), 0)
+
+ # set data - candidate
+ reply = self.nc.edit_config(target='candidate', config=config_xml.format("merge"))
+ self.check_reply_ok(reply)
+
+ # get from candidate
+ reply = self.nc.get_config(source="candidate", filter=filter_xml)
+ self.check_reply_data(reply)
+ interfaces = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']", namespaces=self.ns)
+ self.assertEqual(len(interfaces), 1)
+
+ # default leaf should NOT be present
+ enabled = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']/enabled", namespaces=self.ns)
+ self.assertEqual(len(enabled), 0)
+
+ # get from candidate with with defaults = 'report-all'
+ reply = self.nc.get_config(source="candidate", filter=filter_xml, with_defaults=with_default_report_all)
+ self.check_reply_data(reply)
+ interfaces = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']", namespaces=self.ns)
+ self.assertEqual(len(interfaces), 1)
+
+ # default leaf should be present
+ enabled = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']/enabled", namespaces=self.ns)
+ self.assertEqual(len(enabled), 0) # TODO: change to 1 once this is implemented
+
+ # get from running - should be empty
+ reply = self.nc.get_config(source="running", filter=filter_xml)
+ self.check_reply_data(reply)
+ deltas = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']", namespaces=self.ns)
+ self.assertEqual(len(deltas), 0)
+
+ # commit - should fail, not enabled in running
+ reply = self.nc.commit()
+ self.check_reply_err(reply)
+
+ # delete from candidate
+ reply = self.nc.edit_config(target='candidate', config=config_xml.format("delete"))
+ self.check_reply_ok(reply)
+
+ # get from candidate - should be empty
+ reply = self.nc.get_config(source="candidate", filter=filter_xml)
+ self.check_reply_data(reply)
+ deltas = reply.data.xpath("/nc:rpc-reply/nc:data/if:interfaces/if:interface[if:name='TestInterface']", namespaces=self.ns)
+ self.assertEqual(len(deltas), 0)
+
+if __name__ == '__main__':
+ unittest.main()
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..63a0c2d99
--- /dev/null
+++ b/test/mocks/netconf-pnp-simulator/engine/tests/test_turing_machine.py
@@ -0,0 +1,124 @@
+import unittest
+import nctest
+import os
+
+class TestTuringMachine(nctest.NCTestCase):
+ """ Tests basic NETCONF operations on the turing-machine YANG module. """
+
+ def __init__(self, *args, **kwargs):
+ super(TestTuringMachine, self).__init__(*args, **kwargs)
+ self.ns = {"nc": "urn:ietf:params:xml:ns:netconf:base:1.0", "tm": "http://example.net/turing-machine"}
+
+ def check_deltas_in_data(self, data):
+ deltas = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/*", namespaces=self.ns)
+ self.assertNotEqual(len(deltas), 0)
+ for d in deltas:
+ self.assertTrue(d.tag.endswith("delta"))
+
+ def check_labels_only_in_data(self, data):
+ children = data.xpath("/nc:rpc-reply/nc:data/*", namespaces=self.ns)
+ self.assertNotEqual(len(children), 0)
+ for child in children:
+ self.assertTrue(child.tag.endswith("turing-machine"))
+ children = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/*", namespaces=self.ns)
+ self.assertNotEqual(len(children), 0)
+ for child in children:
+ self.assertTrue(child.tag.endswith("transition-function"))
+ children = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/*", namespaces=self.ns)
+ self.assertNotEqual(len(children), 0)
+ for child in children:
+ self.assertTrue(child.tag.endswith("delta"))
+ children = data.xpath("/nc:rpc-reply/nc:data/tm:turing-machine/tm:transition-function/tm:delta/*", namespaces=self.ns)
+ self.assertNotEqual(len(children), 0)
+ for child in children:
+ self.assertTrue(child.tag.endswith("label"))
+
+ def test_get(self):
+ reply = self.nc.get()
+ self.check_reply_data(reply)
+ self.check_deltas_in_data(reply.data)
+
+ def test_get_config_startup(self):
+ reply = self.nc.get_config(source="startup")
+ self.check_reply_data(reply)
+ self.check_deltas_in_data(reply.data)
+
+ def test_get_config_running(self):
+ reply = self.nc.get_config(source="running")
+ self.check_reply_data(reply)
+ self.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)
+ self.check_reply_data(reply)
+ self.check_deltas_in_data(reply.data)
+ self.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)
+ self.check_reply_data(reply)
+ self.check_deltas_in_data(reply.data)
+ self.check_labels_only_in_data(reply.data)
+
+ @unittest.skipIf(os.environ.get("DOCKER_IMG_TAG") == "latest", "bug in Netopeer2 replace operation")
+ 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))
+ self.check_reply_ok(reply)
+ # get
+ reply = self.nc.get_config(source="running")
+ self.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=self.ns)
+ self.assertEqual(len(deltas), 1)
+ # create already existing - expect error
+ reply = self.nc.edit_config(target='running', config=config_xml.format("create", 9, 99))
+ self.check_reply_err(reply)
+ # replace
+ reply = self.nc.edit_config(target='running', config=config_xml.format("replace", 9, 88))
+ self.check_reply_ok(reply)
+ # get
+ reply = self.nc.get_config(source="running")
+ self.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=self.ns)
+ self.assertEqual(len(states), 1)
+ self.assertEqual(states[0].text, "88")
+ # delete
+ reply = self.nc.edit_config(target='running', config=config_xml.format("delete", 9, 88))
+ self.check_reply_ok(reply)
+ # delete non-existing - expect error
+ reply = self.nc.edit_config(target='running', config=config_xml.format("delete", 9, 88))
+ self.check_reply_err(reply)
+ # get - should be empty
+ reply = self.nc.get_config(source="running")
+ self.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=self.ns)
+ self.assertEqual(len(deltas), 0)
+
+if __name__ == '__main__':
+ unittest.main()