v1/tests/credential_management_api_tests.py

0001 # TODO: the result of the tests depend on the cert/token used for the call. These particular results are for running with user pandasv1
0002
0003 import unittest
0004 from datetime import datetime, timezone
0005
0006 from pandaserver.api.v1.http_client import HttpClient, api_url, api_url_ssl
0007
0008 # Get current UTC time with microseconds. The format needs to be compatible with the one used in the database
0009 now_utc = datetime.now(timezone.utc)
0010 formatted_time = now_utc.strftime("%d.%m.%y %H:%M:%S") + f".{now_utc.microsecond // 1000:02d}"
0011
0012 KEY = "test_key"
0013 VALUE = "test_value"
0014
0015 NO_SSL_RESPONSE = {"success": False, "message": "SSL secure connection is required", "data": None}
0016
0017
0018 class TestSecretManagementAPI(unittest.TestCase):
0019     def setUp(self):
0020         self.http_client = HttpClient()
0021         self.urls = [api_url, api_url_ssl]  # Run tests with both base URLs
0022
0023     def test_set_user_secrets(self):
0024         for url in self.urls:
0025             with self.subTest(base_url=url):
0026                 full_url = f"{url}/creds/set_user_secrets"
0027                 print(f"Testing URL: {full_url}")
0028                 data = {"key": KEY, "value": VALUE}
0029                 status, output = self.http_client.post(full_url, data)
0030                 print(output)
0031                 if url.startswith(api_url):
0032                     expected_response = NO_SSL_RESPONSE
0033                 else:
0034                     expected_response = {"success": True, "message": "OK", "data": None}
0035                 self.assertEqual(output, expected_response)
0036
0037     def test_get_user_secrets(self):
0038         for url in self.urls:
0039             with self.subTest(base_url=url):
0040                 full_url = f"{url}/creds/get_user_secrets"
0041                 print(f"Testing URL: {full_url}")
0042                 data = {"keys": KEY}
0043                 status, output = self.http_client.get(full_url, data)
0044                 print(output)
0045                 if url.startswith(api_url):
0046                     expected_response = NO_SSL_RESPONSE
0047                 else:
0048                     expected_response = {"success": True, "message": "", "data": '{"test_key": "test_value"}'}
0049                 self.assertEqual(output, expected_response)
0050
0051     def test_get_key_pair(self):
0052         for url in self.urls:
0053             with self.subTest(base_url=url):
0054                 full_url = f"{url}/creds/get_key_pair"
0055                 print(f"Testing URL: {full_url}")
0056                 data = {"public_key_name": "a", "private_key_name": "b"}
0057                 status, output = self.http_client.get(full_url, data)
0058                 print(output)
0059                 if url.startswith(api_url):
0060                     expected_response = NO_SSL_RESPONSE
0061                 else:
0062                     expected_response = {
0063                         "success": False,
0064                         "message": "Failed since 'pandasv1' not authorized with 'k' in ATLAS_PANDAMETA.USERS.GRIDPREF",
0065                         "data": None,
0066                     }
0067                 self.assertEqual(output, expected_response)
0068
0069     def test_get_proxy(self):
0070         for url in self.urls:
0071             with self.subTest(base_url=url):
0072                 full_url = f"{url}/creds/get_proxy"
0073                 print(f"Testing URL: {full_url}")
0074                 data = {"role": "atlas", "dn": "atlpilo2"}
0075                 status, output = self.http_client.get(full_url, data)
0076                 print(output)
0077                 if url.startswith(api_url):
0078                     expected_response = NO_SSL_RESPONSE
0079                 else:
0080                     expected_response = {"success": False, "message": "'proxy' not found for atlpilo2", "data": None}
0081                 self.assertEqual(output, expected_response)
0082
0083     def test_get_access_token(self):
0084         for url in self.urls:
0085             with self.subTest(base_url=url):
0086                 full_url = f"{url}/creds/get_access_token"
0087                 print(f"Testing URL: {full_url}")
0088                 data = {"client_name": "pilot_server"}
0089                 status, output = self.http_client.get(full_url, data)
0090                 print(output)
0091                 if url.startswith(api_url):
0092                     expected_response = NO_SSL_RESPONSE
0093                 else:
0094                     expected_response = {"success": False, "message": "failed since token key is invalid for pilot_server", "data": None}
0095                 self.assertEqual(output, expected_response)
0096
0097     def test_get_token_key(self):
0098         for url in self.urls:
0099             with self.subTest(base_url=url):
0100                 full_url = f"{url}/creds/get_token_key"
0101                 print(f"Testing URL: {full_url}")
0102                 data = {"client_name": "pilot_server"}
0103                 status, output = self.http_client.get(full_url, data)
0104                 print(output)
0105                 if url.startswith(api_url):
0106                     expected_response = NO_SSL_RESPONSE
0107                 else:
0108                     del output["data"]
0109                     del output["message"]
0110                     expected_response = {"success": True}
0111                 self.assertEqual(output, expected_response)
0112
0113
0114 # Run tests
0115 if __name__ == "__main__":
0116     unittest.main()