mirror of
https://github.com/radixark/miles.git
synced 2026-10-02 07:14:53 +08:00
Lint every Helm chart from pre-commit and CI
Nothing under charts/ yet; this puts the gate in place first so the charts that follow land already covered. - tools/lint_charts.py vendors each chart's dependencies, then lints every chart with every value combination it supports. It skips when helm is absent locally and fails when CI provides no helm. - Both CI jobs that need helm install the pinned version: the CPU job because chart tests shell out to it, the pre-commit job because the hook does. - check-yaml has to skip templates/: Go-templated YAML is not YAML.
This commit is contained in:
fzyzcjy
@@ -193,3 +193,5 @@ _examples_synced/
|
||||
.DS_Store
|
||||
|
||||
.humanize/
|
||||
|
||||
charts/*/charts/
|
||||
|
||||
@@ -15,6 +15,7 @@ repos:
|
||||
rev: v4.5.0
|
||||
hooks:
|
||||
- id: check-yaml
|
||||
exclude: '^charts/.*/templates/'
|
||||
- id: check-case-conflict
|
||||
- id: detect-private-key
|
||||
- id: check-added-large-files
|
||||
@@ -64,6 +65,13 @@ repos:
|
||||
entry: '"root":'
|
||||
files: ^docs/docs\.json$
|
||||
|
||||
- id: helm-lint
|
||||
name: helm lint every chart
|
||||
language: system
|
||||
entry: python3 tools/kubernetes/lint_charts.py
|
||||
pass_filenames: false
|
||||
files: '^charts/'
|
||||
|
||||
- id: ban-mpu-get
|
||||
name: Ban direct mpu.get_* calls (use ParallelState)
|
||||
language: pygrep
|
||||
|
||||
@@ -50,12 +50,8 @@ thinking_template = "{reasoning_content}"
|
||||
response_format_template: str = (
|
||||
"## Response Format:\n\nYou MUST strictly adhere to the following schema to reply:\n{schema}"
|
||||
)
|
||||
tool_call_template: str = (
|
||||
'<{dsml_token}invoke name="{name}">\n{arguments}\n</{dsml_token}invoke>'
|
||||
)
|
||||
tool_calls_template = (
|
||||
"<{dsml_token}function_calls>\n{tool_calls}\n</{dsml_token}function_calls>"
|
||||
)
|
||||
tool_call_template: str = '<{dsml_token}invoke name="{name}">\n{arguments}\n</{dsml_token}invoke>'
|
||||
tool_calls_template = "<{dsml_token}function_calls>\n{tool_calls}\n</{dsml_token}function_calls>"
|
||||
|
||||
tool_output_template: str = "\n<result>{content}</result>"
|
||||
|
||||
@@ -99,13 +95,9 @@ def encode_arguments_to_dsml(tool_call: Dict[str, str]) -> str:
|
||||
P_dsml_strs = []
|
||||
|
||||
raw_arguments = tool_call["arguments"]
|
||||
arguments = (
|
||||
json.loads(raw_arguments) if isinstance(raw_arguments, str) else raw_arguments
|
||||
)
|
||||
arguments = json.loads(raw_arguments) if isinstance(raw_arguments, str) else raw_arguments
|
||||
if not isinstance(arguments, dict):
|
||||
raise ValueError(
|
||||
"Assistant tool call function.arguments must be a JSON object."
|
||||
)
|
||||
raise ValueError("Assistant tool call function.arguments must be a JSON object.")
|
||||
|
||||
for k, v in arguments.items():
|
||||
p_dsml_str = p_dsml_template.format(
|
||||
@@ -120,20 +112,14 @@ def encode_arguments_to_dsml(tool_call: Dict[str, str]) -> str:
|
||||
return "\n".join(P_dsml_strs)
|
||||
|
||||
|
||||
def decode_dsml_to_arguments(
|
||||
tool_name: str, tool_args: Dict[str, Tuple[str, str]]
|
||||
) -> Dict[str, str]:
|
||||
def decode_dsml_to_arguments(tool_name: str, tool_args: Dict[str, Tuple[str, str]]) -> Dict[str, str]:
|
||||
def _decode_value(key: str, value: str, string: str):
|
||||
if string == "true":
|
||||
value = to_json(value)
|
||||
return f"{to_json(key)}: {value}"
|
||||
|
||||
tool_args_json = (
|
||||
"{"
|
||||
+ ", ".join(
|
||||
[_decode_value(k, v, string=is_str) for k, (v, is_str) in tool_args.items()]
|
||||
)
|
||||
+ "}"
|
||||
"{" + ", ".join([_decode_value(k, v, string=is_str) for k, (v, is_str) in tool_args.items()]) + "}"
|
||||
)
|
||||
return dict(name=tool_name, arguments=tool_args_json)
|
||||
|
||||
@@ -158,13 +144,9 @@ def find_last_user_index(messages: List[Dict[str, Any]]) -> int:
|
||||
return last_user_index
|
||||
|
||||
|
||||
def render_message(
|
||||
index: int, messages: List[Dict[str, Any]], thinking_mode: str, drop_thinking: bool = True
|
||||
) -> str:
|
||||
def render_message(index: int, messages: List[Dict[str, Any]], thinking_mode: str, drop_thinking: bool = True) -> str:
|
||||
if not (0 <= index < len(messages)):
|
||||
raise DS32EncodingError(
|
||||
f"Index {index} out of range for messages list of length {len(messages)}"
|
||||
)
|
||||
raise DS32EncodingError(f"Index {index} out of range for messages list of length {len(messages)}")
|
||||
if thinking_mode not in ["chat", "thinking"]:
|
||||
raise DS32EncodingError(f"Invalid thinking_mode `{thinking_mode}`")
|
||||
|
||||
@@ -190,9 +172,7 @@ def render_message(
|
||||
prompt += "\n\n" + render_tools(tools)
|
||||
|
||||
if response_format:
|
||||
prompt += "\n\n" + response_format_template.format(
|
||||
schema=to_json(response_format)
|
||||
)
|
||||
prompt += "\n\n" + response_format_template.format(schema=to_json(response_format))
|
||||
|
||||
elif role == "developer":
|
||||
if not content:
|
||||
@@ -202,9 +182,7 @@ def render_message(
|
||||
content_developer += "\n\n" + render_tools(tools)
|
||||
|
||||
if response_format:
|
||||
content_developer += "\n\n" + response_format_template.format(
|
||||
schema=to_json(response_format)
|
||||
)
|
||||
content_developer += "\n\n" + response_format_template.format(schema=to_json(response_format))
|
||||
|
||||
content_developer += "\n\n# The user's message is: {}".format(content)
|
||||
|
||||
@@ -236,10 +214,7 @@ def render_message(
|
||||
prev_assistant_idx -= 1
|
||||
assistant_msg = messages[prev_assistant_idx]
|
||||
|
||||
if not (
|
||||
index == 0
|
||||
or (prev_assistant_idx >= 0 and assistant_msg.get("role") == "assistant")
|
||||
):
|
||||
if not (index == 0 or (prev_assistant_idx >= 0 and assistant_msg.get("role") == "assistant")):
|
||||
raise DS32EncodingError(f"Invalid messages at {index}:\n{assistant_msg}")
|
||||
|
||||
tool_call_order = index - prev_assistant_idx
|
||||
@@ -293,10 +268,7 @@ def render_message(
|
||||
# last user turn), which makes the render append-only across user
|
||||
# appends; mirrors encoding_dsv4's assistant gate.
|
||||
if thinking_mode == "thinking" and (not drop_thinking or index > last_user_idx):
|
||||
thinking_part = (
|
||||
thinking_template.format(reasoning_content=reasoning_content or "")
|
||||
+ thinking_end_token
|
||||
)
|
||||
thinking_part = thinking_template.format(reasoning_content=reasoning_content or "") + thinking_end_token
|
||||
|
||||
prompt += assistant_msg_template.format(
|
||||
reasoning=thinking_part,
|
||||
@@ -313,9 +285,7 @@ def drop_thinking_messages(
|
||||
messages: List[Dict[str, Any]], last_user_idx: Optional[int] = None
|
||||
) -> List[Dict[str, Any]]:
|
||||
messages_wo_thinking: List[Dict[str, Any]] = []
|
||||
last_user_idx = (
|
||||
find_last_user_index(messages) if last_user_idx is None else last_user_idx
|
||||
)
|
||||
last_user_idx = find_last_user_index(messages) if last_user_idx is None else last_user_idx
|
||||
for idx, msg in enumerate(messages):
|
||||
role = msg.get("role")
|
||||
if role in ["user", "system", "tool"] or idx >= last_user_idx:
|
||||
@@ -356,9 +326,7 @@ def encode_messages(
|
||||
return prompt
|
||||
|
||||
|
||||
def _read_until_stop(
|
||||
index: int, text: str, stop: List[str]
|
||||
) -> Tuple[int, str, Optional[str]]:
|
||||
def _read_until_stop(index: int, text: str, stop: List[str]) -> Tuple[int, str, Optional[str]]:
|
||||
min_pos = len(text)
|
||||
matched_stop = None
|
||||
|
||||
@@ -382,9 +350,7 @@ def parse_tool_calls(index: int, text: str):
|
||||
tool_calls_end_token = f"</{dsml_token}function_calls>"
|
||||
|
||||
while index < len(text):
|
||||
index, _, stop_token = _read_until_stop(
|
||||
index, text, [f"<{dsml_token}invoke", tool_calls_end_token]
|
||||
)
|
||||
index, _, stop_token = _read_until_stop(index, text, [f"<{dsml_token}invoke", tool_calls_end_token])
|
||||
if _ != ">\n":
|
||||
raise DS32EncodingError("Tool call format error")
|
||||
|
||||
@@ -398,18 +364,14 @@ def parse_tool_calls(index: int, text: str):
|
||||
index, text, [f"<{dsml_token}parameter", f"</{dsml_token}invoke"]
|
||||
)
|
||||
|
||||
p_tool_name = re.findall(
|
||||
r'^\s*name="(.*?)">\n$', tool_name_content, flags=re.DOTALL
|
||||
)
|
||||
p_tool_name = re.findall(r'^\s*name="(.*?)">\n$', tool_name_content, flags=re.DOTALL)
|
||||
if len(p_tool_name) != 1:
|
||||
raise DS32EncodingError("Tool name format error")
|
||||
tool_name = p_tool_name[0]
|
||||
|
||||
tool_args: Dict[str, Tuple[str, str]] = {}
|
||||
while stop_token == f"<{dsml_token}parameter":
|
||||
index, param_content, stop_token = _read_until_stop(
|
||||
index, text, [f"/{dsml_token}parameter"]
|
||||
)
|
||||
index, param_content, stop_token = _read_until_stop(index, text, [f"/{dsml_token}parameter"])
|
||||
|
||||
param_kv = re.findall(
|
||||
r'^ name="(.*?)" string="(true|false)">(.*?)<$',
|
||||
@@ -445,16 +407,12 @@ def parse_message_from_completion_text(text: str, thinking_mode: str):
|
||||
is_thinking, is_tool_calling = thinking_mode == "thinking", False
|
||||
|
||||
if is_thinking:
|
||||
index, content_delta, stop_token = _read_until_stop(
|
||||
index, text, [thinking_end_token, tool_calls_start_token]
|
||||
)
|
||||
index, content_delta, stop_token = _read_until_stop(index, text, [thinking_end_token, tool_calls_start_token])
|
||||
reasoning_content = content_delta
|
||||
if stop_token != thinking_end_token:
|
||||
raise DS32EncodingError("Invalid thinking format")
|
||||
|
||||
index, content_delta, stop_token = _read_until_stop(
|
||||
index, text, [eos_token, tool_calls_start_token]
|
||||
)
|
||||
index, content_delta, stop_token = _read_until_stop(index, text, [eos_token, tool_calls_start_token])
|
||||
summary_content = content_delta
|
||||
if stop_token == tool_calls_start_token:
|
||||
is_tool_calling = True
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
import importlib.util
|
||||
import shutil
|
||||
import subprocess
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.fast.source_scan import REPO_ROOT
|
||||
|
||||
LINT_CHARTS_PATH = REPO_ROOT / "tools" / "kubernetes" / "lint_charts.py"
|
||||
|
||||
requires_helm = pytest.mark.skipif(shutil.which("helm") is None, reason="helm is required to lint charts")
|
||||
|
||||
|
||||
def load_lint_charts():
|
||||
spec = importlib.util.spec_from_file_location("lint_charts", LINT_CHARTS_PATH)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lint_charts(tmp_path, monkeypatch):
|
||||
module = load_lint_charts()
|
||||
monkeypatch.setattr(module, "CHARTS_DIR", tmp_path)
|
||||
monkeypatch.delenv("CI", raising=False)
|
||||
return module
|
||||
|
||||
|
||||
def write_chart(root: Path, name: str, *, library: bool = False, template: str = "") -> Path:
|
||||
chart = root / name
|
||||
(chart / "templates").mkdir(parents=True)
|
||||
(chart / "Chart.yaml").write_text(
|
||||
textwrap.dedent(
|
||||
f"""
|
||||
apiVersion: v2
|
||||
name: {name}
|
||||
version: 0.1.0
|
||||
{"type: library" if library else ""}
|
||||
"""
|
||||
).strip()
|
||||
+ "\n"
|
||||
)
|
||||
(chart / "values.yaml").write_text("{}\n")
|
||||
if template:
|
||||
(chart / "templates" / "object.yaml").write_text(template)
|
||||
return chart
|
||||
|
||||
|
||||
VALID_TEMPLATE = "apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: {{ .Release.Name }}\n"
|
||||
BROKEN_TEMPLATE = "apiVersion: v1\nkind: ConfigMap\n name: broken: [\n"
|
||||
|
||||
|
||||
class TestChartDiscovery:
|
||||
def test_library_charts_are_linted_too(self, lint_charts, tmp_path):
|
||||
"""The hook promises to lint every chart, and a library chart can still fail to parse."""
|
||||
write_chart(tmp_path, "app", template=VALID_TEMPLATE)
|
||||
write_chart(tmp_path, "lib", library=True)
|
||||
|
||||
assert [chart.name for chart in lint_charts.all_charts()] == ["app", "lib"]
|
||||
|
||||
def test_it_looks_in_the_repository_charts_directory(self):
|
||||
"""Every other test points the constant at a tmp_path, so nothing else would notice a typo here."""
|
||||
module = load_lint_charts()
|
||||
|
||||
assert (module.REPO_ROOT / "pyproject.toml").exists()
|
||||
assert module.CHARTS_DIR == REPO_ROOT / "charts"
|
||||
|
||||
def test_a_repo_with_no_charts_is_not_a_failure(self, lint_charts):
|
||||
"""The hook lands before the first chart, and an empty repo must not block every commit."""
|
||||
assert lint_charts.all_charts() == []
|
||||
assert lint_charts.main([]) == 0
|
||||
|
||||
|
||||
@requires_helm
|
||||
class TestLinting:
|
||||
def test_a_valid_chart_passes(self, lint_charts, tmp_path):
|
||||
"""The baseline: a chart helm accepts must not be reported as broken."""
|
||||
write_chart(tmp_path, "app", template=VALID_TEMPLATE)
|
||||
|
||||
assert lint_charts.main([]) == 0
|
||||
|
||||
def test_a_broken_chart_fails(self, lint_charts, tmp_path):
|
||||
"""A template helm cannot parse is exactly what this hook exists to catch."""
|
||||
write_chart(tmp_path, "app", template=BROKEN_TEMPLATE)
|
||||
|
||||
assert lint_charts.main([]) == 1
|
||||
|
||||
def test_every_chart_is_linted_even_after_one_fails(self, lint_charts, tmp_path, capsys):
|
||||
"""Stopping at the first failure costs the contributor one round trip per broken chart."""
|
||||
write_chart(tmp_path, "a-broken", template=BROKEN_TEMPLATE)
|
||||
write_chart(tmp_path, "z-also-broken", template=BROKEN_TEMPLATE)
|
||||
|
||||
assert lint_charts.main([]) == 1
|
||||
errors = capsys.readouterr().err
|
||||
|
||||
assert "a-broken" in errors
|
||||
assert "z-also-broken" in errors
|
||||
|
||||
def test_a_chart_whose_dependencies_cannot_be_vendored_fails(self, lint_charts, tmp_path):
|
||||
"""helm lint only warns about a missing dependency, so the build step is the only guard."""
|
||||
app = write_chart(tmp_path, "app", template=VALID_TEMPLATE)
|
||||
(app / "Chart.yaml").write_text(
|
||||
"apiVersion: v2\nname: app\nversion: 0.1.0\n"
|
||||
'dependencies:\n - name: absent\n version: 0.1.0\n repository: "file://../absent"\n'
|
||||
)
|
||||
(app / "Chart.lock").write_text(
|
||||
'dependencies:\n- name: absent\n repository: "file://../absent"\n version: 0.1.0\n'
|
||||
'digest: sha256:0\ngenerated: "2026-01-01T00:00:00Z"\n'
|
||||
)
|
||||
|
||||
assert lint_charts.main([]) == 1
|
||||
|
||||
def test_every_variant_of_a_chart_is_linted(self, lint_charts, tmp_path):
|
||||
"""A chart is only as good as its worst supported value combination."""
|
||||
chart = write_chart(tmp_path, "app", template=VALID_TEMPLATE)
|
||||
(chart / "values.schema.json").write_text(
|
||||
'{"type": "object", "properties": {"size": {"enum": ["small", "large"]}}}'
|
||||
)
|
||||
lint_charts.VARIANTS["app"] = [["--set", "size=huge"]]
|
||||
|
||||
assert lint_charts.main([]) == 1
|
||||
|
||||
def test_dependencies_are_vendored_before_linting(self, lint_charts, tmp_path):
|
||||
"""The vendored copy is gitignored, so a fresh clone has a lock file and nothing else."""
|
||||
write_chart(tmp_path, "lib", library=True)
|
||||
(tmp_path / "lib" / "templates" / "_helpers.tpl").write_text('{{- define "lib.name" -}}lib{{- end }}\n')
|
||||
app = write_chart(tmp_path, "app", template=VALID_TEMPLATE)
|
||||
(app / "Chart.yaml").write_text(
|
||||
"apiVersion: v2\nname: app\nversion: 0.1.0\n"
|
||||
'dependencies:\n - name: lib\n version: 0.1.0\n repository: "file://../lib"\n'
|
||||
)
|
||||
subprocess.run(["helm", "dependency", "update", str(app)], capture_output=True, check=True)
|
||||
shutil.rmtree(app / "charts")
|
||||
|
||||
assert (app / "Chart.lock").exists()
|
||||
assert lint_charts.main([]) == 0
|
||||
assert list((app / "charts").glob("lib-*.tgz"))
|
||||
|
||||
|
||||
class TestMissingHelm:
|
||||
def test_it_skips_when_helm_is_absent_locally(self, lint_charts, tmp_path, monkeypatch):
|
||||
"""Contributors without helm should not be blocked by a hook that runs on every commit."""
|
||||
write_chart(tmp_path, "app", template=BROKEN_TEMPLATE)
|
||||
monkeypatch.setattr(lint_charts.shutil, "which", lambda _: None)
|
||||
|
||||
assert lint_charts.main([]) == 0
|
||||
|
||||
def test_it_fails_when_ci_provides_no_helm(self, lint_charts, tmp_path, monkeypatch):
|
||||
"""In CI a missing helm means the charts went unchecked, which must not look like success."""
|
||||
write_chart(tmp_path, "app", template=VALID_TEMPLATE)
|
||||
monkeypatch.setattr(lint_charts.shutil, "which", lambda _: None)
|
||||
monkeypatch.setenv("CI", "true")
|
||||
|
||||
assert lint_charts.main([]) == 1
|
||||
@@ -0,0 +1,134 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
CHARTS_DIR = REPO_ROOT / "charts"
|
||||
|
||||
BASE_VALUES: dict[str, list[str]] = {
|
||||
"miles-workbench": ["--set", "objectName=lint-miles-workbench"],
|
||||
}
|
||||
|
||||
SHARED_INFRA_VARIANTS: list[list[str]] = [
|
||||
["--set", "infra.sharedStorage.type=pvc", "--set", "infra.sharedStorage.pvcClaimName=shared"],
|
||||
["--set", "infra.sharedStorage.type=none"],
|
||||
["--set", "infra.paths.repos.miles=alice/miles", "--set", "infra.paths.repos.megatron=alice/Megatron-LM"],
|
||||
]
|
||||
|
||||
VARIANTS: dict[str, list[list[str]]] = {
|
||||
"miles-workbench": [
|
||||
*SHARED_INFRA_VARIANTS,
|
||||
["--set", "rbac.create=false", "--set", "serviceAccount.name=preexisting"],
|
||||
["--set", "rbac.leaderWorkerSets=false"],
|
||||
],
|
||||
"miles-run": [
|
||||
*SHARED_INFRA_VARIANTS,
|
||||
["--set-json", 'run.orchestrator.command=["python","train.py"]'],
|
||||
[
|
||||
"--set-json",
|
||||
'run.staticWorkers=[{"name":"router","objectName":"lint-router",'
|
||||
'"command":["python","-m","router"],"ports":[{"name":"http","port":30000}]}]',
|
||||
],
|
||||
[
|
||||
"--set",
|
||||
"commandJob.enabled=true",
|
||||
"--set",
|
||||
"commandJob.name=convert",
|
||||
"--set",
|
||||
"commandJob.objectName=lint-convert",
|
||||
"--set-json",
|
||||
'commandJob.command=["bash"]',
|
||||
],
|
||||
[
|
||||
"--set-json",
|
||||
'run.inferenceEngines=[{"name":"engine","objectName":"lint-engine","replicas":2,"size":4,'
|
||||
'"command":["python","-m","sglang.launch_server"],"resources":{"limits":{"nvidia.com/gpu":8}}}]',
|
||||
],
|
||||
[
|
||||
"--set-json",
|
||||
'run.inferenceEngines=[{"name":"prefill","objectName":"lint-prefill","replicas":1,"size":4,"command":["python"]},'
|
||||
'{"name":"decode","objectName":"lint-decode","replicas":8,"command":["python"]}]',
|
||||
],
|
||||
[
|
||||
"--set-json",
|
||||
'run.trainerEngines=[{"name":"trainer-engine-actor","objectName":"lint-trainer-engine-actor","replicas":2,"size":2,'
|
||||
'"command":["python","-m","supervisor"]},'
|
||||
'{"name":"trainer-engine-critic","objectName":"lint-trainer-engine-critic","command":["python","-m","supervisor"]}]',
|
||||
],
|
||||
[
|
||||
"--set-json",
|
||||
'run.colocate={"namespace":"lint","release":"lint","trainer_pool_id":"lint-t",'
|
||||
'"inference_pools":[{"pool_id":"lint-e","layout":{"num_inference_cells":2,"num_trainer_cells":2,'
|
||||
'"num_pods_per_inference_cell":1,"num_pods_per_trainer_cell":2,"num_gpus_per_node":8,"gpu_offset":0}},'
|
||||
'{"pool_id":"lint-f","layout":{"num_inference_cells":2,"num_trainer_cells":2,'
|
||||
'"num_pods_per_inference_cell":1,"num_pods_per_trainer_cell":2,"num_gpus_per_node":8,'
|
||||
'"gpu_offset":16}}]}',
|
||||
"--set-json",
|
||||
'run.inferenceEngines=[{"name":"e","objectName":"lint-e","replicas":2,"command":["python"]},'
|
||||
'{"name":"f","objectName":"lint-f","replicas":2,"command":["python"]}]',
|
||||
"--set-json",
|
||||
'run.trainerEngines=[{"name":"t","objectName":"lint-t","replicas":2,"size":2,"command":["python"]}]',
|
||||
],
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
REJECTED_VARIANTS: dict[str, list[list[str]]] = {
|
||||
"miles-workbench": [["--set", "infra.env.PYTHONPATH=/somewhere"]],
|
||||
"miles-run": [["--set", "infra.env.PYTHONPATH=/somewhere"]],
|
||||
}
|
||||
|
||||
|
||||
def run(command: list[str]) -> subprocess.CompletedProcess:
|
||||
print("+ " + " ".join(command), file=sys.stderr)
|
||||
return subprocess.run(command, capture_output=True, text=True)
|
||||
|
||||
|
||||
def all_charts() -> list[Path]:
|
||||
return sorted(chart_yaml.parent for chart_yaml in CHARTS_DIR.glob("*/Chart.yaml"))
|
||||
|
||||
|
||||
def lint_chart(chart: Path) -> bool:
|
||||
if (chart / "Chart.lock").exists():
|
||||
built = run(["helm", "dependency", "build", str(chart)])
|
||||
if built.returncode != 0:
|
||||
print(built.stdout + built.stderr, file=sys.stderr)
|
||||
return False
|
||||
|
||||
ok = True
|
||||
base = BASE_VALUES.get(chart.name, [])
|
||||
for extra in [[], *VARIANTS.get(chart.name, [])]:
|
||||
result = run(["helm", "lint", str(chart), *base, *extra])
|
||||
if result.returncode != 0:
|
||||
print(result.stdout + result.stderr, file=sys.stderr)
|
||||
ok = False
|
||||
for extra in REJECTED_VARIANTS.get(chart.name, []):
|
||||
result = run(["helm", "lint", str(chart), *extra])
|
||||
if result.returncode == 0:
|
||||
print(f"{chart.name} accepted values it must refuse: {extra}", file=sys.stderr)
|
||||
ok = False
|
||||
return ok
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
argparse.ArgumentParser(description="helm lint every chart under charts/").parse_args(argv)
|
||||
|
||||
if shutil.which("helm") is None:
|
||||
message = "helm is not installed"
|
||||
if os.environ.get("CI"):
|
||||
print(f"{message}; CI must provide it", file=sys.stderr)
|
||||
return 1
|
||||
print(f"{message}; skipping chart lint", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
return 0 if all([lint_chart(chart) for chart in all_charts()]) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user