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_backups_at_initialization(tmpdir):
for i in reversed(range(1, 10)):
tmpdir.join('test.log.%d' % i).write(str(i))
logger.start(tmpdir.join('test.log'), backups=1)
assert len(tmpdir.listdir()) == 2
assert tmpdir.join('test.log').read() == ''
assert tmpdir.join('test.log.1').read() == '1'
def test_backups_zfill(tmpdir):
logger.start(tmpdir.join('test.log'), rotation=0, backups=None, format='{message}')
logger.debug('a')
logger.debug('0')
assert tmpdir.join('test.log.1').read() == 'a\n'
for _ in range(10):
logger.debug('0')
logger.debug('b')
logger.debug('0')
assert tmpdir.join('test.log.01').read() == 'b\n'
for _ in range(100):
logger.debug('0')
logger.debug('c')
def test_compression_without_rotation(tmpdir):
import gzip
n = logger.start(tmpdir.join('test.log'), compression=True, format='{message}')
logger.debug("Test")
logger.stop(n)
assert len(tmpdir.listdir()) == 1
archive = tmpdir.join('test.log.gz')
with gzip.open(archive.realpath()) as gz:
assert gz.read().decode('utf8').replace('\r', '') == 'Test\n'
def test_notifier_as_sink():
noti = notifier.email(to="dest@gmail.com")
mock = MagicMock()
noti.provider.notify = mock
logger.start(noti.send)
logger.info("Test")
assert mock.call_count == 1
def test_invalid_compression(compression):
with pytest.raises(ValueError):
logger.start('test.log', compression=compression)
def test_compression_rotation(tmpdir):
import gzip
n = logger.start(tmpdir.join('test.log'), rotation=0, compression=True, format='{message}', backups=5)
for i in range(10):
logger.debug(str(i))
logger.stop(n)
assert tmpdir.join('test.log').read() == '9\n'
for i in range(5):
archive = tmpdir.join('test.log.%d.gz' % (i + 1))
with gzip.open(archive.realpath()) as gz:
assert gz.read().decode('utf8').replace('\r', '') == '%d\n' % (9 - i - 1)
def test_renaming(tmpdir, name, should_rename):
file = tmpdir.join(name)
logger.start(file.realpath(), rotation=0)
assert len(tmpdir.listdir()) == 1
basename = tmpdir.listdir()[0].basename
logger.debug("a")
logger.debug("b")
logger.debug("c")
files = [f.basename for f in tmpdir.listdir()]
renamed = [basename + '.' + str(i) in files for i in [1, 2, 3]]
if should_rename:
assert all(renamed)
else:
assert not any(renamed)
def __init__(self):
#logpath = './log'
#log = setup_logger(logpath, 'archon_logger', 'archon')
logger.start("log/broker.log", rotation="500 MB")
#logger.start("log/broker.log", colorize=True, format="{time} {message}")
logger.debug("init arch")
filename = "apikeys.toml"
self.afacade = facade.Facade()
#in memory data
self.balances = None
self.openorders = list()
self.submitted_orders = list()
self.active_exchanges = None
self.selected_exchange = None
try:
all_conf = parse_toml("conf.toml")
#active_exchanges = all_conf["BROKER"]["active_exchanges"]
def init_logger():
logger.remove() # Remove the pre-configured handler
logger.start(sys.stderr, format="{level} {time:MM-DD HH:mm:ss} {file}:{line} - {message}")