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

# The real mvm-s3 shape (two externally-owned, fixed-IP networks - matches
# terraform/mvm-s3.tfvars), reused by several tests below.
MVM_S3_ROUTER_NETWORKS = {
    "primary": {
        "network_id": "25532efe-2931-4666-a090-0da3a6f18224",
        "subnet_id": "1fa5357c-1f11-4d20-89f4-b27a4fb56e6e",
        "cidr": "172.16.252.8/29",
        "ip_addresses": ["172.16.252.11", "172.16.252.12", "172.16.252.13", "172.16.252.14"],
    },
    "backup": {
        "network_id": "2b4cc25f-55a1-4d36-a3a2-5b16414af31a",
        "subnet_id": "5eacbc86-e15d-4c49-8704-547a719acc23",
        "cidr": "172.16.252.0/29",
        "ip_addresses": ["172.16.252.3", "172.16.252.4", "172.16.252.5", "172.16.252.6"],
    },
}


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


def find_local(doc, name):
    """main.tf has more than one `locals { ... }` block - search all of them."""
    for block in doc.get("locals", []):
        if name in block:
            return block[name]
    return None


def find_data_sources(doc, dtype):
    """Yield (name, attrs) for every data source of a given type in a parsed doc."""
    for block in doc.get("data", []):
        if dtype in block:
            yield from block[dtype].items()


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

REQUIRED_FILES = [
    "main.tf",
    "variables.tf",
    "versions.tf",
    "terraform.tfvars",
    "prod.secrets.tfvars.example",
    "mvm-s3.tfvars",
    "mvm-s3.secrets.tfvars.example",
    "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,router_networks",
    [
        (None, None, None),  # whatever terraform.tfvars already commits to
        (1, ["10.90.0.0/29"], None),
        (4, ["10.90.0.0/28", "10.90.0.16/28", "10.90.0.32/28"], None),
        # mvm-s3-shaped: no project-managed private networks, two
        # externally-owned fixed-IP ones instead.
        (4, [], MVM_S3_ROUTER_NETWORKS),
    ],
)
def test_terraform_init_and_validate_against_real_provider_schema(
    router_count, private_network_cidrs, router_networks
):
    """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/router_networks
    (set via TF_VAR_*, exactly how horizontal scaling and the mvm-s3-style
    fixed-IP deployment are meant to be driven) to prove the delivery
    actually resolves at other scales/shapes, 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)
    if router_networks is not None:
        env["TF_VAR_router_networks"] = json.dumps(router_networks)

    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),
        ([], True),  # optional now - a router_networks-only deployment (e.g. mvm-s3) sets none
        (["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}"


def _router_networks_case(**override):
    net = json.loads(json.dumps(MVM_S3_ROUTER_NETWORKS))  # deep copy
    net["primary"].update(override)
    return net


@pytest.mark.parametrize(
    "router_networks,should_pass",
    [
        (MVM_S3_ROUTER_NETWORKS, True),
        ({}, True),  # optional - a private_network_cidrs-only deployment (e.g. PROD) sets none
        (_router_networks_case(cidr="not-a-cidr"), False),  # cidr must be a valid IPv4 CIDR
        (_router_networks_case(ip_addresses=["not-an-ip"]), False),  # ip must be a valid IPv4 address
        (_router_networks_case(ip_addresses=["10.0.0.1"]), False),  # ip must fall inside its own cidr
    ],
)
def test_router_networks_validation_is_enforced(router_networks, should_pass):
    ok, output = _plan_variables_only({"TF_VAR_router_networks": json.dumps(router_networks)})
    assert ok == should_pass, f"unexpected result for router_networks={router_networks!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}"


@pytest.mark.parametrize(
    "override,should_pass",
    [
        (None, True),  # default null - no override, dynamic lookup is used
        ("11111111-1111-1111-1111-111111111111", True),  # explicit override accepted
    ],
)
def test_default_security_group_id_override_accepted(override, should_pass):
    overrides = {}
    if override is not None:
        overrides["TF_VAR_default_security_group_id"] = override
    ok, output = _plan_variables_only(overrides)
    assert ok == should_pass, f"unexpected result for default_security_group_id={override!r}:\n{output}"


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


@pytest.mark.parametrize("relpath", ["main.tf", "variables.tf", "versions.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)


# ---------------------------------------------------------------------------
# versions.tf: the provider is actually configured (auth variables used to
# be declared in variables.tf but never wired to anything - regression
# guard against that gap reappearing)
# ---------------------------------------------------------------------------


def test_provider_vkcs_wires_all_auth_variables():
    versions = load_tf("versions.tf")
    provider_blocks = versions.get("provider", [])
    vkcs_provider = None
    for block in provider_blocks:
        if "vkcs" in block:
            vkcs_provider = block["vkcs"]
    assert vkcs_provider is not None, "expected a provider \"vkcs\" block in versions.tf"

    expected = {
        "auth_url": "${var.auth_url}",
        "username": "${var.username}",
        "password": "${var.password}",
        "project_id": "${var.project_id}",
        "region": "${var.region}",
        "user_domain_name": "${var.user_domain_name}",
    }
    for attr, expr in expected.items():
        assert vkcs_provider.get(attr) == [expr], (
            f"provider \"vkcs\" must set {attr} = var.{attr} - "
            f"auth variables must not be left orphaned/unwired"
        )


@pytest.mark.parametrize("name", ["auth_url", "user_domain_name", "region"])
def test_auth_variable_has_a_default(name):
    v = find_variable(load_tf("variables.tf"), name)
    assert v is not None, f"variable {name} is missing"
    assert "default" in v, f"{name} should have a sane default so it doesn't have to be set explicitly for the common case"


# ---------------------------------------------------------------------------
# Real secrets must never land in a git-tracked tfvars file - each
# environment's creds belong in its own gitignored *.secrets.tfvars, passed
# explicitly via -var-file (see *.secrets.tfvars.example). Two environments
# now share this terraform/ directory (PROD and mvm-s3), so unlike the old
# single-environment *.auto.tfvars convention, nothing here may rely on
# Terraform's automatic tfvars loading for secrets.
# ---------------------------------------------------------------------------


def test_gitignore_excludes_secrets_tfvars_overlay():
    gitignore_text = (REPO_ROOT / ".gitignore").read_text()
    assert "*.secrets.tfvars" in gitignore_text, (
        "*.secrets.tfvars must be gitignored - real per-environment "
        "credentials are meant to be passed via such a file with an "
        "explicit -var-file, never committed"
    )


@pytest.mark.parametrize(
    "relpath", ["prod.secrets.tfvars.example", "mvm-s3.secrets.tfvars.example"]
)
def test_secrets_tfvars_example_is_not_gitignored(relpath):
    """Each *.example file documents the -var-file pattern for one
    environment and must ship in the repo (unlike the real *.secrets.tfvars
    it documents)."""
    result = subprocess.run(
        ["git", "check-ignore", f"terraform/{relpath}"],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0, f"{relpath} must NOT be gitignored"


@pytest.mark.parametrize("relpath", ["terraform.tfvars", "mvm-s3.tfvars"])
def test_committed_tfvars_have_no_auth_credentials(relpath):
    """terraform.tfvars and mvm-s3.tfvars are committed, non-secret shape
    templates - auth_url/user_domain_name/real credentials belong in each
    environment's gitignored *.secrets.tfvars overlay, not here."""
    tfvars_text = (TERRAFORM_DIR / relpath).read_text()
    active_lines = [
        line for line in tfvars_text.splitlines() if not line.strip().startswith("#")
    ]
    for forbidden in ("auth_url", "user_domain_name"):
        assert not any(re.match(rf"^\s*{forbidden}\s*=", line) for line in active_lines), (
            f"{forbidden} should not be set in the committed {relpath} "
            f"template - use a gitignored *.secrets.tfvars overlay instead"
        )


# ---------------------------------------------------------------------------
# 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 v.get("default") == [[]], (
        "private_network_cidrs must default to [] - a deployment that only "
        "uses var.router_networks (e.g. mvm-s3) needs no project-managed "
        "private networks at all"
    )
    assert len(v.get("validation", [])) >= 2, (
        "expected validations for: valid CIDR syntax, uniqueness"
    )


def test_router_networks_variable():
    v = find_variable(load_tf("variables.tf"), "router_networks")
    assert v is not None, "variable router_networks is missing"
    assert v.get("default") == [{}], (
        "router_networks must default to {} - a deployment that only uses "
        "var.private_network_cidrs (e.g. PROD) needs no externally-owned "
        "fixed-IP networks at all"
    )
    assert len(v.get("validation", [])) >= 3, (
        "expected validations for: valid CIDR syntax, valid IPv4 addresses, "
        "each address falling inside its own cidr"
    )


def test_router_count_pinned_in_tfvars_is_a_valid_number():
    """terraform.tfvars now pins a real router_count for this project's
    actual PROD deployment (a deliberate choice, confirmed with the user) -
    a tfvars-file value always beats a TF_VAR_ environment variable in
    Terraform's precedence order, so TF_VAR_router_count no longer has any
    effect while this stays set. Just sanity-check it's a positive integer,
    not that it's absent."""
    tfvars_text = (TERRAFORM_DIR / "terraform.tfvars").read_text()
    active_lines = [
        line for line in tfvars_text.splitlines() if not line.strip().startswith("#")
    ]
    matches = [line for line in active_lines if re.match(r"^\s*router_count\s*=", line)]
    assert matches, "expected router_count to be pinned in terraform.tfvars for this deployment"
    value = int(matches[0].split("=", 1)[1].strip())
    assert value >= 1


def test_private_network_cidrs_is_set_in_tfvars():
    """private_network_cidrs defaults to [], but PROD's terraform.tfvars
    still pins its real project-managed networks explicitly."""
    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_compute_instances_do_not_set_top_level_image_id():
    """Regression guard: the provider docs say 'Do not specify [image_id]
    if booting from a volume' - doing so anyway caused every already-created
    instance to be flagged for destroy+recreate on the next plan, because
    Nova reports back a sentinel string ("Attempt to boot from volume - no
    image supplied") instead of echoing the image UUID for a volume-booted
    server, which Terraform then sees as configuration drift on a
    ForceNew attribute. The image only belongs inside block_device."""
    main = load_tf("main.tf")
    for name, attrs in find_resources(main, "vkcs_compute_instance"):
        assert "image_id" not in attrs, (
            f"vkcs_compute_instance.{name} must not set top-level image_id "
            f"when booting from a volume via block_device"
        )
        assert attrs["block_device"][0]["source_type"] == ["image"]
        assert "uuid" in attrs["block_device"][0]


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


def test_deployment_is_router_only():
    """This deployment provisions only the router VMs - the demo's
    priv_srv_01/02/03 instances and the shared LAN network/security group
    that only they used were removed as out of scope (confirmed with the
    user)."""
    main = load_tf("main.tf")
    instance_names = {name for name, _ in find_resources(main, "vkcs_compute_instance")}
    assert instance_names == {"router"}, f"expected only the router instance, found {instance_names}"

    network_names = {name for name, _ in find_resources(main, "vkcs_networking_network")}
    assert "lan_net" not in network_names

    secgroup_names = {name for name, _ in find_resources(main, "vkcs_networking_secgroup")}
    assert secgroup_names == {"router_sg"}, f"expected only router_sg, found {secgroup_names}"


def test_router_interfaces_local_merges_both_network_sources():
    """locals.router_interfaces unifies the two ways a router can get a
    private interface: project-managed (private_network_cidrs, network/
    subnet created by this Terraform) and externally-owned, fixed-IP
    (router_networks, referenced by UUID only - e.g. the mvm-s3
    environment)."""
    main = load_tf("main.tf")
    router_interfaces = find_local(main, "router_interfaces")
    assert router_interfaces is not None, "locals.router_interfaces is missing"
    expr = router_interfaces[0]
    assert expr.startswith("${merge("), "router_interfaces must be built via merge(...)"
    assert "for role , cidr in local.private_network_cidr" in expr, (
        "router_interfaces must include the project-managed private_network_cidrs roles"
    )
    assert "for role , net in var.router_networks" in expr, (
        "router_interfaces must include the externally-owned router_networks roles"
    )


def test_router_interface_roles_local_is_keys_of_router_interfaces():
    main = load_tf("main.tf")
    assert find_local(main, "router_interface_roles") == [
        "${keys(local.router_interfaces)}"
    ], "locals.router_interface_roles must be keys(local.router_interfaces)"


def test_router_network_blocks_scale_with_router_interface_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.router_interface_roles}"], (
        "the dynamic private network blocks must iterate "
        "local.router_interface_roles so the NIC count scales with both "
        "private_network_cidrs and router_networks entries"
    )


# ---------------------------------------------------------------------------
# main.tf: the "default" security group UUID is resolved dynamically, not
# hardcoded (it's unique per VK Cloud project)
# ---------------------------------------------------------------------------


def test_default_security_group_data_source_exists():
    main = load_tf("main.tf")
    secgroups = dict(find_data_sources(main, "vkcs_networking_secgroup"))
    assert "default" in secgroups, "expected data.vkcs_networking_secgroup.default"
    assert secgroups["default"]["name"] == ["default"]


def test_default_security_group_id_local_prefers_override_then_lookup():
    main = load_tf("main.tf")
    value = find_local(main, "default_security_group_id")
    assert value == [
        "${coalesce(var.default_security_group_id,"
        "data.vkcs_networking_secgroup.default.id)}"
    ], (
        "locals.default_security_group_id must fall back to the dynamic "
        "lookup unless var.default_security_group_id is explicitly overridden"
    )


def test_no_hardcoded_security_group_uuid_in_main():
    text = (TERRAFORM_DIR / "main.tf").read_text()
    uuid_pattern = re.compile(
        r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I
    )
    assert not uuid_pattern.search(text), (
        "main.tf must not contain a literal UUID (e.g. a hardcoded security "
        "group ID) - it's unique per project and must be resolved dynamically "
        "via data.vkcs_networking_secgroup or overridden via "
        "var.default_security_group_id"
    )


def test_all_instances_use_default_security_group_local():
    main = load_tf("main.tf")
    for name, attrs in find_resources(main, "vkcs_compute_instance"):
        sg_ids = attrs["security_group_ids"][0]
        assert "${local.default_security_group_id}" in sg_ids, (
            f"vkcs_compute_instance.{name} does not reference "
            f"local.default_security_group_id in security_group_ids"
        )


# ---------------------------------------------------------------------------
# main.tf: SSH keypair - Nova keypairs are per-user, not per-project, so a
# keypair uploaded under a different account is invisible to the deploying
# one. var.ssh_public_key lets Terraform register it itself.
# ---------------------------------------------------------------------------


def test_ssh_public_key_variable_defaults_to_null():
    v = find_variable(load_tf("variables.tf"), "ssh_public_key")
    assert v is not None, "variable ssh_public_key is missing"
    assert v["default"] == [None], "ssh_public_key should default to null (register nothing unless supplied)"


def test_keypair_resource_conditionally_registers_public_key():
    main = load_tf("main.tf")
    keypairs = dict(find_resources(main, "vkcs_compute_keypair"))
    assert "router" in keypairs, "expected a vkcs_compute_keypair.router resource"
    kp = keypairs["router"]
    assert kp["count"] == ["${var.ssh_public_key != None ? 1 : 0}"], (
        "the keypair should only be created when ssh_public_key is actually supplied"
    )
    assert kp["name"] == ["${var.ssh_key_name}"]
    assert kp["public_key"] == ["${var.ssh_public_key}"]


def test_ssh_key_name_local_prefers_registered_keypair():
    main = load_tf("main.tf")
    value = find_local(main, "ssh_key_name")
    assert value == [
        "${var.ssh_public_key != None ? vkcs_compute_keypair.router[0].name : var.ssh_key_name}"
    ], (
        "locals.ssh_key_name must use the keypair Terraform registers itself "
        "when ssh_public_key is supplied, falling back to var.ssh_key_name "
        "(an already-existing keypair) otherwise"
    )


def test_all_instances_use_ssh_key_name_local():
    main = load_tf("main.tf")
    for name, attrs in find_resources(main, "vkcs_compute_instance"):
        assert attrs["key_pair"] == ["${local.ssh_key_name}"], (
            f"vkcs_compute_instance.{name} must reference local.ssh_key_name, "
            f"not var.ssh_key_name directly, so it depends on the keypair "
            f"resource when one is registered"
        )


# ---------------------------------------------------------------------------
# main.tf: router private subnets don't need Neutron's DHCP service - with
# config_drive = true, cloud-init learns each interface's address from
# config-drive metadata rather than an actual DHCP exchange, and
# network-init.sh.tpl re-pins it deterministically to ethN afterwards
# ---------------------------------------------------------------------------


def test_router_priv_subnet_has_dhcp_disabled():
    main = load_tf("main.tf")
    subnets = dict(find_resources(main, "vkcs_networking_subnet"))
    assert "router_priv_subnet" in subnets
    assert subnets["router_priv_subnet"]["enable_dhcp"] == [False], (
        "router_priv_subnet must have enable_dhcp = false - no in-guest "
        "DHCP client is ever used on these interfaces"
    )


def test_router_iface_port_ip_address_is_conditional_on_fixed_ips():
    """Regression guard + new behaviour, in one place: a hand-computed
    fixed_ip.ip_address collided with VKCS's own auto-created service ports
    on the network (observed: a "network:dns" port silently consuming an
    address) during a real PROD deployment - so for project-managed roles
    (private_network_cidrs) ip_address must stay null and let Neutron's IPAM
    auto-assign. But externally-owned roles (router_networks, e.g. mvm-s3)
    have their IP pre-agreed by another project's admin, so those must set
    it explicitly. Both cases are driven by the same conditional expression."""
    main = load_tf("main.tf")
    ports = dict(find_resources(main, "vkcs_networking_port"))
    assert "router_iface_port" in ports
    fixed_ip = ports["router_iface_port"]["fixed_ip"][0]
    assert "subnet_id" in fixed_ip
    assert fixed_ip["ip_address"] == [
        "${local.router_interfaces[each.value[1]].fixed_ips == None ? None : "
        "local.router_interfaces[each.value[1]].fixed_ips[each.value[0]]}"
    ], (
        "router_iface_port.fixed_ip.ip_address must stay null when the "
        "role's fixed_ips is null (project-managed roles, IPAM auto-assign) "
        "and pick the per-router static IP otherwise (router_networks roles)"
    )


def test_router_iface_port_has_length_precondition_for_fixed_ips():
    """router_networks.ip_addresses must cover every router - a precondition
    (not just a variable validation{}) gives a clear per-role error at plan
    time instead of an out-of-range index crash."""
    main = load_tf("main.tf")
    ports = dict(find_resources(main, "vkcs_networking_port"))
    port = ports["router_iface_port"]
    assert "lifecycle" in port, "router_iface_port must declare a lifecycle.precondition"
    precondition = port["lifecycle"][0]["precondition"][0]
    assert "fixed_ips" in precondition["condition"][0]
    assert "length(" in precondition["condition"][0]


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


def _configured_router_count():
    """The router_count actually in effect: whatever terraform.tfvars pins,
    else the variable's default (a tfvars value always beats the default)."""
    tfvars_text = (TERRAFORM_DIR / "terraform.tfvars").read_text()
    for line in tfvars_text.splitlines():
        if line.strip().startswith("#"):
            continue
        m = re.match(r"^\s*router_count\s*=\s*(\d+)", line)
        if m:
            return int(m.group(1))
    return find_variable(load_tf("variables.tf"), "router_count")["default"][0]


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 = _configured_router_count()
    for net in networks:
        # ip_address is left unset on each port (Neutron IPAM auto-assigns -
        # see router_iface_port in main.tf), so there's no fixed per-router
        # offset to reserve room for - but VKCS auto-creates its own service
        # ports on the network (observed: one "network:dns" port consuming
        # an address), so there must be room for router_count routers plus
        # at least one such reservation, on top of network/gateway/broadcast.
        assert net.num_addresses >= router_count + 3, (
            f"{net} has too few addresses for {router_count} routers plus "
            f"platform-reserved ports (e.g. VKCS's network:dns service port)"
        )


# ---------------------------------------------------------------------------
# mvm-s3.tfvars: sanity-check the externally-owned network/IP values shipped
# for this environment (no project-managed private networks at all here)
# ---------------------------------------------------------------------------


def _mvm_s3_router_count():
    tfvars_text = (TERRAFORM_DIR / "mvm-s3.tfvars").read_text()
    for line in tfvars_text.splitlines():
        if line.strip().startswith("#"):
            continue
        m = re.match(r"^\s*router_count\s*=\s*(\d+)", line)
        if m:
            return int(m.group(1))
    return find_variable(load_tf("variables.tf"), "router_count")["default"][0]


def test_mvm_s3_tfvars_disables_project_managed_private_networks():
    tfvars = load_tf("mvm-s3.tfvars")
    assert tfvars["private_network_cidrs"][0] == [], (
        "mvm-s3 must not create any project-managed private network - both "
        "of its private interfaces come from router_networks instead"
    )


def test_mvm_s3_tfvars_router_networks_matches_the_diagram():
    """Regression guard: mvm-s3.tfvars must keep shipping exactly the
    network/subnet UUIDs and per-router IPs from the admin's diagram."""
    tfvars = load_tf("mvm-s3.tfvars")
    assert tfvars["router_networks"][0] == MVM_S3_ROUTER_NETWORKS


def test_mvm_s3_tfvars_router_networks_ip_addresses_cover_router_count():
    tfvars = load_tf("mvm-s3.tfvars")
    router_networks = tfvars["router_networks"][0]
    router_count = _mvm_s3_router_count()
    assert router_networks, "mvm-s3.tfvars must set router_networks"
    for role, net in router_networks.items():
        assert len(net["ip_addresses"]) >= router_count, (
            f"router_networks[{role!r}].ip_addresses has fewer entries "
            f"than router_count={router_count}"
        )


def test_mvm_s3_tfvars_router_networks_ips_are_valid_and_dont_overlap():
    tfvars = load_tf("mvm-s3.tfvars")
    router_networks = tfvars["router_networks"][0]
    networks = []
    for role, net in router_networks.items():
        cidr = ipaddress.ip_network(net["cidr"])
        networks.append(cidr)
        for ip in net["ip_addresses"]:
            assert ipaddress.ip_address(ip) in cidr, (
                f"router_networks[{role!r}] ip {ip} does not fall inside {cidr}"
            )
        assert len(net["ip_addresses"]) == len(set(net["ip_addresses"])), (
            f"router_networks[{role!r}].ip_addresses has duplicate entries"
        )
    for i, a in enumerate(networks):
        for b in networks[i + 1 :]:
            assert not a.overlaps(b), f"{a} overlaps {b} in mvm-s3.tfvars router_networks"


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