v1/tests/task_api_tests.py

0001 # Description: Unit tests for the Task 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 # to run the tests with a real JEDI Task ID by setting the environment variable
0009 JEDI_TASK_ID = os.environ.get("JEDI_TASK_ID_TEST", -1)
0010
0011
0012 class TestTaskAPI(unittest.TestCase):
0013     def setUp(self):
0014         self.http_client = HttpClient()
0015
0016     def test_retry(self):
0017         url = f"{api_url_ssl}/task/retry"
0018         print(f"Testing URL: {url}")
0019         data = {"task_id": JEDI_TASK_ID}
0020         status, output = self.http_client.post(url, data)
0021         print(output)
0022         output["status"] = status
0023         output["message"] = ""
0024
0025         if JEDI_TASK_ID == -1:
0026             # Fake task should not be found
0027             expected_response = {"status": 0, "success": False, "data": 2, "message": ""}
0028         else:
0029             # Real task
0030             expected_response = {"status": 0, "success": True, "data": 0, "message": ""}
0031         self.assertEqual(output, expected_response)
0032
0033     def test_enable_jumbo_jobs(self):
0034         url = f"{api_url_ssl}/task/enable_jumbo_jobs"
0035         print(f"Testing URL: {url}")
0036         data = {"task_id": JEDI_TASK_ID, "jumbo_jobs_total": 1}
0037         status, output = self.http_client.post(url, data)
0038         print(output)
0039         output["status"] = status
0040         del output["data"]
0041         del output["message"]
0042
0043         if JEDI_TASK_ID == -1:
0044             # Fake task should not be found
0045             expected_response = {"status": 0, "success": False}
0046         else:
0047             # Real task
0048             expected_response = {"status": 0, "success": True}
0049         self.assertEqual(output, expected_response)
0050
0051     def test_enable_job_cloning(self):
0052         url = f"{api_url_ssl}/task/enable_job_cloning"
0053         print(f"Testing URL: {url}")
0054         data = {"task_id": JEDI_TASK_ID}
0055         status, output = self.http_client.post(url, data)
0056         print(output)
0057         output["status"] = status
0058         del output["data"]
0059         del output["message"]
0060
0061         if JEDI_TASK_ID == -1:
0062             # Fake task should not be found
0063             expected_response = {"status": 0, "success": False}
0064         else:
0065             # Real task
0066             expected_response = {"status": 0, "success": True}
0067         self.assertEqual(output, expected_response)
0068
0069     def test_disable_job_cloning(self):
0070         url = f"{api_url_ssl}/task/disable_job_cloning"
0071         print(f"Testing URL: {url}")
0072         data = {"task_id": JEDI_TASK_ID}
0073         status, output = self.http_client.post(url, data)
0074         print(output)
0075         output["status"] = status
0076         del output["data"]
0077         del output["message"]
0078
0079         if JEDI_TASK_ID == -1:
0080             # Fake task should not be found
0081             expected_response = {"status": 0, "success": False}
0082         else:
0083             # Real task
0084             expected_response = {"status": 0, "success": True}
0085         self.assertEqual(output, expected_response)
0086
0087     def test_increase_attempts(self):
0088         url = f"{api_url_ssl}/task/increase_attempts"
0089         print(f"Testing URL: {url}")
0090         data = {"task_id": JEDI_TASK_ID}
0091         status, output = self.http_client.post(url, data)
0092         print(output)
0093         output["status"] = status
0094         del output["data"]
0095         del output["message"]
0096
0097         if JEDI_TASK_ID == -1:
0098             # Fake task should not be found
0099             expected_response = {"status": 0, "success": False}
0100         else:
0101             # Real task
0102             expected_response = {"status": 0, "success": True}
0103         self.assertEqual(output, expected_response)
0104
0105     def test_resume(self):
0106         url = f"{api_url_ssl}/task/resume"
0107         print(f"Testing URL: {url}")
0108         data = {"task_id": JEDI_TASK_ID}
0109         status, output = self.http_client.post(url, data)
0110         print(output)
0111         output["status"] = status
0112         del output["data"]
0113         del output["message"]
0114
0115         if JEDI_TASK_ID == -1:
0116             # Fake task should not be found
0117             expected_response = {"status": 0, "success": False}
0118         else:
0119             # Real task
0120             expected_response = {"status": 0, "success": True}
0121         self.assertEqual(output, expected_response)
0122
0123     def test_release(self):
0124         url = f"{api_url_ssl}/task/release"
0125         print(f"Testing URL: {url}")
0126         data = {"task_id": JEDI_TASK_ID}
0127         status, output = self.http_client.post(url, data)
0128         print(output)
0129         output["status"] = status
0130         del output["data"]
0131         del output["message"]
0132
0133         if JEDI_TASK_ID == -1:
0134             # Fake task should not be found
0135             expected_response = {"status": 0, "success": False}
0136         else:
0137             # Real task
0138             expected_response = {"status": 0, "success": True}
0139         self.assertEqual(output, expected_response)
0140
0141     def test_reassign(self):
0142         # reassign(req: PandaRequest, task_id: int, site: str = None, cloud: str = None, nucleus: str = None, soft: bool = None, mode: str = None):
0143         url = f"{api_url_ssl}/task/reassign"
0144         print(f"Testing URL: {url}")
0145         data = {"task_id": JEDI_TASK_ID, "site": "CERN"}
0146         status, output = self.http_client.post(url, data)
0147         print(output)
0148         output["status"] = status
0149         del output["data"]
0150         del output["message"]
0151
0152         if JEDI_TASK_ID == -1:
0153             # Fake task should not be found
0154             expected_response = {"status": 0, "success": False}
0155         else:
0156             # Real task
0157             expected_response = {"status": 0, "success": True}
0158         self.assertEqual(output, expected_response)
0159
0160     def test_pause(self):
0161         url = f"{api_url_ssl}/task/pause"
0162         print(f"Testing URL: {url}")
0163         data = {"task_id": JEDI_TASK_ID}
0164         status, output = self.http_client.post(url, data)
0165         print(output)
0166         output["status"] = status
0167         del output["data"]
0168         del output["message"]
0169
0170         if JEDI_TASK_ID == -1:
0171             # Fake task should not be found
0172             expected_response = {"status": 0, "success": False}
0173         else:
0174             # Real task
0175             expected_response = {"status": 0, "success": True}
0176         self.assertEqual(output, expected_response)
0177
0178     def test_kill(self):
0179         url = f"{api_url_ssl}/task/kill"
0180         print(f"Testing URL: {url}")
0181         data = {"task_id": JEDI_TASK_ID}
0182         status, output = self.http_client.post(url, data)
0183         print(output)
0184         output["status"] = status
0185         del output["data"]
0186         del output["message"]
0187
0188         if JEDI_TASK_ID == -1:
0189             # Fake task should not be found
0190             expected_response = {"status": 0, "success": False}
0191         else:
0192             # Real task
0193             expected_response = {"status": 0, "success": True}
0194         self.assertEqual(output, expected_response)
0195
0196     def test_finish(self):
0197         # def finish(req: PandaRequest, task_id: int, soft: bool = False, broadcast: bool = False) -> Dict[str, Any]:
0198         url = f"{api_url_ssl}/task/finish"
0199         print(f"Testing URL: {url}")
0200         data = {"task_id": JEDI_TASK_ID, "soft": False, "broadcast": False}
0201         status, output = self.http_client.post(url, data)
0202         print(output)
0203         output["status"] = status
0204         del output["data"]
0205         del output["message"]
0206
0207         if JEDI_TASK_ID == -1:
0208             # Fake task should not be found
0209             expected_response = {"status": 0, "success": False}
0210         else:
0211             # Real task
0212             expected_response = {"status": 0, "success": True}
0213         self.assertEqual(output, expected_response)
0214
0215     def test_reactivate(self):
0216         # def reactivate(req: PandaRequest, task_id: int, keep_attempt_nr: bool = False, trigger_job_generation: bool = False) -> Dict[str, Any]:
0217         url = f"{api_url_ssl}/task/reactivate"
0218         print(f"Testing URL: {url}")
0219         data = {"task_id": JEDI_TASK_ID}
0220         status, output = self.http_client.post(url, data)
0221         print(output)
0222         output["status"] = status
0223         del output["data"]
0224         del output["message"]
0225
0226         # This method does not distinguish between real and fake tasks
0227         expected_response = {"status": 0, "success": True}
0228         self.assertEqual(output, expected_response)
0229
0230     def test_avalanche(self):
0231         url = f"{api_url_ssl}/task/avalanche"
0232         print(f"Testing URL: {url}")
0233         data = {"task_id": JEDI_TASK_ID}
0234         status, output = self.http_client.post(url, data)
0235         print(output)
0236         output["status"] = status
0237         del output["data"]
0238         del output["message"]
0239
0240         if JEDI_TASK_ID == -1:
0241             # Fake task should not be found
0242             expected_response = {"status": 0, "success": False}
0243         else:
0244             # Real task
0245             expected_response = {"status": 0, "success": True}
0246         self.assertEqual(output, expected_response)
0247
0248     def test_reload_input(self):
0249         url = f"{api_url_ssl}/task/reload_input"
0250         print(f"Testing URL: {url}")
0251         data = {"task_id": JEDI_TASK_ID}
0252         status, output = self.http_client.post(url, data)
0253         print(output)
0254         output["status"] = status
0255         del output["data"]
0256         del output["message"]
0257
0258         if JEDI_TASK_ID == -1:
0259             # Fake task should not be found
0260             expected_response = {"status": 0, "success": False}
0261         else:
0262             # Real task
0263             expected_response = {"status": 0, "success": True}
0264         self.assertEqual(output, expected_response)
0265
0266     def test_reassign_global_share(self):
0267         url = f"{api_url_ssl}/task/reassign_global_share"
0268         print(f"Testing URL: {url}")
0269         data = {"task_id_list": [JEDI_TASK_ID], "share": "Test", "reassign_running_jobs": False}
0270         status, output = self.http_client.post(url, data)
0271         print(output)
0272         output["status"] = status
0273         del output["data"]
0274         del output["message"]
0275
0276         if JEDI_TASK_ID == -1:
0277             # Fake task should not be found
0278             expected_response = {"status": 0, "success": False}
0279         else:
0280             # Real task
0281             expected_response = {"status": 0, "success": True}
0282         self.assertEqual(output, expected_response)
0283
0284     def test_get_status(self):
0285         url = f"{api_url_ssl}/task/get_status"
0286         print(f"Testing URL: {url}")
0287         data = {"task_id": JEDI_TASK_ID}
0288         status, output = self.http_client.get(url, data)
0289         print(output)
0290         output["status"] = status
0291         del output["data"]
0292         del output["message"]
0293
0294         if JEDI_TASK_ID == -1:
0295             # Fake task should not be found
0296             expected_response = {"status": 0, "success": False}
0297         else:
0298             # Real task
0299             expected_response = {"status": 0, "success": True}
0300         self.assertEqual(output, expected_response)
0301
0302     def test_get_details(self):
0303         url = f"{api_url_ssl}/task/get_details"
0304         print(f"Testing URL: {url}")
0305         data = {"task_id": JEDI_TASK_ID, "include_parameters": False, "include_status": False}
0306         status, output = self.http_client.get(url, data)
0307         print(output)
0308         output["status"] = status
0309         del output["data"]
0310         del output["message"]
0311
0312         if JEDI_TASK_ID == -1:
0313             # Fake task should not be found
0314             expected_response = {"status": 0, "success": False}
0315         else:
0316             # Real task
0317             expected_response = {"status": 0, "success": True}
0318         self.assertEqual(output, expected_response)
0319
0320     def test_change_attribute(self):
0321         url = f"{api_url_ssl}/task/change_attribute"
0322         print(f"Testing URL: {url}")
0323         data = {"task_id": JEDI_TASK_ID, "attribute_name": "coreCount", "value": 8}
0324         status, output = self.http_client.post(url, data)
0325         print(output)
0326         output["status"] = status
0327         del output["data"]
0328         del output["message"]
0329
0330         if JEDI_TASK_ID == -1:
0331             # Fake task should not be found
0332             expected_response = {"status": 0, "success": False}
0333         else:
0334             # Real task
0335             expected_response = {"status": 0, "success": True}
0336         self.assertEqual(output, expected_response)
0337
0338     def test_change_modification_time(self):
0339         url = f"{api_url_ssl}/task/change_modification_time"
0340         print(f"Testing URL: {url}")
0341         data = {"task_id": JEDI_TASK_ID, "positive_hour_offset": 1}
0342         status, output = self.http_client.post(url, data)
0343         print(output)
0344         output["status"] = status
0345         del output["data"]
0346         del output["message"]
0347
0348         if JEDI_TASK_ID == -1:
0349             # Fake task should not be found
0350             expected_response = {"status": 0, "success": False}
0351         else:
0352             # Real task
0353             expected_response = {"status": 0, "success": True}
0354         self.assertEqual(output, expected_response)
0355
0356     def test_change_priority(self):
0357         url = f"{api_url_ssl}/task/change_priority"
0358         print(f"Testing URL: {url}")
0359         data = {"task_id": JEDI_TASK_ID, "priority": 1}
0360         status, output = self.http_client.post(url, data)
0361         print(output)
0362         output["status"] = status
0363         del output["data"]
0364         del output["message"]
0365
0366         if JEDI_TASK_ID == -1:
0367             # Fake task should not be found
0368             expected_response = {"status": 0, "success": False}
0369         else:
0370             # Real task
0371             expected_response = {"status": 0, "success": True}
0372         self.assertEqual(output, expected_response)
0373
0374     def test_change_split_rule(self):
0375         url = f"{api_url_ssl}/task/change_split_rule"
0376         print(f"Testing URL: {url}")
0377         data = {"task_id": JEDI_TASK_ID, "attribute_name": "AI", "value": 1}
0378         status, output = self.http_client.post(url, data)
0379         print(output)
0380         output["status"] = status
0381         del output["data"]
0382         del output["message"]
0383
0384         if JEDI_TASK_ID == -1:
0385             # Fake task should not be found
0386             expected_response = {"status": 0, "success": False}
0387         else:
0388             # Real task
0389             expected_response = {"status": 0, "success": True}
0390         self.assertEqual(output, expected_response)
0391
0392     def test_get_tasks_modified_since(self):
0393         # def get_tasks_modified_since(req, since: str, dn: str = "", full: bool = False, min_task_id: int = None, prod_source_label: str = "user") -> Dict[str, Any]:
0394         url = f"{api_url_ssl}/task/get_tasks_modified_since"
0395         print(f"Testing URL: {url}")
0396
0397         # generate the date one week ago
0398         one_week_ago = datetime.now() - timedelta(weeks=1)
0399         one_week_ago_str = one_week_ago.strftime("%Y-%m-%d %H:%M:%S")
0400
0401         data = {"since": one_week_ago_str, "dn": "test", "full": False, "min_task_id": 1, "prod_source_label": "user"}
0402         status, output = self.http_client.get(url, data)
0403         print(output)
0404         output["status"] = status
0405         del output["data"]
0406         del output["message"]
0407
0408         expected_response = {"status": 0, "success": True}
0409         self.assertEqual(output, expected_response)
0410
0411     def test_get_datasets_and_files(self):
0412         # def get_datasets_and_files(req, task_id, dataset_types: List = ("input", "pseudo_input")) -> Dict[str, Any]:
0413         url = f"{api_url_ssl}/task/get_datasets_and_files"
0414         print(f"Testing URL: {url}")
0415         data = {"task_id": JEDI_TASK_ID, "dataset_types": ["input", "inexisting_type"]}
0416         status, output = self.http_client.get(url, data)
0417         print(output)
0418         output["status"] = status
0419         del output["data"]
0420         del output["message"]
0421
0422         if JEDI_TASK_ID == -1:
0423             # Fake task should not be found
0424             expected_response = {"status": 0, "success": False}
0425         else:
0426             # Real task
0427             expected_response = {"status": 0, "success": True}
0428         self.assertEqual(output, expected_response)
0429
0430     def test_get_job_ids(self):
0431         # def get_job_ids(req: PandaRequest, task_id: int) -> Dict[str, Any]:
0432         url = f"{api_url_ssl}/task/get_job_ids"
0433         print(f"Testing URL: {url}")
0434         data = {"task_id": [JEDI_TASK_ID]}
0435         status, output = self.http_client.get(url, data)
0436         print(output)
0437         output["status"] = status
0438         del output["data"]
0439         del output["message"]
0440
0441         # This method does not distinguish between real and fake tasks
0442         expected_response = {"status": 0, "success": True}
0443         self.assertEqual(output, expected_response)
0444
0445     def test_insert_task_parameters(self):
0446         # def insert_task_parameters(req: PandaRequest, task_parameters: Dict, parent_tid: int = None) -> Dict[str, Any]:
0447         url = f"{api_url_ssl}/task/insert_task_parameters"
0448         print(f"Testing URL: {url}")
0449         data = {"task_parameters": {"test": 1, "taskName": "test_task"}, "parent_tid": 1}
0450         status, output = self.http_client.post(url, data)
0451         print(output)
0452         output["status"] = status
0453         del output["data"]
0454         del output["message"]
0455
0456         expected_response = {"status": 0, "success": True}
0457         self.assertEqual(output, expected_response)
0458
0459     def test_get_task_parameters(self):
0460         # def get_task_parameters(req: PandaRequest, task_id: int) -> Dict[str, Any]:
0461         url = f"{api_url_ssl}/task/get_task_parameters"
0462         print(f"Testing URL: {url}")
0463         data = {"task_id": JEDI_TASK_ID}
0464         status, output = self.http_client.get(url, data)
0465         print(output)
0466         output["status"] = status
0467         del output["data"]
0468         del output["message"]
0469
0470         if JEDI_TASK_ID == -1:
0471             # Fake task should not be found
0472             expected_response = {"status": 0, "success": False}
0473         else:
0474             # Real task
0475             expected_response = {"status": 0, "success": True}
0476         self.assertEqual(output, expected_response)
0477
0478     def test_get_detailed_info(self):
0479         # def get_detailed_info(req: PandaRequest, task_id: int) -> Dict[str, Any]:
0480         url = f"{api_url_ssl}/task/get_detailed_info"
0481         print(f"Testing URL: {url}")
0482         data = {"task_id": JEDI_TASK_ID}
0483         status, output = self.http_client.get(url, data)
0484         print(output)
0485         output["status"] = status
0486
0487         if JEDI_TASK_ID == -1:
0488             # Fake task should not be found
0489             del output["data"]
0490             del output["message"]
0491             expected_response = {"status": 0, "success": False}
0492         else:
0493             # Real task - verify the returned dict contains expected keys
0494             task_info = output.get("data", {})
0495             self.assertIn("jediTaskID", task_info)
0496             self.assertIn("status", task_info)
0497             self.assertIn("jobParamsTemplate", task_info)
0498             self.assertIn("taskParams", task_info)
0499             self.assertEqual(task_info["jediTaskID"], JEDI_TASK_ID)
0500             del output["data"]
0501             del output["message"]
0502             expected_response = {"status": 0, "success": True}
0503         self.assertEqual(output, expected_response)
0504
0505     def test_get_parent_detailed_info(self):
0506         # def get_parent_detailed_info(req: PandaRequest, task_id: int) -> Dict[str, Any]:
0507         url = f"{api_url_ssl}/task/get_parent_detailed_info"
0508         print(f"Testing URL: {url}")
0509         data = {"task_id": JEDI_TASK_ID}
0510         status, output = self.http_client.get(url, data)
0511         print(output)
0512         output["status"] = status
0513
0514         if JEDI_TASK_ID == -1:
0515             # Fake task should not be found
0516             del output["data"]
0517             del output["message"]
0518             expected_response = {"status": 0, "success": False}
0519             self.assertEqual(output, expected_response)
0520             return
0521
0522         # For real tasks, parent may or may not exist. Validate shape and consistency.
0523         if output.get("success"):
0524             task_info = output.get("data", {})
0525             self.assertIn("jediTaskID", task_info)
0526             self.assertIn("status", task_info)
0527         else:
0528             self.assertIn(
0529                 output.get("message"),
0530                 {
0531                     "Child task not found",
0532                     "No parent task found",
0533                     "Parent task not found",
0534                     "Error resolving parent task",
0535                     "Parent task details not found or retrieval error",
0536                 },
0537             )
0538
0539         del output["data"]
0540         del output["message"]
0541         expected_response = {"status": 0, "success": output["success"]}
0542         self.assertEqual(output, expected_response)
0543
0544
0545     def test_get_scout_job_descriptions(self):
0546         # def get_scout_job_descriptions(req: PandaRequest, task_id: int) -> Dict:
0547         url = f"{api_url_ssl}/task/get_scout_job_descriptions"
0548         print(f"Testing URL: {url}")
0549         data = {"task_id": JEDI_TASK_ID}
0550         status, output = self.http_client.get(url, data)
0551         print(output)
0552         output["status"] = status
0553
0554         if JEDI_TASK_ID == -1:
0555             # Fake task has no scout jobs, returns empty list
0556             output["message"] = ""
0557             expected_response = {"status": 0, "success": True, "data": [], "message": ""}
0558             self.assertEqual(output, expected_response)
0559         else:
0560             # Real task: success with a list of scout job descriptions
0561             self.assertEqual(status, 0)
0562             self.assertTrue(output["success"])
0563             self.assertIsInstance(output["data"], list)
0564
0565     def test_get_job_descriptions(self):
0566         # def get_job_descriptions(req: PandaRequest, task_id: int, unsuccessful_only: bool = False) -> Dict:
0567         url = f"{api_url_ssl}/task/get_job_descriptions"
0568         print(f"Testing URL: {url}")
0569
0570         for unsuccessful_only in [False, True]:
0571             data = {"task_id": JEDI_TASK_ID, "unsuccessful_only": unsuccessful_only}
0572             status, output = self.http_client.get(url, data)
0573             print(output)
0574             output["status"] = status
0575
0576             if JEDI_TASK_ID == -1:
0577                 # Fake task has no jobs, returns empty list
0578                 output["message"] = ""
0579                 expected_response = {"status": 0, "success": True, "data": [], "message": ""}
0580                 self.assertEqual(output, expected_response)
0581             else:
0582                 # Real task: success with a list of job descriptions
0583                 self.assertEqual(status, 0)
0584                 self.assertTrue(output["success"])
0585                 self.assertIsInstance(output["data"], list)
0586
0587
0588 # Run tests
0589 if __name__ == "__main__":
0590     unittest.main()