Skip to content

Commit 0592af3

Browse files
LillecarlLillecarl
committed
Add unit tests for check_csi_volume_mounts: 14 cases covering subPath, /nix mount, compat driver, initContainers
1 parent 4beff5a commit 0592af3

1 file changed

Lines changed: 263 additions & 0 deletions

File tree

‎pkgs/nixkube/tests/test_csi_volume_checks.py‎

Lines changed: 263 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,263 @@
1+
# SPDX-License-Identifier: MIT
2+
3+
from unittest.mock import AsyncMock, MagicMock, patch
4+
5+
import pytest
6+
from src.csi.server import check_csi_volume_mounts
7+
8+
9+
def make_pod(spec: dict) -> MagicMock:
10+
"""Create a mock Pod with a .raw property returning the given spec dict."""
11+
pod = MagicMock()
12+
pod.raw = {"spec": spec}
13+
return pod
14+
15+
16+
def nixkube_volume(name: str, driver: str = "nixkube") -> dict:
17+
return {"name": name, "csi": {"driver": driver}}
18+
19+
20+
def non_csi_volume(name: str) -> dict:
21+
return {"name": name, "emptyDir": {}}
22+
23+
24+
def container_with_mounts(name: str, *mounts: dict) -> dict:
25+
return {"name": name, "volumeMounts": list(mounts)}
26+
27+
28+
def mount(vol_name: str, mount_path: str, sub_path: str | None = None) -> dict:
29+
m: dict = {"name": vol_name, "mountPath": mount_path}
30+
if sub_path is not None:
31+
m["subPath"] = sub_path
32+
return m
33+
34+
35+
class TestCheckCsiVolumeMounts:
36+
"""Tests for check_csi_volume_mounts() warning event logic."""
37+
38+
@pytest.mark.asyncio
39+
async def test_no_volumes_no_events(self):
40+
"""Pod with no volumes emits no events."""
41+
pod = make_pod({"containers": [], "volumes": []})
42+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
43+
await check_csi_volume_mounts(pod)
44+
mock_event.assert_not_called()
45+
46+
@pytest.mark.asyncio
47+
async def test_non_csi_volume_no_events(self):
48+
"""Non-CSI volumes (emptyDir) are ignored."""
49+
pod = make_pod(
50+
{
51+
"containers": [container_with_mounts("app", mount("data", "/data"))],
52+
"volumes": [non_csi_volume("data")],
53+
}
54+
)
55+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
56+
await check_csi_volume_mounts(pod)
57+
mock_event.assert_not_called()
58+
59+
@pytest.mark.asyncio
60+
async def test_non_nixkube_csi_driver_no_events(self):
61+
"""CSI volumes from other drivers are ignored."""
62+
pod = make_pod(
63+
{
64+
"containers": [container_with_mounts("app", mount("nfs", "/data"))],
65+
"volumes": [{"name": "nfs", "csi": {"driver": "nfs.csi.k8s.io"}}],
66+
}
67+
)
68+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
69+
await check_csi_volume_mounts(pod)
70+
mock_event.assert_not_called()
71+
72+
@pytest.mark.asyncio
73+
async def test_correct_config_no_events(self):
74+
"""Correct nixkube volume (subPath='nix', mountPath='/nix') emits no events."""
75+
pod = make_pod(
76+
{
77+
"containers": [
78+
container_with_mounts("app", mount("nix-store", "/nix", "nix"))
79+
],
80+
"volumes": [nixkube_volume("nix-store")],
81+
}
82+
)
83+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
84+
await check_csi_volume_mounts(pod)
85+
mock_event.assert_not_called()
86+
87+
@pytest.mark.asyncio
88+
async def test_missing_subpath_emits_warning(self):
89+
"""Volume mount without subPath emits MissingSubPath warning."""
90+
pod = make_pod(
91+
{
92+
"containers": [
93+
container_with_mounts("app", mount("nix-store", "/nix"))
94+
],
95+
"volumes": [nixkube_volume("nix-store")],
96+
}
97+
)
98+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
99+
await check_csi_volume_mounts(pod)
100+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
101+
assert "MissingSubPath" in reasons
102+
103+
@pytest.mark.asyncio
104+
async def test_missing_nix_mount_emits_warning(self):
105+
"""Volume mounted elsewhere but not at /nix emits MissingNixMount warning."""
106+
pod = make_pod(
107+
{
108+
"containers": [
109+
container_with_mounts("app", mount("nix-store", "/opt/nix", "nix"))
110+
],
111+
"volumes": [nixkube_volume("nix-store")],
112+
}
113+
)
114+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
115+
await check_csi_volume_mounts(pod)
116+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
117+
assert "MissingNixMount" in reasons
118+
119+
@pytest.mark.asyncio
120+
async def test_no_mounts_at_all_emits_missing_nix_mount(self):
121+
"""nixkube volume with no volumeMounts in any container emits MissingNixMount."""
122+
pod = make_pod(
123+
{
124+
"containers": [{"name": "app", "volumeMounts": []}],
125+
"volumes": [nixkube_volume("nix-store")],
126+
}
127+
)
128+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
129+
await check_csi_volume_mounts(pod)
130+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
131+
assert "MissingNixMount" in reasons
132+
assert "MissingSubPath" not in reasons
133+
134+
@pytest.mark.asyncio
135+
async def test_both_warnings_when_missing_subpath_and_not_at_nix(self):
136+
"""Mount missing subPath and not at /nix triggers both warnings."""
137+
pod = make_pod(
138+
{
139+
"containers": [
140+
container_with_mounts("app", mount("nix-store", "/opt/nix"))
141+
],
142+
"volumes": [nixkube_volume("nix-store")],
143+
}
144+
)
145+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
146+
await check_csi_volume_mounts(pod)
147+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
148+
assert "MissingSubPath" in reasons
149+
assert "MissingNixMount" in reasons
150+
151+
@pytest.mark.asyncio
152+
async def test_compat_driver_name_also_checked(self):
153+
"""nix.csi.store driver is treated the same as nixkube."""
154+
pod = make_pod(
155+
{
156+
"containers": [
157+
container_with_mounts("app", mount("nix-store", "/nix"))
158+
],
159+
"volumes": [nixkube_volume("nix-store", driver="nix.csi.store")],
160+
}
161+
)
162+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
163+
await check_csi_volume_mounts(pod)
164+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
165+
# Mount at /nix but no subPath → MissingSubPath; mount IS at /nix → no MissingNixMount
166+
assert "MissingSubPath" in reasons
167+
assert "MissingNixMount" not in reasons
168+
169+
@pytest.mark.asyncio
170+
async def test_init_container_mount_checked(self):
171+
"""Mounts in initContainers are also checked for missing subPath."""
172+
pod = make_pod(
173+
{
174+
"containers": [],
175+
"initContainers": [
176+
container_with_mounts("init", mount("nix-store", "/nix"))
177+
],
178+
"volumes": [nixkube_volume("nix-store")],
179+
}
180+
)
181+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
182+
await check_csi_volume_mounts(pod)
183+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
184+
assert "MissingSubPath" in reasons
185+
186+
@pytest.mark.asyncio
187+
async def test_init_container_nix_mount_satisfies_missing_nix_check(self):
188+
"""A /nix mount in an initContainer prevents MissingNixMount."""
189+
pod = make_pod(
190+
{
191+
"containers": [],
192+
"initContainers": [
193+
container_with_mounts("init", mount("nix-store", "/nix", "nix"))
194+
],
195+
"volumes": [nixkube_volume("nix-store")],
196+
}
197+
)
198+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
199+
await check_csi_volume_mounts(pod)
200+
mock_event.assert_not_called()
201+
202+
@pytest.mark.asyncio
203+
async def test_multiple_containers_one_missing_subpath(self):
204+
"""Only the container with a missing subPath triggers MissingSubPath."""
205+
pod = make_pod(
206+
{
207+
"containers": [
208+
container_with_mounts("good", mount("nix-store", "/nix", "nix")),
209+
container_with_mounts("bad", mount("nix-store", "/nix")),
210+
],
211+
"volumes": [nixkube_volume("nix-store")],
212+
}
213+
)
214+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
215+
await check_csi_volume_mounts(pod)
216+
missing_subpath_calls = [
217+
c
218+
for c in mock_event.call_args_list
219+
if c.kwargs["reason"] == "MissingSubPath"
220+
]
221+
assert len(missing_subpath_calls) == 1
222+
assert "bad" in missing_subpath_calls[0].kwargs["note"]
223+
224+
@pytest.mark.asyncio
225+
async def test_multiple_nixkube_volumes_checked_independently(self):
226+
"""Each nixkube volume is checked independently."""
227+
pod = make_pod(
228+
{
229+
"containers": [
230+
container_with_mounts(
231+
"app",
232+
mount("vol-a", "/nix", "nix"), # correct
233+
mount("vol-b", "/nix2"), # missing subPath + wrong path
234+
)
235+
],
236+
"volumes": [
237+
nixkube_volume("vol-a"),
238+
nixkube_volume("vol-b"),
239+
],
240+
}
241+
)
242+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
243+
await check_csi_volume_mounts(pod)
244+
reasons = [c.kwargs["reason"] for c in mock_event.call_args_list]
245+
# vol-a is fine, vol-b is missing subPath and missing /nix
246+
assert reasons.count("MissingSubPath") == 1
247+
assert reasons.count("MissingNixMount") == 1
248+
249+
@pytest.mark.asyncio
250+
async def test_events_are_warnings(self):
251+
"""All emitted events have event_type='Warning'."""
252+
pod = make_pod(
253+
{
254+
"containers": [
255+
container_with_mounts("app", mount("nix-store", "/opt/nix"))
256+
],
257+
"volumes": [nixkube_volume("nix-store")],
258+
}
259+
)
260+
with patch("src.csi.server.report_event", new_callable=AsyncMock) as mock_event:
261+
await check_csi_volume_mounts(pod)
262+
for c in mock_event.call_args_list:
263+
assert c.kwargs["event_type"] == "Warning"

0 commit comments

Comments
 (0)