smoketest: flow-accounting: T4437: adjust smoketest to new generated config syntax

1 files changed, 74 insertions, 20 deletions

diff --git a/smoketest/scripts/cli/test_system_flow-accounting.py b/smoketest/scripts/cli/test_system_flow-accounting.py
index 5a73ebc7d..a6eef3fb6 100755
--- a/smoketest/scripts/cli/test_system_flow-accounting.py
+++ b/smoketest/scripts/cli/test_system_flow-accounting.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 #
-# Copyright (C) 2021 VyOS maintainers and contributors
+# Copyright (C) 2021-2022 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
@@ -20,6 +20,8 @@ from base_vyostest_shim import VyOSUnitTestSHIM
 
 from vyos.configsession import ConfigSessionError
 from vyos.ifconfig import Section
+from vyos.template import bracketize_ipv6
+from vyos.template import is_ipv6
 from vyos.util import cmd
 from vyos.util import process_named_running
 from vyos.util import read_file
@@ -103,14 +105,12 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
         agent_address = '192.0.2.2'
 
         sflow_server = {
-            '1.2.3.4' : {
-            },
-            '5.6.7.8' : {
-                'port' : '6000'
-            }
+            '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
@@ -155,6 +155,54 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
 
         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
+            if is_ipv6(tmp_srv):
+                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'
@@ -173,14 +221,13 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
         tmo_udp = '10'
 
         netflow_server = {
-            '11.22.33.44' : {
-            },
-            '55.66.77.88' : {
-                'port' : '6000'
-            }
+            '11.22.33.44' : { },
+            '55.66.77.88' : { 'port' : '6000' },
+            '2001:db8::1' : { },
         }
 
         self.cli_set(['interfaces', 'dummy', dummy_if, 'address', agent_address + '/32'])
+        self.cli_set(['interfaces', 'dummy', dummy_if, 'address', source_address + '/32'])
 
         for interface in Section.interfaces('ethernet'):
             self.cli_set(base_path + ['interface', interface])
@@ -217,23 +264,30 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
 
         tmp = []
         for server, server_config in netflow_server.items():
-            tmp.append(f'nfprobe[nf_{server}]')
+            tmp_srv = server
+            if is_ipv6(tmp_srv):
+                tmp_srv = tmp_srv.replace(':', '.')
+            tmp.append(f'nfprobe[nf_{tmp_srv}]')
         tmp.append('memory')
         self.assertIn('plugins: ' + ','.join(tmp), uacctd)
 
         for server, server_config in netflow_server.items():
-            self.assertIn(f'nfprobe_engine[nf_{server}]: {engine_id}', uacctd)
-            self.assertIn(f'nfprobe_maxflows[nf_{server}]: {max_flows}', uacctd)
-            self.assertIn(f'sampling_rate[nf_{server}]: {sampling_rate}', uacctd)
-            self.assertIn(f'nfprobe_source_ip[nf_{server}]: {source_address}', uacctd)
-            self.assertIn(f'nfprobe_version[nf_{server}]: {version}', uacctd)
+            tmp_srv = server
+            if is_ipv6(tmp_srv):
+                tmp_srv = tmp_srv.replace(':', '.')
+
+            self.assertIn(f'nfprobe_engine[nf_{tmp_srv}]: {engine_id}', uacctd)
+            self.assertIn(f'nfprobe_maxflows[nf_{tmp_srv}]: {max_flows}', uacctd)
+            self.assertIn(f'sampling_rate[nf_{tmp_srv}]: {sampling_rate}', uacctd)
+            self.assertIn(f'nfprobe_source_ip[nf_{tmp_srv}]: {source_address}', uacctd)
+            self.assertIn(f'nfprobe_version[nf_{tmp_srv}]: {version}', uacctd)
 
             if 'port' in server_config:
-                self.assertIn(f'nfprobe_receiver[nf_{server}]: {server}', uacctd)
+                self.assertIn(f'nfprobe_receiver[nf_{tmp_srv}]: {bracketize_ipv6(server)}', uacctd)
             else:
-                self.assertIn(f'nfprobe_receiver[nf_{server}]: {server}:2055', uacctd)
+                self.assertIn(f'nfprobe_receiver[nf_{tmp_srv}]: {bracketize_ipv6(server)}:2055', uacctd)
 
-            self.assertIn(f'nfprobe_timeouts[nf_{server}]: 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.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])