taskbuffer/tests/container_image_dispatch_tests.py

0001 # Description: Unit tests for dispatching to a worker bound to one container image. No database is used
0002 import unittest
0003 from typing import Any
0004
0005 from pandaserver.taskbuffer.db_proxy_mods.job_complex_module import JobComplexModule
0006
0007 IMAGE = "/cvmfs/singularity.opensciencegrid.org/eicweb/eic_xl:26.07.3-stable"
0008
0009
0010 def where_clause(**kwargs: Any) -> tuple[str, dict[str, Any]]:
0011     params: dict[str, Any] = dict(
0012         site_name="SITE",
0013         mem=0,
0014         disk_space=0,
0015         background=False,
0016         resource_type=None,
0017         prod_source_label="managed",
0018         computing_element=None,
0019         is_gu=False,
0020         job_type=None,
0021         prod_user_id=None,
0022         task_id=None,
0023         average_memory_limit=None,
0024         remaining_time=None,
0025     )
0026     params.update(kwargs)
0027     # the method reads nothing from the instance, so one made without
0028     # __init__ (which opens no database connection) is enough
0029     return JobComplexModule.__new__(JobComplexModule).construct_where_clause(**params)
0030
0031
0032 class TestContainerImageWhereClause(unittest.TestCase):
0033     def test_bound_worker_gets_only_its_image(self) -> None:
0034         sql, var_map = where_clause(container_name=IMAGE)
0035         self.assertIn("AND container_name=:containerName ", sql)
0036         self.assertEqual(var_map[":containerName"], IMAGE)
0037
0038     def test_unbound_worker_is_unchanged(self) -> None:
0039         for value in (None, ""):
0040             sql, var_map = where_clause(container_name=value)
0041             self.assertNotIn("container_name", sql)
0042             self.assertNotIn(":containerName", var_map)
0043         self.assertEqual(where_clause(container_name=None), where_clause())
0044
0045     def test_combines_with_the_other_filters(self) -> None:
0046         sql, var_map = where_clause(container_name=IMAGE, task_id=123, excluded_task_ids=[7])
0047         self.assertIn("AND jediTaskID=:taskID ", sql)
0048         self.assertIn("AND container_name=:containerName ", sql)
0049         self.assertIn("jediTaskID NOT IN (", sql)
0050         self.assertEqual(var_map[":taskID"], 123)
0051
0052
0053 if __name__ == "__main__":
0054     unittest.main()