d5c04991bc
* CDK - using native logger with custom formatter * CDK - using native logger(fix exception and add trace) * CDK - using native logger in AirbyteEntrypoint * CDK - CHANGELOG.md * CDK - remove unnecessary imports * CDK - fixing according to PR review * CDK native logger #1279 - annotations * CDK native logger #1279 - fixing according to PR review * CDK standard logger #1279 - tests * CDK standard logger #1279 - reformat * Airbyte CDK #1279 - improve docstrings * Airbyte CDK #1279 - improve log levels * Airbyte CDK #1279 - fix init get source name * Airbyte CDK #1279 - update test licence * Airbyte CDK #1279 - bump version
76 lines
1.9 KiB
Python
76 lines
1.9 KiB
Python
#
|
|
# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
|
|
import json
|
|
from typing import Dict
|
|
|
|
import pytest
|
|
from airbyte_cdk.logger import AirbyteLogFormatter, init_logger
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def logger():
|
|
logger = init_logger("Test logger")
|
|
return logger
|
|
|
|
|
|
def test_formatter(logger, caplog):
|
|
formatter = AirbyteLogFormatter()
|
|
logger.info("Test formatter")
|
|
record = caplog.records[0]
|
|
formatted_record = formatter.format(record)
|
|
formatted_record_data = json.loads(formatted_record)
|
|
assert formatted_record_data.get("type") == "LOG"
|
|
log = formatted_record_data.get("log")
|
|
assert isinstance(log, Dict)
|
|
level = log.get("level")
|
|
message = log.get("message")
|
|
assert level == "INFO"
|
|
assert message == "Test formatter"
|
|
|
|
|
|
def test_trace(logger, caplog):
|
|
logger.trace("Test trace 1")
|
|
record = caplog.records[0]
|
|
assert record.levelname == "TRACE"
|
|
assert record.message == "Test trace 1"
|
|
|
|
|
|
def test_debug(logger, caplog):
|
|
logger.debug("Test debug 1")
|
|
record = caplog.records[0]
|
|
assert record.levelname == "DEBUG"
|
|
assert record.message == "Test debug 1"
|
|
|
|
|
|
def test_info(logger, caplog):
|
|
logger.info("Test info 1")
|
|
logger.info("Test info 2")
|
|
assert len(caplog.records) == 2
|
|
first_record = caplog.records[0]
|
|
assert first_record.levelname == "INFO"
|
|
assert first_record.message == "Test info 1"
|
|
|
|
|
|
def test_warn(logger, caplog):
|
|
logger.warn("Test warn 1")
|
|
record = caplog.records[0]
|
|
assert record.levelname == "WARNING"
|
|
assert record.message == "Test warn 1"
|
|
|
|
|
|
def test_error(logger, caplog):
|
|
logger.error("Test error 1")
|
|
record = caplog.records[0]
|
|
assert record.levelname == "ERROR"
|
|
assert record.message == "Test error 1"
|
|
|
|
|
|
def test_fatal(logger, caplog):
|
|
logger.fatal("Test fatal 1")
|
|
record = caplog.records[0]
|
|
assert record.levelname == "CRITICAL"
|
|
assert record.message == "Test fatal 1"
|