|
20 | 20 | from typing import Any, Callable, Optional, Protocol, TypeVar, Union, cast |
21 | 21 |
|
22 | 22 | import wrapt |
| 23 | +from paramiko.packet import first_arg |
23 | 24 | from typing_extensions import Self |
24 | 25 |
|
25 | 26 | from testcontainers.core.config import testcontainers_config |
@@ -161,7 +162,7 @@ def check_http(container): |
161 | 162 | ) |
162 | 163 |
|
163 | 164 | class LegacyWaitStrategy(WaitStrategy): |
164 | | - def __init__(self, func: Callable[..., Any], instance: Any, args: list[Any], kwargs: dict[str, Any]): |
| 165 | + def __init__(self, func: Callable[..., Any], instance: Any, args: tuple[Any], kwargs: dict[str, Any]): |
165 | 166 | super().__init__() |
166 | 167 | self.func = func |
167 | 168 | self.instance = instance |
@@ -197,12 +198,16 @@ def wait_until_ready(self, container: WaitStrategyTarget) -> Any: |
197 | 198 | logger.debug(f"Connection attempt failed: {e!s}") |
198 | 199 | time.sleep(self._poll_interval) |
199 | 200 |
|
200 | | - @wrapt.decorator # type: ignore[misc] |
201 | | - def wrapper(wrapped: Callable[..., Any], instance: Any, args: list[Any], kwargs: dict[str, Any]) -> Any: |
| 201 | + @wrapt.decorator |
| 202 | + def wrapper(wrapped: Callable[..., Any], instance: Any, args: tuple[Any], kwargs: dict[str, Any]) -> Any: |
202 | 203 | # Use the LegacyWaitStrategy to handle retries with proper timeout |
203 | 204 | strategy = LegacyWaitStrategy(wrapped, instance, args, kwargs) |
204 | 205 | # For backwards compatibility, assume the instance is the container |
205 | | - container = instance if hasattr(instance, "get_container_host_ip") else args[0] if args else None |
| 206 | + try: |
| 207 | + first_arg = args[0] |
| 208 | + except IndexError: |
| 209 | + first_arg = None |
| 210 | + container = instance if hasattr(instance, "get_container_host_ip") else first_arg |
206 | 211 | if container: |
207 | 212 | return strategy.wait_until_ready(container) |
208 | 213 | else: |
|
0 commit comments