Skip to content
GitLab
Projects
Groups
Topics
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Sign in
Toggle navigation
Menu
TFS
controller
Compare revisions
231ab1e65b6f4b69c8c65e01b5e3be050273ec84...f6c040ab31901e0049c80aa0d91625dafd12b78c
Commits (3)
dummy logic is added in "SopCollector" method
· 269da2e9
Waleed Akbar
authored
Apr 29, 2024
269da2e9
New test case are added
· 1c177bb7
Waleed Akbar
authored
Apr 29, 2024
1c177bb7
New test cases are added
· f6c040ab
Waleed Akbar
authored
Apr 29, 2024
f6c040ab
Hide whitespace changes
Inline
Side-by-side
src/telemetry_frontend/service/TelemetryFrontendServiceServicerImpl.py
View file @
f6c040ab
...
...
@@ -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
...
...
src/telemetry_frontend/tests/Messages.py
View file @
f6c040ab
...
...
@@ -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
src/telemetry_frontend/tests/test_unitary.py
View file @
f6c040ab
...
...
@@ -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