1import json 2 3from unittest import TestCase 4from unittest.mock import patch 5from opnsense_cli.api.client import ApiClient 6from opnsense_cli.exceptions.api import APIException 7 8 9class TestApiClient(TestCase): 10 @patch('opnsense_cli.api.client.requests.get') 11 def test_execute_get_success(self, request_mock): 12 api_response_fixture = {'product_id': 'opnsense'} 13 request_mock.return_value.status_code = 200 14 request_mock.return_value.text = json.dumps(api_response_fixture) 15 16 client_args = [ 17 'api_key', 18 'api_secret', 19 'https://127.0.0.1/api', 20 True, 21 '~/.opn-cli/ca.pem', 22 60 23 ] 24 api_config = { 25 "module": "Core", 26 "controller": "firmware", 27 "method": "get", 28 "command": "info", 29 } 30 api_parameters = [] 31 32 client = ApiClient(*client_args) 33 result = client.execute(*api_parameters, **api_config) 34 35 request_mock.assert_called_once_with( 36 'https://127.0.0.1/api/core/firmware/info', 37 verify='~/.opn-cli/ca.pem', 38 auth=('api_key', 'api_secret'), 39 timeout=60 40 ) 41 self.assertEqual(api_response_fixture, result) 42 43 @patch('opnsense_cli.api.client.requests.get') 44 def test_execute_get_failure(self, request_mock): 45 request_mock.return_value.status_code = 400 46 request_mock.return_value.text = { 47 "message": "controller OPNsense\\Core\\Api\\IndexController not found", 48 "status": 400 49 } 50 request_mock.return_value.url = 'https://127.0.0.1/api/not/existing/confusion' 51 52 client_args = [ 53 'api_key', 54 'api_secret', 55 'https://127.0.0.1/api', 56 True, 57 '~/.opn-cli/ca.pem', 58 60 59 ] 60 api_config = { 61 "module": "Not", 62 "controller": "Existing", 63 "method": "get", 64 "command": "confusion", 65 } 66 api_parameters = [] 67 68 client = ApiClient(*client_args) 69 self.assertRaises(APIException, client.execute, *api_parameters, **api_config) 70 request_mock.assert_called_once_with( 71 'https://127.0.0.1/api/not/existing/confusion', 72 verify='~/.opn-cli/ca.pem', 73 auth=('api_key', 'api_secret'), 74 timeout=60 75 ) 76 77 @patch('opnsense_cli.api.client.requests.post') 78 def test_execute_post_json_success(self, request_mock): 79 api_response_fixture = [ 80 {'status': 'ok', 'msg_uuid': '8a0a415a-dbee-410d-be9f-01b90d71ff7c'} 81 ] 82 request_mock.return_value.status_code = 200 83 request_mock.return_value.text = json.dumps(api_response_fixture) 84 85 client_args = [ 86 'api_key2', 87 'api_secret2', 88 'https://127.0.0.1/api', 89 False, 90 '~/.opn-cli/ca.pem', 91 40 92 ] 93 api_config = { 94 "module": "openvpn", 95 "controller": "export", 96 "method": "post", 97 "command": "download", 98 } 99 api_parameters = [ 100 'vpnid', 101 'certref' 102 ] 103 104 api_payload = { 105 "param1": 0, 106 "param2": "test", 107 "paramN": "testN", 108 } 109 110 client = ApiClient(*client_args) 111 result = client.execute(*api_parameters, json=api_payload, **api_config) 112 113 request_mock.assert_called_once_with( 114 'https://127.0.0.1/api/openvpn/export/download/vpnid/certref', 115 json=api_payload, 116 verify=False, 117 auth=('api_key2', 'api_secret2'), 118 timeout=40 119 ) 120 self.assertEqual(api_response_fixture, result) 121 122 def test_execute_failure(self): 123 client_args = [ 124 'api_key3', 125 'api_secret3', 126 'https://127.0.0.1/api', 127 False, 128 '~/.opn-cli/ca.pem', 129 10 130 ] 131 api_config = { 132 "module": "Core", 133 "controller": "Firmware", 134 "method": "head", 135 "command": "reinstall", 136 } 137 api_parameters = [] 138 139 client = ApiClient(*client_args) 140 self.assertRaises(APIException, client.execute, *api_parameters, **api_config) 141