# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See LICENSE for more details.
#
# Copyright (c) 2020 ScyllaDB

import logging
import os
import unittest.mock
from collections import namedtuple
from unittest.mock import patch

import pytest
import yaml

from sdcm import sct_config
from sdcm.keystore import KeyStore
from sdcm.provision.aws.capacity_errors import RegionAMINotFoundError
from sdcm.test_config import TestConfig
from sdcm.utils.common import get_latest_scylla_release


def _get_latest_scylla_release(product="scylla"):
    return get_latest_scylla_release(product=product, verify=False)


def _set_gce_instance_types(monkeypatch):
    """gce_instance_type_db defaults to empty and is now required; set it for gce-backend tests."""
    monkeypatch.setenv("SCT_GCE_INSTANCE_TYPE_DB", "n2-highmem-2")


@pytest.fixture(scope="module")
def monkeymodule():
    """Fixture that provides a monkeypatching with module scope."""
    with pytest.MonkeyPatch.context() as mp:
        yield mp


@pytest.fixture(name="conf", scope="module")
def fixture_conf(monkeymodule):
    monkeymodule.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeymodule.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeymodule.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeymodule.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    yield sct_config.SCTConfiguration()


@pytest.fixture(autouse=True)
def fixture_env(monkeypatch):
    logging.basicConfig(level=logging.ERROR)
    logging.getLogger("botocore").setLevel(logging.CRITICAL)
    logging.getLogger("boto3").setLevel(logging.CRITICAL)
    logging.getLogger("anyconfig").setLevel(logging.ERROR)

    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_USE_MGMT", "false")
    monkeypatch.setenv("SCT_SCYLLA_VERSION", _get_latest_scylla_release())
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")


def test_01_dump_config(conf):
    logging.debug(conf.dump_config())


def test_02_verify_config(conf):
    conf.verify_configuration()
    conf.check_required_files()


def test_05_docker(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_USE_MGMT", "false")
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "2025.1.0")
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    assert "docker_image" in conf.dump_config()
    assert conf.docker_image == "scylladb/scylla"


def test_06a_docker_latest_no_loader(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_USE_MGMT", "false")
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "latest")
    monkeypatch.setenv("SCT_N_LOADERS", "0")
    docker_tag_after_processing = "fake_specific_docker_tag"

    with unittest.mock.patch.object(
        sct_config, "get_specific_tag_of_docker_image", return_value=docker_tag_after_processing
    ):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
    assert conf.scylla_version != "latest"


def test_07_baremetal_exception(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "baremetal")
    conf = sct_config.SCTConfiguration()

    with pytest.raises(AssertionError):
        conf.verify_configuration()


def test_08_baremetal(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "baremetal")
    monkeypatch.setenv("SCT_DB_NODES_PRIVATE_IP", '["1.2.3.4", "1.2.3.5"]')
    monkeypatch.setenv("SCT_DB_NODES_PUBLIC_IP", '["1.2.3.4", "1.2.3.5"]')
    monkeypatch.setenv("SCT_USE_PREINSTALLED_SCYLLA", "true")
    monkeypatch.setenv("SCT_S3_BAREMETAL_CONFIG", "some_config")
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    assert "db_nodes_private_ip" in conf.dump_config()
    assert conf.db_nodes_private_ip == ["1.2.3.4", "1.2.3.5"]


def test_09_unknown_configure(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/unknown_param_in_config.yaml")
    with pytest.raises(ValueError) as exc:
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
    assert exc.value.args[0] == "Unknown configuration key='WTF'"


def test_09_unknown_env(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_WHAT_IS_THAT", "just_made_this_up")
    monkeypatch.setenv("SCT_WHAT_IS_THAT_2", "what is this ?")
    conf = sct_config.SCTConfiguration()

    with pytest.raises(ValueError) as context:
        conf.verify_configuration()
    msg = str(context.value)
    assert "SCT_WHAT_IS_THAT_2=what is this ?" in msg
    assert "SCT_WHAT_IS_THAT=just_made_this_up" in msg


def test_12_scylla_version_repo_ubuntu(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "gce")
    _set_gce_instance_types(monkeypatch)
    monkeypatch.setenv("SCT_SCYLLA_LINUX_DISTRO", "ubuntu-xenial")
    monkeypatch.setenv("SCT_SCYLLA_LINUX_DISTRO_LOADER", "ubuntu-xenial")
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "3.0.3")
    monkeypatch.setenv(
        "SCT_GCE_IMAGE_DB",
        "https://www.googleapis.com/compute/v1/projects/centos-cloud/global/images/family/centos-stream-9",
    )
    expected_repo = "https://s3.amazonaws.com/downloads.scylladb.com/deb/ubuntu/scylla-3.0-xenial.list"
    with (
        unittest.mock.patch.object(sct_config, "get_branch_version", return_value="4.7.dev", clear=True),
        unittest.mock.patch.object(sct_config, "find_scylla_repo", return_value=expected_repo, clear=True),
    ):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
    assert "scylla_repo" in conf.dump_config()
    assert conf.scylla_repo == expected_repo


def test_12_scylla_version_repo_ubuntu_loader_centos(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "gce")
    _set_gce_instance_types(monkeypatch)
    monkeypatch.setenv("SCT_SCYLLA_LINUX_DISTRO", "ubuntu-xenial")
    monkeypatch.setenv("SCT_SCYLLA_LINUX_DISTRO_LOADER", "centos")
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "3.0.3")
    monkeypatch.setenv(
        "SCT_GCE_IMAGE_DB",
        "https://www.googleapis.com/compute/v1/projects/centos-cloud/global/images/family/centos-stream-9",
    )
    expected_repo = "https://s3.amazonaws.com/downloads.scylladb.com/deb/ubuntu/scylla-3.0-xenial.list"
    with (
        unittest.mock.patch.object(sct_config, "get_branch_version", return_value="4.7.dev", clear=True),
        unittest.mock.patch.object(sct_config, "find_scylla_repo", return_value=expected_repo, clear=True),
    ):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
    assert "scylla_repo" in conf.dump_config()
    assert conf.scylla_repo == expected_repo


def test_12_k8s_scylla_version_ubuntu_loader_centos(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "k8s-local-kind")
    monkeypatch.setenv("SCT_SCYLLA_LINUX_DISTRO", "ubuntu-xenial")
    monkeypatch.setenv("SCT_SCYLLA_LINUX_DISTRO_LOADER", "centos")
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    assert "scylla_repo" in conf.dump_config()
    assert not conf.scylla_repo


def test_14_check_rackaware_config_false_loader(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_RACK_AWARE_LOADER", "false")
    monkeypatch.setenv("SCT_SIMULATED_RACKS", "0")
    monkeypatch.setenv("SCT_REGION_NAME", "eu-west-1")
    monkeypatch.setenv("SCT_AVAILABILITY_ZONE", "a,b")
    monkeypatch.setenv("SCT_N_DB_NODES", "2")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv(
        "SCT_TEARDOWN_VALIDATORS",
        """rackaware:
             enabled: true
        """,
    )

    conf = sct_config.SCTConfiguration()
    with pytest.raises(ValueError, match="rack_aware_loader' must be set to True for rackaware validator."):
        conf.verify_configuration()


def test_14_check_rackaware_config_one_racks(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_RACK_AWARE_LOADER", "true")
    monkeypatch.setenv("SCT_REGION_NAME", "eu-west-1")
    monkeypatch.setenv("SCT_AVAILABILITY_ZONE", "a")
    monkeypatch.setenv("SCT_SIMULATED_RACKS", "0")
    monkeypatch.setenv("SCT_N_DB_NODES", "2")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv(
        "SCT_TEARDOWN_VALIDATORS",
        """rackaware:
             enabled: true
        """,
    )

    conf = sct_config.SCTConfiguration()
    with pytest.raises(
        ValueError, match="Rack-aware validation can only be performed in multi-availability zone or multi-"
    ):
        conf.verify_configuration()


def test_14_check_rackaware_config_no_rack_without_loader(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_RACK_AWARE_LOADER", "true")
    monkeypatch.setenv("SCT_REGION_NAME", "eu-west-1 eu-west-2")
    monkeypatch.setenv("SCT_AVAILABILITY_ZONE", "a,b")
    monkeypatch.setenv("SCT_SIMULATED_RACKS", "0")
    monkeypatch.setenv("SCT_N_DB_NODES", "2 2")
    monkeypatch.setenv("SCT_N_LOADERS", "2 2")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy ami-dummy2")
    monkeypatch.setenv("SCT_AMI_ID_LOADER", "ami-loader1 ami-loader2")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv(
        "SCT_TEARDOWN_VALIDATORS",
        """rackaware:
             enabled: true
        """,
    )

    conf = sct_config.SCTConfiguration()
    with pytest.raises(ValueError, match="Rack-aware validation requires zones without loaders."):
        conf.verify_configuration()


def test_14_check_rackaware_config_multi_az(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_RACK_AWARE_LOADER", "true")
    monkeypatch.setenv("SCT_REGION_NAME", "eu-west-1")
    monkeypatch.setenv("SCT_AVAILABILITY_ZONE", "a,b")
    monkeypatch.setenv("SCT_N_DB_NODES", "2")
    monkeypatch.setenv("SCT_N_LOADERS", "1")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv(
        "SCT_TEARDOWN_VALIDATORS",
        """rackaware:
             enabled: true
        """,
    )

    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()


def test_14_check_rackaware_config_multi_region(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_RACK_AWARE_LOADER", "true")
    monkeypatch.setenv("SCT_REGION_NAME", '["eu-west-1", "us-east-1"]')
    monkeypatch.setenv("SCT_N_DB_NODES", "2 2")
    monkeypatch.setenv("SCT_N_LOADERS", "1 0")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy ami-dummy2")
    monkeypatch.setenv("SCT_AMI_ID_LOADER", "ami-loader1 ami-loader2")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv(
        "SCT_TEARDOWN_VALIDATORS",
        """rackaware:
             enabled: true
        """,
    )

    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()


def test_14_check_rackaware_config_multi_az_and_region(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_RACK_AWARE_LOADER", "true")
    monkeypatch.setenv("SCT_REGION_NAME", '["eu-west-1", "us-east-1"]')
    monkeypatch.setenv("SCT_AVAILABILITY_ZONE", "a,b")
    monkeypatch.setenv("SCT_N_DB_NODES", "2 2")
    monkeypatch.setenv("SCT_N_LOADERS", "1 1")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_DB", "i4i.large")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy ami-dummy2")
    monkeypatch.setenv("SCT_AMI_ID_LOADER", "ami-loader1 ami-loader2")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv(
        "SCT_TEARDOWN_VALIDATORS",
        """rackaware:
             enabled: true
        """,
    )

    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()


def test_conf_check_required_files_negative(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/stress_cmd_with_bad_profile.yaml")
    monkeypatch.setenv(
        "SCT_STRESS_CMD",
        """cassandra-stress user profile=/tmp/1232123123123123123.yaml ops'(insert=100)'\
        no-warmup cl=ONE duration=10m -mode cql3 native -rate threads=3000 -errors ignore""",
    )
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    try:
        conf.check_required_files()
    except ValueError as exc:
        assert (
            str(exc) == "Stress command parameter 'stress_cmd' contains profile '/tmp/1232123123123123123.yaml' that"
            " does not exists under data_dir/"
        )


def test_15_new_scylla_repo(monkeypatch):
    centos_repo = (
        "https://s3.amazonaws.com/downloads.scylladb.com/enterprise/rpm/unstable/centos/"
        "9f724fedb93b4734fcfaec1156806921ff46e956-2bdfa9f7ef592edaf15e028faf3b7f695f39ebc1"
        "-525a0255f73d454f8f97f32b8bdd71c8dec35d3d-a6b2b2355c666b1893f702a587287da978aeec22/71/scylla"
        ".repo"
    )

    monkeypatch.delenv("SCT_SCYLLA_VERSION", raising=False)
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "gce")
    _set_gce_instance_types(monkeypatch)
    monkeypatch.setenv("SCT_SCYLLA_REPO", centos_repo)
    monkeypatch.setenv("SCT_NEW_SCYLLA_REPO", centos_repo)
    monkeypatch.setenv("SCT_USER_PREFIX", "testing")
    monkeypatch.setenv(
        "SCT_GCE_IMAGE_DB",
        "https://www.googleapis.com/compute/v1/projects/centos-cloud/global/images/family/centos-stream-9",
    )

    with unittest.mock.patch.object(sct_config, "get_branch_version", return_value="2019.1.1", clear=True):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
        conf._get_target_upgrade_version()
    assert conf.target_upgrade_version == "2019.1.1"


def test_15a_new_scylla_repo_by_scylla_version(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "gce")
    _set_gce_instance_types(monkeypatch)
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "master:latest")
    monkeypatch.setenv("SCT_NEW_VERSION", "master:latest")
    monkeypatch.setenv("SCT_USER_PREFIX", "testing")
    monkeypatch.setenv(
        "SCT_GCE_IMAGE_DB",
        "https://www.googleapis.com/compute/v1/projects/centos-cloud/global/images/family/centos-stream-9",
    )

    resolved_repo_link = "https://s3.amazonaws.com/downloads.scylladb.com/unstable/scylla/master/rpm\
        /centos/2021-06-09T13:12:44Z/scylla.repo"

    with (
        unittest.mock.patch.object(sct_config, "get_branch_version", return_value="666.development", clear=True),
        unittest.mock.patch.object(sct_config, "find_scylla_repo", return_value=resolved_repo_link, clear=True),
    ):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
        conf._get_target_upgrade_version()

    assert conf.scylla_repo == resolved_repo_link
    target_upgrade_version = conf.target_upgrade_version
    assert target_upgrade_version == "666.development" or target_upgrade_version.endswith(".dev")


def test_15b_image_id_by_scylla_version(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "gce")
    _set_gce_instance_types(monkeypatch)
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "master:latest")
    monkeypatch.setenv("SCT_USER_PREFIX", "testing")
    monkeypatch.setenv("SCT_GCE_IMAGE_DB", "")
    monkeypatch.setenv("SCT_USE_PREINSTALLED_SCYLLA", "true")

    resolved_image_link = (
        "https://www.googleapis.com/compute/v1/projects/scylla-images/global/images/"
        "scylla-4-7-dev-0-20220113-8bcd23fa0-1-build-359"
    )
    image = namedtuple("Image", "name self_link")(
        name="scylla-4-7-dev-0-20220113-8bcd23fa0-1-build-359", self_link=resolved_image_link
    )

    with unittest.mock.patch.object(sct_config, "get_branched_gce_images", return_value=[image], clear=True):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
        conf._get_target_upgrade_version()

    assert conf.gce_image_db == resolved_image_link


def test_17_verify_scylla_bench_required_parameters_in_command(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv(
        "SCT_STRESS_CMD", "scylla-bench -workload=sequential -mode=write -replication-factor=3 -partition-count=100"
    )
    monkeypatch.setenv(
        "SCT_STRESS_READ_CMD", "scylla-bench -workload=uniform -mode=read -replication-factor=3 -partition-count=100"
    )

    conf = sct_config.SCTConfiguration()
    assert conf.stress_cmd == [
        "scylla-bench -workload=sequential -mode=write -replication-factor=3 -partition-count=100"
    ]
    assert conf.stress_read_cmd == [
        "scylla-bench -workload=uniform -mode=read -replication-factor=3 -partition-count=100"
    ]


def test_17_1_raise_error_if_scylla_bench_command_dont_have_workload(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv("SCT_STRESS_CMD", "scylla-bench -mode=write -replication-factor=3 -partition-count=100")
    monkeypatch.setenv(
        "SCT_STRESS_READ_CMD", "scylla-bench -workload=uniform -mode=read -replication-factor=3 -partition-count=100"
    )

    err_msg = "Scylla-bench command scylla-bench -mode=write -replication-factor=3 -partition-count=100 doesn't have parameter -workload"

    with pytest.raises(ValueError) as context:
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()

    assert err_msg in str(context.value)


def test_17_2_raise_error_if_scylla_bench_command_dont_have_mode(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv(
        "SCT_STRESS_CMD", "scylla-bench -workload=sequential -mode=write -replication-factor=3 -partition-count=100"
    )
    monkeypatch.setenv(
        "SCT_STRESS_READ_CMD", "scylla-bench -workload=uniform -replication-factor=3 -partition-count=100"
    )

    err_msg = "Scylla-bench command scylla-bench -workload=uniform -replication-factor=3 -partition-count=100 doesn't have parameter -mode"

    with pytest.raises(ValueError) as context:
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()

    assert err_msg in str(context.value)


def test_19_aws_region_with_no_region_data(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_REGION_NAME", "not-exists-2")

    with pytest.raises(ValueError, match="not-exists-2 isn't supported"):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()


def test_21_nested_values(monkeypatch):
    monkeypatch.setenv(
        "SCT_CONFIG_FILES",
        ('["unit_tests/test_configs/minimal_test_case.yaml", "unit_tests/test_data/stress_image_extra_config.yaml"]'),
    )
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-1234")
    monkeypatch.setenv("SCT_STRESS_READ_CMD.0", "cassandra_stress")
    monkeypatch.setenv("SCT_STRESS_READ_CMD.1", "cassandra_stress")
    monkeypatch.setenv("SCT_STRESS_IMAGE", '{"ycsb": "scylladb/something_else"}')
    monkeypatch.setenv("SCT_STRESS_IMAGE.cassandra-stress", "scylla-bench")
    monkeypatch.setenv("SCT_STRESS_IMAGE.scylla-bench", "scylladb/something")

    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    stress_image = conf.stress_image
    assert stress_image["ndbench"] == "scylladb/hydra-loaders:ndbench-jdk8-A"
    assert stress_image["nosqlbench"] == "scylladb/hydra-loaders:nosqlbench-A"

    assert conf.get("stress_image.scylla-bench") == "scylladb/something"
    assert conf.get("stress_image.non-exist") is None

    assert conf.stress_read_cmd == ["cassandra_stress", "cassandra_stress"]


def test_22_get_none(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-1234")
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()

    val = conf.get(None)
    assert val is None


def test_23_1_include_nemesis_selector(monkeypatch):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv(
        "SCT_CONFIG_FILES",
        '["unit_tests/test_configs/minimal_test_case.yaml", "unit_tests/test_configs/nemesis_selector_list.yaml"]',
    )

    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()

    assert conf.nemesis_selector == ["config_changes and topology_changes"]


def test_26_run_fullscan_params_validtion_positive(monkeypatch):
    monkeypatch.setenv(
        "SCT_CONFIG_FILES",
        '["unit_tests/test_configs/minimal_test_case.yaml", "unit_tests/test_configs/positive_fullscan_param.yaml"]',
    )
    sct_config.SCTConfiguration()


def test_27_run_fullscan_params_validtion_negative(monkeypatch):
    monkeypatch.setenv(
        "SCT_CONFIG_FILES",
        '["unit_tests/test_configs/minimal_test_case.yaml", "unit_tests/test_configs/negative_fullscan_param.yaml"]',
    )
    try:
        sct_config.SCTConfiguration()
    except ValueError as exp:
        assert (
            str(exp) == "Config params validation errors:\n\tfield 'mode' must be one of "
            "'('random', 'table', 'partition', 'aggregate', 'table_and_aggregate')' "
            "but got 'agggregate'\n\t"
            "field 'ks_cf' must be an instance of <class 'str'>, but got '1'\n\t"
            "field 'validate_data' must be an instance of <class 'bool'>, but got 'no'\n\t"
            "field 'full_scan_aggregates_operation_limit' must be an instance of <class 'int'>, but got 'a'"
        )


def test_28_number_of_nodes_per_az_must_be_divisable_by_number_of_az(monkeypatch):
    monkeypatch.setenv("SCT_N_DB_NODES", "3 3 2")
    monkeypatch.setenv("SCT_REGION_NAME", "eu-west-1 eu-west-2 us-east-1")
    monkeypatch.setenv("SCT_AVAILABILITY_ZONE", "a,b,c")
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy ami-dummy ami-dummy")
    monkeypatch.setenv("SCT_AMI_ID_LOADER", "ami-loader1 ami-loader2 ami-loader3")
    with pytest.raises(AssertionError) as context:
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()

    assert "should be divisible by number of availability zones" in str(context.value)


def test_35_append_str_options(monkeypatch):
    monkeypatch.setenv("SCT_APPEND_SCYLLA_ARGS", "++ --overprovisioned 5")

    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    assert conf.append_scylla_args == (
        "--blocked-reactor-notify-ms 25 --abort-on-lsa-bad-alloc 1 "
        "--abort-on-seastar-bad-alloc --abort-on-internal-error 1 --abort-on-ebadf 1 "
        "--enable-sstable-key-validation 1 --overprovisioned 5"
    )


def test_35_append_list_options(monkeypatch):
    monkeypatch.setenv("SCT_SCYLLA_D_OVERRIDES_FILES", '["++", "extra_file/scylla.d/io.conf"]')
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    assert conf.scylla_d_overrides_files == ["extra_file/scylla.d/io.conf"]


def test_35_append_unsupported_list_options(monkeypatch):
    monkeypatch.setenv("SCT_AZURE_REGION_NAME", '["++", "euwest"]')
    with pytest.raises(AssertionError) as context:
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
    assert "Option azure_region_name is not appendable" == str(context.value)


def test_35_append_unsupported_str_options(monkeypatch):
    monkeypatch.setenv("SCT_MANAGER_VERSION", "++ new version")
    with pytest.raises(AssertionError) as context:
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()
    assert "Option manager_version is not appendable" == str(context.value)


def test_36_update_config_based_on_version():
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    conf.artifact_scylla_version = "2025.1.0~dev"
    conf.is_enterprise = True
    conf.update_config_based_on_version()


def test_37_validates_single_thread_count_for_all_throttle_steps(monkeypatch):
    monkeypatch.setenv("SCT_PERF_GRADUAL_THREADS", '{"read": 620, "write": [630], "mixed": [500]}')
    monkeypatch.setenv(
        "SCT_PERF_GRADUAL_THROTTLE_STEPS",
        '{"read": ["unthrottled", "unthrottled"], "mixed": [100, "unthrottled"], "write": [200, "unthrottled"]}',
    )
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()


def test_37_raises_error_for_mismatched_thread_and_throttle_step_counts(monkeypatch):
    monkeypatch.setenv("SCT_PERF_GRADUAL_THREADS", '{"read": [620, 630]}')
    monkeypatch.setenv("SCT_PERF_GRADUAL_THROTTLE_STEPS", '{"read": ["100", "200", "unthrottled"]}')
    with pytest.raises(ValueError):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()


def test_37_handles_empty_performance_throughput_parameters(monkeypatch):
    monkeypatch.setenv("SCT_PERF_GRADUAL_THREADS", "{}")
    monkeypatch.setenv("SCT_PERF_GRADUAL_THROTTLE_STEPS", "{}")
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()


def test_37_validates_multiple_thread_counts_per_throttle_step(monkeypatch):
    monkeypatch.setenv("SCT_PERF_GRADUAL_THREADS", '{"write": [400, 420, 440, 460, 480]}')
    monkeypatch.setenv("SCT_PERF_GRADUAL_THROTTLE_STEPS", '{"write": ["100", "200", "300", "400", "unthrottled"]}')
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()


def test_37_raises_error_for_invalid_thread_count_type(monkeypatch):
    monkeypatch.setenv("SCT_PERF_GRADUAL_THREADS", '{"mixed": ["invalid_type"]}')
    monkeypatch.setenv("SCT_PERF_GRADUAL_THROTTLE_STEPS", '{"mixed": ["unthrottled"]}')
    with pytest.raises(ValueError):
        conf = sct_config.SCTConfiguration()
        conf.verify_configuration()


@pytest.mark.parametrize(
    "backend, version, expected_repo",
    (
        ("k8s-eks", "2024.1.21", "scylladb/scylla-enterprise"),
        ("k8s-gke", "2024.1.21", "scylladb/scylla-enterprise"),
        ("k8s-eks", "2025.4.0", "scylladb/scylla"),
        ("k8s-gke", "2025.4.0", "scylladb/scylla"),
        ("docker", "2025.1.0", "scylladb/scylla"),
        ("docker", "2025.1.0-dev", "scylladb/scylla-nightly"),
    ),
)
def test_38_verify_scylla_version_lookup_k8s(monkeypatch, backend, version, expected_repo):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", backend)
    monkeypatch.setenv("SCT_SCYLLA_VERSION", version)
    # k8s-eks requires the operator docker image (a Jenkins runtime param); set it explicitly.
    monkeypatch.setenv("SCT_K8S_SCYLLA_OPERATOR_DOCKER_IMAGE", "scylladb/scylla-operator:latest")
    conf = sct_config.SCTConfiguration()
    conf.verify_configuration()
    assert "docker_image" in conf.dump_config()
    assert conf.get("docker_image") == expected_repo


def test_39_billing_project_from_job_name(monkeypatch):
    """Test that billing_project is set correctly from JOB_NAME."""
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")

    # Test with scylla-5.2 prefix
    monkeypatch.setenv("JOB_NAME", "scylla-5.2/longevity/test")
    conf = sct_config.SCTConfiguration()
    assert conf.get("billing_project") == "5.2"

    # Test with scylladb-5.4 prefix
    monkeypatch.setenv("JOB_NAME", "scylladb-5.4/longevity/test")
    conf = sct_config.SCTConfiguration()
    assert conf.get("billing_project") == "5.4"


def test_39_billing_project_staging_not_set(monkeypatch):
    """Test that billing_project is NOT set to 'staging'."""
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")

    # Test with staging folder - should not set billing_project
    monkeypatch.setenv("JOB_NAME", "scylla-staging/longevity/test")
    conf = sct_config.SCTConfiguration()
    # billing_project should not be set to "staging"
    assert conf.get("billing_project") != "staging"

    # Test with scylladb-staging folder - should not set billing_project
    monkeypatch.setenv("JOB_NAME", "scylladb-staging/longevity/test")
    conf = sct_config.SCTConfiguration()
    # billing_project should not be set to "staging"
    assert conf.get("billing_project") != "staging"


def test_40_aws_emr_defaults_loaded(conf):
    """Test that EMR defaults from aws_emr_config.yaml are loaded for AWS backend."""
    expected_keys = [
        "emr_release_label",
        "emr_instance_type_master",
        "emr_instance_type_core",
        "emr_instance_count_core",
        "emr_applications",
    ]
    dumped = conf.dump_config()
    for key in expected_keys:
        assert key in dumped

    assert conf.get("emr_release_label") == ""
    assert conf.get("emr_instance_type_master") == "m5.xlarge"
    assert conf.get("emr_instance_type_core") == "m5.xlarge"
    assert conf.get("emr_instance_count_core") == 2
    assert conf.get("emr_applications") == ["Spark"]
    assert conf.get("emr_keep_alive") is True


class TestCommonTags:
    """Tests for TestConfig.common_tags() method."""

    def test_jenkins_job_tag_from_build_tag(self, monkeypatch):
        """Test that JenkinsJobTag is set from BUILD_TAG."""
        monkeypatch.setenv("BUILD_TAG", "jenkins-scylla-master-tier1-test-128")
        monkeypatch.setenv("JOB_NAME", "scylla-master/test")
        tags = TestConfig.common_tags()
        assert tags["JenkinsJobTag"] == "jenkins-scylla-master-tier1-test-128"

    @pytest.mark.parametrize(
        "build_tag, expected_jenkins_job",
        [
            ("jenkins-scylla-master-tier1-longevity-12h-test-128", "jenkins-scylla-master-tier1-longevity-12h-test"),
            ("jenkins-scylla-master-test-256", "jenkins-scylla-master-test"),
            ("jenkins-job-name-1", "jenkins-job-name"),
            # Edge case: trailing segment is not digits - remains unchanged
            ("jenkins-no-trailing-dash", "jenkins-no-trailing-dash"),
            ("jenkins", "jenkins"),
            ("-42", ""),
        ],
    )
    def test_jenkins_job_derived_from_build_tag(self, monkeypatch, build_tag, expected_jenkins_job):
        """Test that JenkinsJob is derived from BUILD_TAG by stripping trailing dash+digits."""
        monkeypatch.setenv("BUILD_TAG", build_tag)
        monkeypatch.setenv("JOB_NAME", "scylla-master/test")
        tags = TestConfig.common_tags()
        assert tags["JenkinsJob"] == expected_jenkins_job

    def test_jenkins_job_not_set_without_build_tag(self, monkeypatch):
        """Test that JenkinsJob is not set when BUILD_TAG is not present."""
        monkeypatch.delenv("BUILD_TAG", raising=False)
        monkeypatch.setenv("JOB_NAME", "scylla-master/test")
        tags = TestConfig.common_tags()
        assert "JenkinsJobTag" not in tags
        assert "JenkinsJob" not in tags


def test_vector_store_ami_name_resolved_to_ami_id(monkeypatch):
    """Verify that ami_id_vector_store goes through convert_name_to_ami_if_needed.

    Regression test for scylladb/scylla-cluster-tests#13104 which accidentally
    dropped ami_id_vector_store from the ami_id_params list during the pydantic
    config refactor, causing image name-to-AMI resolution to be skipped.
    """
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-db-scylla")
    monkeypatch.setenv("SCT_AMI_ID_VECTOR_STORE", "vector-store-1-5-0-arm64-2026-03-17t07-07-32z")
    monkeypatch.setenv("SCT_INSTANCE_TYPE_VECTOR_STORE", "i4i.large")
    monkeypatch.setenv("SCT_AMI_VECTOR_STORE_USER", "ubuntu")
    monkeypatch.setenv("SCT_N_VECTOR_STORE_NODES", "1")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")

    resolved_names = {}

    def fake_convert(param, region_names):
        if not param.startswith("ami-"):
            resolved_names[param] = f"ami-{param}"
            return f"ami-{param}"
        return param

    with unittest.mock.patch("sdcm.sct_config.convert_name_to_ami_if_needed", side_effect=fake_convert):
        sct_config.SCTConfiguration()

    assert "vector-store-1-5-0-arm64-2026-03-17t07-07-32z" in resolved_names, (
        "convert_name_to_ami_if_needed was not called for ami_id_vector_store"
    )


def test_migrator_source_hosts_param(monkeypatch):
    """migrator_source_hosts is accepted and returned as a list."""
    monkeypatch.setenv("SCT_MIGRATOR_SOURCE_HOSTS", '["10.0.0.1", "10.0.0.2"]')
    conf = sct_config.SCTConfiguration()
    assert conf.get("migrator_source_hosts") == ["10.0.0.1", "10.0.0.2"]


def test_migrator_source_hosts_and_test_id_mutually_exclusive(monkeypatch):
    """Setting both migrator_source_hosts and migrator_source_test_id raises ValueError."""
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_MIGRATOR_SOURCE_HOSTS", "10.0.0.1")
    monkeypatch.setenv("SCT_MIGRATOR_SOURCE_TEST_ID", "aaaa-bbbb-cccc")
    conf = sct_config.SCTConfiguration()
    with pytest.raises(ValueError, match="mutually exclusive"):
        conf.verify_configuration()


@pytest.mark.parametrize(
    "raw_value,expected",
    [
        ("  2025.1.0  ", "2025.1.0"),
        ("\t5.4.0\n", "5.4.0"),
        ("2025.1.0", "2025.1.0"),
    ],
)
def test_env_var_whitespace_is_stripped(monkeypatch, raw_value, expected):
    """Environment variable values with leading/trailing whitespace are trimmed (SCT-340)."""
    monkeypatch.setenv("SCT_SCYLLA_VERSION", raw_value)
    conf = sct_config.SCTConfiguration()
    assert conf.get("scylla_version") == expected


def _make_aws_conf(monkeypatch, **env):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-source-scylla")
    monkeypatch.setenv("SCT_REGION_NAME", "us-east-1")
    monkeypatch.setenv("SCT_USE_MGMT", "false")
    monkeypatch.setenv("SCT_IGNORE_RESOLVED_PLACEMENT", "1")
    monkeypatch.delenv("SCT_TEST_ID", raising=False)
    for key, value in env.items():
        monkeypatch.setenv(key, value)

    conf = sct_config.SCTConfiguration()
    # start each test from a clean AMI slate so resolve_amis only touches what the test sets up
    for key in conf.ami_id_params:
        conf[key] = ""

    conf._ami_params_snapshot = {}
    conf.target_db_image_ids = []
    return conf


def test_resolve_amis_reresolves_name_intent_via_convert(monkeypatch):
    conf = _make_aws_conf(monkeypatch)
    conf._ami_params_snapshot = {"ami_id_loader": "resolve:ssm:/some/loader/path"}
    conf["ami_id_loader"] = "ami-loader-east"

    with patch("sdcm.sct_config.convert_name_to_ami_if_needed", return_value="ami-loader-west") as mock_convert:
        conf.resolve_amis(["eu-west-1"], source_region="us-east-1")

    mock_convert.assert_called_once_with("resolve:ssm:/some/loader/path", ("eu-west-1",))
    assert conf["ami_id_loader"] == "ami-loader-west"


def test_resolve_amis_remaps_explicit_ami_via_find_equivalent(monkeypatch):
    conf = _make_aws_conf(monkeypatch)
    conf._ami_params_snapshot = {"ami_id_db_scylla": "ami-source-scylla"}
    conf["ami_id_db_scylla"] = "ami-source-scylla"

    with patch(
        "sdcm.sct_config.find_equivalent_ami",
        return_value=[{"region": "eu-west-1", "ami_id": "ami-target-scylla"}],
    ) as mock_equiv:
        conf.resolve_amis(["eu-west-1"], source_region="us-east-1")

    mock_equiv.assert_called_once_with("ami-source-scylla", "us-east-1", target_regions=["eu-west-1"])
    assert conf["ami_id_db_scylla"] == "ami-target-scylla"


def test_resolve_amis_raises_region_ineligible_when_no_equivalent(monkeypatch):
    conf = _make_aws_conf(monkeypatch)
    conf._ami_params_snapshot = {"ami_id_db_scylla": "ami-source-scylla"}
    conf["ami_id_db_scylla"] = "ami-source-scylla"

    with patch("sdcm.sct_config.find_equivalent_ami", return_value=[]):
        with pytest.raises(RegionAMINotFoundError):
            conf.resolve_amis(["eu-west-1"], source_region="us-east-1")


@pytest.fixture(name="placement_logdir")
def fixture_placement_logdir(monkeypatch, tmp_path):
    """Point base_logdir at a tmp dir to not touch ~/sct-results."""
    monkeypatch.setenv("_SCT_LOGDIR", str(tmp_path))
    return tmp_path


def test_write_then_read_resolved_placement_round_trips(placement_logdir):  # noqa: ARG001
    TestConfig.write_resolved_placement("tid-1", region_name="eu-west-1", availability_zone="b")
    data = TestConfig.read_resolved_placement("tid-1")
    assert data["test_id"] == "tid-1"
    assert data["region_name"] == "eu-west-1"
    assert data["availability_zone"] == "b"


def test_read_resolved_placement_returns_none_on_test_id_mismatch(placement_logdir):  # noqa: ARG001
    path = TestConfig.resolved_placement_file_path("tid-2")
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as handle:
        yaml.safe_dump({"test_id": "some-other-id", "region_name": "eu-west-1", "availability_zone": "b"}, handle)
    assert TestConfig.read_resolved_placement("tid-2") is None


@pytest.mark.parametrize(
    "test_id, original_region, region_name, availability_zone, expected",
    [
        (
            "tid-pa",
            "us-east-1",
            "eu-west-1",
            "b",
            {"test_id": "tid-pa", "region_name": "eu-west-1", "availability_zone": "b"},
        ),
        ("tid-pb", "us-east-1", "us-east-1", "a", None),
        ("tid-pc", None, None, None, None),
    ],
    ids=["region_changed_writes", "unchanged_noop", "missing_noop"],
)
def test_persist_resolved_placement_writes_only_when_region_changed(
    placement_logdir, test_id, original_region, region_name, availability_zone, expected
):  # noqa: ARG001
    TestConfig.persist_resolved_placement_if_changed(
        test_id, original_region=original_region, region_name=region_name, availability_zone=availability_zone
    )
    assert TestConfig.read_resolved_placement(test_id) == expected


def _set_aws_overlay_env(monkeypatch, test_id, original_region="us-east-1"):
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "aws")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    monkeypatch.setenv("SCT_AMI_ID_DB_SCYLLA", "ami-dummy")
    monkeypatch.setenv("SCT_REGION_NAME", original_region)
    monkeypatch.setenv("SCT_TEST_ID", test_id)
    monkeypatch.setenv("SCT_USE_MGMT", "false")
    monkeypatch.delenv("SCT_IGNORE_RESOLVED_PLACEMENT", raising=False)


def test_overlay_beats_env_when_test_id_matches(monkeypatch, placement_logdir):  # noqa: ARG001
    test_id = "11111111-1111-1111-1111-111111111111"
    _set_aws_overlay_env(monkeypatch, test_id, original_region="us-east-1")
    TestConfig.write_resolved_placement(test_id, region_name="eu-west-1", availability_zone="b")

    # relocation re-resolves region-bound AMIs; ami-dummy is explicit so it goes through find_equivalent_ami
    with patch(
        "sdcm.sct_config.find_equivalent_ami",
        return_value=[{"region": "eu-west-1", "ami_id": "ami-dummy-west"}],
    ):
        conf = sct_config.SCTConfiguration()

    # region_names is env-first, so the overlay must have rewritten SCT_REGION_NAME
    assert conf.region_names == ["eu-west-1"]
    assert os.environ["SCT_REGION_NAME"] == "eu-west-1"
    assert conf["availability_zone"] == "b"
    # AMIs must have been re-resolved for the relocated region
    assert conf["ami_id_db_scylla"] == "ami-dummy-west"


def test_resolved_placement_with_amis_applies_directly_and_skips_re_resolution(monkeypatch, placement_logdir):  # noqa: ARG001
    """When the handoff carries resolved AMIs, apply them verbatim and do not re-resolve."""
    test_id = "22222222-2222-2222-2222-222222222222"
    _set_aws_overlay_env(monkeypatch, test_id, original_region="us-east-1")
    TestConfig.write_resolved_placement(
        test_id,
        region_name="eu-west-1",
        availability_zone="b",
        amis={"ami_id_db_scylla": "ami-persisted-west"},
    )

    with patch(
        "sdcm.sct_config.find_equivalent_ami",
        return_value=[{"region": "eu-west-1", "ami_id": "ami-reresolved"}],
    ):
        conf = sct_config.SCTConfiguration()

    assert conf.region_names == ["eu-west-1"]
    assert conf["availability_zone"] == "b"
    assert conf["ami_id_db_scylla"] == "ami-persisted-west"


def _set_gce_overlay_env(monkeypatch, test_id, original_datacenter):
    """Env for a GCE run whose placement handoff should be picked up at config load."""
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "gce")
    monkeypatch.setenv("SCT_CONFIG_FILES", "unit_tests/test_configs/minimal_test_case.yaml")
    _set_gce_instance_types(monkeypatch)
    monkeypatch.setenv(
        "SCT_GCE_IMAGE_DB",
        "https://www.googleapis.com/compute/v1/projects/centos-cloud/global/images/family/centos-stream-9",
    )
    monkeypatch.setenv("SCT_GCE_DATACENTER", original_datacenter)
    monkeypatch.setenv("SCT_TEST_ID", test_id)
    monkeypatch.setenv("SCT_USE_MGMT", "false")
    monkeypatch.delenv("SCT_IGNORE_RESOLVED_PLACEMENT", raising=False)


def test_gce_resolved_placement_applies_relocated_datacenter(monkeypatch, placement_logdir):  # noqa: ARG001
    """A GCE region-fallback relocation is picked up by the next hydra command (the Run Test step).

    GCE carries the region in `gce_datacenter` rather than AWS's `region_name`/AMIs, and
    `gce_datacenters` is env-first - so the overlay has to rewrite SCT_GCE_DATACENTER, not just the
    config value, or the split provision->run pipeline snaps back to the original region.
    """
    test_id = "33333333-3333-3333-3333-333333333333"
    _set_gce_overlay_env(monkeypatch, test_id, original_datacenter="us-east1")
    TestConfig.write_resolved_placement(test_id, region_name="us-west1", availability_zone="c")

    conf = sct_config.SCTConfiguration()

    assert os.environ["SCT_GCE_DATACENTER"] == "us-west1"
    assert conf.gce_datacenters == ["us-west1"]
    assert conf["availability_zone"] == "c"
    assert os.environ["SCT_AVAILABILITY_ZONE"] == "c"


def test_gce_resolved_placement_applies_relocated_multi_dc(monkeypatch, placement_logdir):  # noqa: ARG001
    """The handoff round-trips a multi-DC placement, so per-DC relocation survives the step boundary."""
    test_id = "44444444-4444-4444-4444-444444444444"
    _set_gce_overlay_env(monkeypatch, test_id, original_datacenter="us-east1 us-west1")
    TestConfig.write_resolved_placement(test_id, region_name="us-east1 us-east4", availability_zone="b")

    conf = sct_config.SCTConfiguration()

    assert conf.gce_datacenters == ["us-east1", "us-east4"]  # only the exhausted DC moved
    assert os.environ["SCT_GCE_DATACENTER"] == "us-east1 us-east4"


def test_gce_resolved_placement_ignored_when_disabled(monkeypatch, placement_logdir):  # noqa: ARG001
    """SCT_IGNORE_RESOLVED_PLACEMENT keeps the configured GCE region, handoff file or not."""
    test_id = "55555555-5555-5555-5555-555555555555"
    _set_gce_overlay_env(monkeypatch, test_id, original_datacenter="us-east1")
    TestConfig.write_resolved_placement(test_id, region_name="us-west1", availability_zone="c")
    monkeypatch.setenv("SCT_IGNORE_RESOLVED_PLACEMENT", "1")

    conf = sct_config.SCTConfiguration()

    assert conf.gce_datacenters == ["us-east1"]
    assert os.environ["SCT_GCE_DATACENTER"] == "us-east1"


@pytest.mark.parametrize(
    "release_label, expected_error",
    [
        ("emr-7.12.0", "not a native Spark 4.X release"),
        ("emr-spark-8.0.0", None),
    ],
)
def test_verify_emr_spark_mode_native_release_label(monkeypatch, release_label, expected_error):
    """Native Spark mode validates emr-spark-* releases and rejects non-spark release labels."""
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    monkeypatch.setenv("SCT_EMR_RELEASE_LABEL", release_label)

    conf = sct_config.SCTConfiguration()
    if expected_error is None:
        conf.verify_configuration()
    else:
        with pytest.raises(ValueError, match=expected_error):
            conf.verify_configuration()


@pytest.fixture(name="clean_keystore_exports")
def fixture_clean_keystore_exports(monkeypatch):
    """Isolate the process-wide record of keystore env vars SCT exported itself.

    Constructing SCTConfiguration writes SCT_KEYSTORE_* into os.environ, and
    monkeypatch cannot roll back a variable that did not exist beforehand, so the
    teardown clears whatever the test left behind.
    """
    keystore_envs = ("SCT_KEYSTORE_BACKEND", "SCT_KEYSTORE_SM_PREFIX", "SCT_KEYSTORE_SM_REGION")
    monkeypatch.setattr(sct_config, "_KEYSTORE_ENV_EXPORTED", {})
    for env_name in keystore_envs:
        monkeypatch.delenv(env_name, raising=False)
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "docker")
    yield
    for env_name in keystore_envs:
        os.environ.pop(env_name, None)


def _write_keystore_conf(tmp_path, param, value):
    conf_file = tmp_path / f"keystore_{param}_{value.replace('/', '')}.yaml"
    conf_file.write_text(f"{param}: '{value}'\n", encoding="utf-8")
    return str(conf_file)


# All three keystore params share one propagation loop, so each case exercises the
# same code path -- but only `keystore_backend` was covered while the other two
# (both new) went through it untested.
KEYSTORE_PARAM_CASES = [
    pytest.param("keystore_backend", "secretsmanager", "s3", id="backend"),
    pytest.param("keystore_sm_prefix", "sct/", "other/", id="sm_prefix"),
    pytest.param("keystore_sm_region", "us-east-1", "eu-west-1", id="sm_region"),
]


@pytest.mark.parametrize("param, default_value, other_value", KEYSTORE_PARAM_CASES)
def test_keystore_export_does_not_outrank_a_later_config_file(
    monkeypatch,
    tmp_path,
    clean_keystore_exports,  # noqa: ARG001
    param,
    default_value,
    other_value,
):
    """A keystore value this process exported must not win over a later config file.

    SCTConfiguration writes the resolved keystore settings into os.environ so bare
    KeyStore() callers pick them up.  Environment variables outrank config files, so
    without the self-export bookkeeping the first instance's default would silently
    beat every later instance's config file in the same process.
    """
    env_name = f"SCT_{param.upper()}"

    first = sct_config.SCTConfiguration()
    assert first.get(param) == default_value
    assert os.environ[env_name] == default_value

    monkeypatch.setenv("SCT_CONFIG_FILES", _write_keystore_conf(tmp_path, param, other_value))
    second = sct_config.SCTConfiguration()
    assert second.get(param) == other_value
    # and the export is refreshed so KeyStore() agrees with the config file
    assert os.environ[env_name] == other_value


@pytest.mark.parametrize("param, default_value, other_value", KEYSTORE_PARAM_CASES)
def test_keystore_user_env_var_still_beats_config_file(
    monkeypatch,
    tmp_path,
    clean_keystore_exports,  # noqa: ARG001
    param,
    default_value,  # noqa: ARG001
    other_value,
):
    """An env var the user set is a real override: it outranks the config file and is never rewritten."""
    env_name = f"SCT_{param.upper()}"
    monkeypatch.setenv(env_name, other_value)
    monkeypatch.setenv("SCT_CONFIG_FILES", _write_keystore_conf(tmp_path, param, default_value))

    conf = sct_config.SCTConfiguration()
    assert conf.get(param) == other_value
    assert os.environ[env_name] == other_value
    assert env_name not in sct_config._KEYSTORE_ENV_EXPORTED


@pytest.mark.parametrize("param, default_value, other_value", KEYSTORE_PARAM_CASES)
def test_keystore_env_changed_after_export_is_a_user_override(
    monkeypatch,
    clean_keystore_exports,  # noqa: ARG001
    param,
    default_value,
    other_value,
):
    """Only the exact value we exported is ignored; a value changed since is honoured."""
    env_name = f"SCT_{param.upper()}"
    sct_config.SCTConfiguration()
    assert sct_config._KEYSTORE_ENV_EXPORTED[env_name] == default_value

    monkeypatch.setenv(env_name, other_value)
    assert sct_config.SCTConfiguration().get(param) == other_value


@pytest.mark.parametrize("param, default_value, other_value", KEYSTORE_PARAM_CASES)
def test_keystore_marker_is_dropped_once_the_user_overrides(
    monkeypatch,
    tmp_path,
    clean_keystore_exports,  # noqa: ARG001
    param,
    default_value,
    other_value,
):
    """A user override must not be re-armed as a self-export by matching the old value.

    Without dropping the marker, setting the env var back to the value SCT once
    exported would make it look like our own leak again, so a config file would
    silently win over what the user explicitly asked for.
    """
    env_name = f"SCT_{param.upper()}"
    sct_config.SCTConfiguration()
    assert sct_config._KEYSTORE_ENV_EXPORTED[env_name] == default_value

    # The user takes over, then happens to set it back to the exported value.
    monkeypatch.setenv(env_name, other_value)
    sct_config.SCTConfiguration()
    assert env_name not in sct_config._KEYSTORE_ENV_EXPORTED

    monkeypatch.setenv(env_name, default_value)
    monkeypatch.setenv("SCT_CONFIG_FILES", _write_keystore_conf(tmp_path, param, other_value))
    assert sct_config.SCTConfiguration().get(param) == default_value


def test_keystore_env_is_exported_before_init_resolves_xcloud_version(monkeypatch, tmp_path, clean_keystore_exports):  # noqa: ARG001
    """A config-file `keystore_backend` must apply to KeyStore calls made *during* __init__.

    `_resolve_xcloud_version_tag` reaches `cloud_env_credentials` -> bare `KeyStore()`
    from inside __init__.  While the export ran at the end of __init__, that fetch
    still saw the default backend, so the documented `keystore_backend: 's3'` opt-out
    was a no-op for exactly the path that needs it most.
    """
    monkeypatch.setenv("SCT_CLUSTER_BACKEND", "xcloud")
    monkeypatch.setenv("SCT_XCLOUD_PROVIDER", "aws")
    monkeypatch.setenv("SCT_SCYLLA_VERSION", "release:latest")
    monkeypatch.setenv("SCT_CONFIG_FILES", _write_keystore_conf(tmp_path, "keystore_backend", "s3"))

    seen = []
    monkeypatch.setattr(
        sct_config.SCTConfiguration,
        "_resolve_xcloud_version_tag",
        lambda self, version_tag: seen.append(KeyStore()._backend),
    )

    sct_config.SCTConfiguration()

    assert seen == ["s3"], f"KeyStore built during __init__ saw {seen}, config file asked for s3"
