CloudRouterAdvanced/terraform/tests/test_terraform_delivery.py

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

None of these tests run terraform plan/apply or call any cloud API - no
credentials and no network access to VK Cloud itself are required or used.
`terraform init`/`validate` do run for real, but against a project-local
filesystem-mirror copy of the vkcs provider binary (see
setup-local-terraform.sh), so they only need the provider's static schema,
never a live cloud endpoint. Checks performed:

  * the expected files exist,
  * terraform fmt reports the tree as already formatted,
  * the .tf files parse as valid HCL,
  * `terraform init` + `terraform validate` succeed against the real vkcs
    provider schema (via the local filesystem mirror - skipped if that
    mirror hasn't been set up),
  * router VM count and private-interface count are wired to variables
    (not hardcoded), and those variables aren't shadowed by terraform.tfvars,
  * the per-router/per-interface CIDR carving never produces overlapping
    subnets, for a range of router_count / private_interface_count values,
  * 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 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_interface_count",
    [
        (None, None),  # defaults: 2, 2
        (1, 1),
        (4, 3),
    ],
)
def test_terraform_init_and_validate_against_real_provider_schema(
    router_count, private_interface_count
):
    """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.

    Parametrized over router_count/private_interface_count (set via
    TF_VAR_*, exactly how horizontal scaling is meant to be driven) to
    prove the delivery actually validates 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_interface_count is not None:
        env["TF_VAR_private_interface_count"] = str(private_interface_count)

    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}"
        )


# ---------------------------------------------------------------------------
# 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 with sane defaults
# ---------------------------------------------------------------------------


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_interface_count_variable():
    v = find_variable(load_tf("variables.tf"), "private_interface_count")
    assert v is not None, "variable private_interface_count is missing"
    assert v["type"] == ["${number}"]
    assert v["default"] == [2]


@pytest.mark.parametrize("name", ["router_count", "private_interface_count"])
def test_scaling_variable_not_pinned_in_tfvars(name):
    """TF_VAR_<name> env-var overrides only take effect if terraform.tfvars
    doesn't set an active (non-comment) value for the same variable."""
    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(rf"^\s*{name}\s*=", line) for line in active_lines), (
        f"{name} must not be set in terraform.tfvars, or the "
        f"TF_VAR_{name} environment variable would be shadowed"
    )


# ---------------------------------------------------------------------------
# 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 i in range(var.private_interface_count) : "priv${i + 1}"]}'
    ], "locals.private_roles must be generated from var.private_interface_count"


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 private_interface_count"
    )


# ---------------------------------------------------------------------------
# CIDR carving: router_index/role_index -> netnum must never collide, for a
# range of router_count / private_interface_count combinations
# ---------------------------------------------------------------------------


def cidrsubnet(supernet, newbits, netnum):
    """Pure-Python re-implementation of Terraform's cidrsubnet() built-in."""
    net = ipaddress.ip_network(supernet)
    subnets = list(net.subnets(new_prefix=net.prefixlen + newbits))
    return subnets[netnum]


@pytest.mark.parametrize(
    "router_count,private_interface_count",
    [(1, 1), (2, 2), (3, 2), (4, 3), (8, 4), (1, 6)],
)
def test_private_subnet_carving_has_no_collisions(router_count, private_interface_count):
    supernet = "10.90.0.0/16"
    prefix_length = 29
    newbits = prefix_length - ipaddress.ip_network(supernet).prefixlen

    subnets = [
        cidrsubnet(
            supernet, newbits, router_index * private_interface_count + role_index
        )
        for router_index in range(router_count)
        for role_index in range(private_interface_count)
    ]

    assert len(subnets) == len(set(subnets)), "duplicate per-router private subnets"
    for i, a in enumerate(subnets):
        for b in subnets[i + 1 :]:
            assert not a.overlaps(b), f"{a} overlaps {b}"


def test_private_subnet_pool_capacity_is_generous():
    supernet = ipaddress.ip_network("10.90.0.0/16")
    prefix_length = 29
    capacity = 2 ** (prefix_length - supernet.prefixlen)
    assert capacity >= 1000, "default private_supernet/prefix combo has too little headroom"


# ---------------------------------------------------------------------------
# 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}"
    )