summaryrefslogtreecommitdiff
path: root/smoketest/scripts/cli
diff options
context:
space:
mode:
Diffstat (limited to 'smoketest/scripts/cli')
-rw-r--r--smoketest/scripts/cli/test_service_monitoring_network_event.py65
-rwxr-xr-xsmoketest/scripts/cli/test_system_flow-accounting.py107
-rwxr-xr-xsmoketest/scripts/cli/test_system_sflow.py33
3 files changed, 99 insertions, 106 deletions
diff --git a/smoketest/scripts/cli/test_service_monitoring_network_event.py b/smoketest/scripts/cli/test_service_monitoring_network_event.py
new file mode 100644
index 000000000..3c9b4bf7f
--- /dev/null
+++ b/smoketest/scripts/cli/test_service_monitoring_network_event.py
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2024 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import unittest
+from base_vyostest_shim import VyOSUnitTestSHIM
+from vyos.utils.file import read_json
+
+
+base_path = ['service', 'monitoring', 'network-event']
+
+
+def get_logger_config():
+ return read_json('/run/vyos-network-event-logger.conf')
+
+
+class TestMonitoringNetworkEvent(VyOSUnitTestSHIM.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ super(TestMonitoringNetworkEvent, cls).setUpClass()
+
+ # ensure we can also run this test on a live system - so lets clean
+ # out the current configuration :)
+ cls.cli_delete(cls, base_path)
+
+ def tearDown(self):
+ self.cli_delete(base_path)
+ self.cli_commit()
+
+ def test_network_event_log(self):
+ expected_config = {
+ 'event': {
+ 'route': {},
+ 'link': {},
+ 'addr': {},
+ 'neigh': {},
+ 'rule': {},
+ },
+ 'queue_size': '10000'
+ }
+
+ self.cli_set(base_path + ['event', 'route'])
+ self.cli_set(base_path + ['event', 'link'])
+ self.cli_set(base_path + ['event', 'addr'])
+ self.cli_set(base_path + ['event', 'neigh'])
+ self.cli_set(base_path + ['event', 'rule'])
+ self.cli_set(base_path + ['queue-size', '10000'])
+ self.cli_commit()
+ self.assertEqual(expected_config, get_logger_config())
+
+
+if __name__ == '__main__':
+ unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_system_flow-accounting.py b/smoketest/scripts/cli/test_system_flow-accounting.py
index 515134220..9d7942789 100755
--- a/smoketest/scripts/cli/test_system_flow-accounting.py
+++ b/smoketest/scripts/cli/test_system_flow-accounting.py
@@ -97,111 +97,6 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
self.assertIn(f'syslog: {syslog}', uacctd)
self.assertIn(f'plugins: memory', uacctd)
- def test_sflow(self):
- sampling_rate = '4000'
- source_address = '192.0.2.1'
- dummy_if = 'dum3841'
- agent_address = '192.0.2.2'
-
- sflow_server = {
- '1.2.3.4' : { },
- '5.6.7.8' : { 'port' : '6000' },
- }
-
- self.cli_set(['interfaces', 'dummy', dummy_if, 'address', agent_address + '/32'])
- self.cli_set(['interfaces', 'dummy', dummy_if, 'address', source_address + '/32'])
- self.cli_set(base_path + ['disable-imt'])
-
- # You need to configure at least one interface for flow-accounting
- with self.assertRaises(ConfigSessionError):
- self.cli_commit()
- for interface in Section.interfaces('ethernet'):
- self.cli_set(base_path + ['interface', interface])
-
-
- # You need to configure at least one sFlow or NetFlow protocol, or not
- # set "disable-imt" for flow-accounting
- with self.assertRaises(ConfigSessionError):
- self.cli_commit()
-
- self.cli_set(base_path + ['sflow', 'agent-address', agent_address])
- self.cli_set(base_path + ['sflow', 'sampling-rate', sampling_rate])
- self.cli_set(base_path + ['sflow', 'source-address', source_address])
- for server, server_config in sflow_server.items():
- self.cli_set(base_path + ['sflow', 'server', server])
- if 'port' in server_config:
- self.cli_set(base_path + ['sflow', 'server', server, 'port', server_config['port']])
-
- # commit changes
- self.cli_commit()
-
- uacctd = read_file(uacctd_conf)
-
- # when 'disable-imt' is not configured on the CLI it must be present
- self.assertNotIn(f'imt_path: /tmp/uacctd.pipe', uacctd)
- self.assertNotIn(f'imt_mem_pools_number: 169', uacctd)
- self.assertNotIn(f'plugins: memory', uacctd)
-
- for server, server_config in sflow_server.items():
- plugin_name = server.replace('.', '-')
- if 'port' in server_config:
- self.assertIn(f'sfprobe_receiver[sf_{plugin_name}]: {server}', uacctd)
- else:
- self.assertIn(f'sfprobe_receiver[sf_{plugin_name}]: {server}:6343', uacctd)
-
- self.assertIn(f'sfprobe_agentip[sf_{plugin_name}]: {agent_address}', uacctd)
- self.assertIn(f'sampling_rate[sf_{plugin_name}]: {sampling_rate}', uacctd)
- self.assertIn(f'sfprobe_source_ip[sf_{plugin_name}]: {source_address}', uacctd)
-
- self.cli_delete(['interfaces', 'dummy', dummy_if])
-
- def test_sflow_ipv6(self):
- sampling_rate = '100'
- sflow_server = {
- '2001:db8::1' : { },
- '2001:db8::2' : { 'port' : '6000' },
- }
-
- self.cli_set(base_path + ['disable-imt'])
-
- # You need to configure at least one interface for flow-accounting
- with self.assertRaises(ConfigSessionError):
- self.cli_commit()
- for interface in Section.interfaces('ethernet'):
- self.cli_set(base_path + ['interface', interface])
-
-
- # You need to configure at least one sFlow or NetFlow protocol, or not
- # set "disable-imt" for flow-accounting
- with self.assertRaises(ConfigSessionError):
- self.cli_commit()
-
- self.cli_set(base_path + ['sflow', 'sampling-rate', sampling_rate])
- for server, server_config in sflow_server.items():
- self.cli_set(base_path + ['sflow', 'server', server])
- if 'port' in server_config:
- self.cli_set(base_path + ['sflow', 'server', server, 'port', server_config['port']])
-
- # commit changes
- self.cli_commit()
-
- uacctd = read_file(uacctd_conf)
-
- # when 'disable-imt' is not configured on the CLI it must be present
- self.assertNotIn(f'imt_path: /tmp/uacctd.pipe', uacctd)
- self.assertNotIn(f'imt_mem_pools_number: 169', uacctd)
- self.assertNotIn(f'plugins: memory', uacctd)
-
- for server, server_config in sflow_server.items():
- tmp_srv = server
- tmp_srv = tmp_srv.replace(':', '-')
-
- if 'port' in server_config:
- self.assertIn(f'sfprobe_receiver[sf_{tmp_srv}]: {bracketize_ipv6(server)}', uacctd)
- else:
- self.assertIn(f'sfprobe_receiver[sf_{tmp_srv}]: {bracketize_ipv6(server)}:6343', uacctd)
- self.assertIn(f'sampling_rate[sf_{tmp_srv}]: {sampling_rate}', uacctd)
-
def test_netflow(self):
engine_id = '33'
max_flows = '667'
@@ -288,8 +183,8 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
self.assertIn(f'nfprobe_timeouts[nf_{tmp_srv}]: expint={tmo_expiry}:general={tmo_flow}:icmp={tmo_icmp}:maxlife={tmo_max}:tcp.fin={tmo_tcp_fin}:tcp={tmo_tcp_generic}:tcp.rst={tmo_tcp_rst}:udp={tmo_udp}', uacctd)
-
self.cli_delete(['interfaces', 'dummy', dummy_if])
+
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_system_sflow.py b/smoketest/scripts/cli/test_system_sflow.py
index 74c065473..700253e2b 100755
--- a/smoketest/scripts/cli/test_system_sflow.py
+++ b/smoketest/scripts/cli/test_system_sflow.py
@@ -96,6 +96,39 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
for interface in Section.interfaces('ethernet'):
self.assertIn(f'pcap {{ dev={interface} }}', hsflowd)
+ def test_sflow_ipv6(self):
+ sampling_rate = '100'
+ default_polling = '30'
+ default_port = '6343'
+ sflow_server = {
+ '2001:db8::1': {},
+ '2001:db8::2': {'port': '8023'},
+ }
+
+ for interface in Section.interfaces('ethernet'):
+ self.cli_set(base_path + ['interface', interface])
+
+ self.cli_set(base_path + ['sampling-rate', sampling_rate])
+ for server, server_config in sflow_server.items():
+ self.cli_set(base_path + ['server', server])
+ if 'port' in server_config:
+ self.cli_set(base_path + ['server', server, 'port', server_config['port']])
+
+ # commit changes
+ self.cli_commit()
+
+ # verify configuration
+ hsflowd = read_file(hsflowd_conf)
+
+ self.assertIn(f'sampling={sampling_rate}', hsflowd)
+ self.assertIn(f'polling={default_polling}', hsflowd)
+
+ for server, server_config in sflow_server.items():
+ if 'port' in server_config:
+ self.assertIn(f'collector {{ ip = {server} udpport = {server_config["port"]} }}', hsflowd)
+ else:
+ self.assertIn(f'collector {{ ip = {server} udpport = {default_port} }}', hsflowd)
+
def test_vrf(self):
interface = 'eth0'
server = '192.0.2.1'
generated by cgit v1.2.3 (git 2.39.1) at 2025-12-27 09:40:26 +0000