v1/tests/pilot_api_tests.py

0001 # Description: Unit tests for the Pilot API methods
0002 import os
0003 import unittest
0004 from datetime import datetime, timedelta
0005
0006 from pandaserver.api.v1.http_client import HttpClient, api_url_ssl
0007
0008
0009 class TestPilotAPI(unittest.TestCase):
0010     def setUp(self):
0011         self.http_client = HttpClient()
0012
0013     def test_acquire_jobs(self):
0014         url = f"{api_url_ssl}/pilot/acquire_jobs"
0015         print(f"Testing URL: {url}")
0016         data = {
0017             "site_name": "CERN",
0018             "timeout": 60,
0019             "memory": 999999999,
0020             "disk_space": 999999999,
0021             "prod_source_label": "managed",
0022             "node": "aipanda120.cern.ch",
0023             "computing_element": "CERN",
0024             "prod_user_id": None,
0025             "get_proxy_key": None,
0026             "task_id": None,
0027             "n_jobs": 1,
0028             "background": False,
0029             "resource_type": "SCORE",
0030             "harvester_id": "imaginary_harvester",
0031             "worker_id": 12345,
0032             "scheduler_id": "imaginary_scheduler",
0033             "job_type": "user",
0034             "via_topic": False,
0035         }
0036
0037         status, output = self.http_client.post(url, data)
0038         print(output)
0039         output["status"] = status
0040
0041         expected_response = {"status": 0, "success": False, "data": 2, "message": ""}
0042         self.assertEqual(output, expected_response)
0043
0044     def test_acquire_jobs(self):
0045         url = f"{api_url_ssl}/pilot/acquire_jobs"
0046         print(f"Testing URL: {url}")
0047         data = {
0048             "site_name": "CERN",
0049             "timeout": 60,
0050             "memory": 999999999,
0051             "disk_space": 999999999,
0052             "prod_source_label": "managed",
0053             "node": "aipanda120.cern.ch",
0054             "computing_element": "CERN",
0055             "prod_user_id": None,
0056             "get_proxy_key": None,
0057             "task_id": None,
0058             "n_jobs": 1,
0059             "background": False,
0060             "resource_type": "SCORE",
0061             "harvester_id": "imaginary_harvester",
0062             "worker_id": 12345,
0063             "scheduler_id": "imaginary_scheduler",
0064             "job_type": "user",
0065             "via_topic": False,
0066         }
0067
0068         status, output = self.http_client.post(url, data)
0069         print(output)
0070         output["status"] = status
0071
0072         expected_response = {"status": 0, "success": False, "data": 2, "message": ""}
0073         self.assertEqual(output, expected_response)
0074
0075     def test_get_job_status(self):
0076         url = f"{api_url_ssl}/pilot/get_job_status"
0077         print(f"Testing URL: {url}")
0078         data = {
0079             "job_ids": "4674379299",
0080             "timeout": 60,
0081         }
0082
0083         status, output = self.http_client.get(url, data)
0084         print(output)
0085         output["status"] = status
0086
0087         expected_response = {"status": 0, "success": False, "data": 2, "message": ""}
0088         self.assertEqual(output, expected_response)
0089
0090     def test_update_job(self):
0091         url = f"{api_url_ssl}/pilot/update_job"
0092         print(f"Testing URL: {url}")
0093         data = {"job_id": 4674379299, "job_status": "starting"}
0094
0095         status, output = self.http_client.post(url, data)
0096         print(output)
0097         output["status"] = status
0098
0099         expected_response = {"status": 0, "success": False, "data": 2, "message": ""}
0100         self.assertEqual(output, expected_response)
0101
0102     def test_update_worker_node(self):
0103         url = f"{api_url_ssl}/pilot/update_worker_node"
0104         print(f"Testing URL: {url}")
0105         data = {
0106             "site": "CERN",
0107             "panda_queue": "CERN",
0108             "host_name": "slot1@wn1.cern.ch",
0109             "cpu_model": "AMD EPYC 7B12",
0110             "n_logical_cpus": 64,
0111             "n_sockets": 2,
0112             "cores_per_socket": 8,
0113             "threads_per_core": 2,
0114             "cpu_architecture": "x86_64",
0115             "cpu_architecture_level": "x86_64-v3",
0116             "clock_speed": 2.7,
0117             "total_memory": 3350,
0118             "total_local_disk": 75,
0119         }
0120
0121         status, output = self.http_client.post(url, data)
0122         print(output)
0123         output["status"] = status
0124
0125         expected_response = {"status": 0, "success": True, "data": None, "message": "Inserted new worker node."}
0126         self.assertEqual(output, expected_response)
0127
0128     def test_update_worker_node_gpu(self):
0129         url = f"{api_url_ssl}/pilot/update_worker_node_gpu"
0130         print(f"Testing URL: {url}")
0131         data = {
0132             "site": "CERN",
0133             "host_name": "slot1@lxplus940.cern.ch",
0134             "vendor": "NVIDIA",
0135             "model": "Tesla T4",
0136             "architecture": "Turing",
0137             "vram": 15360,
0138             "framework": "CUDA",
0139             "framework_version": "12.9",
0140             "driver_version": "575.51.03",
0141             "count": 1,
0142         }
0143
0144         status, output = self.http_client.post(url, data)
0145         print(output)
0146         output["status"] = status
0147
0148         expected_response = {"status": 0, "success": True, "data": None, "message": "Inserted new worker node GPU."}
0149         self.assertEqual(output, expected_response)
0150
0151
0152 # Run tests
0153 if __name__ == "__main__":
0154     unittest.main()