diff options
Diffstat (limited to 'smoketest/scripts/cli/test_service_https.py')
| -rwxr-xr-x | smoketest/scripts/cli/test_service_https.py | 173 |
1 files changed, 157 insertions, 16 deletions
diff --git a/smoketest/scripts/cli/test_service_https.py b/smoketest/scripts/cli/test_service_https.py index 7fee7158d..4585bb0bd 100755 --- a/smoketest/scripts/cli/test_service_https.py +++ b/smoketest/scripts/cli/test_service_https.py @@ -17,6 +17,7 @@ import unittest import json import psutil +import time from requests import request from urllib3.exceptions import InsecureRequestWarning @@ -35,6 +36,9 @@ from vyos.configsession import ConfigSessionError base_path = ['service', 'https'] pki_base = ['pki'] +address = '127.0.0.1' +key = 'VyOS-key' + cert_data = """ MIICFDCCAbugAwIBAgIUfMbIsB/ozMXijYgUYG80T1ry+mcwCgYIKoZIzj0EAwIw WTELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv @@ -116,6 +120,40 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): # always forward to base class super().tearDown() + def _api_get_background_operations(self): + url = f'https://{address}/retrieve/background-operations' + r = request('POST', url, verify=False, json={'key': key}) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertTrue(body.get('success')) + ops = body.get('data', {}).get('operations', []) + return ops + + def _wait_no_active_operations(self, timeout: int = 30): + # wait until no queued/running operations remain + deadline = time.time() + timeout + statuses = ('queued', 'running') + + while time.time() < deadline: + ops = self._api_get_background_operations() + ops = [op for op in ops if op.get('status') in statuses] + if not ops: + return + sleep(0.25) + self.fail('Timeout waiting for background operations to finish') + + def assertBackgroundOpResponseIsOk(self, response): + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertTrue(body.get('success')) + + data = body.get('data', {}) + self.assertIsInstance(data, dict) + self.assertIn('operation', data) + op = data['operation'] + self.assertIn('op_id', op) + self.assertIn('status', op) + def test_listen_address(self): test_prefix = ['192.0.2.1/26', '2001:db8:1::ffff/64'] test_addr = [ i.split('/')[0] for i in test_prefix ] @@ -356,8 +394,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_add_delete(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/retrieve' payload = {'data': '{"op": "showConfig", "path": []}', 'key': f'{key}'} headers = {} @@ -387,8 +423,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_show(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/show' headers = {} @@ -405,8 +439,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_generate(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/generate' headers = {} @@ -423,8 +455,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_configure(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/configure' headers = {} conf_interface = 'dum0' @@ -449,8 +479,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_config_file(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/config-file' headers = {} @@ -467,8 +495,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_reset(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/reset' headers = {} @@ -485,8 +511,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_image(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/image' headers = {} @@ -528,8 +552,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_config_file_load_http(self): # Test load config from HTTP URL - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/config-file' url_config = f'https://{address}/configure' headers = {} @@ -572,6 +594,125 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): call(f'sudo rm -f {nginx_tmp_site}') call('sudo systemctl reload nginx') + @ignore_warning(InsecureRequestWarning) + def test_api_configure_background(self): + url = f'https://{address}/configure' + conf_interface = 'dum8' + conf_address = '192.0.2.88/32' + + # Enable REST API + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + self.cli_set(base_path + ['api', 'rest']) + self.cli_commit() + + payload_path = [ + 'interfaces', + 'dummy', + conf_interface, + 'address', + ] + params = {'in_background': True} + payload = { + 'data': json.dumps( + {'op': 'set', 'path': payload_path, 'value': conf_address} + ), + 'key': key, + } + + r = request('POST', url, verify=False, params=params, data=payload) + self.assertBackgroundOpResponseIsOk(r) + body = r.json() + op = body.get('data', {}).get('operation', []) + + # Operation should appear as active shortly + ops = self._api_get_background_operations() + self.assertTrue( + any(o.get('op_id') == op.get('op_id') for o in ops), + 'Queued operation is not visible in `/retrieve/background-operations`', + ) + + # Wait until done + self._wait_no_active_operations() + + # Verify config applied (using CLI show) + self.assertIn(conf_address, self.op_mode(['show', 'configuration', 'commands'])) + + @ignore_warning(InsecureRequestWarning) + def test_api_configure_section_background(self): + url = f'https://{address}/configure-section' + conf_interface = 'dum9' + conf_address = '192.0.2.99/32' + + # Enable REST API + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + self.cli_set(base_path + ['api', 'rest']) + self.cli_commit() + + # Configure-section payload: set a full section + # example: set interfaces dummy dum8 address 192.0.2.99/32 + payload = { + 'data': json.dumps( + { + 'op': 'set', + 'path': ['interfaces', 'dummy', conf_interface], + 'section': { + 'address': [conf_address], + }, + } + ), + 'key': key, + } + params = {'in_background': True} + + r = request('POST', url, verify=False, params=params, data=payload) + self.assertBackgroundOpResponseIsOk(r) + + # Wait until done + self._wait_no_active_operations() + + # Verify section applied + self.assertIn(conf_address, self.op_mode(['show', 'configuration', 'commands'])) + + @ignore_warning(InsecureRequestWarning) + def test_api_configure_background_ops_over_max(self): + max_ops = 128 + + # Enable REST API + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + self.cli_set(base_path + ['api', 'rest']) + self.cli_commit() + + op_ids = [] + params = {'in_background': True} + url = f'https://{address}/configure' + + # Create many non-existent configurations to fill the queue. + for i in range(max_ops + 5): + config_name = f'invalid-test-option-{i}' + payload_path = ['system', config_name] + payload = { + 'data': json.dumps( + {'op': 'set', 'path': payload_path, 'value': config_name} + ), + 'key': key, + } + + with self.subTest(payload_path=payload_path): + r = request('POST', url, verify=False, params=params, data=payload) + self.assertBackgroundOpResponseIsOk(r) + + body = r.json() + op = body.get('data', {}).get('operation', []) + op_ids.append(op) + + # Wait for queue to drain + self._wait_no_active_operations(timeout=120) + + # Verify pruning: oldest `op_id` should be absent, and count should be <= `max_ops` + ops = self._api_get_background_operations() + self.assertLessEqual(len(ops), max_ops) + self.assertFalse(any(o.get('op_id') == op_ids[0] for o in ops)) + if __name__ == '__main__': unittest.main(verbosity=5) |