VOL-2642 Add a Makefile, tests, and virtualenv
Convert common python and robot into a CORDRobot python module that can
be installed via standard python tools (pip) and from PyPI
Uses a fork of https://github.com/rasjani/robotframework-importresource,
which has been backported to Python 3.5 (used in Ubuntu 16.04
executors).
Reformatted and moved keywords so resource files are scoped to a
specific topic.
Added tox tests for library consistency
- flake8
- pylint
- robotframework-lint
- Ran robot against installed library to verify it can be loaded and
used
Added basic lint and tests to whole repo
Removed old tests:
- CORD <6.x era: SanityPhyPOD.robot, and onosUtils.py
Change-Id: I61265a9fb04034a086e20be1f7236a8793a218aa
diff --git a/cord-robot/CORDRobot/testCaseUtils.py b/cord-robot/CORDRobot/testCaseUtils.py
new file mode 100755
index 0000000..88f1913
--- /dev/null
+++ b/cord-robot/CORDRobot/testCaseUtils.py
@@ -0,0 +1,244 @@
+#
+# Copyright 2018 the original author or authors.
+#
+# 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.
+#
+
+"""
+Test Case Utils module
+"""
+
+from __future__ import absolute_import
+
+import time
+import subprocess
+import pexpect
+import sys
+
+
+class TestCaseUtils():
+
+ @staticmethod
+ def config_dirs(self, log_dir, root_dir=None, voltha_dir=None):
+ self.dirs["log"] = log_dir
+ self.dirs["root"] = root_dir
+ self.dirs["voltha"] = voltha_dir
+
+ def get_dir(self, directory):
+ return self.dirs.get(directory)
+
+ @staticmethod
+ def remove_leading_line(log_dir, log_file):
+ with open(log_dir + "/" + log_file, "r+") as FILE:
+ lines = FILE.readlines()
+ FILE.seek(0)
+ lines = lines[1:]
+ for line in lines:
+ FILE.write(line)
+ FILE.truncate()
+ FILE.close()
+
+ @staticmethod
+ def write_log_of_voltha_cli_comand(
+ log_dir,
+ log_file1,
+ cmd1,
+ log_file2=None,
+ cmd2=None,
+ log_file3=None,
+ cmd3=None,
+ host="localhost",
+ ):
+ output = open(log_dir + "/" + log_file1, "wb")
+ child = pexpect.spawn(
+ "ssh -p 30110 -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no voltha@%s"
+ % host
+ )
+ child.expect(r"[pP]assword:")
+ child.sendline("admin")
+ child.expect(r"\((\x1b\[\d*;?\d+m){1,2}voltha(\x1b\[\d*;?\d+m){1,2}\)")
+ time.sleep(10)
+ child.sendline(cmd1)
+ i = child.expect(
+ [
+ r"\((\x1b\[\d*;?\d+m){1,2}voltha(\x1b\[\d*;?\d+m){1,2}\)",
+ r"\((\x1b\[\d*;?\d+m){1,2}.*device [0-9a-f]{16}(\x1b\[\d*;?\d+m){1,2}\)",
+ ]
+ )
+ if i == 0:
+ output.write(child.before)
+ output.close()
+ TestCaseUtils.remove_leading_line(log_dir, log_file1)
+ elif i == 1:
+ if log_file2 is not None and cmd2 is not None:
+ output = open(log_dir + "/" + log_file2, "wb")
+ child.sendline(cmd2)
+ child.expect(
+ r"\((\x1b\[\d*;?\d+m){1,2}.*device [0-9a-f]{16}(\x1b\[\d*;?\d+m){1,2}\)"
+ )
+ output.write(child.before)
+ output.close()
+ TestCaseUtils.remove_leading_line(log_dir, log_file2)
+ if log_file3 is not None and cmd3 is not None:
+ output = open(log_dir + "/" + log_file3, "wb")
+ child.sendline(cmd3)
+ child.expect(
+ r"\((\x1b\[\d*;?\d+m){1,2}.*device [0-9a-f]{16}(\x1b\[\d*;?\d+m){1,2}\)"
+ )
+ output.write(child.before)
+ output.close()
+ TestCaseUtils.remove_leading_line(log_dir, log_file3)
+ child.close()
+
+ @staticmethod
+ def write_log_of_onos_cli_command(log_dir, log_file, cmd, host="localhost", port=30115):
+ output = open(log_dir + "/" + log_file, "wb")
+ child = pexpect.spawn(
+ "ssh -p %s -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no karaf@%s"
+ % (port, host)
+ )
+ child.expect(r"[pP]assword:")
+ child.sendline("karaf")
+ # Expected prompt:
+ # onos> (ONOS 1.x)
+ # karaf@root > (ONOS 2.x)
+ child.expect([r'(\x1b\[\d*;?\d+m){1,2}onos> (\x1b\[\d*;?\d+m){1,2}', r'karaf@root >'])
+ child.sendline(cmd)
+ child.expect([r'(\x1b\[\d*;?\d+m){1,2}onos> (\x1b\[\d*;?\d+m){1,2}', r'karaf@root >'])
+
+ output.write(child.before)
+
+ output.close()
+ child.close()
+
+ def get_fields_from_grep_command(self, search_word, log_file):
+ grepCommand = "grep %s %s/%s" % (search_word, self.get_dir("log"), log_file)
+ statusLines = subprocess.getstatusoutput(grepCommand)[1]
+ return statusLines
+
+ @staticmethod
+ def parse_fields(status_line, delimiter):
+ statusList = status_line.split(delimiter)
+ return statusList
+
+ def print_log_file(self, log_file):
+ with open(self.get_dir("log") + "/" + log_file, "r+") as FILE:
+ lines = FILE.readlines()
+ print
+ for line in lines:
+ sys.stdout.write(line)
+
+ @staticmethod
+ def extract_pod_ip_addr(pod_name):
+ proc1 = subprocess.Popen(
+ ["/usr/bin/kubectl", "get", "svc", "--all-namespaces"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ proc2 = subprocess.Popen(
+ ["grep", "-e", pod_name],
+ stdin=proc1.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ proc3 = subprocess.Popen(
+ ["awk", "{print $4}"],
+ stdin=proc2.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+
+ proc1.stdout.close()
+ proc2.stdout.close()
+ out, err = proc3.communicate()
+ return out
+
+ @staticmethod
+ def extract_radius_ip_addr(pod_name):
+ proc1 = subprocess.Popen(
+ ["/usr/bin/kubectl", "describe", "pod", "-n", "voltha", pod_name],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ proc2 = subprocess.Popen(
+ ["grep", "^IP:"],
+ stdin=proc1.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ proc3 = subprocess.Popen(
+ ["awk", "{print $2}"],
+ stdin=proc2.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+
+ proc1.stdout.close()
+ proc2.stdout.close()
+ out, err = proc3.communicate()
+ return out
+
+ @staticmethod
+ def extract_pod_name(short_pod_name):
+ proc1 = subprocess.Popen(
+ ["/usr/bin/kubectl", "get", "pods", "--all-namespaces"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ proc2 = subprocess.Popen(
+ ["grep", "-e", short_pod_name],
+ stdin=proc1.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ proc3 = subprocess.Popen(
+ ["awk", "{print $2}"],
+ stdin=proc2.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+
+ proc1.stdout.close()
+ proc2.stdout.close()
+ out, err = proc3.communicate()
+ return out
+
+ def modify_radius_ip_in_json_using_sed(self, new_ip_addr):
+ sedCommand = (
+ "sed -i '/radiusIp/c\\ \"radiusIp\":\"'%s'\",' %s/tests/atests/build/aaa_json"
+ % (new_ip_addr, self.get_dir("voltha"))
+ )
+ status = subprocess.getstatusoutput(sedCommand)[0]
+ return status
+
+ @staticmethod
+ def discover_rg_pod_name():
+ return TestCaseUtils.extract_pod_name("rg0").strip()
+
+ @staticmethod
+ def retrieve_authorized_users_device_id_and_port_number(status_line):
+ fields = TestCaseUtils.parse_fields(status_line, ",")
+ deviceField = fields[2].strip()
+ deviceStr, equal, deviceId = deviceField.partition("=")
+ device_Id = deviceId
+ portField = fields[4].strip()
+ portNumStr, equal, portNum = portField.partition("=")
+ portNumber = portNum
+ return device_Id, portNumber
+
+ def add_subscriber_access(self, device_id, port_number):
+ TestCaseUtils.write_log_of_onos_cli_command(
+ self.get_dir("log"),
+ "voltha_add_subscriber_access.log",
+ "volt-add-subscriber-access %s %s" % (device_id, port_number),
+ )