run_matrix.py

"""Runner script to execute all interface tests."""

# Copyright 2023 Canonical Ltd.
# See LICENSE file for licensing details.

import json
import logging
import os
import shutil
import subprocess
from collections import namedtuple
from pathlib import Path
from typing import TYPE_CHECKING, Dict, Iterable, List, Literal, Tuple

from github import Github
from interface_tester.collector import collect_tests

if TYPE_CHECKING:
    from interface_tester.collector import _CharmTestConfig, _RoleTestSpec

    _Role = Literal["provider", "requirer"]

    # mapping from charm name (e.g. 'traefik-k8s') to whether the tests are passing or not
    _ResultsPerCharm = Dict[str, bool]
    # mapping from role ('provider'|'requirer) to results per charm
    _ResultsPerRole = Dict[_Role, _ResultsPerCharm]
    # mapping from version names (e.g. 'v1') to results per role
    _ResultsPerVersion = Dict[str, _ResultsPerRole]
    # mapping from interface names (e.g. 'ingress') to results per version
    _ResultsPerInterface = Dict[str, _ResultsPerVersion]

# it is "python -m venv" on some platforms/python versions
MKVENV_CMD = os.getenv("MKVENV_CMD", "python -m virtualenv")

FIXTURE_PATH = "tests/interface/conftest.py"
FIXTURE_IDENTIFIER = "interface_tester"
logging.getLogger().setLevel(logging.INFO)

FixtureSpec = namedtuple("FixtureSpec", "path id")


class SetupError(Exception):
    pass


class InterfaceTestError(Exception):
    pass


def _clone_charm_repo(charm_config: "_CharmTestConfig", charm_path: Path):
    """Clones a charm repository to a local path."""
    logging.info(
        f"Cloning: {charm_config.name} from ({charm_config.url}@{charm_config.branch or 'main'})"
    )
    branch_option = ""
    if charm_config.branch:
        branch_option = f"--branch {charm_config.branch}"
        logging.warning(
            f"custom branch provided for {charm_config.name}; "
            f"this should only be done in staging"
        )
    cmd = f"git clone --quiet --depth 1 {branch_option} {charm_config.url} {charm_path}"

    retcode = subprocess.call(
        cmd,
        shell=True,
        stdout=subprocess.DEVNULL,
    )
    if retcode > 0:
        raise SetupError(
            f"Failed to clone repo {charm_config.url}@{charm_config.branch or 'main'} for {charm_config.name}; "
            "check the charms.yaml config.\n"
            f"\t command: {cmd!r}"
        )


def _prepare_repo(
    charm_config: "_CharmTestConfig",
    interface: str,
    version: int,
    repo: str,
    branch: str,
    root: Path = Path("/tmp/charm-relation-interfaces-tests/"),
) -> Tuple[Path, Path]:
    """Clone the charm repository and create the venv if it hasn't been done already."""
    logging.info(f"Preparing testing environment for: {charm_config.name}")

    # path at which we'll clone the charm repo
    repo_root_path = root / charm_config.name

    # multi-charm repos might have multiple charms in a single repo.
    if charm_root_cfg := charm_config.test_setup.get("charm_root"):
        charm_root_path = repo_root_path / charm_root_cfg
        if not charm_root_path.resolve().is_relative_to(repo_root_path):
            raise SetupError(
                f"charm_root must be a relative path, not {charm_root_cfg!r}"
            )
        logging.info(
            f"charm root configured: preparing to set up charm at {charm_root_path}"
        )
    else:
        # usually, the charm is at the root of the repo.
        charm_root_path = repo_root_path

    if not repo_root_path.exists():
        _clone_charm_repo(charm_config, repo_root_path)
        _pre_run(charm_config, charm_root_path)
        _setup_venv(charm_root_path)
    try:
        fixture_spec = _get_fixture(charm_config, charm_root_path)
    except FileNotFoundError as e:
        raise SetupError(f"unable to get fixture spec from {charm_root_path}") from e
    if not fixture_spec.path.is_file():
        # NOTE: In the future we could probably run the tests without a fixture, assuming
        # that the charm needs no patching at all to work with scenario
        raise SetupError(f"fixture missing for charm {charm_config.name}")
    test_path = _generate_test(
        interface, fixture_spec.path.parent, fixture_spec.id, version, repo, branch
    )
    return charm_root_path, test_path


def _clean(root: Path = Path("/tmp/charm-relation-interfaces-tests/")):
    """Clean the directory used to store repos for the tests."""
    if root.is_dir():
        shutil.rmtree(root)


_TEST_CONTENT = """
# file generated by run_matrix.py
from interface_tester import InterfaceTester
def test_{interface}_interface({fixture_id}: InterfaceTester):
    {fixture_id}.configure(
        interface_name="{interface}",
        interface_version={version},
        repo="{repo}",
        branch="{branch}",
    )
    {fixture_id}.run()
"""


def _generate_test(
    interface: str,
    test_path: Path,
    fixture_id: str,
    version: int,
    repo: str,
    branch: str,
) -> Path:
    """Generate a pytest file for a given charm and interface."""
    logging.info(f"Generating test file for {interface} at {test_path}")
    test_content = _TEST_CONTENT.format(
        interface=interface,
        fixture_id=fixture_id,
        version=version,
        repo=repo,
        branch=branch,
    )
    test_filename = f"interface_test_{interface}.py"
    with open(test_path / test_filename, "w") as file:
        file.write(test_content)
    return test_path / test_filename


def _get_fixture(charm_config: "_CharmTestConfig", charm_path: Path) -> FixtureSpec:
    """Get the tester fixture from a charm."""
    fixture_path = charm_path / FIXTURE_PATH
    fixture_id = FIXTURE_IDENTIFIER
    if charm_config.test_setup:
        if charm_config.test_setup["location"]:
            fixture_path = charm_path / Path(charm_config.test_setup["location"])
        if charm_config.test_setup["identifier"]:
            fixture_id = charm_config.test_setup["identifier"]
    return FixtureSpec(fixture_path, fixture_id)


def _pre_run(charm_config: "_CharmTestConfig", charm_path: Path):
    """Run whatever setup commands were configured in the custom test config."""
    if not charm_config.test_setup:
        return
    timeout = 60  # seconds
    if pre_run_cfg := charm_config.test_setup.get("pre_run"):
        logging.info("Running custom pre_run script...")
        try:
            output = subprocess.check_output(
                pre_run_cfg,
                cwd=charm_path,
                shell=True,
                timeout=timeout,
                text=True,
                check=True,
            )
        except subprocess.CalledProcessError as e:
            logging.error(
                "failed to run pre_run script from %s: %s. The script was:\n\t%r",
                charm_path,
                e.stderr,
                pre_run_cfg,
            )
        except subprocess.TimeoutExpired:
            logging.error(
                "pre_run script from %s timed out after %s. The script was:\n\t%r",
                charm_path,
                timeout,
                pre_run_cfg,
            )
        else:
            logging.debug("Custom pre_run script output: %s", output)


def _setup_venv(charm_path: Path) -> None:
    """Create the venv for a charm and return the path to its python."""
    logging.info(f"Preparing venv for {charm_path}")

    original_wd = os.getcwd()
    os.chdir(charm_path)
    # Create the venv and install the requirements
    try:
        subprocess.check_call(
            f"{MKVENV_CMD} ./.interface-venv", shell=True, stdout=subprocess.DEVNULL
        )
        logging.info(f"Installing dependencies in venv for {charm_path}")

        subprocess.check_call(
            ".interface-venv/bin/python -m pip install setuptools pytest pytest-interface-tester",
            shell=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        subprocess.check_call(
            ".interface-venv/bin/python -m pip install -r requirements.txt",
            shell=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        logging.info(f"Installed {(charm_path/'requirements.txt').read_text()}")

    except subprocess.CalledProcessError as e:
        raise SetupError("venv setup failed") from e
    os.chdir(original_wd)


def _run_test_with_pytest(root: Path, test_path: Path):
    """Run a test file with pytest."""
    logging.info(f"Running tests for {root}")
    original_wd = os.getcwd()
    os.chdir(root)
    try:
        subprocess.check_call(
            f"PYTHONPATH=src:lib .interface-venv/bin/python -m pytest {test_path}",
            shell=True,
        )
    except subprocess.CalledProcessError as e:
        raise InterfaceTestError from e
    os.chdir(original_wd)


def _test_charm(
    charm_config: "_CharmTestConfig",
    interface: str,
    version: int,
    role: str,
    repo: str,
    branch: str,
) -> bool:
    """Run interface tests for a charm."""
    logging.info(f"Running tests for charm: {charm_config.name}")
    try:
        charm_path, test_path = _prepare_repo(
            charm_config, interface, version, repo, branch
        )
    except SetupError:
        logging.warning(
            f"test setup failed for {charm_config.name} {interface} {role}",
            exc_info=True,
        )
        return False

    try:
        _run_test_with_pytest(charm_path, test_path)
    except InterfaceTestError:
        logging.warning(
            f"interface tests for {charm_config.name} {interface} {role} failed",
            exc_info=True,
        )
        return False
    return True


def _test_charms(
    charm_configs: Iterable["_CharmTestConfig"],
    interface: str,
    version: int,
    role: str,
    repo: str,
    branch: str,
) -> "_ResultsPerCharm":
    """Test all charms against this interface and role."""
    logging.info(f"Running tests for {interface}")
    out = {}
    for charm_config in charm_configs:
        success = _test_charm(charm_config, interface, version, role, repo, branch)
        out[charm_config.name] = success
        logging.info(f"Result: {'PASSED' if success else 'FAILED'}")
    return out


def _test_roles(
    tests_per_role: Dict["_Role", "_RoleTestSpec"],
    interface: str,
    version: int,
    repo: str,
    branch: str,
) -> "_ResultsPerRole":
    """Run the tests for each role of this interface."""
    results_per_role: _ResultsPerRole = {}
    role: "_Role"
    for role in ["provider", "requirer"]:
        logging.info(f"Running tests for role: {role}")
        interface_tests = tests_per_role[role]["tests"]
        charm_configs = tests_per_role[role]["charms"]

        if not interface_tests:
            logging.info(f"No tests specified for {interface}/{role}; skipping...")
            results_per_role[role] = {}
        elif not charm_configs:
            logging.info(f"No charms registered for {interface}/{role}; skipping...")
            results_per_role[role] = {}
        else:
            logging.info(
                f"Running {len(interface_tests)} {interface} interface tests on: "
                f"{[charm.name for charm in charm_configs]}..."
            )
            results_per_role[role] = _test_charms(
                charm_configs, interface, version, role, repo, branch
            )
    return results_per_role


def _test_interface_version(
    tests_per_version, interface: str, repo: str, branch: str
) -> "_ResultsPerVersion":
    """Run the tests for each version of this interface."""
    logging.info(f"Running tests for interface: {interface}")
    results_per_version: _ResultsPerVersion = {}
    for version, tests_per_role in tests_per_version.items():
        logging.info(f"Running tests for version: {version}")

        version_int = int(version[1:])
        results_per_version[version] = _test_roles(
            tests_per_role, interface, version_int, repo, branch
        )

    return results_per_version


def run_interface_tests(
    path: Path,
    repo: str,
    branch: str,
    include: str = "*",
    keep_cache: bool = False,
) -> "_ResultsPerInterface":
    """Run the tests for the specified interfaces, defaulting to all."""
    failed = False
    if not keep_cache:
        _clean()
    test_results = {}
    collected = collect_tests(path=path, include=include)
    for interface, version_to_roles in collected.items():
        results_per_version = _test_interface_version(
            version_to_roles, interface, repo, branch
        )
        test_results[interface] = results_per_version

        # Running in GitHub actions with the maintainer set on the test.
        if os.getenv("GITHUB_ACTIONS"):
            for version, tests_per_role in version_to_roles.items():
                maintainer = tests_per_role.get("maintainer")
                if maintainer and test_failed(results_per_version[version]):
                    failed = True
                    create_issue(
                        interface, version, results_per_version[version], maintainer
                    )

    if not collected:
        logging.warning("No tests collected.")

    return test_results, failed


def test_failed(role_result: "_ResultsPerRole"):
    for charm_result in role_result.values():
        if False in charm_result.values():
            return True

    return False


def get_team_members_from_team_slug(slug: str) -> List[str]:
    gh = Github(os.getenv("GITHUB_TOKEN"))
    team = gh.get_organization("canonical").get_team_by_slug(slug)
    if not team:
        return []
    return [m.login for m in team.get_members() if not m.login.endswith("-bot")]


def create_issue(
    interface: str, version: str, result_per_role: "_ResultsPerRole", maintainer: str
):
    gh = Github(os.getenv("GITHUB_TOKEN"))
    repo = gh.get_repo("canonical/charm-relation-interfaces")
    issue_assignees = ["IronCore864", "tonyandrewmeyer"]
    workflow_url = ""
    github_run_id = os.getenv("GITHUB_RUN_ID")
    if github_run_id:
        workflow_url = f"https://github.com/canonical/charm-relation-interfaces/actions/runs/{github_run_id}"
    result = flatten_test_result(result_per_role)
    title = f"Interface test for {interface} {version} failed."
    mention_team_members = ", ".join(
        ["@" + member for member in get_team_members_from_team_slug(maintainer)]
    )
    body = f"""\
Tests for interface {interface} {version} failed.

{result}

See the workflow {workflow_url} for more detail.

{mention_team_members}
"""

    issue = None
    labels = ["Type: Interface Testing"]
    for existing_issue in repo.get_issues(state="open", labels=labels):
        if f"{interface} {version}" in existing_issue.title:
            issue = existing_issue
            break

    if issue:
        issue.create_comment(body)
        print(f"GitHub issue updated: {issue.html_url}")
    else:
        issue = repo.create_issue(
            title=title, body=body, assignees=issue_assignees, labels=labels
        )
        print(f"GitHub issue created: {issue.html_url}")


def flatten_test_result(version_result: "_ResultsPerRole"):
    result = ""

    provider_res = "\n".join(
        f"- {charm}: {res}"
        for charm, res in version_result.get("provider").items()
        if res is False
    )

    if provider_res:
        result = "## Provider\n\n" + provider_res

    requirer_res = "\n".join(
        f"- {charm}: {res}"
        for charm, res in version_result.get("requirer").items()
        if res is False
    )

    if requirer_res:
        result += "## Requirer\n\n" + requirer_res

    return result


def pprint_interface_test_results(test_results: dict):
    """Pretty print the results of interface tests."""
    print("+++ Results +++")
    print(json.dumps(test_results, indent=2))


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--include",
        default="*",
        help="Glob to filter what interfaces to include in the test matrix.",
    )
    parser.add_argument(
        "--keep-cache",
        default=False,
        help="Keep the charm cache intact before running the tests. "
        "This will save some time when running the tests again "
        "(assuming the charms haven't changed).",
    )
    parser.add_argument(
        "--repo",
        default="https://github.com/canonical/charm-relation-interfaces",
        help="The repository where to find the tests, defaults to https://github.com/canonical/charm-relation-interfaces.",
    )
    parser.add_argument(
        "--branch",
        default="main",
        help="The branch of the repo where to find the tests, defaults to main.",
    )
    args = parser.parse_args()

    result, failed = run_interface_tests(
        Path("."), args.repo, args.branch, args.include, args.keep_cache
    )
    pprint_interface_test_results(result)
    exit(1) if failed else exit(0)