Commit 027a3305 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Common - Tests - Mock Context Servicer: 

- Implement optical spectrum reservation validation and overlap checks
parent 6947a344

src/common/tests/MockServicerImpl_Context.py

+56 −0
Original line number Diff line number Diff line
@@ -770,6 +770,61 @@ class MockServicerImpl_Context(ContextServiceServicer): 


    # ----- Optical Spectrum Reservation ------------------------------------------------------------------------------



    def _optical_spectrum_reservation_is_active(self, reservation : OpticalSpectrumReservation) -> bool:

        return reservation.status in {

            OpticalSpectrumReservationStatusEnum.OPTICALSPECTRUMRESERVATIONSTATUS_RESERVED,

            OpticalSpectrumReservationStatusEnum.OPTICALSPECTRUMRESERVATIONSTATUS_CONSUMED,

        }



    def _optical_spectrum_reservation_ranges_overlap(

        self, a_start : int, a_end : int, b_start : int, b_end : int

    ) -> bool:

        return a_start <= b_end and b_start <= a_end



    def _validate_optical_spectrum_reservation(

        self, request : OpticalSpectrumReservation, container_name : str, context : grpc.ServicerContext

    ) -> None:

        reservation_uuid = request.reservation_id.reservation_uuid.uuid

        topology_uuid = request.topology_id.topology_uuid.uuid

        band = str(request.band)

        n_start = int(request.n_start)

        n_end = int(request.n_end)

        requested_link_uuids = {link_id.link_uuid.uuid for link_id in request.optical_link_ids}



        for link_uuid in requested_link_uuids:

            optical_link = self.obj_db.get_entry('optical_link', link_uuid, context)

            slots = getattr(optical_link.optical_details, band)

            for slot_index in range(n_start, n_end + 1):

                if int(slots.get(str(slot_index), 1)) != 1:

                    context.abort(

                        grpc.StatusCode.ALREADY_EXISTS,

                        'slot {:s}:{:d} is not available on optical link({:s})'.format(

                            band, slot_index, link_uuid

                        )

                    )



        for reservation in self.obj_db.get_entries(container_name):

            if reservation.reservation_id.reservation_uuid.uuid == reservation_uuid:

                continue

            if reservation.topology_id.topology_uuid.uuid != topology_uuid:

                continue

            if reservation.band != band:

                continue

            if not self._optical_spectrum_reservation_is_active(reservation):

                continue

            existing_link_uuids = {link_id.link_uuid.uuid for link_id in reservation.optical_link_ids}

            if len(requested_link_uuids.intersection(existing_link_uuids)) == 0:

                continue

            if self._optical_spectrum_reservation_ranges_overlap(

                n_start, n_end, int(reservation.n_start), int(reservation.n_end)

            ):

                context.abort(

                    grpc.StatusCode.ALREADY_EXISTS,

                    'OpticalSpectrumReservation({:s}) already exists; overlapping spectrum reservation'.format(

                        reservation.reservation_id.reservation_uuid.uuid

                    )

                )



    def ListOpticalSpectrumReservations(

        self, request : ContextId, context : grpc.ServicerContext

    ) -> OpticalSpectrumReservationList:
@@ -799,6 +854,7 @@ class MockServicerImpl_Context(ContextServiceServicer): 
        container_name = 'optical_spectrum_reservation[{:s}]'.format(

            str(request.reservation_id.context_id.context_uuid.uuid)

        )

        self._validate_optical_spectrum_reservation(request, container_name, context)

        reply, _ = self._set(

            request, container_name, request.reservation_id.reservation_uuid.uuid, 'reservation_id', TOPIC_CONTEXT

        )