h3cx
f3b1a537f0
Implemented handshake in json server and handoff to endpoint Endpoint handles connector handshake accept/decline then launches both its IO threads alongside its heartbeat thread which contains simple timeout logic, timeout calls connection handles to close, subsequently killing the endpoint
35 lines
1.0 KiB
Python
35 lines
1.0 KiB
Python
import logging
|
|
|
|
import pytest
|
|
from dynalab_core import Core
|
|
from dynalab_core.config import CoreConfig
|
|
from dynalab_core.errors import CoreStateMismatchError
|
|
from test.common import find_available_port
|
|
|
|
|
|
def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None:
|
|
port = find_available_port(8765)
|
|
core = Core(CoreConfig(port=port))
|
|
|
|
with caplog.at_level(logging.DEBUG, logger="dynalab_core"):
|
|
core.start()
|
|
|
|
try:
|
|
with pytest.raises(CoreStateMismatchError):
|
|
core.start()
|
|
finally:
|
|
core.stop()
|
|
|
|
events = [getattr(record, "event", None) for record in caplog.records]
|
|
assert events.count("core.starting") == 1
|
|
assert "core.start_rejected" in events
|
|
assert "core.stopped" in events
|
|
|
|
rejection = next(
|
|
record
|
|
for record in caplog.records
|
|
if getattr(record, "event", None) == "core.start_rejected"
|
|
)
|
|
assert rejection.levelno == logging.WARNING
|
|
assert rejection.core_state == "started"
|