summaryrefslogtreecommitdiff
path: root/smoketest/scripts
diff options
context:
space:
mode:
Diffstat (limited to 'smoketest/scripts')
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_bridge.py118
-rwxr-xr-xsmoketest/scripts/cli/test_protocols_bgp.py26
-rwxr-xr-xsmoketest/scripts/cli/test_service_https.py57
-rwxr-xr-xsmoketest/scripts/cli/test_service_ids.py26
-rwxr-xr-xsmoketest/scripts/cli/test_service_router-advert.py62
5 files changed, 234 insertions, 55 deletions
diff --git a/smoketest/scripts/cli/test_interfaces_bridge.py b/smoketest/scripts/cli/test_interfaces_bridge.py
index 664dc48bc..8f711af20 100755
--- a/smoketest/scripts/cli/test_interfaces_bridge.py
+++ b/smoketest/scripts/cli/test_interfaces_bridge.py
@@ -224,6 +224,51 @@ class BridgeInterfaceTest(BasicInterfaceTest.TestCase):
super().test_vif_8021q_mtu_limits()
def test_bridge_vlan_filter(self):
+ def _verify_members() -> None:
+ # check member interfaces are added on the bridge
+ for interface in self._interfaces:
+ bridge_members = []
+ for tmp in glob(f'/sys/class/net/{interface}/lower_*'):
+ bridge_members.append(os.path.basename(tmp).replace('lower_', ''))
+
+ # We can not use assertListEqual() b/c the position of the interface
+ # names within the list is not fixed
+ self.assertEqual(len(self._members), len(bridge_members))
+ for member in self._members:
+ self.assertIn(member, bridge_members)
+
+ def _check_vlan_filter() -> None:
+ for interface in self._interfaces:
+ tmp = cmd(f'bridge -j vlan show dev {interface}')
+ tmp = json.loads(tmp)
+ self.assertIsNotNone(tmp)
+
+ for interface_status in tmp:
+ ifname = interface_status['ifname']
+ for interface in self._members:
+ vlan_success = 0;
+ if interface == ifname:
+ vlans_status = interface_status['vlans']
+ for vlan_status in vlans_status:
+ vlan_id = vlan_status['vlan']
+ flag_num = 0
+ if 'flags' in vlan_status:
+ flags = vlan_status['flags']
+ for flag in flags:
+ flag_num = flag_num +1
+ if vlan_id == 2:
+ if flag_num == 0:
+ vlan_success = vlan_success + 1
+ else:
+ for id in range(4,10):
+ if vlan_id == id:
+ if flag_num == 0:
+ vlan_success = vlan_success + 1
+ if vlan_id >= 101:
+ if flag_num == 2:
+ vlan_success = vlan_success + 1
+ self.assertGreaterEqual(vlan_success, 7)
+
vif_vlan = 2
# Add member interface to bridge and set VLAN filter
for interface in self._interfaces:
@@ -247,61 +292,50 @@ class BridgeInterfaceTest(BasicInterfaceTest.TestCase):
# commit config
self.cli_commit()
- # Detect the vlan filter function
+ # Verify correct setting of VLAN filter function
for interface in self._interfaces:
tmp = read_file(f'/sys/class/net/{interface}/bridge/vlan_filtering')
self.assertEqual(tmp, '1')
- # Execute the program to obtain status information
- json_data = cmd('bridge -j vlan show', shell=True)
- vlan_filter_status = None
- vlan_filter_status = json.loads(json_data)
-
- if vlan_filter_status is not None:
- for interface_status in vlan_filter_status:
- ifname = interface_status['ifname']
- for interface in self._members:
- vlan_success = 0;
- if interface == ifname:
- vlans_status = interface_status['vlans']
- for vlan_status in vlans_status:
- vlan_id = vlan_status['vlan']
- flag_num = 0
- if 'flags' in vlan_status:
- flags = vlan_status['flags']
- for flag in flags:
- flag_num = flag_num +1
- if vlan_id == 2:
- if flag_num == 0:
- vlan_success = vlan_success + 1
- else:
- for id in range(4,10):
- if vlan_id == id:
- if flag_num == 0:
- vlan_success = vlan_success + 1
- if vlan_id >= 101:
- if flag_num == 2:
- vlan_success = vlan_success + 1
- if vlan_success >= 7:
- self.assertTrue(True)
- else:
- self.assertTrue(False)
+ # Execute the program to obtain status information and verify proper
+ # VLAN filter setup
+ _check_vlan_filter()
- else:
- self.assertTrue(False)
+ # check member interfaces are added on the bridge
+ _verify_members()
+
+ # change member interface description to trigger config update,
+ # VLANs must still exist (T4565)
+ for interface in self._interfaces:
+ for member in self._members:
+ self.cli_set(['interfaces', Section.section(member), member, 'description', f'foo {member}'])
+
+ # commit config
+ self.cli_commit()
# check member interfaces are added on the bridge
+ _verify_members()
+
+ # Execute the program to obtain status information and verify proper
+ # VLAN filter setup
+ _check_vlan_filter()
+
+ # delete all members
+ for interface in self._interfaces:
+ self.cli_delete(self._base_path + [interface, 'member'])
+
+ # commit config
+ self.cli_commit()
+
+ # verify member interfaces are no longer assigned on the bridge
for interface in self._interfaces:
bridge_members = []
for tmp in glob(f'/sys/class/net/{interface}/lower_*'):
bridge_members.append(os.path.basename(tmp).replace('lower_', ''))
+ self.assertNotEqual(len(self._members), len(bridge_members))
for member in self._members:
- self.assertIn(member, bridge_members)
-
- # delete all members
- for interface in self._interfaces:
- self.cli_delete(self._base_path + [interface, 'member'])
+ self.assertNotIn(member, bridge_members)
def test_bridge_vif_members(self):
diff --git a/smoketest/scripts/cli/test_protocols_bgp.py b/smoketest/scripts/cli/test_protocols_bgp.py
index 9c0c93779..009dbc803 100755
--- a/smoketest/scripts/cli/test_protocols_bgp.py
+++ b/smoketest/scripts/cli/test_protocols_bgp.py
@@ -921,5 +921,31 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase):
self.assertIn(f' neighbor {peer_group} peer-group', frrconfig)
self.assertIn(f' neighbor {peer_group} remote-as {remote_asn}', frrconfig)
+ def test_bgp_15_local_as_ebgp(self):
+ # https://phabricator.vyos.net/T4560
+ # local-as allowed only for ebgp peers
+
+ neighbor = '192.0.2.99'
+ remote_asn = '500'
+ local_asn = '400'
+
+ self.cli_set(base_path + ['local-as', ASN])
+ self.cli_set(base_path + ['neighbor', neighbor, 'remote-as', ASN])
+ self.cli_set(base_path + ['neighbor', neighbor, 'local-as', local_asn])
+
+ # check validate() - local-as allowed only for ebgp peers
+ with self.assertRaises(ConfigSessionError):
+ self.cli_commit()
+
+ self.cli_set(base_path + ['neighbor', neighbor, 'remote-as', remote_asn])
+
+ self.cli_commit()
+
+ frrconfig = self.getFRRconfig(f'router bgp {ASN}')
+ self.assertIn(f'router bgp {ASN}', frrconfig)
+ self.assertIn(f' neighbor {neighbor} remote-as {remote_asn}', frrconfig)
+ self.assertIn(f' neighbor {neighbor} local-as {local_asn}', frrconfig)
+
+
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_service_https.py b/smoketest/scripts/cli/test_service_https.py
index 71fb3e177..72c1d4e43 100755
--- a/smoketest/scripts/cli/test_service_https.py
+++ b/smoketest/scripts/cli/test_service_https.py
@@ -138,5 +138,62 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
# Must get HTTP code 401 on missing key (Unauthorized)
self.assertEqual(r.status_code, 401)
+ # GraphQL auth test: a missing key will return status code 400, as
+ # 'key' is a non-nullable field in the schema; an incorrect key is
+ # caught by the resolver, and returns success 'False', so one must
+ # check the return value.
+
+ self.cli_set(base_path + ['api', 'gql'])
+ self.cli_commit()
+
+ gql_url = f'https://{address}/graphql'
+
+ query_valid_key = f"""
+ {{
+ SystemStatus (data: {{key: "{key}"}}) {{
+ success
+ errors
+ data {{
+ result
+ }}
+ }}
+ }}
+ """
+
+ r = request('POST', gql_url, verify=False, headers=headers, json={'query': query_valid_key})
+ success = r.json()['data']['SystemStatus']['success']
+ self.assertTrue(success)
+
+ query_invalid_key = """
+ {
+ SystemStatus (data: {key: "invalid"}) {
+ success
+ errors
+ data {
+ result
+ }
+ }
+ }
+ """
+
+ r = request('POST', gql_url, verify=False, headers=headers, json={'query': query_invalid_key})
+ success = r.json()['data']['SystemStatus']['success']
+ self.assertFalse(success)
+
+ query_no_key = """
+ {
+ SystemStatus (data: {}) {
+ success
+ errors
+ data {
+ result
+ }
+ }
+ }
+ """
+
+ r = request('POST', gql_url, verify=False, headers=headers, json={'query': query_no_key})
+ self.assertEqual(r.status_code, 400)
+
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_service_ids.py b/smoketest/scripts/cli/test_service_ids.py
index 18f1b8ec5..d471eeaed 100755
--- a/smoketest/scripts/cli/test_service_ids.py
+++ b/smoketest/scripts/cli/test_service_ids.py
@@ -24,7 +24,9 @@ from vyos.util import process_named_running
from vyos.util import read_file
PROCESS_NAME = 'fastnetmon'
-FASTNETMON_CONF = '/etc/fastnetmon.conf'
+FASTNETMON_CONF = '/run/fastnetmon/fastnetmon.conf'
+NETWORKS_CONF = '/run/fastnetmon/networks_list'
+EXCLUDED_NETWORKS_CONF = '/run/fastnetmon/excluded_networks_list'
base_path = ['service', 'ids', 'ddos-protection']
class TestServiceIDS(VyOSUnitTestSHIM.TestCase):
@@ -48,7 +50,8 @@ class TestServiceIDS(VyOSUnitTestSHIM.TestCase):
self.assertFalse(process_named_running(PROCESS_NAME))
def test_fastnetmon(self):
- networks = ['10.0.0.0/24', '10.5.5.0/24']
+ networks = ['10.0.0.0/24', '10.5.5.0/24', '2001:db8:10::/64', '2001:db8:20::/64']
+ excluded_networks = ['10.0.0.1/32', '2001:db8:10::1/128']
interfaces = ['eth0', 'eth1']
fps = '3500'
mbps = '300'
@@ -61,6 +64,12 @@ class TestServiceIDS(VyOSUnitTestSHIM.TestCase):
for tmp in networks:
self.cli_set(base_path + ['network', tmp])
+ # optional excluded-network!
+ with self.assertRaises(ConfigSessionError):
+ self.cli_commit()
+ for tmp in excluded_networks:
+ self.cli_set(base_path + ['excluded-network', tmp])
+
# Required interface(s)!
with self.assertRaises(ConfigSessionError):
self.cli_commit()
@@ -86,9 +95,22 @@ class TestServiceIDS(VyOSUnitTestSHIM.TestCase):
self.assertIn(f'threshold_mbps = {mbps}', config)
self.assertIn(f'ban_for_pps = on', config)
self.assertIn(f'threshold_pps = {pps}', config)
+ # default
+ self.assertIn(f'enable_ban = on', config)
+ self.assertIn(f'enable_ban_ipv6 = on', config)
+ self.assertIn(f'ban_time = 1900', config)
tmp = ','.join(interfaces)
self.assertIn(f'interfaces = {tmp}', config)
+
+ network_config = read_file(NETWORKS_CONF)
+ for tmp in networks:
+ self.assertIn(f'{tmp}', network_config)
+
+ excluded_network_config = read_file(EXCLUDED_NETWORKS_CONF)
+ for tmp in excluded_networks:
+ self.assertIn(f'{tmp}', excluded_network_config)
+
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_service_router-advert.py b/smoketest/scripts/cli/test_service_router-advert.py
index 1168c05cd..873be7df0 100755
--- a/smoketest/scripts/cli/test_service_router-advert.py
+++ b/smoketest/scripts/cli/test_service_router-advert.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2019-2020 VyOS maintainers and contributors
+# Copyright (C) 2019-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
@@ -23,11 +23,13 @@ from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.util import read_file
from vyos.util import process_named_running
+PROCESS_NAME = 'radvd'
RADVD_CONF = '/run/radvd/radvd.conf'
interface = 'eth1'
base_path = ['service', 'router-advert', 'interface', interface]
address_base = ['interfaces', 'ethernet', interface, 'address']
+prefix = '::/64'
def get_config_value(key):
tmp = read_file(RADVD_CONF)
@@ -35,18 +37,36 @@ def get_config_value(key):
return tmp[0].split()[0].replace(';','')
class TestServiceRADVD(VyOSUnitTestSHIM.TestCase):
- def setUp(self):
- self.cli_set(address_base + ['2001:db8::1/64'])
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestServiceRADVD, cls).setUpClass()
+
+ # ensure we can also run this test on a live system - so lets clean
+ # out the current configuration :)
+ cls.cli_delete(cls, ['service', 'router-advert'])
+
+ cls.cli_set(cls, address_base + ['2001:db8::1/64'])
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.cli_delete(cls, address_base)
+ super(TestServiceRADVD, cls).tearDownClass()
def tearDown(self):
- self.cli_delete(address_base)
+ # Check for running process
+ self.assertTrue(process_named_running(PROCESS_NAME))
+
self.cli_delete(base_path)
self.cli_commit()
+ # Check for no longer running process
+ self.assertFalse(process_named_running(PROCESS_NAME))
+
def test_common(self):
- self.cli_set(base_path + ['prefix', '::/64', 'no-on-link-flag'])
- self.cli_set(base_path + ['prefix', '::/64', 'no-autonomous-flag'])
- self.cli_set(base_path + ['prefix', '::/64', 'valid-lifetime', 'infinity'])
+ self.cli_set(base_path + ['prefix', prefix, 'no-on-link-flag'])
+ self.cli_set(base_path + ['prefix', prefix, 'no-autonomous-flag'])
+ self.cli_set(base_path + ['prefix', prefix, 'valid-lifetime', 'infinity'])
self.cli_set(base_path + ['other-config-flag'])
# commit changes
@@ -57,7 +77,7 @@ class TestServiceRADVD(VyOSUnitTestSHIM.TestCase):
self.assertEqual(tmp, interface)
tmp = get_config_value('prefix')
- self.assertEqual(tmp, '::/64')
+ self.assertEqual(tmp, prefix)
tmp = get_config_value('AdvOtherConfigFlag')
self.assertEqual(tmp, 'on')
@@ -88,15 +108,19 @@ class TestServiceRADVD(VyOSUnitTestSHIM.TestCase):
tmp = get_config_value('AdvOnLink')
self.assertEqual(tmp, 'off')
- # Check for running process
- self.assertTrue(process_named_running('radvd'))
+ tmp = get_config_value('DeprecatePrefix')
+ self.assertEqual(tmp, 'off')
+
+ tmp = get_config_value('DecrementLifetimes')
+ self.assertEqual(tmp, 'off')
+
def test_dns(self):
nameserver = ['2001:db8::1', '2001:db8::2']
dnssl = ['vyos.net', 'vyos.io']
ns_lifetime = '599'
- self.cli_set(base_path + ['prefix', '::/64', 'valid-lifetime', 'infinity'])
+ self.cli_set(base_path + ['prefix', prefix, 'valid-lifetime', 'infinity'])
self.cli_set(base_path + ['other-config-flag'])
for ns in nameserver:
@@ -127,5 +151,21 @@ class TestServiceRADVD(VyOSUnitTestSHIM.TestCase):
self.assertIn(tmp, config)
+ def test_deprecate_prefix(self):
+ self.cli_set(base_path + ['prefix', prefix, 'valid-lifetime', 'infinity'])
+ self.cli_set(base_path + ['prefix', prefix, 'deprecate-prefix'])
+ self.cli_set(base_path + ['prefix', prefix, 'decrement-lifetime'])
+
+ # commit changes
+ self.cli_commit()
+
+ config = read_file(RADVD_CONF)
+
+ tmp = get_config_value('DeprecatePrefix')
+ self.assertEqual(tmp, 'on')
+
+ tmp = get_config_value('DecrementLifetimes')
+ self.assertEqual(tmp, 'on')
+
if __name__ == '__main__':
unittest.main(verbosity=2)