hla_framework_bd/packages/agebd/tests/test_subs.py

import pytest
import yaml

from agebd.subs import (
    SubstitutionError,
    Substitutions,
    SubstitutionsConfig,
    UnknownTokenError,
    find_legacy_tokens,
    find_tokens,
    read_ioc_port,
)

ENVIRONMENTS = {
    "environments": {
        "prod": {
            "agebd_env": "prod",
            "agebd_env_suffix_upper": "",
            "agebd_env_suffix_lower": "",
        },
        "dev": {
            "agebd_env": "dev",
            "agebd_env_suffix_upper": "-DEV",
            "agebd_env_suffix_lower": "-dev",
        },
    },
    "tokens": {
        "agebd_env": {"source": "environment", "description": "env name"},
        "agebd_env_suffix_upper": {"source": "environment", "description": "upper suffix"},
        "agebd_env_suffix_lower": {"source": "environment", "description": "lower suffix"},
        "ioc_port": {"source": "registry", "description": "ioc port"},
    },
    "locations": {
        "render": ["services/*/current/ioc", "qt"],
        "passthrough": ["templates"],
    },
}

REGISTRY = {
    "next_available_ioc_port": 50002,
    "services": [
        {"name": "Master", "ioc_port": 50000, "status": "active"},
        {"name": "TauBPM", "ioc_port": 50001, "status": "active"},
    ],
}


@pytest.fixture
def config_dir(tmp_path):
    (tmp_path / "environments.yml").write_text(yaml.safe_dump(ENVIRONMENTS))
    (tmp_path / "services_registry.yml").write_text(yaml.safe_dump(REGISTRY))
    return tmp_path


def write_environments(config_dir, environments):
    (config_dir / "environments.yml").write_text(yaml.safe_dump(environments))


def test_for_env_resolves_environment_tokens(config_dir):
    subs = Substitutions.for_env("dev", config_dir=config_dir)

    assert subs["agebd_env"] == "dev"
    assert subs["agebd_env_suffix_upper"] == "-DEV"
    assert subs["agebd_env_suffix_lower"] == "-dev"


def test_prod_suffixes_are_empty(config_dir):
    subs = Substitutions.for_env("prod", config_dir=config_dir)

    assert subs.render_text("AGEBD-TUNE@@agebd_env_suffix_upper@@:AO") == "AGEBD-TUNE:AO"


def test_for_service_adds_ioc_port(config_dir):
    subs = Substitutions.for_service("dev", "taubpm", config_dir=config_dir)

    assert subs["ioc_port"] == "50001"


def test_render_text_substitutes_every_occurrence(config_dir):
    subs = Substitutions.for_env("dev", config_dir=config_dir)

    text = "AGEBD-ALH@@agebd_env_suffix_upper@@:X\nsls-vserv-bd-01@@agebd_env_suffix_lower@@\n"

    assert subs.render_text(text) == "AGEBD-ALH-DEV:X\nsls-vserv-bd-01-dev\n"


def test_render_text_raises_on_misspelled_token(config_dir):
    subs = Substitutions.for_env("dev", config_dir=config_dir)

    with pytest.raises(UnknownTokenError, match="agebd_env_sufix_upper"):
        subs.render_text("AGEBD-TUNE@@agebd_env_sufix_upper@@:AO", origin="TUNE.subs")


def test_render_text_raises_when_registry_token_is_not_loaded(config_dir):
    """for_env() has no ioc_port, so a parameters file must fail loudly."""
    subs = Substitutions.for_env("dev", config_dir=config_dir)

    with pytest.raises(UnknownTokenError, match="ioc_port"):
        subs.render_text("ioc_port: @@ioc_port@@")


def test_render_file_reports_whether_it_changed(config_dir, tmp_path):
    subs = Substitutions.for_env("dev", config_dir=config_dir)
    path = tmp_path / "AGEBD-CPCL-TUNE_main.subs"
    path.write_text("{ AGEBD-TUNE@@agebd_env_suffix_upper@@ }\n")

    assert subs.render_file(path) is True
    assert path.read_text() == "{ AGEBD-TUNE-DEV }\n"
    assert subs.render_file(path) is False


def test_render_tree_skips_cache_directories(config_dir, tmp_path):
    subs = Substitutions.for_env("dev", config_dir=config_dir)
    (tmp_path / "ioc").mkdir()
    (tmp_path / "ioc" / "a.subs").write_text("AGEBD-A@@agebd_env_suffix_upper@@")
    (tmp_path / "__pycache__").mkdir()
    (tmp_path / "__pycache__" / "b.subs").write_text("AGEBD-B@@agebd_env_suffix_upper@@")

    changed = subs.render_tree(tmp_path)

    assert [p.name for p in changed] == ["a.subs"]
    assert (tmp_path / "__pycache__" / "b.subs").read_text().endswith("@@agebd_env_suffix_upper@@")


def test_replacement_value_is_literal(config_dir):
    """A value containing regex backreference syntax must not be interpreted."""
    environments = yaml.safe_load(yaml.safe_dump(ENVIRONMENTS))
    environments["environments"]["dev"]["agebd_env_suffix_upper"] = r"-\1DEV"
    write_environments(config_dir, environments)

    subs = Substitutions.for_env("dev", config_dir=config_dir)

    assert subs.render_text("AGEBD-A@@agebd_env_suffix_upper@@") == r"AGEBD-A-\1DEV"


def test_unknown_environment_lists_the_known_ones(config_dir):
    with pytest.raises(SubstitutionError, match="dev, prod"):
        Substitutions.for_env("staging", config_dir=config_dir)


def test_token_missing_from_one_environment_is_a_load_error(config_dir):
    environments = yaml.safe_load(yaml.safe_dump(ENVIRONMENTS))
    del environments["environments"]["prod"]["agebd_env_suffix_lower"]
    write_environments(config_dir, environments)

    with pytest.raises(SubstitutionError, match="missing token"):
        SubstitutionsConfig.load(config_dir)


def test_undeclared_token_in_an_environment_is_a_load_error(config_dir):
    environments = yaml.safe_load(yaml.safe_dump(ENVIRONMENTS))
    environments["environments"]["dev"]["agebd_env_suffix_mixed"] = "-Dev"
    write_environments(config_dir, environments)

    with pytest.raises(SubstitutionError, match="not declared"):
        SubstitutionsConfig.load(config_dir)


def test_unknown_token_source_is_a_load_error(config_dir):
    environments = yaml.safe_load(yaml.safe_dump(ENVIRONMENTS))
    environments["tokens"]["ioc_port"]["source"] = "somewhere"
    write_environments(config_dir, environments)

    with pytest.raises(SubstitutionError, match="unknown source"):
        SubstitutionsConfig.load(config_dir)


def test_read_ioc_port_is_case_insensitive(config_dir):
    registry_file = config_dir / "services_registry.yml"

    assert read_ioc_port(registry_file, "taubpm") == 50001
    assert read_ioc_port(registry_file, "TauBPM") == 50001
    assert read_ioc_port(registry_file, "TAUBPM") == 50001


def test_read_ioc_port_raises_for_unknown_service(config_dir):
    with pytest.raises(SubstitutionError, match="Nope"):
        read_ioc_port(config_dir / "services_registry.yml", "Nope")


def test_path_classification(config_dir):
    config = SubstitutionsConfig.load(config_dir)

    assert config.is_render_path("services/tune/current/ioc/TUNE.template")
    assert config.is_render_path("qt/A_BD_Tune.ui")
    assert config.is_passthrough_path("templates/service/x/current/ioc/a.jinja")
    assert not config.is_render_path("services/tune/current/app/src/main.py")
    assert not config.is_passthrough_path("services/tune/current/app/src/main.py")


def test_expand_render_paths(config_dir, tmp_path):
    config = SubstitutionsConfig.load(config_dir)
    repo = tmp_path / "repo"
    (repo / "services" / "tune" / "current" / "ioc").mkdir(parents=True)
    (repo / "services" / "tune" / "current" / "app").mkdir(parents=True)
    (repo / "qt").mkdir()

    expanded = {p.relative_to(repo).as_posix() for p in config.expand_render_paths(repo)}

    assert expanded == {"services/tune/current/ioc", "qt"}


def test_find_tokens_ignores_non_token_angle_brackets():
    assert find_tokens("@@b@@ and @@agebd_env@@") == {"b", "agebd_env"}
    assert find_tokens("user@@example.com") == set()
    assert find_tokens("@@ NotAToken @@") == set()


def test_find_legacy_tokens_returns_both_stages():
    """
    It cannot tell the two stages apart on syntax alone -- callers intersect
    the result with the declared deploy-time tokens.
    """
    assert find_legacy_tokens("{{ agebd_env_suffix_upper }} {{service_name_upper}}") == {
        "agebd_env_suffix_upper",
        "service_name_upper",
    }