9765e27635
Type: test Enhance vpp_qemu_utils functions with mutex locking, handle namespace and host interface existence, unique namespace/interface name, error handling and retries, check error code explicitly. Change-Id: I1ea66eeefbc1fee9b58e8b9886f4dd6fd8d33444 Signed-off-by: Ivan Ivanets <iivanets@cisco.com>
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
from config import config
|
|
from asfframework import VppAsfTestCase, VppTestRunner, get_testcase_dirname
|
|
import unittest
|
|
import subprocess
|
|
import os
|
|
from vpp_qemu_utils import (
|
|
create_host_interface,
|
|
delete_all_host_interfaces,
|
|
create_namespace,
|
|
delete_all_namespaces,
|
|
)
|
|
|
|
|
|
@unittest.skipIf(
|
|
"http_static" in config.excluded_plugins, "Exclude HTTP Static Server plugin tests"
|
|
)
|
|
@unittest.skipIf("prom" in config.excluded_plugins, "Exclude Prometheus plugin tests")
|
|
@unittest.skipIf(config.skip_netns_tests, "netns not available or disabled from cli")
|
|
class TestProm(VppAsfTestCase):
|
|
"""Prometheus plugin test"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestProm, cls).setUpClass()
|
|
|
|
cls.ns_history_name = (
|
|
f"{config.tmp_dir}/{get_testcase_dirname(cls.__name__)}/history_ns.txt"
|
|
)
|
|
cls.if_history_name = (
|
|
f"{config.tmp_dir}/{get_testcase_dirname(cls.__name__)}/history_if.txt"
|
|
)
|
|
|
|
try:
|
|
# CleanUp
|
|
delete_all_namespaces(cls.ns_history_name)
|
|
delete_all_host_interfaces(cls.if_history_name)
|
|
|
|
cls.ns_name = create_namespace(cls.ns_history_name)
|
|
|
|
cls.host_if_name, cls.vpp_if_name = create_host_interface(
|
|
cls.if_history_name, cls.ns_name, "10.10.1.1/24"
|
|
)
|
|
except Exception as e:
|
|
cls.logger.warning(f"Unable to complete setup: {e}")
|
|
raise unittest.SkipTest("Skipping tests due to setup failure.")
|
|
|
|
cls.vapi.cli(f"create host-interface name {cls.vpp_if_name}")
|
|
cls.vapi.cli(f"set int state host-{cls.vpp_if_name} up")
|
|
cls.vapi.cli(f"set int ip address host-{cls.vpp_if_name} 10.10.1.2/24")
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
delete_all_namespaces(cls.ns_history_name)
|
|
delete_all_host_interfaces(cls.if_history_name)
|
|
|
|
super(TestProm, cls).tearDownClass()
|
|
|
|
def test_prom(self):
|
|
"""Enable HTTP Static server and prometheus exporter, get stats"""
|
|
self.vapi.cli("http static server uri tcp://0.0.0.0/80 url-handlers")
|
|
self.vapi.cli("prom enable")
|
|
self.sleep(1, "wait for min-scrape-interval to expire")
|
|
|
|
process = subprocess.run(
|
|
[
|
|
"ip",
|
|
"netns",
|
|
"exec",
|
|
self.ns_name,
|
|
"curl",
|
|
f"10.10.1.2/stats.prom",
|
|
],
|
|
capture_output=True,
|
|
)
|
|
self.assertIn(b"TYPE", process.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(testRunner=VppTestRunner)
|