diff --git a/tests/integration/server_rest_api/datex2/datex2_import_api_test.py b/tests/integration/server_rest_api/datex2/datex2_import_api_test.py index 0648b5ea..3b181b8d 100644 --- a/tests/integration/server_rest_api/datex2/datex2_import_api_test.py +++ b/tests/integration/server_rest_api/datex2/datex2_import_api_test.py @@ -177,6 +177,55 @@ def test_push_realtime_v35_async_applies_status_updates( # The celery task deletes its source file once it has been processed. assert list(isolated_datex2_dir.iterdir()) == [] + @staticmethod + def test_static_reimport_after_realtime_push_keeps_status( + db: SQLAlchemy, + test_client: OpenApiFlaskClient, + requests_mock: Mocker, + isolated_datex2_dir: Path, + eager_celery_helper, + ) -> None: + """ + Regression test: a static import that happens *after* realtime updates must not reset + the EVSE status back to UNKNOWN. + + Sequence: + 1. Import static data (all EVSEs start as UNKNOWN). + 2. Push realtime data via the push endpoint (statuses become AVAILABLE/CHARGING/...). + 3. Import static data *again*. + 4. The realtime statuses must still be there - not reset to UNKNOWN. + """ + # 1. Static import - all EVSEs start as UNKNOWN. + _import_static_data(requests_mock) + assert db.session.query(Evse).filter(Evse.status == EvseStatus.UNKNOWN).count() == 57 + + # 2. Push realtime data; the eager celery fixture applies it inline. + realtime_data = _load_test_data('datex2_enbw_realtime_reduced.json') + response = test_client.post( + path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}', + json=realtime_data, + ) + assert response.status_code == HTTPStatus.OK + + db.session.expire_all() + assert db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*2').first().status == EvseStatus.AVAILABLE + assert db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first().status == EvseStatus.CHARGING + + # 3. Static import again - this is what used to wipe the realtime status. + _import_static_data(requests_mock) + + # 4. The realtime statuses must survive the static re-import. + db.session.expire_all() + evse_available = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*2').first() + assert evse_available.status == EvseStatus.AVAILABLE, 'static re-import reset an AVAILABLE EVSE to UNKNOWN' + + evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first() + assert evse_charging.status == EvseStatus.CHARGING, 'static re-import reset a CHARGING EVSE to UNKNOWN' + + # EVSEs that never received a realtime update legitimately stay UNKNOWN. + evse_unchanged = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E916701*2').first() + assert evse_unchanged.status == EvseStatus.UNKNOWN + @staticmethod def test_push_realtime_v35_gzip_encoded( db: SQLAlchemy, @@ -208,6 +257,65 @@ def test_push_realtime_v35_gzip_encoded( evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first() assert evse_charging.status == EvseStatus.CHARGING + @staticmethod + def test_push_realtime_v35_delta_push_applies_directly( + db: SQLAlchemy, + test_client: OpenApiFlaskClient, + requests_mock: Mocker, + isolated_datex2_dir: Path, + stubbed_celery_delay, + ) -> None: + """ + deltaPush payloads are small incremental updates and must be applied synchronously on the + request thread, without persisting a file or queueing a celery task. + """ + _import_static_data(requests_mock) + + realtime_data = _load_test_data('datex2_enbw_realtime_reduced.json') + realtime_data['messageContainer']['exchangeInformation']['exchangeContext']['codedExchangeProtocol'] = { + 'value': 'deltaPush', + } + + response = test_client.post( + path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}', + json=realtime_data, + ) + assert response.status_code == HTTPStatus.OK + + # Stored directly: nothing persisted to disk and no celery task queued. + assert list(isolated_datex2_dir.iterdir()) == [] + stubbed_celery_delay.assert_not_called() + + # EVSE statuses are updated synchronously within the request. + db.session.expire_all() + evse_available = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*2').first() + assert evse_available.status == EvseStatus.AVAILABLE + evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first() + assert evse_charging.status == EvseStatus.CHARGING + + @staticmethod + def test_push_realtime_v35_invalid_payload_returns_400( + db: SQLAlchemy, + test_client: OpenApiFlaskClient, + requests_mock: Mocker, + isolated_datex2_dir: Path, + stubbed_celery_delay, + ) -> None: + """ + A structurally invalid payload must be rejected synchronously with HTTP 400 - before any + file is persisted or task queued - regardless of the (async) exchange protocol. + """ + _import_static_data(requests_mock) + + response = test_client.post( + path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}', + json={}, + ) + + assert response.status_code == HTTPStatus.BAD_REQUEST + assert list(isolated_datex2_dir.iterdir()) == [] + stubbed_celery_delay.assert_not_called() + @staticmethod def test_push_realtime_v35_empty_body_returns_400( db: SQLAlchemy, diff --git a/tests/integration/services/import_services/bnetza/bnetza_excel_service_test.py b/tests/integration/services/import_services/bnetza/bnetza_excel_service_test.py index 4632f25d..be8b3916 100644 --- a/tests/integration/services/import_services/bnetza/bnetza_excel_service_test.py +++ b/tests/integration/services/import_services/bnetza/bnetza_excel_service_test.py @@ -109,7 +109,7 @@ def test_bnetza_excel_import(db: SQLAlchemy, requests_mock: Mocker) -> None: 'presence': PresenceStatus.PRESENT, 'physical_reference': None, 'last_updated': ANY, - 'status_last_updated': None, + 'status_last_updated': ANY, 'max_reservation': None, 'parking_restrictions': [], 'terms_and_conditions': None, diff --git a/webapp/server_rest_api/datex2/datex2_handler.py b/webapp/server_rest_api/datex2/datex2_handler.py index 57bad3d0..d34a88a9 100644 --- a/webapp/server_rest_api/datex2/datex2_handler.py +++ b/webapp/server_rest_api/datex2/datex2_handler.py @@ -38,13 +38,16 @@ def __init__(self, *, import_services: ImportServices, **kwargs): super().__init__(**kwargs) self.import_services = import_services - def validate_realtime_request(self, source_uid: str, key: str | None) -> None: + def validate_realtime_request(self, source_uid: str, key: str | None) -> BaseDatex2V35ImportService: """ Synchronous pre-flight validation for an incoming realtime push: makes sure the source exists, is a DATEX II v3.5 source, and the supplied ``key`` query parameter matches the source's configured ``api_key``. Raises ``NotFoundException`` / ``UnauthorizedException``. + + Returns the resolved import service so the caller can validate and apply the payload. """ - self._authenticate(source_uid, key) + _, service = self._authenticate(source_uid, key) + return service def get_last_modified(self, source_uid: str, key: str | None) -> datetime | None: source, _ = self._authenticate(source_uid, key) diff --git a/webapp/server_rest_api/datex2/datex2_rest_api.py b/webapp/server_rest_api/datex2/datex2_rest_api.py index d7a58a03..9becdafe 100644 --- a/webapp/server_rest_api/datex2/datex2_rest_api.py +++ b/webapp/server_rest_api/datex2/datex2_rest_api.py @@ -16,6 +16,7 @@ along with this program. If not, see . """ +import json from email.utils import format_datetime from http import HTTPStatus from pathlib import Path @@ -41,6 +42,7 @@ from webapp.common.server_auth import skip_basic_auth from webapp.dependencies import dependencies from webapp.server_rest_api.base_blueprint import ServerApiBaseBlueprint +from webapp.services.import_services.exceptions import ImportException from webapp.services.import_services.import_celery import datex2_v3_5_realtime_import_by_file from .datex2_handler import Datex2Handler @@ -101,10 +103,11 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]: summary='Push DATEX II v3.5 realtime EVSE statuses', description=( 'Accepts a Mobilithek-style DATEX II v3.5 JSON payload (`messageContainer.payload[0]` ' - '`aegiEnergyInfrastructureStatusPublication`). The request body is persisted to ' - '`DATEX2_IMPORT_DIR` and processed asynchronously by a Celery worker so very large ' - 'payloads do not block the request thread. The `key` query parameter must match the ' - "source's configured `api_key`." + '`aegiEnergyInfrastructureStatusPublication`). The payload is validated synchronously; ' + '`deltaPush` updates are small incremental changes and are applied directly, while any ' + 'other exchange protocol (e.g. snapshots) is persisted to `DATEX2_IMPORT_DIR` and ' + 'processed asynchronously by a Celery worker so very large payloads do not block the ' + "request thread. The `key` query parameter must match the source's configured `api_key`." ), path=[Parameter('source_uid', schema=StringField())], query=[ @@ -122,8 +125,8 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]: ) def post(self, source_uid: str) -> tuple[Response, HTTPStatus]: # Validate the source + API key synchronously so unauthorized/unknown sources fail fast - # before we persist the payload to disk. - self.datex2_handler.validate_realtime_request( + # before we touch the payload. + service = self.datex2_handler.validate_realtime_request( source_uid=source_uid, key=self.request_helper.get_query_args().get('key'), ) @@ -132,6 +135,24 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]: if not data: raise InputValidationException(message='no realtime payload') + try: + parsed_data = json.loads(data) + except json.JSONDecodeError as e: + raise InputValidationException(message='invalid JSON realtime payload') from e + + # Validate the envelope up-front via the import service so malformed pushes fail fast. + try: + message_container = service.validate_realtime_data(parsed_data) + except ImportException as e: + raise InputValidationException(message=e.message) from e + + # deltaPush updates are small incremental changes - apply them directly. Everything else + # (snapshots etc.) can be large, so hand it off to a celery worker via the import file. + if service.is_delta_push(message_container): + service.store_realtime_data(message_container) + # Mobilithek expects HTTP 200 instead of HTTP 204 + return empty_json_response(), HTTPStatus.OK + base_path = Path(self.config_helper.get('DATEX2_IMPORT_DIR')) if not base_path.is_dir(): base_path.mkdir(parents=True, exist_ok=True) diff --git a/webapp/services/import_services/base_import_service.py b/webapp/services/import_services/base_import_service.py index adbe7b92..71d50d74 100644 --- a/webapp/services/import_services/base_import_service.py +++ b/webapp/services/import_services/base_import_service.py @@ -239,7 +239,7 @@ def save_location_update( self.set_image_list(location, location_update.images, images_by_url) old_charging_stations_by_uid = { - charging_station.id: charging_station for charging_station in location.charging_pool + charging_station.uid: charging_station for charging_station in location.charging_pool } new_charging_stations: list[ChargingStation] = [] for charging_station_update in location_update.charging_pool: @@ -354,10 +354,19 @@ def get_evse( evse.last_updated = evse_update.last_updated or datetime.now(tz=timezone.utc) for key, value in evse_update.to_dict().items(): - if key == 'last_updated': + if key in ['last_updated', 'status', 'status_last_updated', 'presence']: continue setattr(evse, key, value) + # presence is not-nullable; only overwrite it when the update actually carries a value, + # otherwise a static import (which never sets presence) would clear an existing presence. + if evse_update.presence is not None: + evse.presence = evse_update.presence + + if evse_update.status is not None and evse_update.status != evse.status: + evse.status_last_updated = evse_update.status_last_updated or datetime.now(tz=timezone.utc) + evse.status = evse_update.status + old_connectors_by_uid = {connector.uid: connector for connector in evse.connectors} new_connectors: list[Connector] = [] for connector_update in evse_update.connectors: diff --git a/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py b/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py index 82908c7d..771bed9f 100644 --- a/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py +++ b/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py @@ -35,7 +35,7 @@ from webapp.services.import_services.base_import_service import BaseImportService from webapp.services.import_services.exceptions import ImportException from webapp.services.import_services.models import EvseRealtimeUpdate, LocationUpdate -from webapp.shared.datex2.models import MessageContainerWrapperInput +from webapp.shared.datex2.models import MessageContainerWrapperInput, ProtocolTypeEnum from webapp.shared.datex2.v3_5_json_realtime.models.energy_infrastructure_station_status_input import ( EnergyInfrastructureStationStatusInput, ) @@ -169,7 +169,8 @@ def fetch_realtime_data(self): return try: - message_container = self.add_realtime_data(data, result) + message_container = self.validate_realtime_data(data) + self.add_realtime_data(message_container, result) except ImportException as e: logger.error( e.message, @@ -218,16 +219,14 @@ def import_realtime_data_from_file(self, import_file_path: Path) -> None: def import_realtime_data(self, data: dict) -> None: """ - Apply a parsed DATEX II v3.5 realtime payload to this source: parse the - ``messageContainer`` envelope, upsert EVSE statuses, and update the source's + Apply a parsed DATEX II v3.5 realtime payload to this source: validate the + ``messageContainer`` envelope, then upsert EVSE statuses and update the source's ``realtime_data_updated_at`` / ``realtime_status`` accordingly. """ source = self.get_source() - last_modified = datetime.now(tz=timezone.utc) - result = RealtimeResult() try: - message_container = self.add_realtime_data(data, result) + message_container = self.validate_realtime_data(data) except ImportException as e: logger.error( e.message, @@ -236,6 +235,22 @@ def import_realtime_data(self, data: dict) -> None: self.update_source(source, realtime_status=SourceStatus.FAILED) return + self.store_realtime_data(message_container) + + def store_realtime_data(self, message_container: MessageContainerWrapperInput) -> None: + """ + Apply an already-validated DATEX II v3.5 realtime message container to this source: upsert + EVSE statuses and update the source's ``realtime_data_updated_at`` / ``realtime_status``. + + Use this when the envelope has already been validated via :meth:`validate_realtime_data` + (e.g. a synchronous ``deltaPush`` from the REST API); otherwise use :meth:`import_realtime_data`. + """ + source = self.get_source() + last_modified = datetime.now(tz=timezone.utc) + result = RealtimeResult() + + self.add_realtime_data(message_container, result) + self.save_evse_updates(list(result.evse_updates_by_evse.values())) if message_container.messageContainer.payload: @@ -249,13 +264,22 @@ def import_realtime_data(self, data: dict) -> None: realtime_error_count=result.realtime_error_count, realtime_data_updated_at=last_modified, ) + exchange_context = message_container.messageContainer.exchangeInformation.exchangeContext + protocol_type = exchange_context.codedExchangeProtocol.value + protocol_label = f'{protocol_type.value} ' if protocol_type else '' logger.info( - f'Imported DATEX2 realtime data for {self.source_info.uid} via async push with ' + f'Imported DATEX2 realtime {protocol_label}data for {self.source_info.uid} via push with ' f'{result.realtime_success_count} valid EVSEs and {result.realtime_error_count} failed EVSEs.', extra={'attributes': {'type': LogMessageType.IMPORT_LOCATION}}, ) - def add_realtime_data(self, data: dict, result: RealtimeResult) -> MessageContainerWrapperInput: + def validate_realtime_data(self, data: dict) -> MessageContainerWrapperInput: + """ + Validate the raw DATEX II v3.5 realtime envelope and return the parsed message container. + + Raises :class:`ImportException` if the payload cannot be parsed or is missing the + ``messageContainer`` payload / ``aegiEnergyInfrastructureStatusPublication``. + """ try: datex_input: MessageContainerWrapperInput = self.message_container_validator.validate(data) except ValidationError as e: @@ -281,6 +305,18 @@ def add_realtime_data(self, data: dict, result: RealtimeResult) -> MessageContai message=f'missing aegiEnergyInfrastructureStatusPublication in {self.source_info.uid} realtime data', ) + return datex_input + + @staticmethod + def is_delta_push(message_container: MessageContainerWrapperInput) -> bool: + """Return whether the realtime payload uses the DATEX II ``deltaPush`` exchange protocol.""" + return ( + message_container.messageContainer.exchangeInformation.exchangeContext.codedExchangeProtocol.value + is ProtocolTypeEnum.DELTA_PUSH + ) + + def add_realtime_data(self, message_container: MessageContainerWrapperInput, result: RealtimeResult) -> None: + payload = message_container.messageContainer.payload[0] for site_status in payload.aegiEnergyInfrastructureStatusPublication.energyInfrastructureSiteStatus: if site_status.energyInfrastructureStationStatus is UnsetValue: continue @@ -297,8 +333,6 @@ def add_realtime_data(self, data: dict, result: RealtimeResult) -> MessageContai result.evse_updates_by_evse[evse_update.evse_id] = evse_update result.realtime_success_count += 1 - return datex_input - def request_data( self, subscription_id: int,