cariflex/tools/EVerest-main/tests/ocpp_tests/conftest.py

# SPDX-License-Identifier: Apache-2.0
# Copyright Pionix GmbH and Contributors to EVerest

# noinspection PyUnresolvedReferences
from everest.testing.core_utils.fixtures import *
from everest.testing.core_utils.probe_module import ProbeModule
from everest.testing.ocpp_utils.charge_point_utils import OcppTestConfiguration

# pylint: disable-next=unused-import
from everest.testing.ocpp_utils.fixtures import (
    ocpp_config,
    ocpp_version,
    charge_point,
    charge_point_v201,
    charge_point_v21,
    central_system,
    central_system_v16,
    central_system_v201,
    central_system_v21,
    central_system_v16_standalone,
    test_utility,
)

import test_sets.everest_test_utils as everest_test_utils

from typing import Any, Callable

import logging

import pytest



def pytest_addoption(parser):
    # Guard against duplicate registration when tests/conftest.py is also loaded
    # (e.g. when invoked via run-tests.sh with --config-file pointing to tests/pytest.ini).
    try:
        parser.addoption(
            "--everest-prefix",
            action="store",
            default="../../build/dist",
            help="everest prefix path; default = '../../build/dist'",
        )
    except ValueError:
        logging.error("Option --everest-prefix already registered, skipping duplicate registration.")


def pytest_sessionfinish(session, exitstatus):
    pass


@pytest.fixture(scope="session")
def exi_generator():
    certs_path = str(Path(__file__).parent / "test_sets" / "everest-aux" / "certs")
    return everest_test_utils.EXIGenerator(certs_path)


@pytest.fixture
def test_config() -> OcppTestConfiguration:
    return everest_test_utils.load_test_config()


@pytest.fixture
def core_config(request) -> EverestEnvironmentCoreConfiguration:
    everest_prefix = Path(request.config.getoption("--everest-prefix"))

    marker = request.node.get_closest_marker("everest_core_config")

    if marker is None:
        test_function_name = request.function.__name__
        test_module_name = request.module.__name__
        everest_config_path = everest_test_utils.get_everest_config(
            test_function_name, test_module_name
        )
    else:
        everest_config_path = (
            Path(__file__).parent /
            "test_sets/everest-aux/config" / marker.args[0]
        )

    return EverestEnvironmentCoreConfiguration(
        everest_core_path=everest_prefix,
        template_everest_config_path=everest_config_path,
    )


@pytest.fixture
def started_test_controller(test_controller):
    test_controller.start()
    yield test_controller
    test_controller.stop()


@pytest.fixture
def skip_implementation():
    return None


@pytest.fixture
def overwrite_implementation():
    return None


def implement_command(
    module: ProbeModule,
    skip_implementation: dict,
    implementation_id: str,
    command_name: str,
    handler: Callable[[dict], Any],
):
    skip = False
    if skip_implementation:
        if implementation_id in skip_implementation:
            to_skip = skip_implementation[implementation_id]
            if command_name in to_skip:
                logging.info(f"Skipping implementation of {command_name}")
                skip = True
    if not skip:
        module.implement_command(implementation_id, command_name, handler)


@pytest.fixture
def probe_module(
    started_test_controller, everest_core, skip_implementation
) -> ProbeModule:
    # initiate the probe module, connecting to the same runtime session the test controller started
    module = ProbeModule(everest_core.get_runtime_session())

    logging.info(f"hello: {skip_implementation}")

    # implement necessary commands for initialization in the module
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "get_evse",
        lambda arg: {"id": 1, "connectors": [{"id": 1}]},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "enable_disable",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "authorize_response",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "update_allowed_energy_transfer_modes",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "withdraw_authorization",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "reserve",
        lambda arg: False,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "cancel_reservation",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "pause_charging",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "resume_charging",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "stop_transaction",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "force_unlock",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleIso15118Extensions",
        "set_get_certificate_response",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "external_ready_to_start_charging",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorA",
        "set_plug_and_charge_configuration",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "get_evse",
        lambda arg: {"id": 2, "connectors": [{"id": 1}]},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "enable_disable",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "authorize_response",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "update_allowed_energy_transfer_modes",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "withdraw_authorization",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "reserve",
        lambda arg: False,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "cancel_reservation",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "pause_charging",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "resume_charging",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "stop_transaction",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "force_unlock",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "external_ready_to_start_charging",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleConnectorB",
        "set_plug_and_charge_configuration",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSystem",
        "get_boot_reason",
        lambda arg: "PowerUp",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSystem",
        "update_firmware",
        lambda arg: "Accepted",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSystem",
        "allow_firmware_installation",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSystem",
        "upload_logs",
        lambda arg: "Accepted",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSystem",
        "is_reset_allowed",
        lambda arg: True,
    )
    implement_command(
        module, skip_implementation, "ProbeModuleSystem", "reset", lambda arg: None
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSystem",
        "set_system_time",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_leaf_expiry_days_count",
        lambda arg: 42,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_v2g_ocsp_request_data",
        lambda arg: {"ocsp_request_data_list": []},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_mo_ocsp_request_data",
        lambda arg: {"ocsp_request_data_list": []},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "install_ca_certificate",
        lambda arg: "Accepted",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "delete_certificate",
        lambda arg: "Accepted",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "update_leaf_certificate",
        lambda arg: "Accepted",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "verify_certificate",
        lambda arg: "Valid",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_installed_certificates",
        lambda arg: {"status": "Accepted", "certificate_hash_data_chain": []},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "update_ocsp_cache",
        lambda arg: None,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "is_ca_certificate_installed",
        lambda arg: False,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "generate_certificate_signing_request",
        lambda arg: {"status": "Accepted"},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_leaf_certificate_info",
        lambda arg: {"status": "Accepted"},
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_verify_file",
        lambda arg: "",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_verify_location",
        lambda arg: "",
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "verify_file_signature",
        lambda arg: True,
    )
    implement_command(
        module,
        skip_implementation,
        "ProbeModuleSecurity",
        "get_all_valid_certificates_info",
        lambda arg: {"status": "NotFound", "info": []},
    )

    return module


@pytest.fixture()
def ocpp_config_reader(ocpp_config, ocpp_configuration):
    """
    Returns a reader over the final OCPP config (after all adaptations during test setup) for convenience.
    """
    return everest_test_utils.OCPPConfigReader(ocpp_configuration)