fix typing issue

Author: khsrali

src/aiida/engine/processes/calcjobs/manager.py

@@ -57,8 +57,8 @@ def __init__(self, authinfo: AuthInfo, transport_queue: 'TransportQueue', last_u
         self._loop = transport_queue.loop
         self._logger = logging.getLogger(__name__)
 
-        self._jobs_cache: Dict[Hashable, 'JobInfo'] = {}
-        self._job_update_requests: Dict[Hashable, asyncio.Future] = {}  # Mapping: {job_id: Future}
+        self._jobs_cache: Dict[str, 'JobInfo'] = {}
+        self._job_update_requests: Dict[str, asyncio.Future] = {}  # Mapping: {job_id: Future}
         self._last_updated = last_updated
         self._update_handle: Optional[asyncio.TimerHandle] = None
         self._polling_jobs: List[str] = []
@@ -88,7 +88,7 @@ def last_updated(self) -> Optional[float]:
         """
         return self._last_updated
 
-    async def _get_jobs_from_scheduler(self) -> Dict[Hashable, 'JobInfo']:
+    async def _get_jobs_from_scheduler(self) -> Dict[str, 'JobInfo']:
         """Get the current jobs list from the scheduler.
 
         :return: a mapping of job ids to :py:class:`~aiida.schedulers.datastructures.JobInfo` instances
@@ -168,7 +168,7 @@ def request_job_info_update(self, authinfo: AuthInfo, job_id: Hashable) -> Itera
         """
         self._authinfo = authinfo
         # Get or create the future
-        request = self._job_update_requests.setdefault(job_id, asyncio.Future())
+        request = self._job_update_requests.setdefault(str(job_id), asyncio.Future())
         assert not request.done(), 'Expected pending job info future, found in done state.'
 
         try:

tests/engine/test_manager.py

@@ -76,42 +76,59 @@ def test_prevent_racing_condition(self):
         """Test that the `JobsList` prevents racing condition when updating job info.
 
         This test simulates a race condition where:
-        1. Job 'job1' requests an update
-        2. During the scheduler query, a new job 'job2' also requests an update
-        3. JobList must only update about 'job1'
-        4. 'job2' future should be kept pending for the next update cycle
+        1. Job job_id_a requests an update
+        2. During the scheduler query, a new job job_id_b also requests an update
+        3. JobList must only update about job_id_a
+        4. job_id_b future should be kept pending for the next update cycle
         """
         from unittest.mock import patch
 
         from aiida.schedulers.datastructures import JobInfo, JobState
 
         jobs_list = self.jobs_list
 
-        mock_job_info = JobInfo()
-        mock_job_info.job_id = 'job1'
-        mock_job_info.job_state = JobState.RUNNING
+        mock_job_info_a = JobInfo()
+        job_id_a = 'A'
+        mock_job_info_a.job_id = job_id_a
+        mock_job_info_a.job_state = JobState.RUNNING
+
+        mock_job_info_b = JobInfo()
+        job_id_b = 10  # intentionally using int to test str conversion
+        mock_job_info_b.job_id = job_id_b
 
         def mock_get_jobs(**kwargs):
             # Simulate the race: job2 is added to _job_update_requests while we're querying the scheduler
-            jobs_list._job_update_requests.setdefault('job2', asyncio.Future())
+            jobs_list._job_update_requests.setdefault(str(job_id_b), asyncio.Future())
 
-            # Return only job1 (scheduler was queried with only job1)
-            return {'job1': mock_job_info}
+            # Return only job_id_a (scheduler was queried with only job_id_a)
+            return {job_id_a: mock_job_info_a}
 
-        # Request update for job1
-        future1 = jobs_list._job_update_requests.setdefault('job1', asyncio.Future())
+        # Request update for job_id_a
+        future1 = jobs_list._job_update_requests.setdefault(str(job_id_a), asyncio.Future())
 
         # Patch the scheduler's get_jobs
         scheduler = self.auth_info.computer.get_scheduler()
         with patch.object(scheduler.__class__, 'get_jobs', side_effect=mock_get_jobs):
             self.loop.run_until_complete(jobs_list._update_job_info())
 
-        # Verify job1 was resolved correctly
-        assert future1.done(), 'job1 future should be resolved'
-        assert future1.result() == mock_job_info, 'job1 should have the correct JobInfo'
+        # Verify job_id_a was resolved correctly
+        assert future1.done(), 'job_id_a future should be resolved'
+        assert future1.result() == mock_job_info_a, 'job_id_a should have the correct JobInfo'
 
         # Verify job2 was NOT resolved and it has remained in _job_update_requests for the next cycle
-        assert 'job2' in jobs_list._job_update_requests, 'job2 should still be in update requests'
-        future2 = jobs_list._job_update_requests['job2']
-        assert not future2.done(), 'job2 future should NOT be resolved yet (prevented racing bug)'
-        assert len(jobs_list._job_update_requests) == 1, 'Only job2 should remain in update requests'
+        assert str(job_id_b) in jobs_list._job_update_requests, 'job_id_b should still be in update requests'
+        future2 = jobs_list._job_update_requests[str(job_id_b)]
+        assert not future2.done(), 'job_id_b future should NOT be resolved yet (prevented racing bug)'
+        assert len(jobs_list._job_update_requests) == 1, 'Only job_id_b should remain in update requests'
+
+        def mock_get_jobs(**kwargs):
+            # Intentionally return empty dict to simulate job_id_b
+            # not being in the scheduler anymore. Simulated as finished.
+            return {}
+
+        future2 = jobs_list._job_update_requests.get(str(job_id_b))
+
+        with patch.object(scheduler.__class__, 'get_jobs', side_effect=mock_get_jobs):
+            self.loop.run_until_complete(jobs_list._update_job_info())
+
+        assert future2.done(), 'job_id_b future should be resolved'