diff --git a/everest-testing/README.md b/everest-testing/README.md index ccf5d403..f3f49cde 100644 --- a/everest-testing/README.md +++ b/everest-testing/README.md @@ -27,13 +27,17 @@ The core_utils basically provide two fixtures that you can require in your test - **core_config** Core configuration, which is the everest_core path and the configuration path (utilizes the `everest_core_config` marker.) - **probe_module_config** The main fixture `everest_core` can be used to start and stop the everest-core application. -- **ocpp_config** The main fixture `everest_core` can be used to start and stop the everest-core application. +- **ocpp_config** Used to provide the configuration to set up the OCPP (1.6 or 2.0.1) module. +- **evse_security_config** Used to provide the configuration to set up the EvseSecurity module. +- **persistent_storage_config** Used to provide the configuration to set up the PersistentStorage module. ### pytest markers Some OCPP fixtures will parse pytest markers of test cases. The following markers can be used: - **everest_core_config**: Can be used to specify the everest configuration file to be used in this test case - **probe_module**: If set, the ProbeModule will be injected into the config (used by the `probe_module_config` fixture). You may provide the config args as fixture arguments. +- **source_certs_dir**: If set and the default `evse_security_config` fixture is used, this will cause the `EvseSecurity` module configuration to use a temporary certificates folder into which the source certificate folder trees are copied. +- **use_temporary_persistent_store**: If set and the default `persistent_storage_config` fixture is used, this will cause the `PersistentStore` module configuration to use a temporary database. ## OCPP utils @@ -77,7 +81,7 @@ Note: When overriding a fixture, be careful which pytest markers might be ignore ```python -from everest.testing.core_utils.fixtures import everest_core, probe_module_config, test_controller +from everest.testing.core_utils.fixtures import * from everest.testing.ocpp_utils.fixtures import ocpp_config @pytest.fixture @@ -103,6 +107,8 @@ class TestMyEverestModule: ``` +_Note_: The "*" import from `core_utils.fixtures` may ensure backwards compatibility to automatically load new default fixtures in the future! + ## Install diff --git a/everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/evse_security_configuration_visitor.py b/everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/evse_security_configuration_visitor.py new file mode 100644 index 00000000..ba6de406 --- /dev/null +++ b/everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/evse_security_configuration_visitor.py @@ -0,0 +1,78 @@ +from copy import deepcopy +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Dict, Optional + +from everest.testing.core_utils.configuration.everest_configuration_visitors.everst_configuration_visitor import \ + EverestConfigAdjustmentVisitor + + +@dataclass +class EvseSecurityModuleConfiguration: + csms_ca_bundle: Optional[str] = None + mf_ca_bundle: Optional[str] = None + mo_ca_bundle: Optional[str] = None + v2g_ca_bundle: Optional[str] = None + csms_leaf_cert_directory: Optional[str] = None + csms_leaf_key_directory: Optional[str] = None + secc_leaf_cert_directory: Optional[str] = None + secc_leaf_key_directory: Optional[str] = None + private_key_password: Optional[str] = None + + @staticmethod + def default_from_cert_path(certs_directory: Path): + """ Return a default definition of bundles and directory given base directory. """ + return EvseSecurityModuleConfiguration( + csms_ca_bundle=str(certs_directory / "ca/csms/CSMS_ROOT_CA.pem"), + mf_ca_bundle=str(certs_directory / "ca/mf/MF_ROOT_CA.pem"), + mo_ca_bundle=str(certs_directory / "ca/mo/MO_ROOT_CA.pem"), + v2g_ca_bundle=str(certs_directory / "ca/v2g/V2G_ROOT_CA.pem"), + csms_leaf_cert_directory=str(certs_directory / "client/csms"), + csms_leaf_key_directory=str(certs_directory / "client/csms"), + secc_leaf_cert_directory=str(certs_directory / "client/cso"), + secc_leaf_key_directory=str(certs_directory / "client/cso"), + ) + + def to_module_configuration(self) -> Dict: + return {k: v for k, v in asdict(self).items() if v is not None} + + +class EvseSecurityModuleConfigurationVisitor(EverestConfigAdjustmentVisitor): + """ Adjusts the Everest configuration by manipulating the evse security module configuration to point + to the correct (temporary) certificate paths. + + """ + + def __init__(self, + configuration: EvseSecurityModuleConfiguration, + module_id: Optional[str] = None): + """ + + Args: + configuration: module configuration. This will be merged into the template configuration (meaning None values are ignored/taken from the originally provided Everest configuration) + module_id: Id of security module; if None, auto-detected by module type "EvseSecurity + """ + self._security_module_id = module_id + self._configuration = configuration + + def _determine_module_id(self, everest_config: Dict): + if self._security_module_id: + assert self._security_module_id in everest_config[ + "active_modules"], f"Module id {self._security_module_id} not found in EVerest configuration" + return self._security_module_id + else: + try: + return next(k for k, v in everest_config["active_modules"].items() if v["module"] == "EvseSecurity") + except StopIteration: + raise ValueError("No EvseSecurity module found in EVerest configuration") + + def adjust_everest_configuration(self, everest_config: Dict): + + adjusted_config = deepcopy(everest_config) + + module_cfg = adjusted_config["active_modules"][self._determine_module_id(adjusted_config)] + + module_cfg["config_module"] = {**module_cfg["config_module"], + **self._configuration.to_module_configuration()} + + return adjusted_config diff --git a/everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/persistent_store_configuration_visitor.py b/everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/persistent_store_configuration_visitor.py new file mode 100644 index 00000000..537cf0e9 --- /dev/null +++ b/everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/persistent_store_configuration_visitor.py @@ -0,0 +1,47 @@ +from copy import deepcopy +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Dict, Optional + +from everest.testing.core_utils.configuration.everest_configuration_visitors.everst_configuration_visitor import \ + EverestConfigAdjustmentVisitor + + +class PersistentStoreConfigurationVisitor(EverestConfigAdjustmentVisitor): + """ Adjusts the Everest configuration by manipulating the PersistentStore module configuration to point + to the desired (temporary) storage + + """ + + def __init__(self, + sqlite_db_file_path: Path, + module_id: Optional[str] = None): + """ + + Args: + sqlite_db_file_path: database to be used in configuration + module_id: Id of security module; if None, auto-detected by module type "EvseSecurity + """ + self._module_id = module_id + self._sqlite_db_file_path = sqlite_db_file_path + + def _determine_module_id(self, everest_config: Dict): + if self._module_id: + assert self._module_id in everest_config[ + "active_modules"], f"Module id {self._module_id} not found in EVerest configuration" + return self._module_id + else: + try: + return next(k for k, v in everest_config["active_modules"].items() if v["module"] == "PersistentStore") + except StopIteration: + raise ValueError("No PersistentStore module found in EVerest configuration") + + def adjust_everest_configuration(self, everest_config: Dict): + + adjusted_config = deepcopy(everest_config) + + module_cfg = adjusted_config["active_modules"][self._determine_module_id(adjusted_config)] + + module_cfg.setdefault("config_module", {})["sqlite_db_file_path"] = str(self._sqlite_db_file_path) + + return adjusted_config diff --git a/everest-testing/src/everest/testing/core_utils/configuration/everest_environment_setup.py b/everest-testing/src/everest/testing/core_utils/configuration/everest_environment_setup.py index bafa65d2..c3dea9e3 100644 --- a/everest-testing/src/everest/testing/core_utils/configuration/everest_environment_setup.py +++ b/everest-testing/src/everest/testing/core_utils/configuration/everest_environment_setup.py @@ -2,6 +2,7 @@ # Copyright 2020 - 2023 Pionix GmbH and Contributors to EVerest import logging +import shutil from dataclasses import dataclass, field from pathlib import Path from typing import Optional, Dict, List, Union @@ -9,8 +10,13 @@ import yaml from everest.testing.core_utils.common import OCPPVersion -from everest.testing.core_utils.configuration.everest_configuration_visitors.ocpp_module_configuration_visitor import OCPPModuleConfigurationVisitor, \ +from everest.testing.core_utils.configuration.everest_configuration_visitors.evse_security_configuration_visitor import \ + EvseSecurityModuleConfigurationVisitor, EvseSecurityModuleConfiguration +from everest.testing.core_utils.configuration.everest_configuration_visitors.ocpp_module_configuration_visitor import \ + OCPPModuleConfigurationVisitor, \ OCPPModulePaths16, OCPPModulePaths201 +from everest.testing.core_utils.configuration.everest_configuration_visitors.persistent_store_configuration_visitor import \ + PersistentStoreConfigurationVisitor from everest.testing.core_utils.configuration.everest_configuration_visitors.probe_module_configuration_visitor import \ ProbeModuleConfigurationVisitor @@ -31,6 +37,18 @@ class EverestEnvironmentOCPPConfiguration: Path] = None # Path for OCPP config to be used; if not provided, will be determined from everest config +@dataclass +class EverestEnvironmentEvseSecurityConfiguration: + use_temporary_certificates_folder: bool = True # if true, configuration will be adapted to use temporary certifcates folder, this assumes a "default" file tree structure, cf. the EvseSecurityModuleConfiguration dataclass + module_id: Optional[str] = None # if None, auto-detected + source_certificate_directory: Optional[Path] = None # if provided, this will be copied to temporary path and + + +@dataclass +class EverestEnvironmentPersistentStoreConfiguration: + use_temporary_folder: bool = True # if true, a temporary persistant storage folder will be used + + @dataclass class EverestEnvironmentCoreConfiguration: everest_core_path: Path @@ -62,21 +80,26 @@ class EverestTestEnvironmentSetup: @dataclass class _EverestEnvironmentTemporaryPaths: """ Paths of the temporary configuration files / data """ + certs_dir: Path # used by both OCPP and evse security ocpp_config_file: Path ocpp_user_config_file: Path ocpp_database_dir: Path - ocpp_certs_dir: Path ocpp_message_log_directory: Path + persistent_store_db_path: Path def __init__(self, core_config: EverestEnvironmentCoreConfiguration, ocpp_config: Optional[EverestEnvironmentOCPPConfiguration] = None, probe_config: Optional[EverestEnvironmentProbeModuleConfiguration] = None, + evse_security_config: Optional[EverestEnvironmentEvseSecurityConfiguration] = None, + persistent_store_config: Optional[EverestEnvironmentPersistentStoreConfiguration] = None, standalone_module: Optional[str] = None ) -> None: self._core_config = core_config self._ocpp_config = ocpp_config self._probe_config = probe_config + self._evse_security_config = evse_security_config + self._persistent_store_config = persistent_store_config self._standalone_module = standalone_module if not self._standalone_module and self._probe_config: self._standalone_module = self._probe_config.module_id @@ -99,6 +122,9 @@ def setup_environment(self, tmp_path: Path): temporary_paths=temporary_paths ) + if self._evse_security_config: + self._setup_evse_security_configuration(temporary_paths) + @property def everest_core(self) -> EverestCore: assert self._everest_core, "Everest Core not initialized; run 'setup_environment' first" @@ -107,19 +133,23 @@ def everest_core(self) -> EverestCore: def _create_temporary_directory_structure(self, tmp_path: Path) -> _EverestEnvironmentTemporaryPaths: ocpp_config_dir = tmp_path / "ocpp_config" ocpp_config_dir.mkdir(exist_ok=True) - ocpp_certs_dir = ocpp_config_dir / "certs" - ocpp_certs_dir.mkdir(exist_ok=True) + certs_dir = tmp_path / "certs" + certs_dir.mkdir(exist_ok=True) ocpp_logs_dir = ocpp_config_dir / "logs" ocpp_logs_dir.mkdir(exist_ok=True) + persistent_store_dir = tmp_path / "persistent_storage" + persistent_store_dir.mkdir(exist_ok=True) + logging.info(f"temp ocpp config files directory: {ocpp_config_dir}") return self._EverestEnvironmentTemporaryPaths( ocpp_config_file=ocpp_config_dir / "config.json", ocpp_user_config_file=ocpp_config_dir / "user_config.json", ocpp_database_dir=ocpp_config_dir, - ocpp_certs_dir=ocpp_certs_dir, - ocpp_message_log_directory=ocpp_logs_dir + certs_dir=certs_dir, + ocpp_message_log_directory=ocpp_logs_dir, + persistent_store_db_path=persistent_store_dir / "persistent_store.db" ) def _create_ocpp_module_configuration_visitor(self, @@ -129,7 +159,7 @@ def _create_ocpp_module_configuration_visitor(self, ocpp_paths = OCPPModulePaths16( ChargePointConfigPath=str(temporary_paths.ocpp_config_file), MessageLogPath=str(temporary_paths.ocpp_message_log_directory), - CertsPath=str(temporary_paths.ocpp_certs_dir), + CertsPath=str(temporary_paths.certs_dir), UserConfigPath=str(temporary_paths.ocpp_user_config_file), DatabasePath=str(temporary_paths.ocpp_database_dir) # self.temp_ocpp_database_dir.name ) @@ -137,7 +167,7 @@ def _create_ocpp_module_configuration_visitor(self, ocpp_paths = OCPPModulePaths201( ChargePointConfigPath=str(temporary_paths.ocpp_config_file), MessageLogPath=str(temporary_paths.ocpp_message_log_directory), - CertsPath=str(temporary_paths.ocpp_certs_dir), + CertsPath=str(temporary_paths.certs_dir), CoreDatabasePath=str(temporary_paths.ocpp_database_dir), DeviceModelDatabasePath=str(temporary_paths.ocpp_database_dir / "device_model_storage.db"), ) @@ -162,7 +192,7 @@ def _setup_libocpp_configuration(self, source_certs_directory: Path, liboccp_configuration_helper.install_default_ocpp_certificates( source_certs_directory=source_certs_directory, - target_certs_directory=temporary_paths.ocpp_certs_dir) # Path(self.temp_ocpp_certs_dir.name)) + target_certs_directory=temporary_paths.certs_dir) # Path(self.temp_ocpp_certs_dir.name)) liboccp_configuration_helper.generate_ocpp_config( central_system_port=self._ocpp_config.central_system_port, @@ -187,6 +217,18 @@ def _create_everest_configuration_visitors(self, temporary_paths: _EverestEnviro configuration_visitors.append( ProbeModuleConfigurationVisitor(connections=self._probe_config.connections, module_id=self._probe_config.module_id)) + + if self._evse_security_config and self._evse_security_config.use_temporary_certificates_folder: + configuration_visitors.append( + EvseSecurityModuleConfigurationVisitor(module_id=self._evse_security_config.module_id, + configuration=EvseSecurityModuleConfiguration.default_from_cert_path( + temporary_paths.certs_dir))) + + if self._persistent_store_config and self._persistent_store_config.use_temporary_folder: + configuration_visitors.append( + PersistentStoreConfigurationVisitor(sqlite_db_file_path=temporary_paths.persistent_store_db_path) + ) + return configuration_visitors def _determine_configured_charge_point_config_path_from_everest_config(self): @@ -205,3 +247,9 @@ def _determine_configured_charge_point_config_path_from_everest_config(self): raise ValueError(f"unknown OCPP version {self._ocpp_config.ocpp_version}") ocpp_config_path = ocpp_dir / charge_point_config_path return ocpp_config_path + + def _setup_evse_security_configuration(self, temporary_paths: _EverestEnvironmentTemporaryPaths): + """ If configures, copies the source certificate trees""" + if self._evse_security_config and self._evse_security_config.source_certificate_directory: + shutil.copytree(self._evse_security_config.source_certificate_directory / "ca", temporary_paths.certs_dir / "ca", dirs_exist_ok=True) + shutil.copytree(self._evse_security_config.source_certificate_directory / "client", temporary_paths.certs_dir / "client", dirs_exist_ok=True) diff --git a/everest-testing/src/everest/testing/core_utils/fixtures.py b/everest-testing/src/everest/testing/core_utils/fixtures.py index c6e02d81..81c5a728 100644 --- a/everest-testing/src/everest/testing/core_utils/fixtures.py +++ b/everest-testing/src/everest/testing/core_utils/fixtures.py @@ -5,8 +5,10 @@ import pytest -from everest.testing.core_utils.configuration.everest_environment_setup import EverestEnvironmentProbeModuleConfiguration, \ - EverestTestEnvironmentSetup, EverestEnvironmentOCPPConfiguration, EverestEnvironmentCoreConfiguration +from everest.testing.core_utils.configuration.everest_environment_setup import \ + EverestEnvironmentProbeModuleConfiguration, \ + EverestTestEnvironmentSetup, EverestEnvironmentOCPPConfiguration, EverestEnvironmentCoreConfiguration, \ + EverestEnvironmentEvseSecurityConfiguration, EverestEnvironmentPersistentStoreConfiguration from everest.testing.core_utils.controller.everest_test_controller import EverestTestController from everest.testing.core_utils.everest_core import EverestCore @@ -43,13 +45,31 @@ def core_config(request) -> EverestEnvironmentCoreConfiguration: def ocpp_config(request) -> Optional[EverestEnvironmentOCPPConfiguration]: return None +@pytest.fixture +def evse_security_config(request) -> Optional[EverestEnvironmentEvseSecurityConfiguration]: + source_certs_dir_marker = request.node.get_closest_marker("source_certs_dir") + if source_certs_dir_marker: + return EverestEnvironmentEvseSecurityConfiguration(source_certificate_directory=Path(source_certs_dir_marker.args[0])) + return None + +@pytest.fixture +def persistent_store_config(request) -> Optional[EverestEnvironmentPersistentStoreConfiguration]: + persistent_store_marker = request.node.get_closest_marker("use_temporary_persistent_store") + if persistent_store_marker: + return EverestEnvironmentPersistentStoreConfiguration(use_temporary_folder=True) + return None + + @pytest.fixture def everest_core(request, tmp_path, core_config: EverestEnvironmentCoreConfiguration, - ocpp_config: EverestEnvironmentOCPPConfiguration, - probe_module_config: EverestEnvironmentProbeModuleConfiguration) -> EverestCore: + ocpp_config: Optional[EverestEnvironmentOCPPConfiguration], + probe_module_config: Optional[EverestEnvironmentProbeModuleConfiguration], + evse_security_config: Optional[EverestEnvironmentEvseSecurityConfiguration], + persistent_store_config: Optional[EverestEnvironmentPersistentStoreConfiguration], + ) -> EverestCore: """Fixture that can be used to start and stop everest-core""" standalone_module_marker = request.node.get_closest_marker('standalone_module') @@ -58,6 +78,8 @@ def everest_core(request, core_config=core_config, ocpp_config=ocpp_config, probe_config=probe_module_config, + evse_security_config=evse_security_config, + persistent_store_config=persistent_store_config, standalone_module=standalone_module_marker )