Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_changing_connection_name_doesnt_switch_logger_if_external_logger_used(memory_connection_class):
from moler.connection import Connection
moler_conn = Connection()
connection = memory_connection_class(moler_connection=moler_conn,
name="ABC",
logger_name="conn.ABC")
assert connection.logger.name == "conn.ABC"
connection.name = "DEF"
assert connection.logger.name == "conn.ABC"
def test_can_use_default_logger_based_on_connection_name():
from moler.connection import Connection
import logging
moler_conn = Connection(name="ABC")
assert isinstance(moler_conn.logger, logging.Logger)
assert moler_conn.logger.name == "moler.connection.ABC"
moler_conn = Connection()
assert isinstance(moler_conn.logger, logging.Logger)
assert moler_conn.logger.name == "moler.connection.{}".format(moler_conn.name)
def test_can_generate_connection_name_if_none_given():
from moler.connection import Connection
from moler.helpers import instance_id
moler_conn = Connection()
assert moler_conn.name == instance_id(moler_conn)
def test_overwrites_moler_connection_name_with_own_one(memory_connection_class):
from moler.connection import Connection
moler_conn = Connection()
# during construction
connection = memory_connection_class(moler_connection=moler_conn, name="web_srv")
assert moler_conn.name == "web_srv"
# during direct attribute set
connection.name = "http_srv"
assert moler_conn.name == "http_srv"
def test_can_use_default_logger_based_on_connection_name(memory_connection_class):
from moler.connection import Connection
import logging
moler_conn = Connection()
connection = memory_connection_class(moler_connection=moler_conn, name="ABC")
assert isinstance(connection.logger, logging.Logger)
assert connection.logger.name == "moler.connection.ABC.io"
moler_conn = Connection()
connection = memory_connection_class(moler_connection=moler_conn)
assert isinstance(connection.logger, logging.Logger)
assert connection.logger.name == "moler.connection.{}.io".format(connection.name)
def test_can_use_default_logger_based_on_connection_name():
from moler.connection import Connection
import logging
moler_conn = Connection(name="ABC")
assert isinstance(moler_conn.logger, logging.Logger)
assert moler_conn.logger.name == "moler.connection.ABC"
moler_conn = Connection()
assert isinstance(moler_conn.logger, logging.Logger)
assert moler_conn.logger.name == "moler.connection.{}".format(moler_conn.name)
def test_changing_connection_name_switches_logger_if_default_logger_used():
from moler.connection import Connection
# default logger generated internally by connection
moler_conn = Connection(name="ABC")
assert moler_conn.logger.name == "moler.connection.ABC"
moler_conn.name = "DEF"
assert moler_conn.logger.name == "moler.connection.DEF"
# default logger via default naming
moler_conn = Connection(name="ABC", logger_name="moler.connection.ABC")
moler_conn.name = "DEF"
assert moler_conn.logger.name == "moler.connection.DEF"
def test_can_switch_off_logging(memory_connection_class):
from moler.connection import Connection
moler_conn = Connection()
connection = memory_connection_class(moler_connection=moler_conn,
logger_name=None)
assert connection.logger is None
def test_can_use_provided_logger(memory_connection_class):
from moler.connection import Connection
import logging
moler_conn = Connection()
connection = memory_connection_class(moler_connection=moler_conn,
logger_name="conn.web_srv")
assert isinstance(connection.logger, logging.Logger)
assert connection.logger.name == "conn.web_srv"
def test_can_encode_data_for_external_io_needs__encoder_via_composition():
from moler.connection import Connection
class ExternalIO(object):
def __init__(self):
self.sent_data = []
def write(self, bytes):
self.sent_data.append(bytes)
used_io = ExternalIO()
# compose Moler connection with external encoder
hexlified_connection = Connection(how2send=used_io.write,
encoder=lambda data: binascii.hexlify(bytearray(source=data, encoding="utf-8")))
hexlified_connection.send(data="hi") # 'h' is ASCII 0x68 'i' is ASCII 0x69
assert b'6869' in used_io.sent_data