v1/tests/async_process_api_tests.py

0001 # TODO: the secure-endpoint results depend on the cert used for the call.
0002 # These results assume the cert's DN is in the `allowAsyncRequest` list.
0003
0004 import socket
0005 import unittest
0006 import uuid
0007
0008 from pandaserver.api.v1.http_client import HttpClient, api_url, api_url_ssl
0009
0010 NO_SSL_RESPONSE = {"success": False, "message": "SSL secure connection is required", "data": None}
0011
0012
0013 class TestAsyncProcessAPI(unittest.TestCase):
0014     def setUp(self):
0015         self.http_client = HttpClient()
0016         self.urls = [api_url, api_url_ssl]
0017
0018     def test_submit_grep_request_no_ssl(self):
0019         full_url = f"{api_url}/async_process/submit_grep_request"
0020         print(f"Testing URL: {full_url}")
0021         data = {"pattern": "ERROR", "log_filename": "panda-server.log", "service_name": "server"}
0022         status, output = self.http_client.post(full_url, data)
0023         print(output)
0024         self.assertEqual(output, NO_SSL_RESPONSE)
0025
0026     def test_submit_grep_request_missing_target(self):
0027         full_url = f"{api_url_ssl}/async_process/submit_grep_request"
0028         print(f"Testing URL: {full_url}")
0029         data = {"pattern": "ERROR", "log_filename": "panda-server.log"}
0030         status, output = self.http_client.post(full_url, data)
0031         print(output)
0032         expected_response = {
0033             "success": False,
0034             "message": "exactly one of service_name or machine_name must be provided",
0035             "data": None,
0036         }
0037         self.assertEqual(output, expected_response)
0038
0039     def test_submit_grep_request_both_targets(self):
0040         full_url = f"{api_url_ssl}/async_process/submit_grep_request"
0041         print(f"Testing URL: {full_url}")
0042         data = {
0043             "pattern": "ERROR",
0044             "log_filename": "panda-server.log",
0045             "service_name": "server",
0046             "machine_name": socket.getfqdn(),
0047         }
0048         status, output = self.http_client.post(full_url, data)
0049         print(output)
0050         expected_response = {
0051             "success": False,
0052             "message": "exactly one of service_name or machine_name must be provided",
0053             "data": None,
0054         }
0055         self.assertEqual(output, expected_response)
0056
0057     def test_submit_grep_request_invalid_filename(self):
0058         full_url = f"{api_url_ssl}/async_process/submit_grep_request"
0059         print(f"Testing URL: {full_url}")
0060         data = {"pattern": "ERROR", "log_filename": "../etc/passwd", "service_name": "server"}
0061         status, output = self.http_client.post(full_url, data)
0062         print(output)
0063         expected_response = {
0064             "success": False,
0065             "message": "invalid log_filename: must not contain path separators",
0066             "data": None,
0067         }
0068         self.assertEqual(output, expected_response)
0069
0070     def test_submit_grep_request_success(self):
0071         full_url = f"{api_url_ssl}/async_process/submit_grep_request"
0072         print(f"Testing URL: {full_url}")
0073         data = {
0074             "pattern": "ERROR",
0075             "log_filename": "panda-server.log",
0076             "machine_name": socket.getfqdn(),
0077         }
0078         status, output = self.http_client.post(full_url, data)
0079         print(output)
0080         self.assertTrue(output["success"])
0081         self.assertIsInstance(output["data"], dict)
0082         request_id = output["data"]["request_id"]
0083         self.assertIsInstance(request_id, str)
0084         self.assertEqual(len(request_id), 36)
0085
0086     def test_get_result_not_found(self):
0087         missing_id = str(uuid.uuid4())
0088         for url in self.urls:
0089             with self.subTest(base_url=url):
0090                 full_url = f"{url}/async_process/get_result"
0091                 print(f"Testing URL: {full_url}")
0092                 data = {"request_id": missing_id}
0093                 status, output = self.http_client.get(full_url, data)
0094                 print(output)
0095                 expected_response = {
0096                     "success": False,
0097                     "message": f"request_id '{missing_id}' not found",
0098                     "data": None,
0099                 }
0100                 self.assertEqual(output, expected_response)
0101
0102     def test_get_result_pending(self):
0103         submit_url = f"{api_url_ssl}/async_process/submit_grep_request"
0104         print(f"Testing URL: {submit_url}")
0105         submit_data = {
0106             "pattern": "ERROR",
0107             "log_filename": "panda-server.log",
0108             "machine_name": socket.getfqdn(),
0109         }
0110         status, submit_output = self.http_client.post(submit_url, submit_data)
0111         print(submit_output)
0112         if not submit_output.get("success"):
0113             raise unittest.SkipTest(f"submit_grep_request did not succeed: {submit_output.get('message')}")
0114         request_id = submit_output["data"]["request_id"]
0115
0116         get_url = f"{api_url_ssl}/async_process/get_result"
0117         print(f"Testing URL: {get_url}")
0118         status, output = self.http_client.get(get_url, {"request_id": request_id})
0119         print(output)
0120         self.assertTrue(output["success"])
0121         self.assertIsInstance(output["data"], dict)
0122         self.assertEqual(output["data"]["overall_status"], "pending")
0123         self.assertIsInstance(output["data"]["expected_machines"], list)
0124         self.assertIsInstance(output["data"]["results"], list)
0125
0126
0127 # Run tests
0128 if __name__ == "__main__":
0129     unittest.main()