(env: Mapping[str, str], timeout: float = 0.5)
| 267 | |
| 268 | |
| 269 | def dapr_sidecar_available(env: Mapping[str, str], timeout: float = 0.5) -> bool: |
| 270 | endpoint = env.get("DAPR_HTTP_ENDPOINT", "http://127.0.0.1:3500") |
| 271 | parsed = urlparse(endpoint) |
| 272 | host = parsed.hostname or "127.0.0.1" |
| 273 | if parsed.port is not None: |
| 274 | port = parsed.port |
| 275 | elif parsed.scheme == "https": |
| 276 | port = 443 |
| 277 | else: |
| 278 | port = 80 |
| 279 | |
| 280 | try: |
| 281 | with socket.create_connection((host, port), timeout=timeout): |
| 282 | return True |
| 283 | except OSError: |
| 284 | return False |
| 285 | |
| 286 | |
| 287 | def prerequisite_skip_reasons( |
no test coverage detected