mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
1196 lines
39 KiB
Python
1196 lines
39 KiB
Python
"""Tests for knoe.core.env pure utility functions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import configparser
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
# ---------------------------------------------------------------------------
|
|
# config.py helpers
|
|
# ---------------------------------------------------------------------------
|
|
from knoe.config import (
|
|
_parse_bool as cfg_parse_bool,
|
|
_expand_path as cfg_expand_path,
|
|
_expand_cfg_value as cfg_expand_cfg_value,
|
|
_preserve_cfg_expr_for_persistence,
|
|
_collect_cfg_vars as cfg_collect_cfg_vars,
|
|
_filter_cfg_values_for_persistence,
|
|
_normalize_cfg_value_for_persistence,
|
|
_extract_yaml_scalar_from_text as cfg_extract_yaml,
|
|
_load_properties,
|
|
_write_k3s_kubeconfig,
|
|
_merge_kubeconfig,
|
|
normalize_version,
|
|
get_resource_path as cfg_get_resource_path,
|
|
get_docker_build_platform_args as cfg_get_docker_args,
|
|
_is_knoe_secret as cfg_is_knoe_secret,
|
|
_is_openbao_ref as cfg_is_openbao_ref,
|
|
_encrypt_cfg_secret,
|
|
_resolve_secret_value,
|
|
_update_knoe_cfg_value,
|
|
get_properties,
|
|
get_config_value,
|
|
setup_logging,
|
|
)
|
|
# ---------------------------------------------------------------------------
|
|
# env.py helpers
|
|
# ---------------------------------------------------------------------------
|
|
from knoe.core.env import (
|
|
_bool_str,
|
|
_build_required_port_forwards,
|
|
_clean_yaml_value,
|
|
_cluster_env_radio_value,
|
|
_collect_images_from_files,
|
|
_detect_ansible_k3s_settings,
|
|
_detect_ansible_storage_mounts,
|
|
_detect_ansible_node_labels,
|
|
_deployment_mode_from_env,
|
|
_deployment_target_label,
|
|
_expand_cfg_value,
|
|
_expand_path,
|
|
_extract_inline_vault_block,
|
|
_extract_yaml_scalar_from_text,
|
|
_find_kubeconfig_file,
|
|
_format_ollama_host,
|
|
_is_openbao_ref,
|
|
_is_knoe_secret,
|
|
_k3d_knoe_data_volume_args,
|
|
_local_registry_enabled,
|
|
_looks_like_k8s_bearer_token,
|
|
_normalize_cluster_env,
|
|
_normalize_k3s_token,
|
|
_openbao_placeholder,
|
|
_parse_ansible_inventory_hosts,
|
|
_parse_bool,
|
|
_parse_internal_a_records,
|
|
_parse_ollama_host,
|
|
_parse_yaml_scalar_values,
|
|
_pf_extract_id,
|
|
_pf_mapping_str,
|
|
_pf_upsert_mapping,
|
|
_registry_image_ref_exists,
|
|
_resolve_host_ip,
|
|
_collect_cfg_vars as env_collect_cfg_vars,
|
|
get_resource_path,
|
|
is_apple_silicon,
|
|
)
|
|
|
|
|
|
# ===== _is_knoe_secret / _is_openbao_ref =====
|
|
|
|
|
|
class TestSecretDetection:
|
|
def test_is_knoe_secret_true(self):
|
|
assert _is_knoe_secret("${KNOE_SECRET:abc}") is True
|
|
|
|
def test_is_knoe_secret_false(self):
|
|
assert _is_knoe_secret("plain") is False
|
|
assert _is_knoe_secret("") is False
|
|
assert _is_knoe_secret(None) is False
|
|
|
|
def test_is_openbao_ref_true(self):
|
|
assert _is_openbao_ref("${OPENBAO:kv/knoe/ns/db#password}") is True
|
|
|
|
def test_is_openbao_ref_false(self):
|
|
assert _is_openbao_ref("plain") is False
|
|
assert _is_openbao_ref("") is False
|
|
assert _is_openbao_ref(None) is False
|
|
|
|
def test_cfg_variants(self):
|
|
assert cfg_is_knoe_secret("${KNOE_SECRET:x}") is True
|
|
assert cfg_is_openbao_ref("${OPENBAO:x}") is True
|
|
|
|
|
|
# ===== _bool_str / _parse_bool =====
|
|
|
|
|
|
class TestBoolHelpers:
|
|
def test_bool_str(self):
|
|
assert _bool_str(True) == "true"
|
|
assert _bool_str(False) == "false"
|
|
assert _bool_str(1) == "true"
|
|
assert _bool_str(0) == "false"
|
|
|
|
@pytest.mark.parametrize(
|
|
"val,expected",
|
|
[
|
|
("true", True),
|
|
("True", True),
|
|
("1", True),
|
|
("yes", True),
|
|
("y", True),
|
|
("on", True),
|
|
("false", False),
|
|
("False", False),
|
|
("0", False),
|
|
("no", False),
|
|
("n", False),
|
|
("off", False),
|
|
(None, False),
|
|
("", False),
|
|
("maybe", False),
|
|
],
|
|
)
|
|
def test_parse_bool(self, val, expected):
|
|
assert _parse_bool(val) == expected
|
|
|
|
def test_parse_bool_default(self):
|
|
assert _parse_bool(None, default=True) is True
|
|
assert _parse_bool("junk", default=True) is True
|
|
|
|
def test_cfg_parse_bool(self):
|
|
assert cfg_parse_bool(True) is True
|
|
assert cfg_parse_bool(False) is False
|
|
assert cfg_parse_bool("true") is True
|
|
assert cfg_parse_bool("false") is False
|
|
assert cfg_parse_bool(None, default=True) is True
|
|
|
|
|
|
# ===== _normalize_cluster_env family =====
|
|
|
|
|
|
class TestClusterEnvNormalization:
|
|
@pytest.mark.parametrize(
|
|
"inp,expected",
|
|
[
|
|
("dev", "dev"),
|
|
("k3d", "dev"),
|
|
("k3d-dev", "dev"),
|
|
("knoe-dev-cluster", "dev"),
|
|
("k3d-knoe-dev-cluster", "dev"),
|
|
("k3d-custom", "dev"),
|
|
("service", "service"),
|
|
("k3s", "service"),
|
|
("k3s-service", "service"),
|
|
("knoe-service-cluster", "service"),
|
|
("knoe-service-x", "service"),
|
|
("prod", "prod"),
|
|
("production", "prod"),
|
|
("k8s", "prod"),
|
|
("knoe-prod-cluster", "prod"),
|
|
("knoe-prod-x", "prod"),
|
|
("", ""),
|
|
(None, ""),
|
|
("unknown", "unknown"),
|
|
],
|
|
)
|
|
def test_normalize_cluster_env(self, inp, expected):
|
|
assert _normalize_cluster_env(inp) == expected
|
|
|
|
def test_cluster_env_radio_value(self):
|
|
assert _cluster_env_radio_value("k3d") == "dev"
|
|
assert _cluster_env_radio_value("k3s") == "service"
|
|
assert _cluster_env_radio_value("k8s") == "prod"
|
|
assert _cluster_env_radio_value("unknown") == "unknown"
|
|
assert _cluster_env_radio_value(None) == ""
|
|
|
|
def test_deployment_target_label(self):
|
|
assert _deployment_target_label("dev") == "knoe-dev-cluster"
|
|
assert _deployment_target_label("k3s") == "knoe-service-cluster"
|
|
assert _deployment_target_label("k8s") == "knoe-prod-cluster"
|
|
assert _deployment_target_label(None) == ""
|
|
|
|
def test_deployment_mode_from_env(self):
|
|
assert _deployment_mode_from_env("dev") == "k3d"
|
|
assert _deployment_mode_from_env("service") == "k3s"
|
|
assert _deployment_mode_from_env("prod") == "k8s"
|
|
assert _deployment_mode_from_env("") == ""
|
|
assert _deployment_mode_from_env(None) == ""
|
|
|
|
|
|
# ===== _normalize_k3s_token / _looks_like_k8s_bearer_token =====
|
|
|
|
|
|
class TestTokenHelpers:
|
|
def test_normalize_k3s_token_plain(self):
|
|
assert _normalize_k3s_token("abc123") == "abc123"
|
|
|
|
def test_normalize_k3s_token_empty(self):
|
|
assert _normalize_k3s_token("") == ""
|
|
assert _normalize_k3s_token(None) == ""
|
|
assert _normalize_k3s_token(" ") == ""
|
|
|
|
def test_normalize_k3s_token_openbao(self):
|
|
assert _normalize_k3s_token("${OPENBAO:kv/x}") == ""
|
|
|
|
def test_normalize_k3s_token_knoe_secret(self):
|
|
assert _normalize_k3s_token("${KNOE_SECRET:junk}") == ""
|
|
|
|
def test_looks_like_bearer_jwt(self):
|
|
assert _looks_like_k8s_bearer_token("a.b.c") is True
|
|
|
|
def test_looks_like_bearer_bootstrap(self):
|
|
assert _looks_like_k8s_bearer_token("abcdef.1234567890abcdef") is True
|
|
|
|
def test_not_bearer_node_token(self):
|
|
assert _looks_like_k8s_bearer_token("K1::node::xyz") is False
|
|
|
|
def test_not_bearer_empty(self):
|
|
assert _looks_like_k8s_bearer_token("") is False
|
|
assert _looks_like_k8s_bearer_token(None) is False
|
|
|
|
def test_not_bearer_short_parts(self):
|
|
assert _looks_like_k8s_bearer_token("a.b") is False
|
|
|
|
|
|
# ===== _parse_ollama_host / _format_ollama_host =====
|
|
|
|
|
|
class TestOllamaHelpers:
|
|
def test_parse_ollama_host_basic(self):
|
|
host, port = _parse_ollama_host("http://myhost:1234")
|
|
assert host == "myhost"
|
|
assert port == "1234"
|
|
|
|
def test_parse_ollama_host_no_scheme(self):
|
|
host, port = _parse_ollama_host("myhost:5555")
|
|
assert host == "myhost"
|
|
assert port == "5555"
|
|
|
|
def test_parse_ollama_host_default_port(self):
|
|
host, port = _parse_ollama_host("http://myhost")
|
|
assert host == "myhost"
|
|
assert port == "11434"
|
|
|
|
def test_parse_ollama_host_empty(self):
|
|
assert _parse_ollama_host("") == ("", "")
|
|
|
|
def test_format_ollama_host(self):
|
|
assert _format_ollama_host("myhost", "1234") == "http://myhost:1234"
|
|
|
|
def test_format_ollama_host_empty(self):
|
|
assert _format_ollama_host("", "1234") == ""
|
|
|
|
def test_format_ollama_host_https(self):
|
|
assert _format_ollama_host("https://myhost", "443") == "https://myhost:443"
|
|
|
|
def test_format_ollama_host_no_port(self):
|
|
assert _format_ollama_host("myhost", "") == "http://myhost"
|
|
|
|
|
|
# ===== _openbao_placeholder =====
|
|
|
|
|
|
class TestOpenbaoPlaceholder:
|
|
def test_with_key(self):
|
|
r = _openbao_placeholder("myns", "db", "password")
|
|
assert r == "${OPENBAO:kv/knoe/myns/db#password}"
|
|
|
|
def test_without_key(self):
|
|
r = _openbao_placeholder("myns", "db")
|
|
assert r == "${OPENBAO:kv/knoe/myns/db}"
|
|
|
|
def test_default_namespace(self):
|
|
r = _openbao_placeholder("", "db", "pw")
|
|
assert "default" in r
|
|
|
|
|
|
# ===== _expand_path / _expand_cfg_value =====
|
|
|
|
|
|
class TestExpandHelpers:
|
|
def test_expand_path_none(self):
|
|
assert _expand_path(None) == ""
|
|
|
|
def test_expand_path_tilde(self):
|
|
result = _expand_path("~/test")
|
|
assert "~" not in result
|
|
|
|
def test_expand_cfg_value_none(self):
|
|
assert _expand_cfg_value(None) == ""
|
|
|
|
def test_expand_cfg_value_openbao(self):
|
|
val = "${OPENBAO:x}"
|
|
assert _expand_cfg_value(val) == val
|
|
|
|
def test_expand_cfg_value_vars(self):
|
|
assert (
|
|
_expand_cfg_value("${HOME}/data", cfg_vars={"HOME": "/opt"}) == "/opt/data"
|
|
)
|
|
|
|
def test_expand_cfg_value_recursive(self):
|
|
assert _expand_cfg_value("$A", cfg_vars={"A": "$B", "B": "final"}) == "final"
|
|
|
|
def test_cfg_expand_path(self):
|
|
assert cfg_expand_path(None) == ""
|
|
assert cfg_expand_path("") == ""
|
|
|
|
def test_cfg_expand_cfg_value(self):
|
|
assert cfg_expand_cfg_value("${X}", {"X": "hello"}) == "hello"
|
|
assert cfg_expand_cfg_value("", {}) == ""
|
|
assert cfg_expand_cfg_value(None, {}) is None
|
|
|
|
def test_preserve_cfg_expr_for_persistence_keeps_home_token(self):
|
|
value = _preserve_cfg_expr_for_persistence(
|
|
"${HOME}/dev/knoe/data/staging",
|
|
"/Users/chrisfu/dev/knoe/data/staging",
|
|
)
|
|
assert value == "${HOME}/dev/knoe/data/staging"
|
|
|
|
def test_preserve_cfg_expr_for_persistence_uses_expanded_without_tokens(self):
|
|
value = _preserve_cfg_expr_for_persistence(
|
|
"/Users/chrisfu/dev/knoe/data/staging",
|
|
"/Users/chrisfu/dev/knoe/data/staging",
|
|
)
|
|
assert value == "/Users/chrisfu/dev/knoe/data/staging"
|
|
|
|
|
|
# ===== _collect_cfg_vars =====
|
|
|
|
|
|
class TestCollectCfgVars:
|
|
def test_env_collect(self):
|
|
cfg = configparser.ConfigParser(interpolation=None)
|
|
cfg.optionxform = str
|
|
cfg.add_section("Global")
|
|
cfg["Global"]["NS"] = "knoe-test"
|
|
result = env_collect_cfg_vars(cfg)
|
|
assert result["NS"] == "knoe-test"
|
|
|
|
def test_cfg_collect(self):
|
|
cfg = configparser.ConfigParser(interpolation=None)
|
|
cfg.optionxform = str
|
|
cfg.add_section("Global")
|
|
cfg["Global"]["FOO"] = "bar"
|
|
cfg.add_section("System Environment")
|
|
cfg["System Environment"]["BAZ"] = "qux"
|
|
result = cfg_collect_cfg_vars(cfg)
|
|
assert "FOO" in result or "foo" in result
|
|
assert "BAZ" in result or "baz" in result
|
|
|
|
|
|
# ===== _clean_yaml_value =====
|
|
|
|
|
|
class TestCleanYamlValue:
|
|
def test_plain(self):
|
|
assert _clean_yaml_value(" hello ") == "hello"
|
|
|
|
def test_quoted(self):
|
|
assert _clean_yaml_value('"hello"') == "hello"
|
|
assert _clean_yaml_value("'hello'") == "hello"
|
|
|
|
def test_comment(self):
|
|
assert _clean_yaml_value("val # comment") == "val"
|
|
|
|
def test_none(self):
|
|
assert _clean_yaml_value(None) == ""
|
|
|
|
|
|
# ===== YAML helpers =====
|
|
|
|
|
|
class TestYamlHelpers:
|
|
def test_extract_yaml_scalar(self):
|
|
text = "name: test\nport: 8080\n# comment\n"
|
|
assert _extract_yaml_scalar_from_text(text, "name") == "test"
|
|
assert _extract_yaml_scalar_from_text(text, "port") == "8080"
|
|
assert _extract_yaml_scalar_from_text(text, "missing") == ""
|
|
|
|
def test_cfg_extract_yaml(self):
|
|
text = 'key: "value"\nother: plain'
|
|
assert cfg_extract_yaml(text, "key") == "value"
|
|
assert cfg_extract_yaml(text, "missing") == ""
|
|
|
|
def test_extract_inline_vault_block(self):
|
|
text = "secret: !vault |\n $ANSIBLE_VAULT;1.1;AES256\n 6162\nnext_key: val"
|
|
result = _extract_inline_vault_block(text, "secret")
|
|
assert "$ANSIBLE_VAULT" in result
|
|
|
|
def test_extract_inline_vault_no_match(self):
|
|
assert _extract_inline_vault_block("key: value", "key") == ""
|
|
assert _extract_inline_vault_block("key: value", "missing") == ""
|
|
|
|
|
|
def test_parse_yaml_scalar_values(self, tmp_path):
|
|
f = tmp_path / "test.yaml"
|
|
f.write_text("---\nname: test\nport: 8080\n# comment\n- item\n")
|
|
result = _parse_yaml_scalar_values(f, {"name", "port"})
|
|
assert result["name"] == "test"
|
|
assert result["port"] == "8080"
|
|
|
|
def test_parse_yaml_scalar_missing_file(self, tmp_path):
|
|
f = tmp_path / "missing.yaml"
|
|
assert _parse_yaml_scalar_values(f, {"key"}) == {}
|
|
|
|
|
|
# ===== _parse_internal_a_records =====
|
|
|
|
|
|
class TestParseInternalARecords:
|
|
def test_basic(self, tmp_path):
|
|
f = tmp_path / "dns.yml"
|
|
f.write_text(
|
|
"knoe_domain: example.com\n"
|
|
"knoe_internal_a_records:\n"
|
|
" - fqdn: host1.example.com\n"
|
|
" ipv4: 10.0.0.1\n"
|
|
" - fqdn: host2.example.com\n"
|
|
" ipv4: 10.0.0.2\n"
|
|
)
|
|
domain, ip_map, fqdn_map = _parse_internal_a_records(f)
|
|
assert domain == "example.com"
|
|
assert fqdn_map["host1.example.com"] == "10.0.0.1"
|
|
assert "host1" in ip_map
|
|
|
|
def test_missing_file(self, tmp_path):
|
|
f = tmp_path / "missing.yml"
|
|
domain, ip_map, fqdn_map = _parse_internal_a_records(f)
|
|
assert domain == ""
|
|
assert ip_map == {}
|
|
|
|
|
|
# ===== _parse_ansible_inventory_hosts =====
|
|
|
|
|
|
class TestParseInventory:
|
|
def test_basic(self, tmp_path):
|
|
f = tmp_path / "hosts"
|
|
f.write_text("[servers]\nhost1\nhost2\n\n[workers]\nhost3\n")
|
|
groups = _parse_ansible_inventory_hosts(f)
|
|
assert "servers" in groups
|
|
assert "host1" in groups["servers"]
|
|
assert "host3" in groups["workers"]
|
|
|
|
def test_missing(self, tmp_path):
|
|
f = tmp_path / "missing"
|
|
assert _parse_ansible_inventory_hosts(f) == {}
|
|
|
|
def test_comments(self, tmp_path):
|
|
f = tmp_path / "hosts"
|
|
f.write_text("# comment\n[grp]\n; comment\nhost1\n")
|
|
groups = _parse_ansible_inventory_hosts(f)
|
|
assert groups["grp"] == ["host1"]
|
|
|
|
|
|
class TestDetectAnsibleK3sSettings:
|
|
def test_children_groups_fallback_to_servers(self, tmp_path):
|
|
inventory_path = tmp_path / "inventory"
|
|
(inventory_path / "host_vars").mkdir(parents=True)
|
|
|
|
# Host vars are intentionally empty; the detection should still infer a
|
|
# server host from the k3s_servers group.
|
|
(inventory_path / "host_vars" / "myrddin.knoe.org.yml").write_text("\n")
|
|
(inventory_path / "host_vars" / "pi.knoe.org.yml").write_text("\n")
|
|
|
|
groups = {
|
|
"k3s_hosts:children": ["k3s_servers", "k3s_agents"],
|
|
"k3s_servers": ["myrddin.knoe.org"],
|
|
"k3s_agents": ["pi.knoe.org"],
|
|
}
|
|
res = _detect_ansible_k3s_settings(
|
|
inventory_path=inventory_path, groups=groups, domain="knoe.org", ip_map={}
|
|
)
|
|
assert res["server_host"] == "myrddin.knoe.org"
|
|
assert res["server_url"] == "https://myrddin.knoe.org:6443"
|
|
|
|
|
|
class TestDetectAnsibleStorageMounts:
|
|
def test_parses_iscsi_mounts_from_host_vars(self, tmp_path):
|
|
inventory = tmp_path / "inventory"
|
|
hv = inventory / "host_vars"
|
|
hv.mkdir(parents=True)
|
|
|
|
(hv / "myrddin.knoe.org.yml").write_text(
|
|
"""
|
|
iscsi_targets:
|
|
- name: syno
|
|
mounts:
|
|
- name: d001
|
|
path: /synology/d001
|
|
k3s_service_node_labels:
|
|
- knoe.org/node-role=db
|
|
- knoe.org/storage=synology
|
|
""".lstrip()
|
|
)
|
|
(hv / "pi.knoe.org.yml").write_text(
|
|
"""
|
|
iscsi_targets:
|
|
- name: syno
|
|
mounts:
|
|
- name: d003
|
|
path: /synology/d003
|
|
""".lstrip()
|
|
)
|
|
|
|
mounts = _detect_ansible_storage_mounts(inventory)
|
|
assert mounts["d001"]["host"] == "myrddin.knoe.org"
|
|
assert mounts["d001"]["path"] == "/synology/d001"
|
|
assert mounts["d003"]["host"] == "pi.knoe.org"
|
|
assert mounts["d003"]["path"] == "/synology/d003"
|
|
|
|
labels = _detect_ansible_node_labels(inventory)
|
|
assert labels["myrddin.knoe.org"]["knoe.org/node-role"] == "db"
|
|
assert labels["myrddin.knoe.org"]["knoe.org/storage"] == "synology"
|
|
|
|
|
|
# ===== _resolve_host_ip =====
|
|
|
|
|
|
class TestResolveHostIp:
|
|
def test_direct(self):
|
|
assert _resolve_host_ip("h1", "example.com", {"h1": "1.2.3.4"}) == "1.2.3.4"
|
|
|
|
def test_fqdn(self):
|
|
assert (
|
|
_resolve_host_ip("h1", "example.com", {"h1.example.com": "1.2.3.4"})
|
|
== "1.2.3.4"
|
|
)
|
|
|
|
def test_short_from_fqdn(self):
|
|
assert (
|
|
_resolve_host_ip("h1.example.com", "example.com", {"h1": "1.2.3.4"})
|
|
== "1.2.3.4"
|
|
)
|
|
|
|
def test_empty(self):
|
|
assert _resolve_host_ip("", "", {}) == ""
|
|
|
|
|
|
# ===== Port-forward helpers =====
|
|
|
|
|
|
class TestPortForwardHelpers:
|
|
def test_pf_extract_id(self):
|
|
assert _pf_extract_id("id=grafana;namespace=monitoring") == "grafana"
|
|
assert _pf_extract_id("namespace=x") == ""
|
|
assert _pf_extract_id("") == ""
|
|
|
|
def test_pf_mapping_str(self):
|
|
s = _pf_mapping_str("test", "ns", "svc/mysvc", 8080, 80)
|
|
assert "id=test" in s
|
|
assert "namespace=ns" in s
|
|
assert "hostPort=8080" in s
|
|
|
|
def test_pf_upsert_new(self):
|
|
section = {}
|
|
prefix = "PF_"
|
|
mapping = "id=grafana;namespace=monitoring;hostPort=3000;servicePort=80"
|
|
assert _pf_upsert_mapping(section, prefix, mapping) is True
|
|
assert "PF_1" in section
|
|
|
|
def test_pf_upsert_update(self):
|
|
section = {
|
|
"PF_1": "id=grafana;namespace=monitoring;hostPort=3000;servicePort=80"
|
|
}
|
|
new = "id=grafana;namespace=monitoring;hostPort=3001;servicePort=80"
|
|
assert _pf_upsert_mapping(section, "PF_", new) is True
|
|
assert section["PF_1"] == new
|
|
|
|
def test_pf_upsert_duplicate(self):
|
|
mapping = "id=grafana;namespace=monitoring;hostPort=3000;servicePort=80"
|
|
section = {"PF_1": mapping}
|
|
assert _pf_upsert_mapping(section, "PF_", mapping) is False
|
|
|
|
def test_pf_upsert_none_section(self):
|
|
assert _pf_upsert_mapping(None, "PF_", "id=x") is False
|
|
|
|
def test_pf_upsert_no_id(self):
|
|
assert _pf_upsert_mapping({}, "PF_", "no-id-here") is False
|
|
|
|
|
|
# ===== _build_required_port_forwards =====
|
|
|
|
|
|
class TestBuildPortForwards:
|
|
def test_basic_k3d(self):
|
|
mappings = _build_required_port_forwards(
|
|
"k3d", "default", "argocd", "myns", "5432", False, "supabase"
|
|
)
|
|
assert len(mappings) >= 8
|
|
ids = [_pf_extract_id(m) for m in mappings]
|
|
assert "argocd" in ids
|
|
assert "grafana" in ids
|
|
assert "postgres" in ids
|
|
assert "openbao" in ids
|
|
|
|
def test_with_supabase(self):
|
|
mappings = _build_required_port_forwards(
|
|
"k3d", "default", "argocd", "myns", "5432", True, "supabase"
|
|
)
|
|
ids = [_pf_extract_id(m) for m in mappings]
|
|
assert "supabase-kong" in ids
|
|
assert "supabase-studio" in ids
|
|
|
|
def test_defaults(self):
|
|
mappings = _build_required_port_forwards("k3d", "", "", "", "", False, "")
|
|
assert len(mappings) >= 8
|
|
|
|
|
|
# ===== _k3d_knoe_data_volume_args =====
|
|
|
|
|
|
class TestK3dVolumeArgs:
|
|
def test_empty(self):
|
|
assert _k3d_knoe_data_volume_args("") == []
|
|
assert _k3d_knoe_data_volume_args(None) == []
|
|
|
|
def test_valid(self, tmp_path):
|
|
result = _k3d_knoe_data_volume_args(str(tmp_path / "data"))
|
|
assert result[0] == "--volume"
|
|
assert "storage@all" in result[1]
|
|
|
|
|
|
# ===== _find_kubeconfig_file =====
|
|
|
|
|
|
class TestFindKubeconfig:
|
|
def test_from_env(self, tmp_path):
|
|
kc = tmp_path / "kubeconfig"
|
|
kc.write_text("test")
|
|
env = {"KUBECONFIG": str(kc)}
|
|
assert _find_kubeconfig_file(env) == str(kc)
|
|
|
|
def test_prefers_project_root_knoe_k3s_over_etc_secrets(
|
|
self, tmp_path, monkeypatch
|
|
):
|
|
# If both exist, prefer the Ansible-fetched kubeconfig in project root.
|
|
import knoe.core.env as env_mod
|
|
|
|
fake_home = tmp_path / "home"
|
|
fake_home.mkdir(parents=True, exist_ok=True)
|
|
monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home))
|
|
|
|
fake_root = tmp_path / "root"
|
|
(fake_root / "etc" / "secrets").mkdir(parents=True, exist_ok=True)
|
|
etc_kc = fake_root / "etc" / "secrets" / "k3s.kubeconfig"
|
|
etc_kc.write_text("etc")
|
|
knoe_kc = fake_root / "knoe-k3s.kubeconfig"
|
|
knoe_kc.write_text("root")
|
|
|
|
monkeypatch.setattr(env_mod, "PROJECT_ROOT", fake_root)
|
|
|
|
# Must pass a non-empty mapping; `_find_kubeconfig_file({})` falls back
|
|
# to `os.environ` because an empty dict is falsy.
|
|
assert _find_kubeconfig_file({"KUBECONFIG": ""}) == str(knoe_kc)
|
|
|
|
def test_none_found(self):
|
|
env = {"KUBECONFIG": "/nonexistent/path"}
|
|
result = _find_kubeconfig_file(env)
|
|
# May return '' or another found path
|
|
assert isinstance(result, str)
|
|
|
|
|
|
# ===== _local_registry_enabled =====
|
|
|
|
|
|
class TestLocalRegistryEnabled:
|
|
def test_explicit_env(self):
|
|
with mock.patch.dict(
|
|
os.environ, {"PROLE_ENABLE_LOCAL_REGISTRY": "true"}, clear=False
|
|
):
|
|
assert _local_registry_enabled() is True
|
|
|
|
def test_mode_hint(self):
|
|
with mock.patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("PROLE_ENABLE_LOCAL_REGISTRY", None)
|
|
os.environ.pop("ENABLE_LOCAL_REGISTRY", None)
|
|
assert _local_registry_enabled("k3d") is True
|
|
assert _local_registry_enabled("k3s") is True
|
|
|
|
def test_explicit_false(self):
|
|
with mock.patch.dict(
|
|
os.environ, {"PROLE_ENABLE_LOCAL_REGISTRY": "false"}, clear=False
|
|
):
|
|
assert _local_registry_enabled() is False
|
|
|
|
|
|
# ===== _collect_images_from_files =====
|
|
|
|
|
|
class TestCollectImages:
|
|
def test_basic(self, tmp_path):
|
|
f = tmp_path / "deploy.yaml"
|
|
f.write_text("spec:\n image: nginx:latest\n image: postgres:16\n")
|
|
result = _collect_images_from_files([f])
|
|
assert "nginx:latest" in result
|
|
assert "postgres:16" in result
|
|
|
|
def test_skip_vars(self, tmp_path):
|
|
f = tmp_path / "deploy.yaml"
|
|
f.write_text(" image: ${REGISTRY}/app:latest\n")
|
|
result = _collect_images_from_files([f])
|
|
assert len(result) == 0
|
|
|
|
def test_missing_file(self, tmp_path):
|
|
result = _collect_images_from_files([tmp_path / "missing.yaml"])
|
|
assert result == set()
|
|
|
|
|
|
# ===== get_resource_path =====
|
|
|
|
|
|
class TestGetResourcePath:
|
|
def test_returns_path(self):
|
|
p = get_resource_path("img/test.png")
|
|
assert isinstance(p, Path)
|
|
|
|
def test_cfg_variant(self):
|
|
p = cfg_get_resource_path("img/test.png")
|
|
assert isinstance(p, Path)
|
|
|
|
|
|
# ===== normalize_version =====
|
|
|
|
|
|
class TestNormalizeVersion:
|
|
def test_semver(self):
|
|
assert normalize_version("1.2.3") == "01.02.03"
|
|
|
|
def test_with_prefix(self):
|
|
assert normalize_version("v1.2.3") == "01.02.03"
|
|
|
|
def test_short(self):
|
|
assert normalize_version("5") == "05.00.00"
|
|
|
|
def test_empty(self):
|
|
assert normalize_version("") == ""
|
|
|
|
def test_no_numbers(self):
|
|
assert normalize_version("abc") == "abc"
|
|
|
|
|
|
# ===== _load_properties =====
|
|
|
|
|
|
class TestLoadProperties:
|
|
def test_basic(self, tmp_path):
|
|
f = tmp_path / "test.properties"
|
|
f.write_text("# comment\nkey1=val1\nkey2 = val2\n\n")
|
|
result = _load_properties(f)
|
|
assert result["key1"] == "val1"
|
|
assert result["key2"] == "val2"
|
|
|
|
def test_missing(self, tmp_path):
|
|
assert _load_properties(tmp_path / "missing") == {}
|
|
|
|
|
|
# ===== _write_k3s_kubeconfig =====
|
|
|
|
|
|
class TestWriteK3sKubeconfig:
|
|
def test_basic(self, tmp_path):
|
|
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path), mock.patch.dict(
|
|
os.environ, {"KNOE_SERVICE": "", "KNOE_HOME": ""}, clear=False
|
|
):
|
|
result = _write_k3s_kubeconfig("https://server:6443", "mytoken")
|
|
assert result.exists()
|
|
content = result.read_text()
|
|
assert "server: https://server:6443" in content
|
|
assert "mytoken" in content
|
|
|
|
def test_no_scheme(self, tmp_path):
|
|
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path), mock.patch.dict(
|
|
os.environ, {"KNOE_SERVICE": "", "KNOE_HOME": ""}, clear=False
|
|
):
|
|
result = _write_k3s_kubeconfig("server:6443", "tok")
|
|
content = result.read_text()
|
|
assert "server: https://server:6443" in content
|
|
|
|
|
|
# ===== _merge_kubeconfig =====
|
|
|
|
|
|
class TestMergeKubeconfig:
|
|
def test_merge_into_empty_home(self, tmp_path):
|
|
"""Merge a standalone kubeconfig when no ~/.kube/config exists yet."""
|
|
kube_dir = tmp_path / ".kube"
|
|
standalone = tmp_path / "standalone.yaml"
|
|
standalone.write_text(
|
|
"apiVersion: v1\nkind: Config\n"
|
|
"clusters:\n- cluster:\n server: https://srv:6443\n"
|
|
" insecure-skip-tls-verify: true\n name: knoe-k3s\n"
|
|
"contexts:\n- context:\n cluster: knoe-k3s\n user: knoe-k3s\n"
|
|
" name: knoe-k3s\ncurrent-context: knoe-k3s\n"
|
|
"users:\n- name: knoe-k3s\n user:\n token: tok123\n"
|
|
)
|
|
with mock.patch("knoe.config.Path.home", return_value=tmp_path):
|
|
result = _merge_kubeconfig(str(standalone))
|
|
if result:
|
|
merged = (kube_dir / "config").read_text()
|
|
assert "knoe-k3s" in merged
|
|
assert "tok123" in merged
|
|
|
|
def test_merge_adds_context_to_existing(self, tmp_path):
|
|
"""Merge adds the knoe-k3s context alongside an existing context."""
|
|
kube_dir = tmp_path / ".kube"
|
|
kube_dir.mkdir()
|
|
existing = kube_dir / "config"
|
|
existing.write_text(
|
|
"apiVersion: v1\nkind: Config\n"
|
|
"clusters:\n- cluster:\n server: https://other:6443\n name: other-cluster\n"
|
|
"contexts:\n- context:\n cluster: other-cluster\n user: other-user\n"
|
|
" name: other-ctx\ncurrent-context: other-ctx\n"
|
|
"users:\n- name: other-user\n user:\n token: othertoken\n"
|
|
)
|
|
standalone = tmp_path / "standalone.yaml"
|
|
standalone.write_text(
|
|
"apiVersion: v1\nkind: Config\n"
|
|
"clusters:\n- cluster:\n server: https://srv:6443\n"
|
|
" insecure-skip-tls-verify: true\n name: knoe-k3s\n"
|
|
"contexts:\n- context:\n cluster: knoe-k3s\n user: knoe-k3s\n"
|
|
" name: knoe-k3s\ncurrent-context: knoe-k3s\n"
|
|
"users:\n- name: knoe-k3s\n user:\n token: tok123\n"
|
|
)
|
|
with mock.patch("knoe.config.Path.home", return_value=tmp_path):
|
|
result = _merge_kubeconfig(str(standalone))
|
|
if result:
|
|
merged = existing.read_text()
|
|
assert "knoe-k3s" in merged
|
|
assert "other-ctx" in merged
|
|
|
|
def test_merge_returns_false_when_kubectl_missing(self, tmp_path):
|
|
"""Returns False when kubectl is not available."""
|
|
standalone = tmp_path / "standalone.yaml"
|
|
standalone.write_text("apiVersion: v1\nkind: Config\n")
|
|
with mock.patch(
|
|
"knoe.config.Path.home", return_value=tmp_path
|
|
), mock.patch("subprocess.run", side_effect=FileNotFoundError):
|
|
result = _merge_kubeconfig(str(standalone))
|
|
assert result is False
|
|
|
|
|
|
# ===== _encrypt_cfg_secret / _resolve_secret_value =====
|
|
|
|
|
|
class TestSecretEncryption:
|
|
def test_encrypt_empty(self):
|
|
assert _encrypt_cfg_secret("") == ""
|
|
assert _encrypt_cfg_secret(None) == ""
|
|
|
|
def test_encrypt_openbao_passthrough(self):
|
|
val = "${OPENBAO:kv/x}"
|
|
assert _encrypt_cfg_secret(val) == val
|
|
|
|
def test_resolve_plain(self):
|
|
assert _resolve_secret_value("hello") == "hello"
|
|
|
|
|
|
# ===== setup_logging =====
|
|
|
|
|
|
class TestSetupLogging:
|
|
def test_default(self):
|
|
setup_logging()
|
|
|
|
def test_verbose(self):
|
|
setup_logging(verbose=True)
|
|
|
|
def test_debug(self):
|
|
setup_logging(debug=True)
|
|
|
|
|
|
# ===== get_properties / get_config_value =====
|
|
|
|
|
|
class TestProperties:
|
|
def test_get_properties(self):
|
|
props = get_properties()
|
|
assert isinstance(props, dict)
|
|
|
|
def test_get_config_value_default(self):
|
|
assert get_config_value("nonexistent_key_xyz", "fallback") == "fallback"
|
|
|
|
|
|
# ===== _update_knoe_cfg_value =====
|
|
|
|
|
|
class TestUpdateKnoeCfgValue:
|
|
def test_creates_section(self, tmp_path):
|
|
cfg_file = tmp_path / "conf" / "knoe.cfg"
|
|
cfg_file.parent.mkdir(parents=True)
|
|
cfg_file.write_text("[Global]\nFOO = bar\n")
|
|
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path):
|
|
_update_knoe_cfg_value("NewSection", "KEY", "VALUE")
|
|
content = cfg_file.read_text()
|
|
assert "NewSection" in content
|
|
assert "KEY" in content
|
|
|
|
def test_missing_cfg(self, tmp_path):
|
|
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path):
|
|
_update_knoe_cfg_value("Sec", "K", "V") # should not raise
|
|
|
|
def test_writes_through_knoe_conf_symlink(self, tmp_path):
|
|
conf_dir = tmp_path / "knoe-conf"
|
|
dev_dir = conf_dir / "dev"
|
|
svc_dir = conf_dir / "service"
|
|
dev_dir.mkdir(parents=True)
|
|
svc_dir.mkdir(parents=True)
|
|
|
|
dev_base = dev_dir / "knoe.cfg"
|
|
svc_base = svc_dir / "knoe.cfg"
|
|
dev_base.write_text("[Global]\nFOO = base\n")
|
|
svc_base.write_text("[Global]\nFOO = service\n")
|
|
|
|
entry = conf_dir / "knoe.cfg"
|
|
os.symlink(str(dev_base), str(entry))
|
|
|
|
with mock.patch.dict(os.environ, {"KNOE_CONF": str(conf_dir)}):
|
|
_update_knoe_cfg_value("Global", "FOO", "updated")
|
|
|
|
assert "FOO = updated" in dev_base.read_text()
|
|
assert svc_base.read_text() == "[Global]\nFOO = service\n"
|
|
|
|
def test_service_mode_drops_placeholder_endpoint_value(self, tmp_path):
|
|
cfg_file = tmp_path / "conf" / "knoe.cfg"
|
|
cfg_file.parent.mkdir(parents=True)
|
|
cfg_file.write_text("[Global]\nSERVICE_HOSTNAME = old-host\n")
|
|
|
|
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path):
|
|
_update_knoe_cfg_value(
|
|
"Global",
|
|
"SERVICE_HOSTNAME",
|
|
"${SERVICE_HOSTNAME}",
|
|
mode="k3s",
|
|
)
|
|
|
|
content = cfg_file.read_text()
|
|
assert "SERVICE_HOSTNAME" not in content
|
|
|
|
|
|
class TestPersistenceFilters:
|
|
def test_service_mode_drops_generated_endpoints_and_localhost(self):
|
|
values = {
|
|
"API_SERVICE_ENDPOINT": "http://localhost:8081",
|
|
"SERVICE_HOSTNAME": "k3d.localhost",
|
|
"PROLE_K3S_SERVER": "https://myrddin.knoe.org:6443",
|
|
"SERVICE_NAMESPACE": "knoe-system",
|
|
}
|
|
|
|
filtered = _filter_cfg_values_for_persistence(
|
|
"Global",
|
|
values,
|
|
mode="k3s",
|
|
explicit_keys={"PROLE_K3S_SERVER"},
|
|
)
|
|
|
|
assert filtered == {"PROLE_K3S_SERVER": "https://myrddin.knoe.org:6443"}
|
|
|
|
def test_service_mode_keeps_explicit_service_namespace(self):
|
|
filtered = _filter_cfg_values_for_persistence(
|
|
"Global",
|
|
{"SERVICE_NAMESPACE": "knoe-system"},
|
|
mode="k3s",
|
|
explicit_keys={"SERVICE_NAMESPACE"},
|
|
)
|
|
|
|
assert filtered == {"SERVICE_NAMESPACE": "knoe-system"}
|
|
|
|
def test_dev_mode_keeps_localhost_endpoint(self):
|
|
value = _normalize_cfg_value_for_persistence(
|
|
"Global",
|
|
"API_SERVICE_ENDPOINT",
|
|
"http://localhost:8081",
|
|
mode="k3d",
|
|
)
|
|
assert value == "http://localhost:8081"
|
|
|
|
|
|
# ===== is_apple_silicon / get_docker_build_platform_args =====
|
|
|
|
|
|
class TestPlatformHelpers:
|
|
def test_is_apple_silicon(self):
|
|
result = is_apple_silicon()
|
|
assert isinstance(result, bool)
|
|
|
|
def test_docker_args_env_override(self):
|
|
with mock.patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": "linux/arm64"}):
|
|
args = cfg_get_docker_args()
|
|
assert args == ["--platform", "linux/arm64"]
|
|
|
|
def test_docker_args_dev(self):
|
|
with mock.patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("PROLE_DOCKER_PLATFORM", None)
|
|
args = cfg_get_docker_args("dev")
|
|
assert isinstance(args, list)
|
|
|
|
|
|
# ===== _render_knoe_cfg =====
|
|
|
|
|
|
class TestRenderKnoeCfg:
|
|
def test_minimal(self):
|
|
from knoe.core.env import _render_knoe_cfg
|
|
|
|
result = _render_knoe_cfg(
|
|
inputs={
|
|
"env_setup.KNOE_HOME": "/opt/knoe",
|
|
"env_setup.NAMESPACE": "test-ns",
|
|
},
|
|
globals_to_save={"KNOE_HOME": "/opt/knoe", "NAMESPACE": "test-ns"},
|
|
sections={},
|
|
generated_at="2025-01-01 00:00:00",
|
|
)
|
|
assert "[Global]" in result
|
|
assert "KNOE_HOME" in result
|
|
assert "test-ns" in result or "NAMESPACE" in result
|
|
|
|
def test_with_sections(self):
|
|
from knoe.core.env import _render_knoe_cfg
|
|
|
|
result = _render_knoe_cfg(
|
|
inputs={
|
|
"env_setup.KNOE_HOME": "/opt/knoe",
|
|
"env_setup.KNOE_CONF": "/opt/knoe/conf",
|
|
"env_setup.PROLE_DATA": "/opt/knoe/data",
|
|
"env_setup.PROLE_LOGS": "/opt/knoe/logs",
|
|
"env_setup.KNOE_SERVICE": "/opt/knoe/etc",
|
|
"env_setup.NAMESPACE": "myns",
|
|
"init_password.db_namespace": "myns",
|
|
"init_password.db_username": "knoe",
|
|
"init_cluster.cluster_env": "dev",
|
|
"init_cluster.supabase_enabled": "false",
|
|
"init_cluster.kerberos_enabled": "false",
|
|
"init_cluster.at_rest_encryption_enabled": "false",
|
|
},
|
|
globals_to_save={"KNOE_HOME": "/opt/knoe", "NAMESPACE": "myns"},
|
|
sections={
|
|
"Database Creation": {"DB_USER": "knoe", "NAMESPACE": "myns"},
|
|
"Docker Build": {},
|
|
"Kerberos Authentication": {"ENABLED": "false"},
|
|
},
|
|
)
|
|
assert "[Database Creation]" in result
|
|
assert "[Inputs]" in result
|
|
|
|
def test_does_not_inject_placeholder_filesystem_paths(self):
|
|
"""New policy: knoe.cfg should not introduce `${...}` placeholders for filesystem paths."""
|
|
|
|
from knoe.core.env import _render_knoe_cfg
|
|
|
|
result = _render_knoe_cfg(
|
|
inputs={
|
|
# Simulate an operator/session where KNOE_HOME is known, but we
|
|
# do not want the renderer to normalize filesystem paths into
|
|
# placeholders.
|
|
"env_setup.KNOE_HOME": "",
|
|
"env_setup.KNOE_CONF": "",
|
|
"env_setup.PROLE_DATA": "",
|
|
"env_setup.PROLE_LOGS": "",
|
|
"env_setup.KNOE_SERVICE": "",
|
|
"env_setup.NAMESPACE": "test-ns",
|
|
},
|
|
globals_to_save={"KNOE_HOME": "/opt/knoe", "NAMESPACE": "test-ns"},
|
|
sections={},
|
|
generated_at="2025-01-01 00:00:00",
|
|
)
|
|
|
|
# Must not emit placeholder filesystem paths like `${KNOE_HOME}`.
|
|
assert "${KNOE_HOME}" not in result
|
|
assert "${KNOE_CONF}" not in result
|
|
assert "${PROLE_DATA}" not in result
|
|
assert "${PROLE_LOGS}" not in result
|
|
assert "${KNOE_SERVICE}" not in result
|
|
|
|
# Must not inject derived filesystem keys that rely on placeholder expansion.
|
|
assert "ANSIBLE_INFRASTRUCTURE" not in result
|
|
assert "ANSIBLE_INVENTORY" not in result
|
|
|
|
|
|
class TestRegistryManifestProbe:
|
|
def test_invalid_ref_returns_false(self):
|
|
assert _registry_image_ref_exists("") is False
|
|
assert _registry_image_ref_exists("alpine:latest") is False
|
|
|
|
def test_head_200_true_and_no_get(self, monkeypatch):
|
|
import http.client
|
|
|
|
requests: list[tuple[str, str]] = []
|
|
|
|
class _Resp:
|
|
def __init__(self, status: int):
|
|
self.status = status
|
|
|
|
def read(self):
|
|
return b""
|
|
|
|
class _Conn:
|
|
def __init__(self, host, port, timeout=None):
|
|
self.host = host
|
|
self.port = port
|
|
self.timeout = timeout
|
|
self._method = None
|
|
self._path = None
|
|
|
|
def request(self, method, path, headers=None):
|
|
requests.append((method, path))
|
|
self._method = method
|
|
self._path = path
|
|
|
|
def getresponse(self):
|
|
assert self._method is not None
|
|
return _Resp(200)
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(http.client, "HTTPConnection", _Conn)
|
|
|
|
assert _registry_image_ref_exists("myrddin.knoe.org:5000/knoe-db:latest") is True
|
|
assert requests and requests[0][0] == "HEAD"
|
|
assert all(m != "GET" for m, _ in requests)
|
|
|
|
def test_head_404_false(self, monkeypatch):
|
|
import http.client
|
|
|
|
class _Resp:
|
|
def __init__(self, status: int):
|
|
self.status = status
|
|
|
|
def read(self):
|
|
return b""
|
|
|
|
class _Conn:
|
|
def __init__(self, host, port, timeout=None):
|
|
self._method = None
|
|
|
|
def request(self, method, path, headers=None):
|
|
self._method = method
|
|
|
|
def getresponse(self):
|
|
return _Resp(404)
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(http.client, "HTTPConnection", _Conn)
|
|
assert _registry_image_ref_exists("myrddin.knoe.org:5000/knoe-db:latest") is False
|
|
|
|
def test_head_405_falls_back_to_get(self, monkeypatch):
|
|
import http.client
|
|
|
|
requests: list[str] = []
|
|
|
|
class _Resp:
|
|
def __init__(self, status: int):
|
|
self.status = status
|
|
|
|
def read(self):
|
|
return b""
|
|
|
|
class _Conn:
|
|
def __init__(self, host, port, timeout=None):
|
|
self._method = None
|
|
|
|
def request(self, method, path, headers=None):
|
|
self._method = method
|
|
requests.append(method)
|
|
|
|
def getresponse(self):
|
|
assert self._method is not None
|
|
if self._method == "HEAD":
|
|
return _Resp(405)
|
|
return _Resp(200)
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(http.client, "HTTPConnection", _Conn)
|
|
assert _registry_image_ref_exists("myrddin.knoe.org:5000/knoe-db:latest") is True
|
|
assert requests == ["HEAD", "GET"]
|