# -*- coding: utf-8 -*- import pendulum import pytest from tsstats import events from tsstats.log import TimedLog, _bundle_logs, _parse_line, parse_logs from tsstats.template import render_servers testlog_path = 'tsstats/tests/res/test.log' static_timestamp = pendulum.datetime(2015, 5, 18, 15, 52, 52, 685612) @pytest.fixture def clients(): return list(parse_logs(testlog_path, online_dc=False))[0].clients def test_log_client_count(clients): assert len(clients) == 3 def test_log_onlinetime(clients): assert clients['1'].onlinetime == pendulum.duration( seconds=402, microseconds=149208) assert clients['2'].onlinetime == pendulum.duration( seconds=19, microseconds=759644) def test_log_kicks(clients): assert clients['UIDClient1'].kicks == 1 def test_log_pkicks(clients): assert clients['2'].pkicks == 1 def test_log_bans(clients): assert clients['UIDClient1'].bans == 1 def test_log_pbans(clients): assert clients['2'].pbans == 1 @pytest.mark.parametrize("logs,bundled", [ ( ['l1.log', 'l2.log'], {'': [TimedLog('l1.log', None), TimedLog('l2.log', None)]} ), ( [ 'ts3server_2016-06-06__14_22_09.527229_1.log', 'ts3server_2017-07-07__15_23_10.638340_1.log' ], { '1': [ TimedLog( 'ts3server_2016-06-06__14_22_09.527229_1.log', pendulum.datetime( year=2016, month=6, day=6, hour=14, minute=22, second=9, microsecond=527229 ) ), TimedLog( 'ts3server_2017-07-07__15_23_10.638340_1.log', pendulum.datetime( year=2017, month=7, day=7, hour=15, minute=23, second=10, microsecond=638340 ) ) ] } ) ]) def test_log_bundle(logs, bundled): assert _bundle_logs(logs) == bundled def test_log_client_online(): current_time = pendulum.now() pendulum.set_test_now(current_time) clients = list(parse_logs(testlog_path))[0].clients old_onlinetime = int(clients['1'].onlinetime.total_seconds()) pendulum.set_test_now(current_time.add(seconds=2)) # add 2s to .now() clients = list(parse_logs(testlog_path))[0].clients assert int(clients['1'].onlinetime.total_seconds()) == old_onlinetime + 2 def test_parse_groups(): server = list(parse_logs('tsstats/tests/res/test.log.groups')) assert len(server) == 0 def test_parse_utf8(output): servers = parse_logs(testlog_path + '.utf8') render_servers(servers, output) def test_parse_invalid_line(): assert _parse_line('INVALID') == [] @pytest.mark.parametrize('line,expected_events', [ ( "client connected 'Client1'(id:1) from 1.2.3.4:1234", [ events.connect(static_timestamp, '1') ] ), ( "client disconnected 'Client1'(id:1) reason 'reasonmsg=ByeBye!'", [ events.disconnect(static_timestamp, '1') ] ), ( "client disconnected 'Client1'(id:1) reason 'invokerid=1" " invokername=Client2 invokeruid=UIDClient2 reasonmsg'", [ events.disconnect(static_timestamp, '1'), events.nick(None, 'UIDClient2', 'Client2'), events.kick(None, 'UIDClient2', '1') ] ), ( "client disconnected 'Client1'(id:1) reason 'invokerid=2 " "invokername=Client2 invokeruid=UIDClient2 reasonmsg bantime=0'", [ events.disconnect(static_timestamp, '1'), events.nick(None, 'UIDClient2', 'Client2'), events.ban(None, 'UIDClient2', '1') ] ) ]) def test_parse_line(line, expected_events): line = '2015-05-18 15:52:52.685612|INFO |VirtualServerBase| 3| ' + line expected_events.insert(0, events.nick(None, '1', 'Client1')) expected_events = [ event._replace(timestamp=static_timestamp) for event in expected_events ] assert _parse_line(line) == expected_events