(mz: MaterializeApplication, pod_name: str)
| 13 | |
| 14 | |
| 15 | def assert_pod_properties(mz: MaterializeApplication, pod_name: str) -> None: |
| 16 | pod_desc = json.loads(mz.kubectl("get", "-o", "json", "pod", pod_name)) |
| 17 | node_selector = pod_desc["spec"]["nodeSelector"] |
| 18 | |
| 19 | expected_disk_selection_mode = "true" |
| 20 | |
| 21 | assert any( |
| 22 | "disk" in k and v == expected_disk_selection_mode |
| 23 | for k, v in node_selector.items() |
| 24 | ), f"pod {pod_name} does not have the disk={expected_disk_selection_mode} selector" |
| 25 | assert not any( |
| 26 | "availability-zone" in k for k in node_selector.keys() |
| 27 | ), f"pod {pod_name} has a availability-zone selector" |
| 28 | |
| 29 | |
| 30 | def test_system_clusters(mz: MaterializeApplication) -> None: |
no test coverage detected