service id check in offered services has been added

flake8

iso15118/secc/states/iso15118_20_states.py

@@ -600,15 +600,26 @@ async def process_message(
             )
         )
 
+        is_found = False
         for offered_service in self.comm_session.matched_services_v20:
             if offered_service.service.id == service_detail_req.service_id:
                 offered_service.parameter_sets = service_parameter_list.parameter_sets
+                is_found = True
+                break
+        if is_found:
+            response_code = ResponseCode.OK
+        else:
+            # [V2G20-464] The message "ServiceDetailRes" shall contain the
+            # ResponseCode "FAILED_ServiceIDInvalid" if the ServiceID contained
+            # in the ServiceDetailReq message was not part of the offered
+            # EnergyTransferServiceList or VASList during ServiceDiscovery.
+            response_code = ResponseCode.FAILED_SERVICE_ID_INVALID
 
         service_detail_res = ServiceDetailRes(
             header=MessageHeader(
                 session_id=self.comm_session.session_id, timestamp=time.time()
             ),
-            response_code=ResponseCode.OK,
+            response_code=response_code,
             service_id=service_detail_req.service_id,
             service_parameter_list=service_parameter_list,
         )

iso15118/secc/states/iso15118_2_states.py

@@ -409,6 +409,29 @@ async def process_message(
 
         service_detail_req: ServiceDetailReq = msg.body.service_detail_req
 
+        is_found = False
+        for service_details in self.comm_session.offered_services:
+            if service_detail_req.service_id == service_details.service_id:
+                is_found = True
+                break
+
+        if not is_found:
+            # [V2G2-425] The SECC shall respond with a negative ResponseCode
+            # 'FAILED_ServiceIDInvalid'
+            # if the EVCC provided a not previously retrieved ServiceID in the
+            # ServiceDetailsReq.
+            error_service_detail_res = ServiceDetailRes(
+                response_code=ResponseCode.FAILED_SERVICE_ID_INVALID,
+                service_id=service_detail_req.service_id,
+            )
+            self.create_next_message(
+                None,
+                error_service_detail_res,
+                Timeouts.V2G_SECC_SEQUENCE_TIMEOUT,
+                Namespace.ISO_V2_MSG_DEF,
+            )
+            return
+
         parameter_set: List[ParameterSet] = []
 
         # Certificate installation service

tests/iso15118_20/secc/test_iso15118_20_states.py

@@ -0,0 +1,74 @@
+from unittest.mock import Mock, patch
+
+import pytest
+
+from iso15118.secc.comm_session_handler import SECCCommunicationSession
+from iso15118.secc.controller.simulator import SimEVSEController
+from iso15118.secc.states.iso15118_20_states import ServiceDetail
+from iso15118.shared.messages.enums import EnergyTransferModeEnum, Protocol, ServiceV20
+from iso15118.shared.messages.iso15118_20.common_messages import (
+    MatchedService,
+    Service,
+    ServiceList,
+)
+from iso15118.shared.messages.iso15118_20.common_types import ResponseCode
+from iso15118.shared.notifications import StopNotification
+from tests.dinspec.secc.test_dinspec_secc_states import MockWriter
+from tests.iso15118_20.secc.test_messages import get_v2g_message_service_detail_req
+
+
+@patch("iso15118.shared.states.EXI.to_exi", new=Mock(return_value=b"01"))
+@pytest.mark.asyncio
+class TestEvScenarios:
+    @pytest.fixture(autouse=True)
+    def _comm_session(self):
+        self.comm_session = Mock(spec=SECCCommunicationSession)
+        self.comm_session.session_id = "F9F9EE8505F55838"
+        self.comm_session.selected_energy_mode = EnergyTransferModeEnum.DC_EXTENDED
+        self.comm_session.selected_charging_type_is_ac = False
+        self.comm_session.stop_reason = StopNotification(False, "pytest")
+        self.comm_session.protocol = Protocol.ISO_15118_20_AC
+        self.comm_session.writer = MockWriter()
+
+    @pytest.mark.parametrize(
+        "service_id_input, response_code",
+        [
+            (1, ResponseCode.OK),
+            (5, ResponseCode.OK),
+            (2, ResponseCode.FAILED_SERVICE_ID_INVALID),
+        ],
+    )
+    async def test_service_detail_service_id_is_in_offered_list(
+        self, service_id_input, response_code
+    ):
+        # [V2G20-464] The message "ServiceDetailRes" shall contain the
+        # ResponseCode "FAILED_ServiceIDInvalid" if the ServiceID contained
+        # in the ServiceDetailReq message was not part of the offered
+        # EnergyTransferServiceList or VASList during ServiceDiscovery.
+
+        self.comm_session.matched_services_v20 = []
+        self.comm_session.evse_controller = await SimEVSEController.create()
+        service_ids = [1, 5]
+        offered_energy_services: ServiceList = ServiceList(services=[])
+        for service_id in service_ids:
+            offered_energy_services.services.append(
+                Service(service_id=service_id, free_service=False)
+            )
+
+        for energy_service in offered_energy_services.services:
+            self.comm_session.matched_services_v20.append(
+                MatchedService(
+                    service=ServiceV20.get_by_id(energy_service.service_id),
+                    is_energy_service=True,
+                    is_free=energy_service.free_service,
+                    # Parameter sets are available with ServiceDetailRes
+                    parameter_sets=[],
+                )
+            )
+
+        service_details = ServiceDetail(self.comm_session)
+        await service_details.process_message(
+            message=get_v2g_message_service_detail_req(service_id_input)
+        )
+        assert service_details.message.response_code is response_code
+        assert isinstance(self.comm_session.current_state, ServiceDetail)

tests/iso15118_20/secc/test_messages.py

@@ -0,0 +1,15 @@
+import time
+
+from iso15118.shared.messages.iso15118_20.common_messages import ServiceDetailReq
+from iso15118.shared.messages.iso15118_20.common_types import MessageHeader
+from tests.tools import MOCK_SESSION_ID
+
+
+def get_v2g_message_service_detail_req(service_list: int) -> ServiceDetailReq:
+    return ServiceDetailReq(
+        header=MessageHeader(
+            session_id=MOCK_SESSION_ID,
+            timestamp=time.time(),
+        ),
+        service_id=service_list,
+    )

tests/secc/states/test_iso15118_2_states.py

@@ -11,6 +11,7 @@
     CurrentDemand,
     PaymentDetails,
     PowerDelivery,
+    ServiceDetail,
     ServiceDiscovery,
     SessionStop,
     Terminate,
@@ -29,7 +30,9 @@
 from iso15118.shared.messages.iso15118_2.datatypes import (
     ACEVSEStatus,
     CertificateChain,
-    ChargingSession,
+    ServiceCategory,
+    ServiceDetails,
+    ServiceName,
 )
 from iso15118.shared.security import get_random_bytes
 from iso15118.shared.states import Pause
@@ -51,6 +54,7 @@
     get_power_delivery_req_charging_profile_out_of_boundary,
     get_v2g_message_power_delivery_req,
     get_v2g_message_power_delivery_req_charging_profile_in_boundary_valid,
+    get_v2g_message_service_detail_req,
 )
 
 
@@ -508,16 +512,35 @@ async def get_ac_evse_status_patch():
         assert charging_status_res.ac_evse_status == await get_ac_evse_status_patch()
 
     @pytest.mark.parametrize(
-        "charging_session, expected_next_state",
+        "service_id, response_code",
         [
-            (ChargingSession.PAUSE, Pause),
-            (ChargingSession.TERMINATE, Terminate),
+            (2, ResponseCode.OK),
+            (3, ResponseCode.FAILED_SERVICE_ID_INVALID),
         ],
     )
-    async def test_session_stop_req(self, charging_session, expected_next_state):
-        # V2G2-718
-        session_stop = SessionStop(self.comm_session)
-        await session_stop.process_message(
-            message=get_dummy_v2g_session_stop_req(charging_session)
+    async def test_service_detail_service_id_is_in_offered_list(
+        self, service_id, response_code
+    ):
+        self.comm_session.selected_auth_option = AuthEnum.PNC_V2
+        self.comm_session.config.free_charging_service = False
+        self.comm_session.writer = Mock()
+        self.comm_session.writer.get_extra_info = Mock()
+
+        cert_install_service = ServiceDetails(
+            service_id=2,
+            service_name=ServiceName.CERTIFICATE,
+            service_category=ServiceCategory.CERTIFICATE,
+            free_service=True,
+        )
+
+        self.comm_session.offered_services = []
+        self.comm_session.offered_services.append(cert_install_service)
+        service_details = ServiceDetail(self.comm_session)
+        await service_details.process_message(
+            message=get_v2g_message_service_detail_req(service_id=service_id)
+        )
+        assert isinstance(self.comm_session.current_state, ServiceDetail)
+        assert (
+            service_details.message.body.service_detail_res.response_code
+            is response_code
         )
-        assert session_stop.next_state == expected_next_state

tests/secc/states/test_messages.py

@@ -14,6 +14,7 @@
     ChargingStatusReq,
     PaymentDetailsReq,
     PowerDeliveryReq,
+    ServiceDetailReq,
     ServiceDiscoveryReq,
     SessionStopReq,
     WeldingDetectionReq,
@@ -510,10 +511,9 @@ def get_dummy_charging_status_req() -> V2GMessage:
     )
 
 
-def get_dummy_v2g_session_stop_req(charging_session: ChargingSession) -> V2GMessage:
-    session_stop_req = SessionStopReq(charging_session=charging_session)
-
+def get_v2g_message_service_detail_req(service_id) -> V2GMessage:
+    service_detail_req = ServiceDetailReq(service_id=service_id)
     return V2GMessage(
         header=MessageHeader(session_id=MOCK_SESSION_ID),
-        body=Body(session_stop_req=session_stop_req),
+        body=Body(service_detail_req=service_detail_req),
     )