Source code for tests.conftest

# conftest.py
#
# SPDX-FileCopyrightText: <text>Copyright 2025-2026 Arm Limited and/or its
# affiliates <open-source-office@arm.com></text>
#
# SPDX-License-Identifier: MIT

import logging
import os
import re
from pathlib import Path
import pytest
from datetime import datetime

from importlib import import_module, resources

from test_automation.utils.auto_platform_base import AutoTestPlatformBase
from test_automation.configs.config import Config
from test_automation.targets.fpga.fpga_controller import (
    load_fpga_controller_from_yaml,
)

from test_automation.targets.registry import get_factory, DriverBundle



logger = logging.getLogger(__name__)


# -------------
# Small helpers
# -------------




def _compute_run_paths(config: pytest.Config) -> dict:
    """
    Create a unique run directory and return run metadata dict.

    :param pytestconfig: Pytest configuration object.
    :returns: Dictionary.
    """
    platform_opt = str(config.getoption("--platform") or "unknown").lower()
    logs_root = Path("logs")
    run_id = datetime.now().strftime("%Y%m%d-%H%M%S")
    run_dir = logs_root / f"{platform_opt}_{run_id}"
    run_dir.mkdir(parents=True, exist_ok=True)
    return {
        "run_id": run_id,
        "platform": platform_opt,
        "log_dir": run_dir,
        "prefix": platform_opt or "fvp_boot",
    }





def _load_target_plugin(config: pytest.Config) -> None:
    """
    Dynamically import target plugin (e.g., for FVP) if selected.

    :param pytestconfig: Pytest configuration object.
    :returns: None.
    """
    try:
        cfg = Config(
            path=str(Path(config.getoption("--config")).expanduser()),
            build_dir=config.getoption("--build-dir"),
            fvp_binary=config.getoption("--fvp-binary"),
        )
        plat_key = str(config.getoption("--platform")).lower()
        plat_dict = cfg.get_platform(plat_key)
        ptype = str(plat_dict.get("type", "")).lower()
        if ptype == "fvp":
            import_module("test_automation.targets.fvp.plugin")
    except Exception as e:
        logger.debug("Skipping dynamic target plugin load: %s", e)





def _apply_debug_logging(config: pytest.Config) -> None:
    """
    Enable CLI logging if --debug-logs was passed.

    :param pytestconfig: Pytest configuration object.
    :returns: None.
    """
    if not config.getoption("--debug-logs"):
        return
    config.option.log_cli = True
    config.option.log_cli_level = "DEBUG"
    config.option.log_cli_format = (
        "%(asctime)s %(levelname)s %(name)s:%(lineno)d - %(message)s"
    )
    config.option.log_cli_date_format = "%H:%M:%S"





def _prepare_bundle_paths(
    pytestconfig: pytest.Config, bundle: DriverBundle
) -> dict:
    """
    Ensure bundle paths (manager + driver) are set up; return run paths dict.

    If pytest_configure already created run paths, reuse them; otherwise,
    compute a fallback here.
    """
    rp = getattr(pytestconfig, "_arm_run_paths", None)
    if not rp:
        # Fallback when pytest_configure didn't set it (e.g., external usage)
        platform_opt = str(
            getattr(pytestconfig, "getoption", lambda *_: "unknown")(
                "--platform", "unknown"
            )
        ).lower()
        run_id = datetime.now().strftime("%Y%m%d-%H%M%S")
        log_dir = Path("logs") / f"{platform_opt}-{run_id}"
        log_dir.mkdir(parents=True, exist_ok=True)
        rp = {
            "run_id": run_id,
            "platform": platform_opt,
            "log_dir": log_dir,
            "prefix": platform_opt or "fvp_boot",
        }
        # set attribute directly instead of setattr
        pytestconfig._arm_run_paths = rp

    # Telnet manager path
    mgr = getattr(bundle, "manager", None)
    if mgr and hasattr(mgr, "set_log_dir"):
        mgr.set_log_dir(str(Path(rp["log_dir"]).expanduser().resolve()))

    # Driver paths/prefix
    drv = bundle.driver
    if hasattr(drv, "log_dir"):
        drv.log_dir = Path(rp["log_dir"]).expanduser().resolve()
        drv.log_dir.mkdir(parents=True, exist_ok=True)
    if hasattr(drv, "log_prefix"):
        drv.log_prefix = rp["prefix"]

    return rp





def _power_on(bundle: DriverBundle, ptype: str, plat: dict) -> None:
    """Bring the platform up and wait until ready."""
    logger.info(
        "Powering on platform type=%s name=%s", ptype, plat.get("name")
    )
    getattr(bundle.driver, "init", lambda: None)()
    bundle.driver.start()
    logger.info("Waiting for platform to become ready…")
    bundle.driver.wait_ready()
    logger.info("Platform ready")





def _cleanup_manager(mgr) -> None:
    """Best-effort session cleanup on the manager."""
    for method in ("close_sessions", "stop_all"):
        try:
            getattr(mgr, method)()
            # Once one cleanup method succeeds, we're done.
            return
        except Exception:
            # Try the next method if this one fails.
            continue





def _power_off(bundle: DriverBundle) -> None:
    """Attempt graceful power-off and session cleanup."""
    logger.info("Powering off platform")
    try:
        bundle.driver.stop()
    except Exception:
        # Don't hide the error; log it and still attempt cleanup.
        logger.exception("Error while stopping device; going with cleanup")
    finally:
        mgr = getattr(bundle, "manager", None)
        if mgr:
            _cleanup_manager(mgr)



# ------------------------------
# Initial configuration + conditional plugin load
# ------------------------------




def pytest_configure(config: pytest.Config) -> None:
    """
    Configure pytest logging and conditionally load target plugins.

    :param config: Pytest configuration object.
    :returns: None.
    """
    _apply_debug_logging(config)

    # Create per-run logs root + save metadata for other hooks/fixtures.
    run_paths = _compute_run_paths(config)
    config._arm_run_paths = run_paths

    # Load the proper target plugin if the selected platform requires it.
    _load_target_plugin(config)



# ------------------------------
# Shared config/platform loaders
# ------------------------------


@pytest.fixture(scope="session")


def cfg(pytestconfig: pytest.Config) -> Config:
    """
    Load the YAML configuration once per test session.

    :param pytestconfig: Pytest configuration object.
    :returns: Parsed :class:`Config` instance.
    """
    config_path = Path(pytestconfig.getoption("--config")).expanduser()
    build_dir = pytestconfig.getoption("--build-dir")
    fvp_binary = pytestconfig.getoption("--fvp-binary")
    logger.debug(
        "Loading YAML config from %s (build-dir=%r, fvp-binary=%r)",
        config_path,
        build_dir,
        fvp_binary,
    )
    try:
        return Config(
            path=str(config_path), build_dir=build_dir, fvp_binary=fvp_binary
        )
    except FileNotFoundError:
        pytest.skip(f"Config file not found: {config_path}")



# Helper reduce fixture complexity


def _available_platform_names(cfg: Config) -> list:
    """
    Return a list of platform 'name' values from cfg.platforms, or an empty
    list if cfg.platforms is not present.

    :param cfg: Parsed configuration object.
    :returns: List of platform names defined in the configuration.
    """
    if not hasattr(cfg, "platforms"):
        return []
    return [p.get("name") for p in cfg.platforms.values()]





def _apply_cfg2_overrides(platform_config: dict) -> None:
    """Temporarily inject cfg2 specific configs before boot."""
    if os.getenv("RD_ASPEN_VARIANT", "").lower() != "cfg2":
        return

    si_cl1_prompt = _get_cfg2_si_cluster1_prompt()
    if not si_cl1_prompt:
        return

    extra_arg = "-C css.smb.si.terminal_uart_si_cluster1.start_telnet=0"
    args = platform_config.setdefault("arguments", [])
    if extra_arg not in args:
        args.append(extra_arg)

    required_terminals = platform_config.setdefault("required_terminals", [])
    if "terminal_uart_si_cluster1" not in required_terminals:
        required_terminals.append("terminal_uart_si_cluster1")

    prompts = platform_config.setdefault("prompts", {})
    shell_map = prompts.setdefault("shell_map", {})
    shell_map["terminal_uart_si_cluster1"] = si_cl1_prompt

    port_map = platform_config.setdefault("port_map", {})
    port_map["si_cl1"] = "terminal_uart_si_cluster1"



@pytest.fixture(scope="session")


def auto_platform_config_data(
    pytestconfig: pytest.Config, cfg: Config
) -> dict:
    """
    Resolve and return the selected platform dictionary from the YAML config.

    :param pytestconfig: Pytest configuration object.
    :param cfg: Parsed configuration object.
    :returns: Platform configuration dictionary.
    """
    platform_name = str(pytestconfig.getoption("--platform")).lower()

    # Fetch the platform config if not present then skip
    try:
        platform_config = cfg.get_platform(platform_name)  # YAML platform dict
    except KeyError:
        names = _available_platform_names(cfg)
        pytest.skip(
            f"Platform '{platform_name}' not found. Available: {names}"
        )
    # Temporary workaround for cfg2 until runtime config exists
    _apply_cfg2_overrides(platform_config)
    # Determine target (support 'type' key as fallback for fvp)
    target = str(
        platform_config.get("target") or platform_config.get("type", "")
    ).lower()

    # Enforce host requirement only for FPGA target
    if target == "fpga":
        host = pytestconfig.getoption("--host") or platform_config.get("host")
        if not host:
            raise pytest.UsageError(
                (
                    f"Platform '{platform_name}' with target='{target}' "
                    "requires a host. Provide it via "
                    "--host <hostname> while invoking pytest."
                )
            )
        platform_config["host"] = host

    logger.debug("Loaded platform keys: %s", list(platform_config.keys()))
    return platform_config



# ------------------------------
# Core bundle: build/start/login/teardown once
# ------------------------------


@pytest.fixture(scope="session")


def platform_bundle(pytestconfig, cfg: Config, auto_platform_config_data):
    """
    Build the platform bundle once from the config.

    :param pytestconfig: Pytest configuration object.
    :param cfg: Parsed configuration object.
    :param auto_platform_config_data: Platform configuration dictionary.
    :returns: Yields an initialized :class:`DriverBundle`.
    """
    plat = auto_platform_config_data
    ptype = str(plat.get("type", "")).lower()

    factory = get_factory(ptype)
    if not factory:
        pytest.skip(f"No platform plugin registered for type={ptype!r}")

    bundle: DriverBundle = factory(plat, cfg)

    # type-specific env export (e.g., FVP_BINARY)
    if callable(getattr(bundle, "export_env", None)):
        bundle.export_env(pytestconfig, plat)

    # Paths and prefixes for logs and sessions
    _prepare_bundle_paths(pytestconfig, bundle)

    # Power on/ready/optional login
    _power_on(bundle, ptype, plat)
    if bundle.manager and callable(getattr(bundle, "login_primary", None)):
        bundle.login_primary(bundle.manager, plat)

    try:
        yield bundle
    finally:
        _power_off(bundle)



# ------------------------------
# Thin fixtures used by tests
# ------------------------------


@pytest.fixture(scope="session")


def platform_driver(platform_bundle):
    """
    Provide the platform driver object for tests.

    :param platform_bundle: Initialized driver bundle.
    :returns: Driver instance from the bundle.
    """
    return platform_bundle.driver



@pytest.fixture(scope="session")


def session_manager(platform_bundle):
    """
    Provide the session/console manager for tests.

    :param platform_bundle: Initialized driver bundle.
    :returns: Session manager object.
    """
    if platform_bundle.manager is None:
        pytest.skip("This platform does not expose a session manager")
    return platform_bundle.manager



# ------------------------------
# Utility / metadata fixtures
# ------------------------------


@pytest.fixture(scope="session")


def platform_name(platform_base_obj, cfg: Config) -> str:
    """
    Normalize the selected platform name to a canonical short name.

    :param platform_base_obj: Initialized :class:`AutoTestPlatformBase`.
    :param cfg: Parsed configuration object.
    :returns: Canonical name like ``'aspne'``.
    """
    available = list(cfg.platforms.keys())
    logger.debug("Available platforms in YAML: %s", available)

    selected_platform = platform_base_obj.platform or ""
    logger.debug("Selected platform: %s", selected_platform)

    if selected_platform not in available:
        logger.warning(
            "Selected platform %s not found in available platforms",
            selected_platform,
        )
        return "unknown"

    if re.search(r"rd_aspen", selected_platform, re.IGNORECASE):
        return "rd_aspen"

    raise RuntimeError(
        f"Platform '{selected_platform}' is defined in config but "
        f"does not match expected patterns (rd_aspen)."
    )



@pytest.fixture(scope="session")


def platform_base_obj(
    session_manager, auto_platform_config_data, pytestconfig
) -> AutoTestPlatformBase:
    """
    Construct the base platform helper object once per session.

    :param session_manager: Session/console manager instance.
    :param auto_platform_config_data: Platform configuration dictionary.
    :param pytestconfig: Pytest configuration object.
    :returns: Initialized :class:`AutoTestPlatformBase` instance.
    """
    cli_platform = (pytestconfig.getoption("--platform") or "").lower()
    return AutoTestPlatformBase(
        session_manager, auto_platform_config_data, cli_platform
    )



@pytest.fixture(scope="session")


def fpga_device(request):
    """
    Session-scoped FPGA device fixture.

    - Initializes and starts the FPGA platform
    - Waits for ready state
    - Starts UART keepalive
    - Ensures graceful shutdown and log collection
    """
    config_path = Path(request.config.getoption("--config"))
    platform = request.config.getoption("--platform")
    host = request.config.getoption("--host")

    logger.info(
        "pytest --config used: %s; looking for platform: %s",
        config_path,
        platform,
    )

    dev = load_fpga_controller_from_yaml(config_path, platform, host=host)

    logger.info("Initializing FPGA device...")
    dev.init()
    logger.info("Starting FPGA device...")
    dev.start()
    logger.info("Waiting for ready state...")
    dev.wait_ready()

    uart_log_path, vuart, pts = dev.get_login_uart()
    dev.net.start_uart_keepalive(vuart=vuart, pts=pts, interval_s=25.0)

    yield dev
    logger.info("Test teardown: stopping UART keepalive and stopping device")
    dev.net.stop_uart_keepalive()
    dev.stop()
    logger.info("FPGA controller stopped and logs collected")



# ------------------------------
# CLI options
# ------------------------------




def pytest_addoption(parser) -> None:
    """
    Register custom CLI options used by the test framework.

    :param parser: Pytest argument parser.
    :returns: None.
    """
    default_cfg = (
        resources.files("test_automation.configs") / "standalone_config.yaml"
    )
    parser.addoption(
        "--platform",
        action="store",
        required=True,
        help="Platform key in the config (e.g., fvp_rd_aspen, fpga_rd_aspen)",
    )
    parser.addoption(
        "--config",
        action="store",
        default=str(default_cfg),
        help="Specify the configuration file to load)",
    )
    parser.addoption(
        "--build-dir",
        action="store",
        default=None,
        help="Optional directory to substitute for BUILD_IMAGE_DIR.",
    )
    parser.addoption(
        "--fvp-binary",
        action="store",
        default=None,
        help="Optional path to substitute for FVP_BINARY.",
    )
    parser.addoption(
        "--debug-logs",
        action="store_true",
        help="Enable live DEBUG logging for this run",
    )
    parser.addoption(
        "--host",
        action="store",
        default=None,
        help=(
            "Hostname/IP for platforms that require a remote host (e.g. fpga)"
        ),
    )





def _get_cfg2_si_cluster1_prompt() -> str | None:
    """
    Return the expected prompt for si_cluster1 based on build type.
    """
    image_name = os.getenv("OPTIONAL_EXISTING_IMAGE_FILENAME", "").lower()

    if "baremetal" in image_name:
        return r"Secondary CPU core 1 \(MPID:0x10100\) is up"

    if "bsp" in image_name:
        return "Hello World"

    return None