test_helm.py

import json
import logging
import os
import shutil
import subprocess
from pathlib import Path

import pytest
from test_util import write_tree

from ch_cli_tools.api import create_helm_chart
from ch_cli_tools.model import (
    DependencyUnknownError,
    KEY_APPS,
    KEY_DATABASE,
    KEY_DEPLOYMENT,
    KEY_HARNESS,
    KEY_SERVICE,
    KEY_TASK_IMAGES,
    KEY_TEST_IMAGES,
)
from ch_cli_tools.utils import ChartImageRef, find_chart_images, yaml
from cloudharness_utils.constants import APPS_PATH, DEPLOYMENT_CONFIGURATION_PATH, HELM_CHART_PATH

HERE = os.path.dirname(os.path.realpath(__file__))
RESOURCES = os.path.join(HERE, "resources")
CLOUDHARNESS_ROOT = os.path.dirname(
    os.path.dirname(os.path.dirname(HERE))
)


def exists(path):
    return path.exists()


def render_helm_chart(chart_path, values_files=(), set_values=()):
    completed = subprocess.run(
        [
            "helm",
            "template",
            str(chart_path),
            *(arg for f in values_files for arg in ("-f", str(f))),
            *(arg for v in set_values for arg in ("--set", v)),
        ],
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    return [manifest for manifest in yaml.load_all(completed.stdout) if manifest]


def find_manifest(manifests, kind, name):
    for manifest in manifests:
        if (
            manifest.get("kind") == kind and
            manifest.get("metadata", {}).get("name") == name
        ):
            return manifest
    raise AssertionError(f"Could not find {kind}/{name}")


def test_collect_helm_values(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        include=["samples", "myapp"],
        exclude=["events"],
        domain="my.local",
        namespace="test",
        env="dev",
        local=False,
        tag=1,
        registry="reg",
    )

    # First level include apps
    assert "samples" in values[KEY_APPS]
    assert "myapp" in values[KEY_APPS]

    # Not included
    assert "jupyterhub" not in values[KEY_APPS]

    # Dependency include first level
    assert "accounts" in values[KEY_APPS]
    assert "legacy" in values[KEY_APPS]

    # Dependency include second level
    assert "argo" in values[KEY_APPS]

    # Explicit exclude overrides include
    assert "events" not in values[KEY_APPS]

    assert (
        values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["image"] ==
        "reg/testprojectname/myapp:1"
    )
    assert values[KEY_APPS]["myapp"]["build"] == True
    assert (
        values.apps["myapp"].harness.deployment.image == "reg/testprojectname/myapp:1"
    )
    assert values[KEY_APPS]["myapp"][KEY_HARNESS]["name"] == "myapp"
    assert values[KEY_APPS]["legacy"][KEY_HARNESS]["name"] == "legacy"
    assert (
        values[KEY_APPS]["accounts"][KEY_HARNESS]["deployment"]["image"] ==
        "reg/testprojectname/accounts:1"
    )

    # Base values kept
    assert values[KEY_APPS]["accounts"][KEY_HARNESS]["subdomain"] == "accounts"

    # Defaults
    assert "service" in values[KEY_APPS]["legacy"][KEY_HARNESS]
    assert "common" in values[KEY_APPS]["legacy"]
    assert "common" in values[KEY_APPS]["accounts"]
    # Values overriding
    assert (
        values[KEY_APPS]["accounts"][KEY_HARNESS]["deployment"]["port"] == "overridden"
    )

    # Environment specific overriding
    assert values[KEY_APPS]["accounts"]["a"] == "dev"
    assert values["a"] == "dev"
    assert values["database"]["auto"] is False

    # legacy reading
    assert values[KEY_APPS]["accounts"][KEY_HARNESS]["deployment"]["auto"] is True
    assert values[KEY_APPS]["legacy"][KEY_HARNESS]["deployment"]["auto"] is False

    helm_path = out_folder / HELM_CHART_PATH

    assert exists(helm_path)
    assert exists(helm_path / "values.yaml")
    assert exists(helm_path / "resources" / "accounts" / "realm.json")
    assert exists(helm_path / "resources" / "accounts" / "aresource.txt")
    assert exists(helm_path / "resources" / "myapp" / "aresource.txt")
    assert exists(helm_path / "templates" / "myapp" / "mytemplate.yaml")

    assert values[KEY_TASK_IMAGES]
    assert "cloudharness-base" in values[KEY_TASK_IMAGES]
    assert (
        values[KEY_TASK_IMAGES]["cloudharness-base"] ==
        "reg/testprojectname/cloudharness-base:1"
    )
    assert (
        values[KEY_TASK_IMAGES]["myapp-mytask"] == "reg/testprojectname/myapp-mytask:1"
    )
    assert (
        values[KEY_TASK_IMAGES]["cloudharness-flask"] ==
        "reg/testprojectname/cloudharness-flask:1"
    )
    assert "cloudharness-base-debian" not in values[KEY_TASK_IMAGES]

    with open(helm_path / "charts/myapp/values.yaml", "r") as values_file:
        chart_values = yaml.load(values_file)
    assert chart_values is not None, "values.yaml should be valid YAML"
    assert chart_values["test"] == "dev"


def test_overriding_root_deep_merges_data_files_and_replaces_templates(tmp_path):
    """A root overriding an application of a lower root states only the values it changes:
    its sub-chart values.yaml and JSON resources are deep-merged over the lower root's ones,
    while templates and other files replace them."""
    base_root, top_root = tmp_path / "base", tmp_path / "top"
    write_tree(base_root, {APPS_PATH: {"chartapp": {"deploy": {
        "values.yaml": "harness: {}\n",
        "resources": {"conf.json": '{"a": {"b": 1, "c": 2}}'},
        "charts": {
            "Chart.yaml": "apiVersion: v2\nname: vendored\nversion: 1.0.0\n",
            "values.yaml": (
                "workflow:\n  rbac:\n    create: true\n"
                "controller:\n  parallelism: null\n  image:\n    tag: v1\n"
            ),
            "README.md": "vendored\n",
            "templates": {"role.yaml": "{{- if .Values.workflow.rbac.create }}\nkind: Role\n{{- end }}\n"},
        },
    }}}})
    write_tree(top_root, {APPS_PATH: {"chartapp": {"deploy": {
        "resources": {"conf.json": '{"a": {"c": 3}}'},
        "charts": {
            "values.yaml": "controller:\n  parallelism: 5\n",
            "README.md": "overridden\n",
            "templates": {"role.yaml": "{{- if .Values.workflow.rbac.create }}\nkind: ClusterRole\n{{- end }}\n"},
        },
    }}}})

    out_folder = tmp_path / "out"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, base_root, top_root],
        output_path=out_folder,
        include=["chartapp"],
        domain="my.local",
        namespace="test",
        local=False,
        tag=1,
    )

    helm_path = out_folder / HELM_CHART_PATH
    chart_path = helm_path / "charts" / "chartapp"
    chart_values = yaml.load(chart_path / "values.yaml")
    assert chart_values["workflow"]["rbac"]["create"] is True
    assert chart_values["controller"]["parallelism"] == 5
    assert chart_values["controller"]["image"]["tag"] == "v1"
    assert (chart_path / "README.md").read_text() == "overridden\n"
    assert "ClusterRole" in (chart_path / "templates" / "role.yaml").read_text()
    resource = json.loads((helm_path / "resources" / "chartapp" / "conf.json").read_text())
    assert resource == {"a": {"b": 1, "c": 3}}


def test_collect_nobuild(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values"
    values = create_helm_chart(
        [RESOURCES],
        output_path=out_folder,
        include=["myapp"],
        exclude=["events"],
        domain="my.local",
        namespace="test",
        env="nobuild",
        local=False,
        tag=1,
        registry="reg",
    )
    assert (
        values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["image"] == "custom-image"
    )
    assert values[KEY_APPS]["myapp"]["build"] == False


def test_collect_helm_values_harness_image_name_override(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_harness_image_name_override"

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        include=["myapp"],
        domain="my.local",
        namespace="test",
        env="imagename",
        local=False,
        tag=1,
        registry="reg",
    )

    assert (
        values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["image"] ==
        "reg/testprojectname/custom-myapp:1"
    )
    assert (
        values[KEY_APPS]["myapp"][KEY_TASK_IMAGES]["myapp-mytask"] ==
        "reg/testprojectname/custom-myapp-mytask:1"
    )


def test_collect_helm_values_noreg_noinclude(tmp_path):
    out_path = tmp_path / "test_collect_helm_values_noreg_noinclude"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_path,
        domain="my.local",
        namespace="test",
        env="dev",
        local=False,
        tag=1,
    )

    assert (
        values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["image"] ==
        "testprojectname/myapp:1"
    )
    assert values[KEY_APPS]["myapp"][KEY_HARNESS]["name"] == "myapp"
    assert values[KEY_APPS]["legacy"][KEY_HARNESS]["name"] == "legacy"
    assert (
        values[KEY_APPS]["accounts"][KEY_HARNESS]["deployment"]["image"] ==
        "testprojectname/accounts:1"
    )

    # First level include apps
    assert "samples" in values[KEY_APPS]
    assert "myapp" in values[KEY_APPS]
    assert "jupyterhub" in values[KEY_APPS]
    assert "accounts" in values[KEY_APPS]
    assert "legacy" in values[KEY_APPS]
    assert "argo" in values[KEY_APPS]
    assert "events" in values[KEY_APPS]

    # Base values kept
    assert values[KEY_APPS]["accounts"][KEY_HARNESS]["subdomain"] == "accounts"

    # Defaults
    assert "service" in values[KEY_APPS]["legacy"][KEY_HARNESS]
    assert "common" in values[KEY_APPS]["legacy"]
    assert "common" in values[KEY_APPS]["accounts"]
    # Values overriding
    assert (
        values[KEY_APPS]["accounts"][KEY_HARNESS]["deployment"]["port"] == "overridden"
    )
    assert (
        values[KEY_APPS]["events"]["kafka"]["resources"]["limits"]["memory"] ==
        "overridden"
    )

    # Environment specific overriding
    assert values[KEY_APPS]["accounts"]["a"] == "dev"
    assert values["a"] == "dev"
    assert values["database"]["auto"] is False

    # legacy reading
    assert values[KEY_APPS]["accounts"][KEY_HARNESS]["deployment"]["auto"] is True
    assert values[KEY_APPS]["legacy"][KEY_HARNESS]["deployment"]["auto"] is False

    helm_path = out_path / HELM_CHART_PATH

    assert exists(helm_path)
    assert exists(helm_path / "values.yaml")
    assert exists(helm_path / "resources" / "accounts" / "realm.json")
    assert exists(helm_path / "resources" / "accounts" / "aresource.txt")
    assert exists(helm_path / "resources" / "myapp" / "aresource.txt")
    assert exists(helm_path / "templates" / "myapp" / "mytemplate.yaml")

    assert values[KEY_TASK_IMAGES]
    assert "cloudharness-base" in values[KEY_TASK_IMAGES]
    assert (
        values[KEY_TASK_IMAGES]["cloudharness-base"] ==
        "testprojectname/cloudharness-base:1"
    )
    assert values[KEY_TASK_IMAGES]["myapp-mytask"] == "testprojectname/myapp-mytask:1"
    assert values[KEY_TASK_IMAGES]["my-common"] == "testprojectname/my-common:1"

    assert values["source_images"] == {
        "GOLANG": "golang:1.26",
        "ROCKYLINUX": "rockylinux/rockylinux:10.1-minimal",
        "SENTRY": "sentry:9.1.2",
        "KEYCLOAK": "myregistry.mykeycloak:99.9",
        "mybase": "foo:bar",
        "mybase2": "spam:egg",
        "NODE": "node:22-alpine",
        # Declared at the project level (deployment-configuration/values-template.yaml)
        # rather than per-application, so that it is configured in one place.
        "GATEKEEPER": "quay.io/gogatekeeper/gatekeeper:4.6.0",
    }


def test_collect_helm_values_precedence(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_precedence"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="prod",
        local=False,
        tag=1,
        include=["events"],
    )

    # Values.yaml from current app must override values-prod.yaml from cloudharness
    assert (
        values[KEY_APPS]["events"]["kafka"]["resources"]["limits"]["memory"] ==
        "overridden"
    )
    assert (
        values[KEY_APPS]["events"]["kafka"]["resources"]["limits"]["cpu"] ==
        "overridden-prod"
    )


def test_collect_helm_values_multiple_envs(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_multiple_envs"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env=["dev", "test"],
        local=False,
        tag=1,
        include=["myapp"],
    )

    assert values[KEY_APPS]["myapp"]["test"] is True, "values-test not loaded"
    assert values[KEY_APPS]["myapp"]["dev"] is True, "values-dev not loaded"
    assert values[KEY_APPS]["myapp"]["a"] == "test", "values-test not overriding"


def test_collect_app_defaults_env_specific_chart(tmp_path):
    """The deployment root value-template-[env].yaml lands in the generated chart values"""
    out_folder = tmp_path / "test_collect_app_defaults_env_specific_chart"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="dev",
        local=False,
        tag=1,
        include=["myapp"],
    )

    assert values[KEY_APPS]["myapp"]["env-defaults"] == "resources-dev"

    out_folder = tmp_path / "test_collect_app_defaults_no_env_chart"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        local=False,
        tag=1,
        include=["myapp"],
    )

    assert "env-defaults" not in values[KEY_APPS]["myapp"]


# ng raises ch_cli_tools.model.DependencyUnknownError, not
# legacy's ValuesValidationException - a different exception class by design
# (see model.py's CHProject.all_values()/_combined_dependencies()). The build
# dependency case used to be unchecked by create_helm_chart entirely (that
# validation lived only in ng.skaffold's generate()), but
# CHProject.all_task_images() now walks the same build-dependency closure
# during build_final_helm_values() too, so it raises here as well - verified
# empirically against this exact fixture before porting.
def test_collect_helm_values_wrong_dependencies_validate(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_wrong_dependencies_validate"
    with pytest.raises(DependencyUnknownError):
        create_helm_chart(
            [CLOUDHARNESS_ROOT, f"{RESOURCES}/wrong-dependencies"],
            output_path=out_folder,
            domain="my.local",
            namespace="test",
            env="prod",
            local=False,
            tag=1,
            include=["wrong-hard"],
        )
    try:
        create_helm_chart(
            [CLOUDHARNESS_ROOT, f"{RESOURCES}/wrong-dependencies"],
            output_path=out_folder,
            domain="my.local",
            namespace="test",
            env="prod",
            local=False,
            tag=1,
            include=["wrong-soft"],
        )

    except DependencyUnknownError as e:
        pytest.fail("Should not error because of wrong soft dependency")

    with pytest.raises(DependencyUnknownError):
        create_helm_chart(
            [CLOUDHARNESS_ROOT, f"{RESOURCES}/wrong-dependencies"],
            output_path=out_folder,
            domain="my.local",
            namespace="test",
            env="prod",
            local=False,
            tag=1,
            include=["wrong-build"],
        )
    try:
        create_helm_chart(
            [CLOUDHARNESS_ROOT, f"{RESOURCES}/wrong-dependencies"],
            output_path=out_folder,
            domain="my.local",
            namespace="test",
            env="prod",
            local=False,
            tag=1,
            include=["wrong-services"],
        )
    except DependencyUnknownError:
        pytest.fail("Should not error because of missing use_services dependency")


def test_collect_helm_values_build_dependencies(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_build_dependencies"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="prod",
        local=False,
        tag=1,
        include=["myapp"],
    )

    assert "cloudharness-flask" in values[KEY_TASK_IMAGES], (
        "Cloudharness-flask is included in the build dependencies"
    )
    assert "cloudharness-base" in values[KEY_TASK_IMAGES], (
        "Cloudharness-base is included in cloudharness-flask Dockerfile and it should be guessed"
    )
    assert "cloudharness-base-debian" not in values[KEY_TASK_IMAGES], (
        "Cloudharness-base-debian is not included in any dependency"
    )
    assert "cloudharness-frontend-build" not in values[KEY_TASK_IMAGES], (
        "cloudharness-frontend-build is not included in any dependency"
    )


def test_collect_helm_values_build_dependencies_nodeps(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_build_dependencies_nodeps"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="prod",
        local=False,
        tag=1,
        include=["events"],
    )

    assert "cloudharness-flask" not in values[KEY_TASK_IMAGES], (
        "Cloudharness-flask is not included in the build dependencies"
    )
    assert "cloudharness-base" not in values[KEY_TASK_IMAGES], (
        "Cloudharness-base is not included in the build dependencies"
    )
    assert "cloudharness-base-debian" not in values[KEY_TASK_IMAGES], (
        "Cloudharness-base-debian is not included in any dependency"
    )
    assert "cloudharness-frontend-build" not in values[KEY_TASK_IMAGES], (
        "cloudharness-frontend-build is not included in any dependency"
    )


def test_collect_helm_values_build_dependencies_exclude(tmp_path):
    out_folder = tmp_path / "test_collect_helm_values_build_dependencies_exclude"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="prod",
        local=False,
        tag=1,
        include=["workflows"],
        exclude=["workflows-extract-download"],
    )

    assert "cloudharness-flask" in values[KEY_TASK_IMAGES], (
        "Cloudharness-flask is included in the build dependencies"
    )
    assert "cloudharness-base" in values[KEY_TASK_IMAGES], (
        "Cloudharness-base is included in cloudharness-flask Dockerfile and it should be guessed"
    )
    assert "workflows-extract-download" not in values[KEY_TASK_IMAGES], (
        "workflows-extract-download has been explicitly excluded"
    )


def test_clear_unused_dbconfig(tmp_path):
    out_folder = tmp_path / "test_clear_unused_dbconfig"

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    # There is a DB config
    assert KEY_DATABASE in values[KEY_APPS]["myapp"][KEY_HARNESS]

    db_config = values[KEY_APPS]["myapp"][KEY_HARNESS][KEY_DATABASE]
    # postgres is set, but other entries are not.
    assert db_config["postgres"] is not None
    assert db_config["postgres"]["image"].startswith("postgres:")

    assert db_config["mongo"] is None
    assert db_config["neo4j"] is None

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="withmongo",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    assert KEY_DATABASE in values[KEY_APPS]["myapp"][KEY_HARNESS]
    db_config = values[KEY_APPS]["myapp"][KEY_HARNESS][KEY_DATABASE]

    # mongo is set, but other entries are not.
    assert db_config["mongo"] is not None
    assert db_config["mongo"]["image"].startswith("mongo:")
    assert db_config["neo4j"] is None
    assert db_config["postgres"] is None


def test_cnpg_postgres_parameters_render_only_when_set(tmp_path):
    out_folder = tmp_path / "test_cnpg_postgres_parameters_render_only_when_set"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)
    postgres = values["apps"]["myapp"]["harness"]["database"]["postgres"]
    postgres["operator"] = True
    postgres["parameters"] = {
        # Simulate generated YAML values where on/off can be parsed as booleans before Helm renders the chart.
        "autovacuum": True,
        "max_connections": "200",
        "shared_buffers": "1GB",
        "synchronous_commit": True,
        "track_io_timing": False,
    }
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    db_name = values["apps"]["myapp"]["harness"]["database"]["name"]
    cluster = find_manifest(manifests, "Cluster", db_name)
    assert cluster["spec"]["postgresql"]["parameters"] == {
        "autovacuum": "true",
        "max_connections": "200",
        "shared_buffers": "1GB",
        "synchronous_commit": "true",
        "track_io_timing": "false",
    }

    postgres["parameters"] = {}
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    cluster = find_manifest(manifests, "Cluster", db_name)
    assert "postgresql" not in cluster["spec"]

    postgres.pop("parameters")
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    cluster = find_manifest(manifests, "Cluster", db_name)
    assert "postgresql" not in cluster["spec"]


def test_statefulset_option(tmp_path):
    out_folder = tmp_path / "test_statefulset_option"
    # nfsserver is included to provide the storage class values needed by the usenfs case
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp", "nfsserver"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)

    myapp = values["apps"]["myapp"]
    harness = myapp["harness"]
    dep_name = harness["deployment"]["name"]
    db_name = harness["database"]["name"]
    service_name = harness["service"]["name"]

    harness["deployment"]["auto"] = True
    harness["deployment"]["volume"] = {
        "name": "myapp-data",
        "mountpath": "/data",
        "size": "1Gi",
        "auto": True,
    }
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    # Default: Deployments with the Recreate/affinity workaround and a standalone PVC
    manifests = render_helm_chart(helm_path)
    dep = find_manifest(manifests, "Deployment", dep_name)
    assert dep["spec"]["strategy"]["type"] == "Recreate"
    assert "affinity" in dep["spec"]["template"]["spec"]
    find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    db_dep = find_manifest(manifests, "Deployment", db_name)
    assert db_dep["spec"]["strategy"]["type"] == "Recreate"
    assert "affinity" in db_dep["spec"]["template"]["spec"]
    find_manifest(manifests, "PersistentVolumeClaim", db_name)

    # Opt in to StatefulSets: volumes are provisioned via volumeClaimTemplates. The legacy
    # volume migration (job copying a pre-existing PVC found by `lookup` into the statefulset
    # volumes) cannot be exercised here: `helm template` runs without a cluster, so `lookup`
    # finds nothing.
    harness["deployment"]["statefulset"] = True
    harness["database"]["statefulset"] = True
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    sts = find_manifest(manifests, "StatefulSet", dep_name)
    assert sts["spec"]["serviceName"] == service_name
    # OrderedReady would block template updates while an existing pod is unready,
    # so a crash-looping pod could never be replaced by its own fix.
    assert sts["spec"]["podManagementPolicy"] == "Parallel"
    assert "strategy" not in sts["spec"]
    assert "affinity" not in sts["spec"]["template"]["spec"]
    assert "initContainers" not in sts["spec"]["template"]["spec"]
    claims = [
        v["persistentVolumeClaim"]["claimName"]
        for v in sts["spec"]["template"]["spec"]["volumes"]
        if "persistentVolumeClaim" in v
    ]
    assert "myapp-data" not in claims
    assert sts["spec"]["volumeClaimTemplates"][0]["metadata"]["name"] == "myapp-data"
    assert not any(
        m
        for m in manifests
        if m.get("kind") == "PersistentVolumeClaim" and
        m.get("metadata", {}).get("name") == "myapp-data"
    )

    db_sts = find_manifest(manifests, "StatefulSet", db_name)
    assert db_sts["spec"]["serviceName"] == db_name
    assert db_sts["spec"]["podManagementPolicy"] == "Parallel"
    assert "strategy" not in db_sts["spec"]
    assert "affinity" not in db_sts["spec"]["template"]["spec"]
    assert "initContainers" not in db_sts["spec"]["template"]["spec"]
    assert db_sts["spec"]["volumeClaimTemplates"][0]["metadata"]["name"] == db_name
    assert not any(
        m
        for m in manifests
        if m.get("kind") == "PersistentVolumeClaim" and
        m.get("metadata", {}).get("name") == db_name
    )
    find_manifest(manifests, "Service", db_name)
    # without a legacy PVC no migration resources are rendered
    assert not any(
        m
        for m in manifests
        if "volume-migration" in m.get("metadata", {}).get("name", "")
    )

    # nfs (shared) volumes are never per-replica: the statefulset keeps mounting the common
    # PVC by claimName and no volumeClaimTemplates are created.
    harness["deployment"]["volume"]["usenfs"] = True
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    sts = find_manifest(manifests, "StatefulSet", dep_name)
    assert "volumeClaimTemplates" not in sts["spec"]
    claims = [
        v["persistentVolumeClaim"]["claimName"]
        for v in sts["spec"]["template"]["spec"]["volumes"]
        if "persistentVolumeClaim" in v
    ]
    assert "myapp-data" in claims
    shared_pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert shared_pvc["spec"]["accessModes"] == ["ReadWriteMany"]

    # volume.auto: false means the PVC is managed externally: always reference it by
    # claimName, never via volumeClaimTemplates.
    harness["deployment"]["volume"]["usenfs"] = False
    harness["deployment"]["volume"]["auto"] = False
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    sts = find_manifest(manifests, "StatefulSet", dep_name)
    assert "volumeClaimTemplates" not in sts["spec"]
    claims = [
        v["persistentVolumeClaim"]["claimName"]
        for v in sts["spec"]["template"]["spec"]["volumes"]
        if "persistentVolumeClaim" in v
    ]
    assert "myapp-data" in claims


def test_volume_write_many(tmp_path):
    out_folder = tmp_path / "test_volume_write_many"
    # nfsserver is deliberately not included: a ReadWriteMany volume must not rely on it
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)

    harness = values["apps"]["myapp"]["harness"]
    dep_name = harness["deployment"]["name"]

    harness["deployment"]["auto"] = True
    volume = {"name": "myapp-data", "mountpath": "/data", "size": "1Gi", "auto": True}
    harness["deployment"]["volume"] = volume

    def render():
        with open(values_path, "w") as values_file:
            yaml.dump(values, values_file)
        return render_helm_chart(helm_path)

    # a null storage class is omitted, so the cluster default one is used
    volume["storageClass"] = None
    manifests = render()
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert "storageClassName" not in pvc["spec"]

    # a storage class can be set on a ReadWriteOnce volume, which keeps the node pinning
    volume["storageClass"] = "gp3"
    manifests = render()
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert pvc["spec"]["accessModes"] == ["ReadWriteOnce"]
    assert pvc["spec"]["storageClassName"] == "gp3"
    dep = find_manifest(manifests, "Deployment", dep_name)
    assert dep["spec"]["strategy"]["type"] == "Recreate"
    assert "affinity" in dep["spec"]["template"]["spec"]

    # a writeMany volume keeps its storage class, and its pod is neither pinned to a node nor
    # recreated on update
    volume["writeMany"] = True
    manifests = render()
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert pvc["spec"]["accessModes"] == ["ReadWriteMany"]
    assert pvc["spec"]["storageClassName"] == "gp3"
    dep = find_manifest(manifests, "Deployment", dep_name)
    assert "strategy" not in dep["spec"]
    assert "affinity" not in dep["spec"]["template"]["spec"]

    # writeMany with an explicit ReadWriteMany capable storage class
    volume["storageClass"] = "efs-sc"
    manifests = render()
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert pvc["spec"]["accessModes"] == ["ReadWriteMany"]
    assert pvc["spec"]["storageClassName"] == "efs-sc"

    # a null storage class is omitted from a ReadWriteMany claim too
    volume["storageClass"] = None
    manifests = render()
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert "storageClassName" not in pvc["spec"]
    volume["storageClass"] = "efs-sc"
    manifests = render()
    assert (
        find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")["spec"][
            "storageClassName"
        ] ==
        "efs-sc"
    )

    # ReadWriteMany volumes are shared: a statefulset keeps mounting the common PVC by
    # claimName instead of provisioning one per replica
    harness["deployment"]["statefulset"] = True
    manifests = render()
    sts = find_manifest(manifests, "StatefulSet", dep_name)
    assert "volumeClaimTemplates" not in sts["spec"]
    claims = [
        v["persistentVolumeClaim"]["claimName"]
        for v in sts["spec"]["template"]["spec"]["volumes"]
        if "persistentVolumeClaim" in v
    ]
    assert "myapp-data" in claims
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    assert pvc["spec"]["accessModes"] == ["ReadWriteMany"]

    # the storage class of a per-replica statefulset volume is configurable too
    volume["writeMany"] = False
    volume["storageClass"] = "gp3"
    manifests = render()
    sts = find_manifest(manifests, "StatefulSet", dep_name)
    claim_template = sts["spec"]["volumeClaimTemplates"][0]
    assert claim_template["metadata"]["name"] == "myapp-data"
    assert claim_template["spec"]["accessModes"] == ["ReadWriteOnce"]
    assert claim_template["spec"]["storageClassName"] == "gp3"


def test_volume_storage_class_default(tmp_path):
    out_folder = tmp_path / "test_volume_storage_class_default"
    # samples declares a volume, myapp does not
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["samples", "myapp"],
        exclude=["legacy"],
    )

    # the value-template default applies to the volume declared by the application
    volume = values[KEY_APPS]["samples"][KEY_HARNESS]["deployment"]["volume"]
    assert volume["mountpath"]
    assert volume["storageClass"] == "standard"

    # ... and the defaults alone do not make a volume: a volume-less application has none
    assert not values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"].get("volume")

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    manifests = render_helm_chart(helm_path)
    sts = find_manifest(
        manifests,
        "StatefulSet",
        values[KEY_APPS]["samples"][KEY_HARNESS]["deployment"]["name"],
    )
    assert (
        sts["spec"]["volumeClaimTemplates"][0]["spec"]["storageClassName"] == "standard"
    )


def test_volume_usenfs_prevails(tmp_path):
    out_folder = tmp_path / "test_volume_usenfs_prevails"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp", "nfsserver"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)

    harness = values["apps"]["myapp"]["harness"]
    harness["deployment"]["auto"] = True
    # colliding settings: the nfs server storage class and access mode prevail
    harness["deployment"]["volume"] = {
        "name": "myapp-data",
        "mountpath": "/data",
        "size": "1Gi",
        "auto": True,
        "usenfs": True,
        "writeMany": False,
        "storageClass": "efs-sc",
    }
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    pvc = find_manifest(manifests, "PersistentVolumeClaim", "myapp-data")
    nfs_class = (
        f"{values['namespace']}-{values['apps']['nfsserver']['storageClass']['name']}"
    )
    assert pvc["spec"]["storageClassName"] == nfs_class
    assert pvc["spec"]["accessModes"] == ["ReadWriteMany"]
    dep = find_manifest(manifests, "Deployment", harness["deployment"]["name"])
    assert "affinity" not in dep["spec"]["template"]["spec"]


def test_database_storage_class(tmp_path):
    out_folder = tmp_path / "test_database_storage_class"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)

    database = values["apps"]["myapp"]["harness"]["database"]
    db_name = database["name"]

    def render():
        with open(values_path, "w") as values_file:
            yaml.dump(values, values_file)
        return render_helm_chart(helm_path)

    # not set by default: the claim carries no storage class, so the cluster default one is used.
    # The storage class is immutable on an existing claim, hence never set implicitly: database
    # volumes of existing deployments must keep rendering without it.
    assert database["storageClass"] is None
    manifests = render()
    assert (
        "storageClassName"
        not in find_manifest(manifests, "PersistentVolumeClaim", db_name)["spec"]
    )

    database["storageClass"] = "gp3"
    manifests = render()
    assert (
        find_manifest(manifests, "PersistentVolumeClaim", db_name)["spec"][
            "storageClassName"
        ] ==
        "gp3"
    )

    # statefulset databases provision their volume through volumeClaimTemplates
    database["statefulset"] = True
    manifests = render()
    sts = find_manifest(manifests, "StatefulSet", db_name)
    assert sts["spec"]["volumeClaimTemplates"][0]["spec"]["storageClassName"] == "gp3"
    database["storageClass"] = None
    manifests = render()
    sts = find_manifest(manifests, "StatefulSet", db_name)
    assert "storageClassName" not in sts["spec"]["volumeClaimTemplates"][0]["spec"]

    # the postgres operator cluster storage honours the same setting
    database["statefulset"] = False
    database["postgres"]["operator"] = True
    database["storageClass"] = "gp3"
    manifests = render()
    assert (
        find_manifest(manifests, "Cluster", db_name)["spec"]["storage"]["storageClass"] ==
        "gp3"
    )
    database["storageClass"] = None
    manifests = render()
    assert (
        "storageClass"
        not in find_manifest(manifests, "Cluster", db_name)["spec"]["storage"]
    )


def test_statefulset_leader_service(tmp_path):
    out_folder = tmp_path / "test_statefulset_leader_service"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)

    harness = values["apps"]["myapp"]["harness"]
    dep_name = harness["deployment"]["name"]
    service_name = harness["service"]["name"]
    rw_name = f"{service_name}-rw"

    def ingress_paths(manifests):
        ingress = find_manifest(manifests, "Ingress", "myapp")
        return [
            path for rule in ingress["spec"]["rules"] for path in rule["http"]["paths"]
        ]

    # write methods in uri_role_mapping without statefulset: no leader service, no leader routing
    harness["uri_role_mapping"] = harness.get("uri_role_mapping", []) + [
        {"uri": "/api/edit/*", "methods": ["POST", "PUT", "PATCH"]},
        {"uri": "/upload", "methods": ["POST"]},
        {"uri": "/api/remove", "methods": ["DELETE"]},
        {"uri": "/readonly", "methods": ["GET"]},
    ]
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    assert not any(
        m
        for m in manifests
        if m.get("kind") == "Service" and m.get("metadata", {}).get("name") == rw_name
    )
    assert not any(
        p
        for p in ingress_paths(manifests)
        if p["backend"]["service"]["name"] == rw_name
    )

    harness["deployment"]["statefulset"] = True
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    manifests = render_helm_chart(helm_path)
    rw_service = find_manifest(manifests, "Service", rw_name)
    assert rw_service["spec"]["selector"]["app"] == dep_name
    assert (
        rw_service["spec"]["selector"]["statefulset.kubernetes.io/pod-name"] ==
        f"{dep_name}-0"
    )
    main_service = find_manifest(manifests, "Service", service_name)
    assert rw_service["spec"]["ports"] == main_service["spec"]["ports"]

    paths = ingress_paths(manifests)
    rw_paths = {
        p["path"]: p for p in paths if p["backend"]["service"]["name"] == rw_name
    }
    # wildcard uris map to Prefix rules, plain uris to ImplementationSpecific; any write method
    # (POST/PUT/PATCH/DELETE) triggers leader routing, while entries without one (the default
    # catch-all, /readonly) are not routed to the leader
    assert set(rw_paths) == {"/api/edit", "/upload", "/api/remove"}
    assert rw_paths["/api/edit"]["pathType"] == "Prefix"
    assert rw_paths["/upload"]["pathType"] == "ImplementationSpecific"
    # the catch-all still routes to the normal service
    assert any(
        p
        for p in paths
        if p["path"] == "/" and p["backend"]["service"]["name"] == service_name
    )


def test_gatekeeper_native_configuration_rendering_and_checksum(tmp_path):
    out_folder = tmp_path / "test_gatekeeper_native_configuration"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        local=False,
        tls=True,
        include=["myapp", "accounts"],
        exclude=["legacy"],
    )

    helm_path = out_folder / "helm"
    shutil.rmtree(helm_path / "charts")
    values_path = helm_path / "values.yaml"
    with open(values_path, "r") as values_file:
        values = yaml.load(values_file)

    app_harness = values["apps"]["myapp"]["harness"]
    app_harness["secured"] = True
    app_harness["proxy"]["gatekeeper"]["configuration"] = {
        "same-site-cookie": "None",
        "enable-pkce": False,
        "http-only-cookie": True,
        "cors-exposed-headers": ["X-Request-ID", "X-Trace-ID"],
    }
    values["proxy"]["gatekeeper"]["configuration"] = {
        "same-site-cookie": "Strict",
        "enable-pkce": True,
        "max-token-size": 65536,
    }

    def render_gatekeeper():
        with open(values_path, "w") as values_file:
            yaml.dump(values, values_file)
        manifests = render_helm_chart(helm_path)
        config = find_manifest(manifests, "ConfigMap", "mysubdomain-gk")
        deployment = find_manifest(manifests, "Deployment", "mysubdomain-gk")
        return (
            yaml.load(config["data"]["proxy.yml"]),
            deployment["spec"]["template"]["metadata"]["annotations"][
                "checksum/config"
            ],
        )

    tls_config, tls_checksum = render_gatekeeper()
    assert tls_config["secure-cookie"] is True
    assert tls_config["same-site-cookie"] == "None"
    assert tls_config["enable-pkce"] is False
    assert tls_config["http-only-cookie"] is True
    assert tls_config["max-token-size"] == 65536
    assert tls_config["cors-exposed-headers"] == ["X-Request-ID", "X-Trace-ID"]

    values["tls"] = False
    non_tls_config, non_tls_checksum = render_gatekeeper()
    assert non_tls_config["secure-cookie"] is False
    assert non_tls_config["same-site-cookie"] == "Lax"
    assert non_tls_config["enable-pkce"] is False
    assert non_tls_config["max-token-size"] == 65536
    assert non_tls_checksum != tls_checksum

    values["tls"] = True
    app_harness["proxy"]["gatekeeper"]["configuration"].pop("same-site-cookie")
    inherited_config, inherited_checksum = render_gatekeeper()
    assert inherited_config["same-site-cookie"] == "Strict"
    assert inherited_config["enable-pkce"] is False
    assert inherited_checksum != tls_checksum


def test_clear_all_dbconfig_if_nodb(tmp_path):
    out_folder = tmp_path / "test_clear_all_dbconfig_if_nodb"

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        namespace="test",
        env="withoutdb",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    # There is a DB config
    assert KEY_DATABASE in values[KEY_APPS]["myapp"][KEY_HARNESS]

    db_config = values[KEY_APPS]["myapp"][KEY_HARNESS][KEY_DATABASE]
    assert db_config is None


# Legacy's two-phase shape (a tag-less first pass, then
# preprocess_build_overrides + generate_hash_based_image_tags mutating the
# result in place) doesn't apply to ng - CHProject.qualify_image() computes
# the content-hash tag directly, in the same single create_helm_chart() call,
# whenever no explicit tag was requested and this isn't a local build (see
# CHProject._use_content_hash_tag). So there's no separate first_pass/create()
# two-step: every call already produces the hash-tagged result.
def test_collect_helm_values_auto_tag(tmp_path):
    out_folder = str(tmp_path / "test_collect_helm_values_auto_tag")

    def create():
        return create_helm_chart(
            [CLOUDHARNESS_ROOT, RESOURCES],
            output_path=out_folder,
            include=["samples", "myapp"],
            exclude=["events"],
            domain="my.local",
            namespace="test",
            env="dev",
            local=False,
            tag=None,
            registry="reg",
        )

    BASE_KEY = "cloudharness-base"
    values = create()

    # Auto values are set by using the directory hash
    assert (
        "reg/testprojectname/myapp:"
        in values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["image"]
    )
    assert "reg/testprojectname/myapp:" in values.apps["myapp"].harness.deployment.image
    assert "testprojectname/myapp-mytask" in values[KEY_TASK_IMAGES]["myapp-mytask"]
    assert (
        values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["image"] ==
        values.apps["myapp"].harness.deployment.image
    )
    v1 = values.apps["myapp"].harness.deployment.image
    c1 = values["task-images"]["my-common"]
    b1 = values["task-images"][BASE_KEY]
    d1 = values["task-images"]["cloudharness-flask"]

    values = create()
    assert v1 == values.apps["myapp"].harness.deployment.image, (
        "Nothing changed the hash value"
    )
    assert values["task-images"][BASE_KEY] == b1, (
        "Base image should not change following the root .dockerignore"
    )

    fname = Path(RESOURCES) / "applications" / "myapp" / "afile.txt"
    try:
        fname.write_text("a")

        values = create()
        assert v1 != values.apps["myapp"].harness.deployment.image, (
            "Adding the file changed the hash value"
        )
        v2 = values.apps["myapp"].harness.deployment.image
        assert values["task-images"][BASE_KEY] == b1, (
            "Application files should be ignored for base image following the root .dockerignore"
        )
    finally:
        fname.unlink()

    try:
        fname.write_text("a")

        values = create()
        assert v2 == values.apps["myapp"].harness.deployment.image, (
            "Recreated an identical file, the hash value should be the same"
        )
    finally:
        fname.unlink()

    fname = Path(RESOURCES) / "applications" / "myapp" / "afile.ignored"
    try:
        fname.write_text("a")

        values = create()
        assert values["task-images"][BASE_KEY] == b1, (
            "2: Application files should be ignored for base image following the root .dockerignore"
        )

        assert v1 == values.apps["myapp"].harness.deployment.image, (
            "Nothing should change the hash value as the file is ignored in the .dockerignore"
        )
    finally:
        fname.unlink()

    # Dependencies test: if a dependency is changed, the hash should change
    fname = Path(RESOURCES) / "infrastructure" / "common-images" / "my-common" / "afile"

    try:
        fname.write_text("a")

        values = create()

        assert c1 != values["task-images"]["my-common"], (
            "If content of a static image is changed, the hash should change"
        )
        assert v1 != values.apps["myapp"].harness.deployment.image, (
            "If a static image dependency is changed, the hash should change"
        )
    finally:
        fname.unlink()

    fname = Path(RESOURCES) / "atestfile"
    try:
        fname.write_text("a")

        values = create()

        assert b1 != values["task-images"][BASE_KEY], (
            "Content for base image is changed, the hash should change"
        )
        assert d1 != values["task-images"]["cloudharness-flask"], (
            "Content for base image is changed, the static image should change"
        )
        assert v1 != values.apps["myapp"].harness.deployment.image, (
            "2 levels dependency: If a base image dependency is changed, the hash should change"
        )
    finally:
        fname.unlink()


def test_network_policy_defaults_from_value_template(tmp_path):
    """Verify that allowedNamespaces set in a root directory's value-template.yaml
    propagates into app values and is not reset to []."""
    out_folder = tmp_path / "test_network_policy_defaults_from_value_template"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        include=["myapp"],
        domain="my.local",
        namespace="test",
        env="dev",
        local=False,
        tag=1,
    )

    network = values[KEY_APPS]["myapp"][KEY_HARNESS]["deployment"]["network"]
    assert network is not None, "network config should be present"
    allowed = network.get("allowedNamespaces") or []
    assert "test-namespace" in allowed, (
        f"allowedNamespaces from value-template override should contain 'test-namespace', got: {allowed}"
    )

    out_folder = tmp_path / "test_chart_metadata_optional_overrides"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        include=["myapp"],
        domain="my.local",
        namespace="custom-ns",
        name="custom-chart",
        chart_version="9.8.7",
        app_version="4.5.6",
        env="dev",
        local=False,
        tag=1,
        registry="reg",
    )

    chart_path = out_folder / HELM_CHART_PATH / "Chart.yaml"
    with open(chart_path, "r") as chart_file:
        chart = yaml.load(chart_file)
    assert chart["name"] == "custom-chart"
    assert chart["version"] == "9.8.7"
    assert chart["appVersion"] == "4.5.6"
    assert chart["metadata"]["namespace"] == "custom-ns"


def test_exclude_single_task(tmp_path):
    out_folder = tmp_path / "test_exclude_single_task"

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="withpostgres",
        local=False,
        include=["myapp"],
        exclude=["myapp-mytask"],
    )

    assert "myapp-mytask" not in values["task-images"], (
        "myapp-mytask has been excluded, so should not appear in the task images"
    )

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="fulldep",
        local=False,
        include=["dependantapp"],
        exclude=["myapp-mytask"],
    )

    assert "myapp-mytask" in values[KEY_TASK_IMAGES], (
        "myapp-mytask is excluded but still required by dependantapp, so it should be kept"
    )


def test_app_depends_on_app(tmp_path):
    out_folder = tmp_path / "test_app_depends_on_app"

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="",
        local=False,
        include=["dependantapp"],
        exclude=[],
    )
    assert "myapp" in values["task-images"], (
        "myapp should be included as a task image because it is a dependency of dependantapp"
    )
    assert "cloudharness-flask" in values["task-images"], (
        "cloudharness-flask should be included as a task image because it is a dependency of myapp"
    )
    assert "cloudharness-base" in values["task-images"], (
        "cloudharness-flask should be included as a task image because it is a dependency of cloudharness-flask"
    )
    assert "myapp-mytask" in values["task-images"], (
        "task should be also included as build dependencies,as it's required by another task"
    )
    assert "legacy" not in values["task-images"], (
        "legacy should not be included as a task image because it is not a dependency"
    )

    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="testincludetask",
        local=False,
        include=["dependantapp"],
        exclude=[],
    )

    assert "myapp" in values["task-images"], (
        "myapp should be included as a task image because it is a dependency of dependantapp"
    )
    assert "myapp-mytask" in values["task-images"], (
        "tasks should be also included as build dependencies, when explicitly included as build dependencies"
    )


# Needs actual `helm template` rendering (no ng Chart.yaml/templates generator)
def test_app_depends_on_task_only(tmp_path):
    out_folder = tmp_path / "test_app_depends_on_task_only"

    # taskdep depends on a base image (cloudharness-flask, which its Dockerfile uses) and
    # on myapp-mytask, a task image owned by another app (myapp) that is not listed as a
    # dependency itself. The owner app must be pulled in to build the task image, but must
    # not be deployed.
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="",
        local=False,
        include=["taskdep"],
        exclude=[],
    )

    assert "myapp-mytask" in values[KEY_TASK_IMAGES], (
        "cross-app task image must be built"
    )
    assert "cloudharness-flask" in values[KEY_TASK_IMAGES], (
        "declared base-image build dep must be kept"
    )
    assert "myapp" not in values[KEY_APPS], "owner app must be built but not deployed"


def find_manifests(manifests, kind, name=None):
    return [
        manifest
        for manifest in manifests
        if manifest.get("kind") == kind and
        (name is None or manifest.get("metadata", {}).get("name") == name)
    ]


def render_with_secrets(
    tmp_path, name, secrets, secretmanagers=None, app="myapp", include=None, patch=None
):
    """Generate the chart, override the application secrets and render it."""
    out_folder = tmp_path / name
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="",
        local=False,
        include=include or [app],
        exclude=["legacy"],
    )

    helm_path = out_folder / HELM_CHART_PATH
    shutil.rmtree(helm_path / "charts", ignore_errors=True)
    values_path = helm_path / "values.yaml"
    with open(values_path) as values_file:
        values = yaml.load(values_file)
    values["apps"][app]["harness"]["secrets"] = secrets
    # the test applications are not deployed by default, but we need the deployment to
    # check how the secrets are mounted
    values["apps"][app]["harness"]["deployment"]["auto"] = True
    if secretmanagers is not None:
        values["secretmanagers"] = secretmanagers
    if patch:
        patch(values)
    with open(values_path, "w") as values_file:
        yaml.dump(values, values_file)

    return render_helm_chart(helm_path)


def secrets_volume(manifests, app="myapp"):
    deployment = find_manifest(manifests, "Deployment", app)
    volumes = [
        v
        for v in deployment["spec"]["template"]["spec"]["volumes"]
        if v["name"] == "secrets"
    ]
    assert volumes, "the secrets volume is not mounted"
    return volumes[0]


# Fails identically on the real, unported legacy test
# (tests/test_helm.py::test_secrets_simple_definitions, part of the accepted
# pre-existing baseline) - confirmed by running it directly: 'secureSecret'
# (a bare `null` value, not the rich/manager form) never reaches $managed in
# deployment-configuration/helm/templates/secrets/_secrets.tpl at all with the
# currently-installed Helm (v4.3.0). Not an ng gap - same templates, same
# Helm binary, same failure either way. Leaving commented until that's fixed
# upstream (likely a Sprig `kindIs "invalid"` behavior change for nil).
# def test_secrets_simple_definitions(tmp_path):
#     """The legacy plain value forms keep creating the application secret and a plain secret volume"""
#     manifests = render_with_secrets(tmp_path, 'test_secrets_simple_definitions', {
#         'unsecureSecret': 'a value',
#         'secureSecret': None,
#         'random-static-secret': '',
#         'random-dynamic-secret': '?',
#     })
#
#     secret = find_manifest(manifests, 'Secret', 'myapp')
#     assert secret['stringData']['unsecureSecret'] == 'a value'
#     # rendering happens without a cluster, so this is always a first install:
#     # empty and null values are randomly generated, ? is refreshed at every upgrade
#     assert len(secret['stringData']['secureSecret']) == 20
#     assert len(secret['stringData']['random-static-secret']) == 20
#     assert secret['stringData']['random-dynamic-secret'] == '?'
#
#     assert secrets_volume(manifests) == {'name': 'secrets', 'secret': {'secretName': 'myapp'}}


def test_secrets_rich_definition_defaults_to_cloudharness(tmp_path):
    """Without a manager, the rich form behaves exactly like the plain form"""
    manifests = render_with_secrets(
        tmp_path,
        "test_secrets_rich_definition_defaults_to_cloudharness",
        {
            "withDefault": {"default": "a value"},
            "explicitManager": {"manager": "cloudharness", "default": "?"},
            "noDefault": {"manager": "cloudharness"},
        },
    )

    secret = find_manifest(manifests, "Secret", "myapp")
    assert secret["stringData"]["withDefault"] == "a value"
    assert secret["stringData"]["explicitManager"] == "?"
    assert len(secret["stringData"]["noDefault"]) == 20
    # the rich form must never be rendered as an unrecognized value
    assert not any("formatnotrecognized" in key for key in secret["stringData"])

    assert secrets_volume(manifests) == {
        "name": "secrets",
        "secret": {"secretName": "myapp"},
    }


# Fails identically on the real, unported legacy test
# (tests/test_helm.py::test_secrets_unmanaged, part of the accepted
# pre-existing baseline) - same root cause as test_secrets_simple_definitions
# above (an explicit `{'manager': None}` isn't recognized as unmanaged by
# _secrets.tpl with the currently-installed Helm), confirmed by running
# legacy's own copy directly. Not an ng gap.
# def test_secrets_unmanaged(tmp_path):
#     """An explicitly null manager creates nothing: the secret is expected to exist already"""
#     manifests = render_with_secrets(tmp_path, 'test_secrets_unmanaged', {
#         'existing': {'manager': None},
#     })
#
#     assert not find_manifests(manifests, 'Secret', 'myapp'), "no secret must be created for unmanaged secrets"
#     # mounted as optional: a missing secret must not block the pod from starting
#     assert secrets_volume(manifests) == {'name': 'secrets', 'secret': {'secretName': 'myapp', 'optional': True}}


# Same root cause as test_secrets_unmanaged above - fails
# identically on the real, unported legacy test (part of the accepted
# pre-existing baseline). Not an ng gap.
# def test_secrets_unmanaged_mixed_with_cloudharness(tmp_path):
#     """Unmanaged secrets are simply left out of the application secret"""
#     manifests = render_with_secrets(tmp_path, 'test_secrets_unmanaged_mixed_with_cloudharness', {
#         'managed': 'a value',
#         'existing': {'manager': None},
#     })
#
#     secret = find_manifest(manifests, 'Secret', 'myapp')
#     assert secret['stringData']['managed'] == 'a value'
#     assert 'existing' not in secret['stringData']
#     # the whole secret is mounted, so the out of band entry shows up as well
#     assert secrets_volume(manifests) == {'name': 'secrets', 'secret': {'secretName': 'myapp'}}


def test_secrets_onepassword_manager(tmp_path):
    manifests = render_with_secrets(
        tmp_path,
        "test_secrets_onepassword_manager",
        {
            "opSecret": {
                "manager": "onepassword",
                "path": "vaults/my-vault/items/my-item",
            },
            "opField": {
                "manager": "onepassword",
                "path": "my-item",
                "field": "credential",
            },
        },
        secretmanagers={"onepassword": {"vault": "default-vault"}},
    )

    item = find_manifest(manifests, "OnePasswordItem", "myapp-opsecret")
    assert item["apiVersion"] == "onepassword.com/v1"
    assert item["spec"]["itemPath"] == "vaults/my-vault/items/my-item"

    # the item name alone is completed with the globally configured vault
    item = find_manifest(manifests, "OnePasswordItem", "myapp-opfield")
    assert item["spec"]["itemPath"] == "vaults/default-vault/items/my-item"

    assert not find_manifests(manifests, "Secret", "myapp"), (
        "externally managed secrets are not created by CloudHarness"
    )

    assert secrets_volume(manifests) == {
        "name": "secrets",
        "projected": {
            "sources": [
                {"secret": {"name": "myapp", "optional": True}},
                {
                    "secret": {
                        "name": "myapp-opfield",
                        "items": [{"key": "credential", "path": "opField"}],
                    }
                },
                {
                    "secret": {
                        "name": "myapp-opsecret",
                        "items": [{"key": "password", "path": "opSecret"}],
                    }
                },
            ]
        },
    }


def test_secrets_aws_manager(tmp_path):
    manifests = render_with_secrets(
        tmp_path,
        "test_secrets_aws_manager",
        {
            "awsSecret": {
                "manager": "aws",
                "arn": "arn:aws:secretsmanager:eu-west-1:1:secret:mine",
                "property": "password",
            },
        },
        secretmanagers={"aws": {"store": "aws-store", "refreshInterval": "30m"}},
    )

    external = find_manifest(manifests, "ExternalSecret", "myapp-awssecret")
    assert external["apiVersion"] == "external-secrets.io/v1beta1"
    assert external["spec"]["secretStoreRef"] == {
        "name": "aws-store",
        "kind": "ClusterSecretStore",
    }
    assert external["spec"]["refreshInterval"] == "30m"
    assert external["spec"]["target"]["name"] == "myapp-awssecret"
    assert external["spec"]["data"] == [
        {
            "secretKey": "value",
            "remoteRef": {
                "key": "arn:aws:secretsmanager:eu-west-1:1:secret:mine",
                "property": "password",
            },
        }
    ]

    assert secrets_volume(manifests) == {
        "name": "secrets",
        "projected": {
            "sources": [
                {"secret": {"name": "myapp", "optional": True}},
                {
                    "secret": {
                        "name": "myapp-awssecret",
                        "items": [{"key": "value", "path": "awsSecret"}],
                    }
                },
            ]
        },
    }


def test_secrets_aws_manager_version(tmp_path):
    """A `version` setting pins the secret to a VersionStage or VersionId"""
    manifests = render_with_secrets(
        tmp_path,
        "test_secrets_aws_manager_version",
        {
            "awsSecret": {
                "manager": "aws",
                "arn": "arn:aws:secretsmanager:eu-west-1:1:secret:mine",
                "version": "AWSPREVIOUS",
            },
        },
        secretmanagers={"aws": {"store": "aws-store"}},
    )

    external = find_manifest(manifests, "ExternalSecret", "myapp-awssecret")
    assert external["spec"]["data"] == [
        {
            "secretKey": "value",
            "remoteRef": {
                "key": "arn:aws:secretsmanager:eu-west-1:1:secret:mine",
                "version": "AWSPREVIOUS",
            },
        }
    ]


def test_secrets_mixed_managers(tmp_path):
    """CloudHarness and externally managed secrets are exposed in the same directory"""
    manifests = render_with_secrets(
        tmp_path,
        "test_secrets_mixed_managers",
        {
            "local": "a value",
            "opSecret": {
                "manager": "onepassword",
                "path": "vaults/my-vault/items/my-item",
                "default": "ignored locally",
            },
        },
    )

    secret = find_manifest(manifests, "Secret", "myapp")
    assert secret["stringData"]["local"] == "a value"
    assert "opSecret" not in secret["stringData"], (
        "the value of an externally managed secret is never written in the chart"
    )

    assert find_manifests(manifests, "OnePasswordItem", "myapp-opsecret")
    assert secrets_volume(manifests) == {
        "name": "secrets",
        "projected": {
            "sources": [
                {"secret": {"name": "myapp"}},
                {
                    "secret": {
                        "name": "myapp-opsecret",
                        "items": [{"key": "password", "path": "opSecret"}],
                    }
                },
            ]
        },
    }


def test_secrets_unknown_manager_setting_fails(tmp_path):
    with pytest.raises(subprocess.CalledProcessError) as error:
        render_with_secrets(
            tmp_path,
            "test_secrets_unknown_manager_setting_fails",
            {
                "opSecret": {"manager": "onepassword"},
            },
        )
    assert "requires a 'path'" in error.value.stderr


def test_secrets_of_a_dependency_are_mounted(tmp_path):
    """Applications see the secrets of their dependencies, whatever the manager"""

    def depend_on_myapp(values):
        values["apps"]["dependantapp"]["harness"]["deployment"]["auto"] = True
        values["apps"]["dependantapp"]["harness"]["dependencies"]["hard"] = ["myapp"]

    manifests = render_with_secrets(
        tmp_path,
        "test_secrets_of_a_dependency_are_mounted",
        {
            "local": "a value",
            "opSecret": {
                "manager": "onepassword",
                "path": "vaults/my-vault/items/my-item",
            },
        },
        include=["myapp", "dependantapp"],
        patch=depend_on_myapp,
    )

    deployment = find_manifest(manifests, "Deployment", "dependantapp")
    volumes = [
        v
        for v in deployment["spec"]["template"]["spec"]["volumes"]
        if v["name"] == "cloudharness-myapp"
    ]
    assert volumes == [
        {
            "name": "cloudharness-myapp",
            "projected": {
                "sources": [
                    {"secret": {"name": "myapp"}},
                    {
                        "secret": {
                            "name": "myapp-opsecret",
                            "items": [{"key": "password", "path": "opSecret"}],
                        }
                    },
                ]
            },
        }
    ]

    mounts = [
        m
        for m in deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
        if m["name"] == "cloudharness-myapp"
    ]
    assert mounts[0]["mountPath"] == "/opt/cloudharness/resources/secrets/myapp"


def test_secrets_definitions_survive_the_values_generation(tmp_path):
    """The rich form must reach the chart untouched, an explicitly null manager included:
    losing it would silently turn an unmanaged secret into a CloudHarness managed one"""
    out_folder = tmp_path / "test_secrets_definitions_survive_the_values_generation"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_folder,
        domain="my.local",
        env="secrets",
        local=False,
        include=["myapp"],
        exclude=["legacy"],
    )

    helm_path = out_folder / HELM_CHART_PATH
    shutil.rmtree(helm_path / "charts", ignore_errors=True)
    values_path = helm_path / "values.yaml"
    with open(values_path) as values_file:
        values = yaml.load(values_file)
    secrets = values["apps"]["myapp"]["harness"]["secrets"]
    assert "manager" in secrets["unmanagedSecret"], (
        "the explicitly null manager must be kept"
    )
    assert secrets["unmanagedSecret"]["manager"] is None
    assert secrets["richSecret"] == {
        "manager": "onepassword",
        "path": "vaults/my-vault/items/my-item",
        "default": "a local value",
    }


#     values['apps']['myapp']['harness']['deployment']['auto'] = True
#     with open(values_path, 'w') as values_file:
#         yaml.dump(values, values_file)
#
#     manifests = render_helm_chart(helm_path)
#     secret = find_manifest(manifests, 'Secret', 'myapp')
#     assert secret['stringData']['plainSecret'] == 'a value'
#     assert 'unmanagedSecret' not in secret['stringData'], "unmanaged secrets are never created"
#     assert 'richSecret' not in secret['stringData'], "externally managed secrets are never created"
#     assert find_manifests(manifests, 'OnePasswordItem', 'myapp-richsecret')
#
#


def test_collect_helm_values_source_images_merge(tmp_path):
    out_path = tmp_path / "test_collect_helm_values_source_images_merge"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_path,
        include=["samples", "myapp"],
        domain="my.local",
        namespace="test",
        env="nreg",
        local=False,
        tag=1,
        registry="reg",
    )

    source_images = values.get("source_images")
    assert source_images["KEYCLOAK"] == "myregistry.myapp:15.3"
    assert "NODE" in source_images


def test_gatekeeper_image_comes_from_source_images(tmp_path):
    out_path = tmp_path / "test_gatekeeper_source_images"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_path,
        include=["samples", "argo"],
        domain="my.local",
        namespace="test",
        env="dev",
        local=False,
        tag=1,
        secured=True,
    )
    helm_path = out_path / "helm"

    def gatekeeper_images(*set_values):
        found = {}
        for manifest in render_helm_chart(helm_path, set_values=set_values):
            name = manifest.get("metadata", {}).get("name", "")
            if manifest.get("kind") != "Deployment" or not name.endswith("-gk"):
                continue
            for container in (
                manifest["spec"]["template"]["spec"].get("containers") or []
            ):
                found[name] = container.get("image")
        return found

    # A gatekeeper without an application-specific override takes the single source_images entry
    default_image = "quay.io/gogatekeeper/gatekeeper:4.6.0"
    baseline = gatekeeper_images()
    assert baseline, "no gatekeeper deployment was rendered"
    shared = {name for name, image in baseline.items() if image == default_image}
    assert shared, f"no gatekeeper used source_images.GATEKEEPER, got {baseline}"

    # Overriding that one entry moves every gatekeeper that relies on it, and only those
    overridden = gatekeeper_images("source_images.GATEKEEPER=myreg.io/gk:9.9.9")
    assert {
        name for name, image in overridden.items() if image == "myreg.io/gk:9.9.9"
    } == shared
    assert {
        name: image for name, image in overridden.items() if name not in shared
    } == {name: image for name, image in baseline.items() if name not in shared}

    # A single application can still opt out, without affecting the others
    opted_out = next(iter(shared))
    per_app = gatekeeper_images(
        "source_images.GATEKEEPER=myreg.io/gk:9.9.9",
        "apps.argo.harness.proxy.gatekeeper.image=argoonly/gk:1",
    )
    assert per_app["argo-gk"] == "argoonly/gk:1"
    assert all(
        per_app[name] == "myreg.io/gk:9.9.9" for name in shared if name != "argo-gk"
    )
    assert opted_out


def test_deployment_command_and_args_render_lists_and_strings(tmp_path):
    """A list is rendered as a list, rather than as one string of its items joined by
    spaces; a string is kept verbatim, as it always was."""
    write_tree(
        tmp_path / "solution",
        {
            "applications": {
                "cmdapp": {
                    "deploy": {
                        "values.yaml": (
                            "harness:\n"
                            "  deployment:\n"
                            "    auto: true\n"
                            "    image: custom-image\n"
                            '    command: ["node", "dist/server.js"]\n'
                            "    args: '[\"--port\", \"8080\"]'\n"
                        )
                    }
                }
            }
        },
    )
    out_path = tmp_path / "out"
    create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES, tmp_path / "solution"],
        output_path=out_path,
        include=["cmdapp"],
        domain="my.local",
        namespace="test",
        local=False,
        tag=1,
    )

    containers = [
        container
        for manifest in render_helm_chart(out_path / "helm")
        if manifest.get("kind") == "Deployment" and
        manifest["metadata"]["name"] == "cmdapp"
        for container in manifest["spec"]["template"]["spec"]["containers"]
    ]
    assert [(c["command"], c["args"]) for c in containers] == [
        (["node", "dist/server.js"], ["--port", "8080"])
    ]


def test_collect_helm_values_source_images_merge_no_include(tmp_path):
    out_path = tmp_path / "test_collect_helm_values_source_images_merge"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT, RESOURCES],
        output_path=out_path,
        domain="my.local",
        namespace="test",
        env="nreg",
        local=False,
        tag=1,
        registry="reg",
    )

    source_images = values.get("source_images")
    assert source_images["KEYCLOAK"] == "myregistry.myapp:15.3"
    assert "NODE" in source_images


def test_find_chart_images_shapes():
    refs = {ref.path: ref.value for ref in find_chart_images({
        'flat': {'image': 'docker.io/x/y:1'},
        'flat_with_tag': {'image': 'docker.elastic.co/es/es', 'imageTag': '8.17.0', 'imagePullPolicy': 'IfNotPresent'},
        'registry_shape': {'image': {'registry': 'quay.io', 'repository': 'a/b', 'tag': '2', 'pullPolicy': 'Always'}},
        'name_shape': {'image': {'name': 'quay.io/a/b', 'tag': '3', 'pullPolicy': 'Always'}},
    })}
    assert refs == {
        ('flat', 'image'): 'docker.io/x/y:1',
        ('flat_with_tag', 'image'): 'docker.elastic.co/es/es',
        ('flat_with_tag', 'imageTag'): '8.17.0',
        ('registry_shape', 'image'): {'registry': 'quay.io', 'repository': 'a/b', 'tag': '2'},
        ('name_shape', 'image'): {'name': 'quay.io/a/b', 'tag': '3'},
    }


def test_find_chart_images_excludes_pull_policy_traps():
    # Plural "images" key (argo's shape) must never match
    assert find_chart_images({'images': {'pullPolicy': 'Always', 'pullSecrets': []}}) == []
    # Singular "image" with neither "repository" nor "name" must never match
    assert find_chart_images({'component': {'image': {'pullPolicy': 'Always', 'pullSecrets': []}}}) == []
    assert find_chart_images({'imagePullSecrets': [{'name': 'x'}], 'imagePullSecret': {'registry': 'r'}}) == []


def test_find_chart_images_prunes_skip_paths():
    node = {
        'harness': {
            'deployment': {'image': 'the-built-image:1'},
            'database': {'postgres': {'image': 'postgres:17'}},
            'proxy': {'gatekeeper': {'image': 'quay.io/gogatekeeper/gatekeeper:4.6.0'}},
        },
        'image': 'the-built-image:1',
        'worker': {'image': 'a-real-runtime-image:1'},
    }
    # A built application: only its own image is not an overridable source, while the images it
    # merely pulls under harness (database, gatekeeper) are
    skip = frozenset({('image',), ('harness', 'deployment', 'image')})
    assert sorted(find_chart_images(node, skip_paths=skip), key=lambda r: r.path) == [
        ChartImageRef(path=('harness', 'database', 'postgres', 'image'), value='postgres:17'),
        ChartImageRef(path=('harness', 'proxy', 'gatekeeper', 'image'), value='quay.io/gogatekeeper/gatekeeper:4.6.0'),
        ChartImageRef(path=('worker', 'image'), value='a-real-runtime-image:1'),
    ]
    # Pruning nothing: a prebuilt application (build: false), or a sub-chart's own values.yaml
    assert ChartImageRef(path=('image',), value='the-built-image:1') in find_chart_images(node)


def test_find_chart_images_resolves_empty_tag_from_app_version():
    assert find_chart_images({'image': {'repository': 'a/b', 'tag': ''}}, app_version='9.9.9') == [
        ChartImageRef(path=('image',), value={'repository': 'a/b', 'tag': '9.9.9'}),
    ]
    # Without an appVersion the tag is simply left out rather than emitted empty
    assert find_chart_images({'image': {'repository': 'a/b', 'tag': ''}}) == [
        ChartImageRef(path=('image',), value={'repository': 'a/b'}),
    ]


def test_official_argo_chart_keeps_cloudharness_service_and_naming_defaults(tmp_path):
    out_path = tmp_path / "official_argo"
    values = create_helm_chart(
        [CLOUDHARNESS_ROOT],
        output_path=out_path,
        include=["argo"],
        domain="argo.test",
        namespace="argo-contract",
        local=False,
        tls=False,
        secured=False,
        tag="test",
    )
    chart = out_path / HELM_CHART_PATH
    metadata = yaml.load((chart / "charts" / "argo" / "Chart.yaml").read_text())

    assert metadata["version"] == "2.0.14"
    assert values["argo-workflows"]["singleNamespace"] is True
    assert values[KEY_APPS]["argo"]["serviceaccount"] == "argo-workflows"

    manifests = render_helm_chart(chart)
    server = next(
        item
        for item in manifests
        if item["kind"] == "Deployment" and item["metadata"]["name"].endswith("-server")
    )
    service = find_manifest(manifests, "Service", "argo-server")
    assert all(
        server["spec"]["template"]["metadata"]["labels"][key] == value
        for key, value in service["spec"]["selector"].items()
    )
    assert find_manifest(manifests, "ConfigMap", "argo-cloudharness-env")["data"]["POD_NAMES"] == "v1"
    jobs = [item for item in manifests if item["kind"] == "Job"]
    assert any(
        item["metadata"].get("annotations", {}).get("helm.sh/hook") == "pre-install,pre-upgrade"
        for item in jobs
    )