Respect enabled=False on the direct-call path

tenacity/__init__.py

@@ -514,6 +514,9 @@ def __call__(
         *args: t.Any,
         **kwargs: t.Any,
     ) -> WrappedFnReturnT:
+        if not self.enabled:
+            return fn(*args, **kwargs)
+
         self.begin()
 
         retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs)

tenacity/asyncio/__init__.py

@@ -111,10 +111,15 @@ def __init__(
     async def __call__(  # type: ignore[override]
         self, fn: WrappedFn, *args: t.Any, **kwargs: t.Any
     ) -> WrappedFnReturnT:
+        is_async = _utils.is_coroutine_callable(fn)
+        if not self.enabled:
+            if is_async:
+                return await fn(*args, **kwargs)  # type: ignore[no-any-return]
+            return fn(*args, **kwargs)  # type: ignore[return-value]
+
         self.begin()
 
         retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs)
-        is_async = _utils.is_coroutine_callable(fn)
         while True:
             do = await self.iter(retry_state=retry_state)
             if isinstance(do, DoAttempt):

tests/test_asyncio.py

@@ -205,6 +205,43 @@ async def test_enabled_false_aiter_succeeds_on_first_attempt(self) -> None:
                 call_count += 1
         assert call_count == 1
 
+    @asynctest
+    async def test_enabled_false_call_raises_original_exception(self) -> None:
+        """When enabled=False, awaiting the controller directly raises the original
+        exception, not a RetryError, and the coroutine executes exactly once."""
+        call_count = 0
+
+        async def always_fails() -> None:
+            nonlocal call_count
+            call_count += 1
+            raise ValueError("fail")
+
+        retrying = AsyncRetrying(
+            enabled=False,
+            stop=stop_after_attempt(5),
+        )
+        with pytest.raises(ValueError, match="fail"):
+            await retrying(always_fails)
+        assert call_count == 1
+
+    @asynctest
+    async def test_enabled_false_call_succeeds_on_first_attempt(self) -> None:
+        """When enabled=False, awaiting the controller directly runs the coroutine
+        once and returns its result."""
+        call_count = 0
+
+        async def succeeds() -> str:
+            nonlocal call_count
+            call_count += 1
+            return "ok"
+
+        retrying = AsyncRetrying(
+            enabled=False,
+            stop=stop_after_attempt(5),
+        )
+        assert await retrying(succeeds) == "ok"
+        assert call_count == 1
+
 
 @unittest.skipIf(not have_trio, "trio not installed")
 class TestTrio(unittest.TestCase):

tests/test_tenacity.py

@@ -1590,6 +1590,43 @@ def test_enabled_false_iter_succeeds_on_first_attempt(self) -> None:
                 call_count += 1
         assert call_count == 1
 
+    def test_enabled_false_call_raises_original_exception(self) -> None:
+        """When enabled=False, calling the controller directly raises the original
+        exception, not a RetryError, and the function executes exactly once."""
+        call_count = 0
+
+        def fails() -> None:
+            nonlocal call_count
+            call_count += 1
+            raise ValueError("fail")
+
+        retrying = Retrying(
+            enabled=False,
+            stop=tenacity.stop_after_attempt(5),
+            wait=tenacity.wait_none(),
+        )
+        with pytest.raises(ValueError, match="fail"):
+            retrying(fails)
+        assert call_count == 1
+
+    def test_enabled_false_call_succeeds_on_first_attempt(self) -> None:
+        """When enabled=False, calling the controller directly runs the function once
+        and returns its result."""
+        call_count = 0
+
+        def succeeds() -> str:
+            nonlocal call_count
+            call_count += 1
+            return "ok"
+
+        retrying = Retrying(
+            enabled=False,
+            stop=tenacity.stop_after_attempt(5),
+            wait=tenacity.wait_none(),
+        )
+        assert retrying(succeeds) == "ok"
+        assert call_count == 1
+
 
 class TestRetryWith:
     def test_redefine_wait(self) -> None: