CloudRouterAdvanced/terraform/tests/test_terraform_delivery.py

"""Local integrity tests for the terraform/ delivery.

None of these tests run terraform apply or call any cloud API - no
credentials and no network access to VK Cloud itself are required or used.

Two different kinds of "real terraform" checks are used here, deliberately:

  * `terraform init` + `terraform validate` against the actual vkcs provider
    schema (via a project-local filesystem-mirror copy of the provider
    binary - see setup-local-terraform.sh) - this proves the config's
    resource/attribute shapes are compatible with the real provider.
    IMPORTANT: `terraform validate` does NOT enforce custom variable
    `validation { ... }` blocks for externally-supplied values (verified
    empirically against this terraform binary - only `plan`/`apply` do), so
    it says nothing about whether e.g. private_network_cidrs is actually
    checked for uniqueness.
  * `terraform plan` against an isolated copy of just variables.tf (no
    provider, no resources) to exercise those variable validation blocks
    for real, entirely offline.

Other checks performed: expected files exist, `terraform fmt` is clean, the
.tf files parse as valid HCL, router VM count and the private-network CIDR
list actually drive the resource/NIC count (not hardcoded), the example
CIDRs in terraform.tfvars don't overlap and have room for router_count
hosts, and the post-install script template only contains the intended
Terraform interpolations and renders to syntactically valid bash.

Run via: venv/bin/pytest terraform/tests -v
(first run terraform/tests/setup-local-terraform.sh once to provision the
 local terraform binary + vkcs provider mirror used by the init/validate test)
"""
import ipaddress
import json
import os
import re
import shutil
import subprocess
import tempfile
from pathlib import Path

import hcl2
import pytest

TERRAFORM_DIR = Path(__file__).resolve().parent.parent
REPO_ROOT = TERRAFORM_DIR.parent
TERRAFORM_BIN = (
    str(REPO_ROOT / "venv" / "bin" / "terraform")
    if (REPO_ROOT / "venv" / "bin" / "terraform").is_file()
    else shutil.which("terraform")
)
CLI_CONFIG_FILE = REPO_ROOT / "venv" / "terraform.d" / "cli-config.tfrc"
CHECKOV_BIN = (
    str(REPO_ROOT / "venv" / "bin" / "checkov")
    if (REPO_ROOT / "venv" / "bin" / "checkov").is_file()
    else shutil.which("checkov")
)


def load_tf(relpath):
    with open(TERRAFORM_DIR / relpath) as f:
        return hcl2.load(f)


def find_resources(doc, rtype):
    """Yield (name, attrs) for every resource of a given type in a parsed doc."""
    for block in doc.get("resource", []):
        if rtype in block:
            yield from block[rtype].items()


def find_variable(doc, name):
    for block in doc.get("variable", []):
        if name in block:
            return block[name]
    return None


# ---------------------------------------------------------------------------
# Delivery layout
# ---------------------------------------------------------------------------

REQUIRED_FILES = [
    "main.tf",
    "variables.tf",
    "terraform.tfvars",
    "scripts/network-init.sh.tpl",
]


@pytest.mark.parametrize("relpath", REQUIRED_FILES)
def test_required_file_exists(relpath):
    assert (TERRAFORM_DIR / relpath).is_file(), f"missing terraform/{relpath}"


def test_legacy_post_install_script_removed():
    assert not (TERRAFORM_DIR / "scripts" / "network-init.sh").exists(), (
        "old non-templated network-init.sh should have been replaced by "
        "network-init.sh.tpl"
    )


# ---------------------------------------------------------------------------
# terraform fmt - the one check here that actually shells out to the
# terraform binary itself ("средствами terraform"); everything else below
# is local HCL parsing / pure-Python re-implementation of the expressions.
# ---------------------------------------------------------------------------


def test_terraform_fmt_clean():
    assert TERRAFORM_BIN, "no terraform binary found (checked venv/bin and PATH)"
    result = subprocess.run(
        [TERRAFORM_BIN, "fmt", "-check", "-diff", "-recursive", str(TERRAFORM_DIR)],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, (
        f"terraform fmt found unformatted files:\n{result.stdout}\n{result.stderr}"
    )


@pytest.mark.parametrize(
    "router_count,private_network_cidrs",
    [
        (None, None),  # whatever terraform.tfvars already commits to
        (1, ["10.90.0.0/29"]),
        (4, ["10.90.0.0/28", "10.90.0.16/28", "10.90.0.32/28"]),
    ],
)
def test_terraform_init_and_validate_against_real_provider_schema(
    router_count, private_network_cidrs
):
    """Real `terraform init` + `terraform validate` against the actual vkcs
    provider, using the project-local filesystem-mirror copy of the
    provider binary (downloaded from its GitHub releases by
    setup-local-terraform.sh, bypassing the region-blocked HashiCorp
    registry). Runs in a throwaway temp copy of terraform/ so it never
    leaves .terraform/ or .terraform.lock.hcl behind in the real delivery.
    No credentials are supplied and no cloud API is contacted - validate
    only needs the provider's static schema to type-check the config (it
    does not enforce the variable validation{} blocks - see
    test_variable_validations_are_enforced_by_plan for that).

    Parametrized over router_count/private_network_cidrs (set via TF_VAR_*,
    exactly how horizontal scaling is meant to be driven) to prove the
    delivery actually resolves at other scales, not just the defaults.
    """
    assert TERRAFORM_BIN, "no terraform binary found (checked venv/bin and PATH)"
    if not CLI_CONFIG_FILE.is_file():
        pytest.skip(
            "local vkcs provider mirror not set up - run "
            "terraform/tests/setup-local-terraform.sh once"
        )

    env = dict(os.environ)
    env["TF_CLI_CONFIG_FILE"] = str(CLI_CONFIG_FILE)
    if router_count is not None:
        env["TF_VAR_router_count"] = str(router_count)
    if private_network_cidrs is not None:
        env["TF_VAR_private_network_cidrs"] = json.dumps(private_network_cidrs)

    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        for item in TERRAFORM_DIR.iterdir():
            if item.name == "tests":
                continue
            if item.is_dir():
                shutil.copytree(item, tmp_path / item.name)
            else:
                shutil.copy2(item, tmp_path / item.name)

        init = subprocess.run(
            [TERRAFORM_BIN, f"-chdir={tmp_path}", "init", "-backend=false", "-input=false"],
            capture_output=True,
            text=True,
            env=env,
        )
        assert init.returncode == 0, f"terraform init failed:\n{init.stdout}\n{init.stderr}"

        validate = subprocess.run(
            [TERRAFORM_BIN, f"-chdir={tmp_path}", "validate"],
            capture_output=True,
            text=True,
            env=env,
        )
        assert validate.returncode == 0, (
            f"terraform validate failed:\n{validate.stdout}\n{validate.stderr}"
        )


def _plan_variables_only(var_overrides):
    """Run `terraform plan` against an isolated copy of just variables.tf -
    no provider, no resources, so this never touches any cloud API. Used to
    exercise variable validation{} blocks for real: `terraform validate`
    does not enforce them for externally-supplied values in this terraform
    version (verified empirically), only `plan`/`apply` do.

    Returns (success: bool, combined stdout+stderr: str).
    """
    assert TERRAFORM_BIN, "no terraform binary found (checked venv/bin and PATH)"
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        shutil.copy2(TERRAFORM_DIR / "variables.tf", tmp_path / "variables.tf")

        env = dict(os.environ)
        env.pop("TF_CLI_CONFIG_FILE", None)
        env.update(
            {
                "TF_VAR_username": "dummy",
                "TF_VAR_password": "dummy",
                "TF_VAR_project_id": "dummy",
                "TF_VAR_ssh_key_name": "dummy",
                "TF_VAR_private_network_cidrs": json.dumps(["10.90.0.0/29"]),
            }
        )
        env.update(var_overrides)

        init = subprocess.run(
            [TERRAFORM_BIN, f"-chdir={tmp_path}", "init", "-backend=false", "-input=false"],
            capture_output=True,
            text=True,
            env=env,
        )
        assert init.returncode == 0, f"terraform init failed:\n{init.stdout}\n{init.stderr}"

        plan = subprocess.run(
            [TERRAFORM_BIN, f"-chdir={tmp_path}", "plan", "-input=false"],
            capture_output=True,
            text=True,
            env=env,
        )
        return plan.returncode == 0, plan.stdout + plan.stderr


@pytest.mark.parametrize(
    "cidrs,should_pass",
    [
        (["10.90.0.0/29", "10.90.0.8/29"], True),
        ([], False),  # must contain at least one CIDR
        (["not-a-cidr"], False),  # must be a valid IPv4 CIDR
        (["10.90.0.0/29", "10.90.0.0/29"], False),  # must be unique
    ],
)
def test_private_network_cidrs_validation_is_enforced(cidrs, should_pass):
    ok, output = _plan_variables_only({"TF_VAR_private_network_cidrs": json.dumps(cidrs)})
    assert ok == should_pass, f"unexpected result for private_network_cidrs={cidrs!r}:\n{output}"


@pytest.mark.parametrize("count,should_pass", [(2, True), (0, False), (-1, False)])
def test_router_count_validation_is_enforced(count, should_pass):
    ok, output = _plan_variables_only({"TF_VAR_router_count": str(count)})
    assert ok == should_pass, f"unexpected result for router_count={count}:\n{output}"


# ---------------------------------------------------------------------------
# HCL parses cleanly
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("relpath", ["main.tf", "variables.tf", "images.tf"])
def test_hcl_file_parses(relpath):
    # images.tf is intentionally fully commented out (disabled helper data
    # source) - it must still parse cleanly, just possibly to an empty doc.
    doc = load_tf(relpath)
    assert isinstance(doc, dict)


# ---------------------------------------------------------------------------
# variables.tf: the scaling knobs exist as expected
# ---------------------------------------------------------------------------


def test_router_count_variable():
    v = find_variable(load_tf("variables.tf"), "router_count")
    assert v is not None, "variable router_count is missing"
    assert v["type"] == ["${number}"]
    assert v["default"] == [2]


def test_private_network_cidrs_variable():
    v = find_variable(load_tf("variables.tf"), "private_network_cidrs")
    assert v is not None, "variable private_network_cidrs is missing"
    assert v["type"] == ["${list(string)}"]
    assert "default" not in v, (
        "private_network_cidrs must NOT have a default - the admin is "
        "required to pass it explicitly"
    )
    assert len(v.get("validation", [])) >= 3, (
        "expected validations for: non-empty, valid CIDR syntax, uniqueness"
    )


def test_router_count_not_pinned_in_tfvars():
    """TF_VAR_router_count only takes effect if terraform.tfvars doesn't set
    an active (non-comment) value for it."""
    tfvars_text = (TERRAFORM_DIR / "terraform.tfvars").read_text()
    active_lines = [
        line for line in tfvars_text.splitlines() if not line.strip().startswith("#")
    ]
    assert not any(re.match(r"^\s*router_count\s*=", line) for line in active_lines), (
        "router_count must not be set in terraform.tfvars, or the "
        "TF_VAR_router_count environment variable would be shadowed"
    )


def test_private_network_cidrs_is_set_in_tfvars():
    """Unlike router_count, private_network_cidrs has no default, so
    terraform.tfvars must actively set it for a working example deployment."""
    tfvars_text = (TERRAFORM_DIR / "terraform.tfvars").read_text()
    active_lines = [
        line for line in tfvars_text.splitlines() if not line.strip().startswith("#")
    ]
    assert any(re.match(r"^\s*private_network_cidrs\s*=", line) for line in active_lines), (
        "private_network_cidrs has no default and must be set in terraform.tfvars"
    )


# ---------------------------------------------------------------------------
# main.tf: router VM count and NIC count are wired to those variables, not
# hardcoded
# ---------------------------------------------------------------------------


def test_router_resource_uses_count_variable():
    main = load_tf("main.tf")
    instances = dict(find_resources(main, "vkcs_compute_instance"))
    assert "router" in instances, "expected a single vkcs_compute_instance.router resource"
    assert instances["router"]["count"] == ["${var.router_count}"]


def test_no_legacy_hardcoded_router_resources():
    main = load_tf("main.tf")
    instance_names = {name for name, _ in find_resources(main, "vkcs_compute_instance")}
    assert instance_names.isdisjoint({"router1", "router2"}), (
        "found legacy hardcoded router1/router2 instances instead of the "
        "count-based router resource"
    )
    port_names = {name for name, _ in find_resources(main, "vkcs_networking_port")}
    assert port_names.isdisjoint({"lan_port1", "lan_port2"}), (
        "found legacy hardcoded lan_port1/lan_port2 instead of the "
        "for_each-based router_priv_port"
    )


def test_private_roles_local_driven_by_variable():
    main = load_tf("main.tf")
    locals_block = main["locals"][0]
    assert locals_block["private_roles"] == [
        '${[for idx in range(length(var.private_network_cidrs)) : "priv${idx + 1}"]}'
    ], "locals.private_roles must be generated from length(var.private_network_cidrs)"


def test_router_network_blocks_scale_with_private_roles():
    main = load_tf("main.tf")
    router = dict(find_resources(main, "vkcs_compute_instance"))["router"]
    dynamic_network = router["dynamic"][0]["network"]
    assert dynamic_network["for_each"] == ["${local.private_roles}"], (
        "the dynamic private network blocks must iterate local.private_roles "
        "so the NIC count scales with the number of private_network_cidrs entries"
    )


# ---------------------------------------------------------------------------
# terraform.tfvars example CIDRs: sanity-check the values actually shipped
# (no auto-carving anymore - these come straight from the admin/example)
# ---------------------------------------------------------------------------


def test_tfvars_private_network_cidrs_do_not_overlap_and_have_room_for_routers():
    tfvars = load_tf("terraform.tfvars")
    raw_cidrs = tfvars["private_network_cidrs"][0]
    networks = [ipaddress.ip_network(c) for c in raw_cidrs]
    assert networks, "terraform.tfvars must set at least one private_network_cidrs entry"

    for i, a in enumerate(networks):
        for b in networks[i + 1 :]:
            assert not a.overlaps(b), f"{a} overlaps {b} in terraform.tfvars"

    router_count_default = find_variable(load_tf("variables.tf"), "router_count")["default"][0]
    for net in networks:
        # offset scheme: cidrhost(cidr, router_index + 2) for router_index in
        # 0..router_count-1, so we need at least router_count + 2 addresses.
        assert net.num_addresses >= router_count_default + 2, (
            f"{net} has too few addresses for {router_count_default} routers "
            f"(offset scheme needs router_count + 2)"
        )


# ---------------------------------------------------------------------------
# network-init.sh.tpl: only the intended Terraform interpolations remain
# un-escaped, and the rendered result is syntactically valid bash
# ---------------------------------------------------------------------------


def _script_template_text():
    return (TERRAFORM_DIR / "scripts" / "network-init.sh.tpl").read_text()


def test_network_init_template_only_intended_interpolations():
    text = _script_template_text()
    # A real Terraform interpolation is "${" not preceded by another "$".
    # Pre-existing bash brace-expansions must be escaped as "$${".
    real_interpolations = re.findall(r"(?<!\$)\$\{([^}]*)\}", text)
    assert real_interpolations, "expected at least the pi.cidr/pi.name interpolations"
    for expr in real_interpolations:
        assert expr.startswith("pi."), (
            f"unexpected un-escaped Terraform interpolation in the script "
            f"template: ${{{expr}}} (bash brace-expansions must use $${{...}})"
        )


def test_network_init_template_has_for_directive():
    text = _script_template_text()
    assert "%{ for pi in private_interfaces" in text
    assert "%{ endfor" in text


def test_network_init_template_renders_to_valid_bash():
    """Simulate templatefile() rendering (unescape $${ -> ${, substitute a
    fake private_interfaces list for the %{ for } directive) and check the
    result is syntactically valid bash. Fully offline, no cloud calls."""
    text = _script_template_text()
    rendered = text.replace("$${", "${")

    for_block = re.compile(
        r"%\{ for pi in private_interfaces ~\}\n.*?\n%\{ endfor ~\}\n", re.S
    )
    assert for_block.search(rendered), "template for-directive not found for rendering"
    rendered = for_block.sub(
        'PRIV_TARGETS+=("10.90.0.8/29|eth1")\n'
        'PRIV_TARGETS+=("10.90.0.16/29|eth2")\n'
        'PRIV_TARGETS+=("10.90.0.24/29|eth3")\n',
        rendered,
    )

    assert "%{" not in rendered
    assert "${pi." not in rendered

    result = subprocess.run(
        ["bash", "-n"], input=rendered, capture_output=True, text=True
    )
    assert result.returncode == 0, f"rendered script has a bash syntax error:\n{result.stderr}"


# ---------------------------------------------------------------------------
# checkov smoke check (documented limitation: checkov ships no policies for
# the vkcs provider, so this only guards against the tool itself breaking -
# it intentionally does not assert resource_count > 0)
# ---------------------------------------------------------------------------


def test_checkov_runs_offline_without_error():
    if CHECKOV_BIN is None:
        pytest.skip("checkov not installed in this environment")
    result = subprocess.run(
        [CHECKOV_BIN, "-d", str(TERRAFORM_DIR), "--framework", "terraform",
         "--skip-download", "--compact", "-o", "json"],
        capture_output=True,
        text=True,
    )
    assert result.returncode in (0, 1), (
        f"checkov exited unexpectedly ({result.returncode}):\n{result.stderr}"
    )