(args: argparse.Namespace)
| 392 | |
| 393 | |
| 394 | def get_resource_id(args: argparse.Namespace): |
| 395 | env_kubectl = make_env_kubectl(args) |
| 396 | |
| 397 | def try_get_resource_id(): |
| 398 | resource_id = ( |
| 399 | env_kubectl( |
| 400 | "get", |
| 401 | "materialize", |
| 402 | args.environment_name, |
| 403 | "-o=jsonpath={.status.resourceId}", |
| 404 | ) |
| 405 | .decode() |
| 406 | .strip() |
| 407 | ) |
| 408 | assert resource_id != "" |
| 409 | return resource_id |
| 410 | |
| 411 | return retry( |
| 412 | try_get_resource_id, |
| 413 | exception_types=(AssertionError, subprocess.CalledProcessError), |
| 414 | ) |
| 415 | |
| 416 | |
| 417 | def make_env_kubectl(args: argparse.Namespace, namespace: str | None = None): |
no test coverage detected