nat66: T4605: Refactor NAT66 to use python module for parsing rules

* Rename table to vyos_nat
* Refactor tests to use `verify_nftables` format

1 files changed, 26 insertions, 37 deletions

diff --git a/smoketest/scripts/cli/test_nat66.py b/smoketest/scripts/cli/test_nat66.py
index 537b094a4..6cf7ca0a1 100755
--- a/smoketest/scripts/cli/test_nat66.py
+++ b/smoketest/scripts/cli/test_nat66.py
@@ -71,12 +71,12 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         self.cli_commit()
 
         nftables_search = [
-            ['oifname "eth1"', 'ip6 saddr fc00::/64', 'snat prefix to fc01::/64'],
-            ['oifname "eth1"', 'ip6 saddr fc00::/64', 'masquerade'],
-            ['oifname "eth1"', 'ip6 saddr fc00::/64', 'return']
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', f'snat prefix to {translation_prefix}'],
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', 'masquerade'],
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', 'return']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_source_nat66_address(self):
         source_prefix = 'fc00::/64'
@@ -88,25 +88,11 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         # check validate() - outbound-interface must be defined
         self.cli_commit()
 
-        tmp = cmd('sudo nft -j list table ip6 nat')
-        data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp))
-
-        for idx in range(0, len(data_json)):
-            data = data_json[idx]
-
-            self.assertEqual(data['chain'], 'POSTROUTING')
-            self.assertEqual(data['family'], 'ip6')
-            self.assertEqual(data['table'], 'nat')
-
-            iface = dict_search('match.right', data['expr'][0])
-            address = dict_search('match.right.prefix.addr', data['expr'][2])
-            mask = dict_search('match.right.prefix.len', data['expr'][2])
-            snat_address = dict_search('snat.addr', data['expr'][3])
+        nftables_search = [
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', f'snat to {translation_address}']
+        ]
 
-            self.assertEqual(iface, 'eth1')
-            # check for translation address
-            self.assertEqual(snat_address, translation_address)
-            self.assertEqual(f'{address}/{mask}', source_prefix)
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_destination_nat66(self):
         destination_address = 'fc00::1'
@@ -129,7 +115,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
             ['iifname "eth1"', 'ip6 saddr fc02::1', 'ip6 daddr fc00::1', 'return']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_destination_nat66_protocol(self):
         translation_address = '2001:db8:1111::1'
@@ -153,7 +139,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
             ['iifname "eth1"', 'tcp dport 4545', 'ip6 saddr 2001:db8:2222::/64', 'tcp sport 8080', 'dnat to 2001:db8:1111::1:5555']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_destination_nat66_prefix(self):
         destination_prefix = 'fc00::/64'
@@ -165,22 +151,25 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         # check validate() - outbound-interface must be defined
         self.cli_commit()
 
-        tmp = cmd('sudo nft -j list table ip6 nat')
-        data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp))
+        nftables_search = [
+            ['iifname "eth1"', f'ip6 daddr {destination_prefix}', f'dnat prefix to {translation_prefix}']
+        ]
+
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
-        for idx in range(0, len(data_json)):
-            data = data_json[idx]
+    def test_destination_nat66_without_translation_address(self):
+        self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1'])
+        self.cli_set(dst_path + ['rule', '1', 'destination', 'port', '443'])
+        self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp'])
+        self.cli_set(dst_path + ['rule', '1', 'translation', 'port', '443'])
 
-            self.assertEqual(data['chain'], 'PREROUTING')
-            self.assertEqual(data['family'], 'ip6')
-            self.assertEqual(data['table'], 'nat')
+        self.cli_commit()
 
-            iface = dict_search('match.right', data['expr'][0])
-            translation_address = dict_search('dnat.addr.prefix.addr', data['expr'][3])
-            translation_mask = dict_search('dnat.addr.prefix.len', data['expr'][3])
+        nftables_search = [
+            ['iifname "eth1"', 'tcp dport 443', 'dnat to :443']
+        ]
 
-            self.assertEqual(f'{translation_address}/{translation_mask}', translation_prefix)
-            self.assertEqual(iface, 'eth1')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_source_nat66_required_translation_prefix(self):
         # T2813: Ensure translation address is specified
@@ -222,7 +211,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
             ['oifname "eth1"', 'ip6 saddr 2001:db8:2222::/64', 'tcp dport 9999', 'tcp sport 8080', 'snat to 2001:db8:1111::1:80']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_nat66_no_rules(self):
         # T3206: deleting all rules but keep the direction 'destination' or