add additional tests for interpolated string and positive int validation

Author: pnilan

airbyte_cdk/sources/declarative/async_job/job_tracker.py

@@ -17,9 +17,9 @@ class ConcurrentJobLimitReached(Exception):
 
 
 class JobTracker:
-    def __init__(self, limit: Union[int, InterpolatedString], config: Mapping[str, Any]):
-        if isinstance(limit, InterpolatedString):
-            limit = int(limit.eval(config=config, json_loads=json.loads))
+    def __init__(self, limit: Union[int, str], config: Mapping[str, Any] = {}):
+        if isinstance(limit, str):
+            limit = int(InterpolatedString(limit, parameters={}).eval(config=config))
 
         if limit < 1:
             raise ValueError(f"Invalid max concurrent jobs limit: {limit}. Minimum value is 1.")

unit_tests/sources/declarative/async_job/test_job_orchestrator.py

@@ -301,7 +301,7 @@ def test_given_exception_when_start_job_and_skip_this_exception(
     def test_given_jobs_failed_more_than_max_attempts_when_create_and_get_completed_partitions_then_free_job_budget(
         self, mock_sleep: MagicMock
     ) -> None:
-        job_tracker = JobTracker(1)
+        job_tracker = JobTracker(1, config={})
         jobs = [self._an_async_job(str(i), _A_STREAM_SLICE) for i in range(_MAX_NUMBER_OF_ATTEMPTS)]
         self._job_repository.start.side_effect = jobs
         self._job_repository.update_jobs_status.side_effect = _status_update_per_jobs(

unit_tests/sources/declarative/async_job/test_job_tracker.py

@@ -15,7 +15,9 @@
 
 class JobTrackerTest(TestCase):
     def setUp(self) -> None:
-        self._tracker = JobTracker(_LIMIT, config={})
+        self._tracker = JobTracker(
+            limit="{{config['max_concurrent_jobs']}}", config={"max_concurrent_jobs": _LIMIT}
+        )
 
     def test_given_limit_reached_when_remove_job_then_can_get_intent_again(self) -> None:
         intents = self._reach_limit()
@@ -39,3 +41,15 @@ def test_given_limit_reached_when_add_job_then_limit_is_still_reached(self) -> N
 
     def _reach_limit(self) -> List[str]:
         return [self._tracker.try_to_get_intent() for i in range(_LIMIT)]
+
+
+def test_given_limit_is_interpolated_string_when_init_then_limit_is_int():
+    tracker = JobTracker(
+        limit="{{config['max_concurrent_jobs']}}", config={"max_concurrent_jobs": _LIMIT}
+    )
+    assert tracker._limit == _LIMIT
+
+
+def test_given_limit_is_less_than_1_when_init_then_raise_value_error():
+    with pytest.raises(ValueError):
+        JobTracker(limit="-1", config={})