services:
netopeer2:
- image: nexus3.onap.org:10001/onap/integration/simulators/netconf-pnp-simulator:2.6.1
+ image: nexus3.onap.org:10001/onap/integration/simulators/netconf-pnp-simulator:2.6.2
container_name: mynetconf
restart: always
ports:
COPY --from=build /opt/ /opt/
ENV LD_LIBRARY_PATH=/opt/lib:/opt/lib64
+ENV PYTHONPATH=/opt/lib/python3.7/site-packages
COPY config/ /config
VOLUME /config
ENV HOME=/home/netconf
VOLUME $HOME/.local/share/virtualenvs
+# This is NOT a robust health check but it does help tox-docker to detect when
+# it can start the tests.
+HEALTHCHECK --interval=1s --start-period=2s --retries=10 CMD test -f /run/netopeer2-server.pid
EXPOSE 830
-tag: "2.6.1"
+tag: "2.6.2"
command=$prog $model
redirect_stderr=true
autorestart=true
-environment=PATH=$PROG_PATH,PYTHONPATH=/opt/lib/python3.7/site-packages,PYTHONUNBUFFERED="1"
+environment=PATH=$PROG_PATH,PYTHONUNBUFFERED="1"
EOF
}
--- /dev/null
+Borrowed from https://github.com/sysrepo/sysrepo-netopeer2-smoketests
+with some minor fixes
--- /dev/null
+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)
--- /dev/null
+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
--- /dev/null
+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()
--- /dev/null
+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()
--- /dev/null
+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()
--- /dev/null
+#-
+# ============LICENSE_START=======================================================
+# Copyright (C) 2020 Nordix Foundation.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+# ============LICENSE_END=========================================================
+
+[tox]
+requires = tox-docker
+skipsdist = True
+
+[testenv]
+changedir = tests
+docker =
+ {env:CONTAINER_PUSH_REGISTRY}/{env:DOCKER_NAME}:{env:DOCKER_IMAGE_TAG}
+
+deps =
+ ncclient
+ discover
+commands = discover -v
services:
netconf-pnp-simulator:
- image: nexus3.onap.org:10001/onap/integration/simulators/netconf-pnp-simulator:2.6.1
+ image: nexus3.onap.org:10001/onap/integration/simulators/netconf-pnp-simulator:2.6.2
container_name: netconf-pnp-simulator
restart: always
ports:
Code Review / integration.git / commitdiff