usov_i-org/bec
1
0
Fork 0
mirror of https://github.com/ivan-usov-org/bec.git synced 2025-04-21 18:20:01 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: bec_service test

Browse Source
This commit is contained in:
usov_i 2023-11-09 15:02:59 +01:00
parent 06f2d781ae
commit 97d3d1f18f
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
bec_lib/tests/test_bec_service.py
View File

@ -4,10 +4,10 @@ from unittest import mock
import pytest import pytest
import bec_lib import bec_lib
from bec_lib import BECMessage from bec_lib import messages
from bec_lib.bec_service import BECService from bec_lib.bec_service import BECService
from bec_lib.BECMessage import BECStatus
from bec_lib.endpoints import MessageEndpoints from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import BECStatus
from bec_lib.service_config import ServiceConfig from bec_lib.service_config import ServiceConfig
# pylint: disable=no-member # pylint: disable=no-member
@ -121,8 +121,8 @@ def test_bec_service_update_existing_services():
f'{MessageEndpoints.service_status("service2")}:val'.encode(), f'{MessageEndpoints.service_status("service2")}:val'.encode(),
] ]
service_msgs = [ service_msgs = [
BECMessage.StatusMessage(name="service1", status=BECStatus.RUNNING, info={}, metadata={}), messages.StatusMessage(name="service1", status=BECStatus.RUNNING, info={}, metadata={}),
BECMessage.StatusMessage(name="service2", status=BECStatus.IDLE, info={}, metadata={}), messages.StatusMessage(name="service2", status=BECStatus.IDLE, info={}, metadata={}),
] ]
connector_cls = mock.MagicMock() connector_cls = mock.MagicMock()
connector_cls().producer().keys.return_value = service_keys connector_cls().producer().keys.return_value = service_keys
@ -144,7 +144,7 @@ def test_bec_service_update_existing_services_ignores_wrong_msgs():
f'{MessageEndpoints.service_status("service2")}:val'.encode(), f'{MessageEndpoints.service_status("service2")}:val'.encode(),
] ]
service_msgs = [ service_msgs = [
BECMessage.StatusMessage(name="service1", status=BECStatus.RUNNING, info={}, metadata={}), messages.StatusMessage(name="service1", status=BECStatus.RUNNING, info={}, metadata={}),
None, None,
] ]
connector_cls = mock.MagicMock() connector_cls = mock.MagicMock()