from pathlib import Path
import pytest
import yaml
from server import DepsResolver
# Helper to write yaml
def write_yaml(path: Path, content: dict):
with open(path, "w") as f:
yaml.dump(content, f)
@pytest.fixture
def resolver(tmp_path):
# Create a dummy structure
# root/
# kustomization.yaml
# deployment.yaml
# service.yaml
# base/
# kustomization.yaml
# base-res.yaml
# root/kustomization.yaml
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["deployment.yaml", "base"],
},
)
(tmp_path / "deployment.yaml").touch()
(tmp_path / "service.yaml").touch() # Unused
# base/kustomization.yaml
base_dir = tmp_path / "base"
base_dir.mkdir()
write_yaml(
base_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["base-res.yaml"],
},
)
(base_dir / "base-res.yaml").touch()
return DepsResolver(root_dir=tmp_path)
def test_collect_paths(resolver, tmp_path):
# Verify that files are collected correctly
# Note: _collect_paths is called in __init__
def test_paths(*paths) -> list[Path]:
return [tmp_path / p for p in paths]
assert sorted(resolver.kustomizations) == test_paths(
"base/kustomization.yaml",
"kustomization.yaml",
)
assert sorted(resolver.paths) == test_paths(
"base/base-res.yaml",
"base/kustomization.yaml",
"deployment.yaml",
"kustomization.yaml",
"service.yaml",
)
def test_compute_dependencies_simple(resolver, tmp_path):
# Test dependencies of root kustomization.yaml
deps = resolver.compute_dependencies(tmp_path / "kustomization.yaml")
# Expected: deployment.yaml (file), base (dir - resolved to base/kustomization.yaml)
# Note: DepsResolver returns relative paths to root_dir
expected_deps = {Path("deployment.yaml"), Path("base/kustomization.yaml")}
assert set(deps) == expected_deps
def test_compute_dependencies_nested(resolver, tmp_path):
# Test dependencies of base/kustomization.yaml
deps = resolver.compute_dependencies(tmp_path / "base" / "kustomization.yaml")
expected_deps = {Path("base/base-res.yaml")}
assert set(deps) == expected_deps
def test_nested_structure_dict(tmp_path):
# Create a kustomization with dict structure (e.g. patchesStrategicMerge or
# similar if it used dicts, or just verifying the recursive collection works
# on generic dicts as implemented)
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"patches": [{"path": "patch.yaml", "target": {"kind": "Deployment"}}],
},
)
(tmp_path / "patch.yaml").touch()
resolver = DepsResolver(root_dir=tmp_path)
deps = resolver.compute_dependencies(tmp_path / "kustomization.yaml")
assert Path("patch.yaml") in deps
def test_invalid_kustomization(tmp_path):
# Invalid apiVersion
write_yaml(
tmp_path / "kustomization.yaml",
{"apiVersion": "invalid", "kind": "Kustomization"},
)
resolver = DepsResolver(root_dir=tmp_path)
with pytest.raises(ValueError, match="not a valid Kustomization file"):
resolver.compute_dependencies(tmp_path / "kustomization.yaml")
def test_missing_file(tmp_path):
# Reference to non-existent file should be ignored.
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["missing.yaml"],
},
)
resolver = DepsResolver(root_dir=tmp_path)
deps = resolver.compute_dependencies(tmp_path / "kustomization.yaml")
assert deps == []
def test_recursive_dependencies_simple(tmp_path):
# Create a three-level nested structure:
# root/ -> base/ -> common/
# common/kustomization.yaml
common_dir = tmp_path / "common"
common_dir.mkdir()
write_yaml(
common_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["common-res.yaml"],
},
)
(common_dir / "common-res.yaml").touch()
# base/kustomization.yaml
base_dir = tmp_path / "base"
base_dir.mkdir()
write_yaml(
base_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["base-res.yaml", "../common"],
},
)
(base_dir / "base-res.yaml").touch()
# root/kustomization.yaml
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["deployment.yaml", "base"],
},
)
(tmp_path / "deployment.yaml").touch()
resolver = DepsResolver(root_dir=tmp_path)
# Non-recursive should only get immediate dependencies
deps_non_recursive = set(
resolver.compute_dependencies(tmp_path / "kustomization.yaml", recursive=False)
)
expected_non_recursive = {
Path("deployment.yaml"),
Path("base/kustomization.yaml"),
}
assert deps_non_recursive == expected_non_recursive
# Recursive should get all transitive dependencies
deps_recursive = set(
resolver.compute_dependencies(tmp_path / "kustomization.yaml", recursive=True)
)
assert deps_recursive == {
Path("deployment.yaml"),
Path("base/kustomization.yaml"),
Path("base/base-res.yaml"),
Path("common/kustomization.yaml"),
Path("common/common-res.yaml"),
}
def test_recursive_dependencies_multiple_branches(tmp_path):
# Create a structure with multiple branches:
# root/ -> base1/ -> shared/
# -> base2/ -> shared/
# shared/kustomization.yaml
shared_dir = tmp_path / "shared"
shared_dir.mkdir()
write_yaml(
shared_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["shared-res.yaml"],
},
)
(shared_dir / "shared-res.yaml").touch()
# base1/kustomization.yaml
base1_dir = tmp_path / "base1"
base1_dir.mkdir()
write_yaml(
base1_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["base1-res.yaml", "../shared"],
},
)
(base1_dir / "base1-res.yaml").touch()
# base2/kustomization.yaml
base2_dir = tmp_path / "base2"
base2_dir.mkdir()
write_yaml(
base2_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["base2-res.yaml", "../shared"],
},
)
(base2_dir / "base2-res.yaml").touch()
# root/kustomization.yaml
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["base1", "base2"],
},
)
resolver = DepsResolver(root_dir=tmp_path)
deps_recursive = set(
resolver.compute_dependencies(tmp_path / "kustomization.yaml", recursive=True)
)
# Should include all files, with shared/ counted once (no duplicates)
expected = {
Path("base1/kustomization.yaml"),
Path("base1/base1-res.yaml"),
Path("base2/kustomization.yaml"),
Path("base2/base2-res.yaml"),
Path("shared/kustomization.yaml"),
Path("shared/shared-res.yaml"),
}
assert deps_recursive == expected
def test_recursive_dependencies_circular(tmp_path):
# Create a circular dependency: a/ -> b/ -> a/
# This should not cause infinite loop
a_dir = tmp_path / "a"
a_dir.mkdir()
b_dir = tmp_path / "b"
b_dir.mkdir()
# a/kustomization.yaml references b/
write_yaml(
a_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["a-res.yaml", "../b"],
},
)
(a_dir / "a-res.yaml").touch()
# b/kustomization.yaml references a/
write_yaml(
b_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["b-res.yaml", "../a"],
},
)
(b_dir / "b-res.yaml").touch()
resolver = DepsResolver(root_dir=tmp_path)
# Should handle circular dependency without infinite loop
deps_recursive = set(
resolver.compute_dependencies(a_dir / "kustomization.yaml", recursive=True)
)
# Should include files from both a/ and b/, but not loop infinitely
# Note: a/kustomization.yaml appears because b/ references it, but it won't be
# recursively expanded again (preventing infinite loop)
expected = {
Path("a/a-res.yaml"),
Path("b/kustomization.yaml"),
Path("b/b-res.yaml"),
Path("a/kustomization.yaml"), # Referenced by b/, but not expanded again
}
assert deps_recursive == expected
def test_recursive_with_patches(tmp_path):
# Test recursive resolution with patches field (dict structure)
base_dir = tmp_path / "base"
base_dir.mkdir()
write_yaml(
base_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["deployment.yaml"],
},
)
(base_dir / "deployment.yaml").touch()
# root/kustomization.yaml with patches
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["base"],
"patches": [{"path": "patch.yaml", "target": {"kind": "Deployment"}}],
},
)
(tmp_path / "patch.yaml").touch()
resolver = DepsResolver(root_dir=tmp_path)
deps_recursive = set(
resolver.compute_dependencies(tmp_path / "kustomization.yaml", recursive=True)
)
# Should include patch.yaml, base kustomization, and deployment from base
expected = {
Path("patch.yaml"),
Path("base/kustomization.yaml"),
Path("base/deployment.yaml"),
}
assert deps_recursive == expected
def test_reverse_dependencies_simple(tmp_path):
# Create a simple structure:
# root/kustomization.yaml -> deployment.yaml
# We want to find which Kustomization depends on deployment.yaml
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["deployment.yaml"],
},
)
(tmp_path / "deployment.yaml").touch()
resolver = DepsResolver(root_dir=tmp_path)
reverse_deps = set(
resolver.compute_dependencies(
tmp_path / "deployment.yaml", reverse=True, recursive=False
)
)
# deployment.yaml is used by root kustomization
assert reverse_deps == {Path("kustomization.yaml")}
def test_reverse_dependencies_multiple_dependents(tmp_path):
# Create structure where multiple Kustomizations depend on the same file:
# base/kustomization.yaml -> shared.yaml
# overlay/kustomization.yaml -> shared.yaml
base_dir = tmp_path / "base"
base_dir.mkdir()
overlay_dir = tmp_path / "overlay"
overlay_dir.mkdir()
(tmp_path / "shared.yaml").touch()
write_yaml(
base_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../shared.yaml"],
},
)
write_yaml(
overlay_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../shared.yaml"],
},
)
resolver = DepsResolver(root_dir=tmp_path)
reverse_deps = set(
resolver.compute_dependencies(
tmp_path / "shared.yaml", reverse=True, recursive=False
)
)
# Both kustomizations depend on shared.yaml
assert reverse_deps == {
Path("base/kustomization.yaml"),
Path("overlay/kustomization.yaml"),
}
def test_reverse_dependencies_nested(tmp_path):
# Create structure:
# common/kustomization.yaml -> common.yaml
# base/kustomization.yaml -> common/
# overlay/kustomization.yaml -> base/
# Test reverse dependencies of common.yaml with recursive=False and recursive=True
common_dir = tmp_path / "common"
common_dir.mkdir()
base_dir = tmp_path / "base"
base_dir.mkdir()
overlay_dir = tmp_path / "overlay"
overlay_dir.mkdir()
(common_dir / "common.yaml").touch()
write_yaml(
common_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["common.yaml"],
},
)
write_yaml(
base_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../common"],
},
)
write_yaml(
overlay_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../base"],
},
)
resolver = DepsResolver(root_dir=tmp_path)
# Non-recursive: only common/kustomization.yaml depends on common.yaml directly
reverse_deps_non_recursive = set(
resolver.compute_dependencies(
tmp_path / "common" / "common.yaml", reverse=True, recursive=False
)
)
assert reverse_deps_non_recursive == {Path("common/kustomization.yaml")}
# Recursive: base/ depends on common/, and overlay/ depends on base/
reverse_deps_recursive = set(
resolver.compute_dependencies(
tmp_path / "common" / "common.yaml", reverse=True, recursive=True
)
)
assert reverse_deps_recursive == {
Path("common/kustomization.yaml"),
Path("base/kustomization.yaml"),
Path("overlay/kustomization.yaml"),
}
def test_reverse_dependencies_of_kustomization(tmp_path):
# Test reverse dependencies when the input is itself a Kustomization file
# base/kustomization.yaml
# overlay/kustomization.yaml -> base/
base_dir = tmp_path / "base"
base_dir.mkdir()
overlay_dir = tmp_path / "overlay"
overlay_dir.mkdir()
write_yaml(
base_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["resource.yaml"],
},
)
(base_dir / "resource.yaml").touch()
write_yaml(
overlay_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../base"],
},
)
resolver = DepsResolver(root_dir=tmp_path)
reverse_deps = set(
resolver.compute_dependencies(
tmp_path / "base" / "kustomization.yaml", reverse=True, recursive=False
)
)
# overlay/kustomization.yaml depends on base/kustomization.yaml
assert reverse_deps == {Path("overlay/kustomization.yaml")}
def test_reverse_dependencies_no_dependents(tmp_path):
# Test file that has no dependents
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["resource.yaml"],
},
)
(tmp_path / "resource.yaml").touch()
(tmp_path / "unused.yaml").touch()
resolver = DepsResolver(root_dir=tmp_path)
reverse_deps = resolver.compute_dependencies(
tmp_path / "unused.yaml", reverse=True, recursive=False
)
# unused.yaml has no dependents
assert reverse_deps == []
def test_reverse_dependencies_recursive_diamond(tmp_path):
# Test diamond dependency structure:
# root/ -> left/ -> shared/
# -> right/ -> shared/
# When querying reverse deps of shared/, should get left, right, and root
shared_dir = tmp_path / "shared"
shared_dir.mkdir()
left_dir = tmp_path / "left"
left_dir.mkdir()
right_dir = tmp_path / "right"
right_dir.mkdir()
write_yaml(
shared_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["resource.yaml"],
},
)
(shared_dir / "resource.yaml").touch()
write_yaml(
left_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../shared"],
},
)
write_yaml(
right_dir / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["../shared"],
},
)
write_yaml(
tmp_path / "kustomization.yaml",
{
"apiVersion": "kustomize.config.k8s.io/v1beta1",
"kind": "Kustomization",
"resources": ["left", "right"],
},
)
resolver = DepsResolver(root_dir=tmp_path)
reverse_deps_recursive = set(
resolver.compute_dependencies(
tmp_path / "shared" / "kustomization.yaml", reverse=True, recursive=True
)
)
# All three should be in the transitive reverse dependencies
assert reverse_deps_recursive == {
Path("left/kustomization.yaml"),
Path("right/kustomization.yaml"),
Path("kustomization.yaml"),
}
