43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
import pytest
|
|
|
|
from services import database_service_base
|
|
from services.database_service_base import DatabaseServiceBase
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_init_success_log_is_once_per_service(monkeypatch):
|
|
logs = []
|
|
|
|
async def fake_get_db_manager():
|
|
return object()
|
|
|
|
def fake_info(device_id, session_id, message):
|
|
logs.append((device_id, session_id, message))
|
|
|
|
DatabaseServiceBase._logged_initialized_services.clear()
|
|
monkeypatch.setattr(database_service_base, "get_db_manager", fake_get_db_manager)
|
|
monkeypatch.setattr(database_service_base.session_logger, "info", fake_info)
|
|
|
|
first = DatabaseServiceBase(service_name="binding_service")
|
|
second = DatabaseServiceBase(service_name="binding_service")
|
|
|
|
await first._init_database()
|
|
await second._init_database()
|
|
|
|
assert logs == [
|
|
(
|
|
"system",
|
|
"binding_service",
|
|
"binding_service数据库连接初始化成功",
|
|
)
|
|
]
|
|
|
|
|
|
def test_binding_service_singleton_is_used_by_router_and_mqtt_handler():
|
|
from banban.routers import bindings
|
|
from banban.service.binding import binding_service
|
|
from handlers import mqtt_handler
|
|
|
|
assert bindings.binding_service is binding_service
|
|
assert mqtt_handler.binding_service is binding_service
|