Service logs

Service logs formatter

Testsuite provides plugin to pretty-print service logs (colorize, simplify, etc…). It only reads logfiles when testcase fails with no extra overhead.

It also support live-logs when pytest is started with -s option. In this case logs would be printed from special thread.

Fixtures

servicelogs_register_logfile

testsuite.plugins.servicelogs.servicelogs_register_logfile(path: pathlib.Path, *, title: str, formatter_factory)

Registers logfile into testsuite.

Parameters:

• path – path to logfile

• title – logfile name

• formatter_factory – Log formatter factory

async def test_formatter(servicelogs_register_logfile):
    logfile = pathlib.Path(...)

    def formatter_factory():
        # formatter may have its own context
        def formatter(line: bytes) -> str | None:
            return line.encode('utf-8')
        return formatter

    servicelogs_register_logfile(
        logfile, title='My service logs',
        formatter_factory=formatter_factory
    )

servicelogs_register_flusher

testsuite.plugins.servicelogs.servicelogs_register_flusher(func)

Registers async function that would be called to flush service buffer in case of failure.

async def test_flusher(servicelogs_register_flusher):
    @servicelogs_register_flusher
    async def flusher():
        ...

service_logs_update_position

testsuite.plugins.servicelogs.service_logs_update_position()

Seeks all registered logfiles to the end. Useful to skip initial service logs.

Options

–service-livelogs-disable

Disables service live logs.

Service logs capture

Logcapture allows to intercepts service logs on demand with context manager.

It starts tcp server and read logs sent by server.

class testsuite.logcapture.LogLevel(*values)

Represents log level as IntEnum, which supports comparison.

Available levels are: TRACE, DEBUG, INFO, WARNING, ERROR, CRITICAL, NONE

classmethod from_string(level: str) → LogLevel

Parse log level from the string.

class testsuite.logcapture.CaptureServer

__init__(*, log_level: LogLevel, parse_line: Callable[[bytes], dict])

Capture server.

async with capture(*, log_level: LogLevel | None = None, timeout: float = 10.0) → AsyncIterator[Capture]

Starts logs capture. Returns Capture object.

property default_log_level: LogLevel

Returns default log level specified on object creation.

getsocknames() → list[tuple]

Return list of server socket names.

async with start(*args, **kwargs) → AsyncIterator[CaptureServer]

Starts capture logs asyncio server.

Arguments are directly passed to asyncio.start_server. Server could be started only once. Capture server is returned. Server is closed when contextmanager is finished.

await wait_for_client(timeout: float = 10.0)

Waits for logserver client to connect.

class testsuite.logcapture.Capture(logs: CapturedLogs)

select(**query) → list[dict]

Select logs matching query.

Could only be used after capture contextmanager block.

async with logcapture_server.capture() as capture:
    ...
records = capture.select(text='Message to capture')

subscribe(**query)

Subscribe to records matching query. Returns decorator function. subscribe() may only be used within capture() block. Callqueue is returned.

async with logcapture_server.capture() as capture:
    @capture.subscribe(text='Message to capture')
    def log_event(link, **other):
        ...
    ...
    assert log_event.wait_call()