prole/tests/installer/test_cleanup_cnpg_storage.py
chrisfu 2505497f0f Standardize namespace and cluster config flow
- Replace ambiguous NAMESPACE handling with SERVICE_NAMESPACE, DATABASE_NAMESPACE, and CLUSTER_NAME across config parsing/defaults, environment setup, actions, and UI bindings.

- Persist DATABASE_NAMESPACE and CLUSTER_NAME from Database Browser on Next and use them consistently in CNPG deployment/runtime logic.

- Update milestones/tests and include related config/tooling artifacts (cleanup_cnpg_storage tool, service config/version/network updates).

Co-authored-by: Junie <junie@jetbrains.com>
2026-03-23 16:52:40 -07:00

127 lines
4.0 KiB
Python

from __future__ import annotations
import pytest
from prole.tools import cleanup_cnpg_storage as cleanup
def test_parse_cli_args_defaults_to_dry_run():
args = cleanup.parse_cli_args(
["inspect", "--database-namespace", "knoe-db"]
)
assert args.command == "inspect"
assert args.database_namespace == "knoe-db"
assert args.cluster_name is None
assert args.dry_run is True
def test_parse_cli_args_cleanup_cluster_requires_cluster_name():
with pytest.raises(SystemExit):
cleanup.parse_cli_args(
["cleanup-cluster", "--database-namespace", "knoe-db"]
)
def test_select_resources_matches_namespace_cluster_and_cnpg_hints():
snapshot = cleanup.ResourceSnapshot(
clusters=[
{
"metadata": {
"name": "prole-db",
"namespace": "knoe-db",
"labels": {"prole.io/cluster": "prole-db"},
}
}
],
pvcs=[
{
"metadata": {
"name": "prole-db-1",
"namespace": "knoe-db",
"labels": {
"cnpg.io/cluster": "prole-db",
"prole.io/namespace": "knoe-db",
},
},
"status": {"phase": "Bound"},
}
],
pvs=[
{
"metadata": {
"name": "synology-iscsi-knoe-db-prole-db-data",
"labels": {
"synology.storage/role": "data",
"prole.io/namespace": "knoe-db",
"prole.io/cluster": "prole-db",
},
},
"spec": {
"storageClassName": "synology-iscsi",
"claimRef": {
"namespace": "knoe-db",
"name": "prole-db-1",
},
"local": {
"path": "/synology/d001/knoe/knoe-db/knoe-db/prole-db/data"
},
},
"status": {"phase": "Released"},
}
],
pods=[
{
"metadata": {
"name": "prole-db-1",
"namespace": "knoe-db",
"labels": {"cnpg.io/cluster": "prole-db"},
},
"status": {"phase": "Running"},
}
],
jobs=[
{
"metadata": {
"name": "prole-db-init-job",
"namespace": "knoe-db",
"labels": {"cnpg.io/cluster": "prole-db"},
}
}
],
)
filters = cleanup.MatchFilters(namespace="knoe-db", cluster_name="prole-db")
selected = cleanup.select_resources(snapshot, filters)
assert [item.name for item in selected.clusters] == ["prole-db"]
assert [item.name for item in selected.pvcs] == ["prole-db-1"]
assert [item.name for item in selected.pvs] == [
"synology-iscsi-knoe-db-prole-db-data"
]
assert [item.name for item in selected.pods] == ["prole-db-1"]
assert [item.name for item in selected.jobs] == ["prole-db-init-job"]
assert any(
reason.startswith("spec.claimRef.namespace=")
for reason in selected.pvs[0].reasons
)
def test_released_pv_prefers_claimref_patch_by_default():
pv = {
"spec": {
"claimRef": {"namespace": "knoe-db", "name": "prole-db-1"}
},
"status": {"phase": "Released"},
}
assert cleanup._should_patch_pv_claim_ref(pv, delete_pv=False) is True
assert cleanup._should_delete_pv(pv, delete_pv=False, force=False) is False
def test_delete_pv_requires_force_for_non_released_phase():
pv_bound = {"status": {"phase": "Bound"}, "spec": {}}
assert cleanup._should_delete_pv(pv_bound, delete_pv=True, force=False) is False
assert cleanup._should_delete_pv(pv_bound, delete_pv=True, force=True) is True