| 32 | |
| 33 | |
| 34 | def _exists( |
| 35 | resource: str, should_exist: bool, context: str, namespace: str | None |
| 36 | ) -> None: |
| 37 | cmd = ["kubectl", "get", "--output", "name", resource, "--context", context] |
| 38 | |
| 39 | if namespace is not None: |
| 40 | cmd.extend(["--namespace", namespace]) |
| 41 | |
| 42 | ui.progress(f'running {" ".join(cmd)} ... ') |
| 43 | |
| 44 | try: |
| 45 | result = subprocess.run(cmd, capture_output=True, encoding="ascii") |
| 46 | result.check_returncode() |
| 47 | if should_exist: |
| 48 | ui.progress("success!", finish=True) |
| 49 | else: |
| 50 | raise UIError(f"{resource} exists, but expected it not to") |
| 51 | except subprocess.CalledProcessError as e: |
| 52 | # A bit gross, but it should be safe enough in practice. |
| 53 | if "(NotFound)" in e.stderr: |
| 54 | if should_exist: |
| 55 | ui.progress("error!", finish=True) |
| 56 | raise UIError(f"{resource} does not exist, but expected it to") |
| 57 | else: |
| 58 | ui.progress("success!", finish=True) |
| 59 | else: |
| 60 | ui.progress(finish=True) |
| 61 | raise UIError(f"kubectl failed: {e}") |