Commits (3)
......@@ -43,7 +43,7 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def StopCollector(self, request : CollectorId, grpc_context: grpc.ServicerContext # type: ignore
) -> Empty: # type: ignore
request.collector_id.uuid = ""
return Empty()
def SelectCollectors(self, request : CollectorFilter, contextgrpc_context: grpc.ServicerContext # type: ignore
......
......@@ -13,12 +13,11 @@
# limitations under the License.
from common.proto import telemetry_frontend_pb2
# from common.proto.kpi_manager_pb2 import kpi_id
# from common.proto.kpi_sample_types_pb2 import KpiSampleType
from common.proto.kpi_sample_types_pb2 import KpiSampleType
def collector_id():
def create_collector_id(coll_id_str):
_collector_id = telemetry_frontend_pb2.CollectorId()
_collector_id.collector_id.uuid = str(1)
_collector_id.collector_id.uuid = str(coll_id_str)
return _collector_id
def create_collector_request(coll_id_str):
......@@ -29,6 +28,37 @@ def create_collector_request(coll_id_str):
_create_collector_request.interval_s = float(-1)
return _create_collector_request
def create_collector_filter(filter):
def create_collector_request_a():
_create_collector_request_a = telemetry_frontend_pb2.Collector()
_create_collector_request_a.kpi_id.kpi_id.uuid = "-1"
return _create_collector_request_a
def create_collector_request_b(str_kpi_id, coll_duration_s, coll_interval_s):
_create_collector_request_b = telemetry_frontend_pb2.Collector()
_create_collector_request_b.kpi_id.kpi_id.uuid = str_kpi_id
_create_collector_request_b.duration_s = coll_duration_s
_create_collector_request_b.interval_s = coll_interval_s
return _create_collector_request_b
def create_collector_filter():
_create_collector_filter = telemetry_frontend_pb2.CollectorFilter()
\ No newline at end of file
new_collector_id = _create_collector_filter.collector_id.add()
new_collector_id.collector_id.uuid = "COLL1"
new_kpi_id = _create_collector_filter.kpi_id.add()
new_kpi_id.kpi_id.uuid = "KPI1"
new_device_id = _create_collector_filter.device_id.add()
new_device_id.device_uuid.uuid = 'DEV1'
new_service_id = _create_collector_filter.service_id.add()
new_service_id.service_uuid.uuid = 'SERV1'
new_slice_id = _create_collector_filter.slice_id.add()
new_slice_id.slice_uuid.uuid = 'SLC1'
new_endpoint_id = _create_collector_filter.endpoint_id.add()
new_endpoint_id.endpoint_uuid.uuid = 'END1'
new_connection_id = _create_collector_filter.connection_id.add()
new_connection_id.connection_uuid.uuid = 'CON1'
_create_collector_filter.kpi_sample_type.append(KpiSampleType.KPISAMPLETYPE_PACKETS_RECEIVED)
return _create_collector_filter
def create_collector_list():
_create_collector_list = telemetry_frontend_pb2.CollectorList()
return _create_collector_list
\ No newline at end of file
......@@ -16,6 +16,7 @@ import os
import pytest
import logging
from typing import Union
from common.proto.context_pb2 import Empty
from common.Constants import ServiceNameEnum
from common.proto.telemetry_frontend_pb2 import CollectorId, CollectorList
......@@ -28,7 +29,8 @@ from common.Settings import (
from telemetry_frontend.client.TelemetryFrontendClient import TelemetryFrontendClient
from telemetry_frontend.service.TelemetryFrontendService import TelemetryFrontendService
from telemetry_frontend.tests.Messages import create_collector_request
from telemetry_frontend.tests.Messages import ( create_collector_id, create_collector_request,
create_collector_filter, create_collector_request_a, create_collector_request_b)
from device.client.DeviceClient import DeviceClient
from device.service.DeviceService import DeviceService
......@@ -168,14 +170,26 @@ def test_start_collector(telemetryFrontend_client):
LOGGER.debug(str(response))
assert isinstance(response, CollectorId)
def test_start_collector_a(telemetryFrontend_client):
LOGGER.warning('test_start_collector requesting')
response = telemetryFrontend_client.StartCollector(create_collector_request_a())
LOGGER.debug(str(response))
assert isinstance(response, CollectorId)
def test_start_collector_b(telemetryFrontend_client):
LOGGER.warning('test_start_collector requesting')
response = telemetryFrontend_client.StartCollector(create_collector_request_b('1',10,2))
LOGGER.debug(str(response))
assert isinstance(response, CollectorId)
def test_stop_collector(telemetryFrontend_client):
LOGGER.warning('test_stop_collector requesting')
response = telemetryFrontend_client.StopCollector('1')
response = telemetryFrontend_client.StopCollector(create_collector_id("1"))
LOGGER.debug(str(response))
assert isinstance(response, Empty)
def test_select_collectors(telemetryFrontend_client):
LOGGER.warning('test_select_collector requesting')
response = telemetryFrontend_client.SelectCollectors()
response = telemetryFrontend_client.SelectCollectors(create_collector_filter())
LOGGER.debug(str(response))
assert isinstance(response, CollectorList)
\ No newline at end of file