Loading tests/commands/test_stats.py +34 −31 Original line number Diff line number Diff line Loading @@ -8,10 +8,9 @@ import sbws.commands.stats from datetime import date import os import time from tests.common import MockPastlyLogger def init_directory(dname): def init_directory(dname, log): p = create_parser() args = p.parse_args('-d {} -vvvv init'.format(dname).split()) conf = get_config(args, log_fn=log.debug) Loading Loading @@ -63,10 +62,7 @@ def add_two_fresh_results(dname): fd.write('{}\n'.format(str(r2))) log = MockPastlyLogger() def test_stats_uninitted(tmpdir, capsys): def test_stats_uninitted(tmpdir, log): ''' An un-initialized .sbws directory should fail hard and exit immediately ''' Loading @@ -79,17 +75,16 @@ def test_stats_uninitted(tmpdir, capsys): assert e.code == 1 else: assert None, 'Should have failed' captured = capsys.readouterr() lines = captured.out.strip().split('\n') lines = [l for l in log.test_get_logged_lines()] assert 'Sbws isn\'t initialized. Try sbws init' == lines[-1] def test_stats_initted(tmpdir, capsys): def test_stats_initted(tmpdir, log): ''' An initialized but rather empty .sbws directory should fail about missing ~/.sbws/datadir ''' init_directory(tmpdir) init_directory(tmpdir, log) p = create_parser() args = p.parse_args('-d {} -vvvv stats'.format(tmpdir).split()) conf = get_config(args, log_fn=log.debug) Loading @@ -99,33 +94,31 @@ def test_stats_initted(tmpdir, capsys): assert e.code == 1 else: assert None, 'Should have failed' captured = capsys.readouterr() lines = captured.out.strip().split('\n') lines = [l for l in log.test_get_logged_lines()] assert '{}/datadir does not exist'.format(tmpdir) == lines[-1] def test_stats_stale_result(tmpdir, capsys): def test_stats_stale_result(tmpdir, log): ''' An initialized .sbws directory with no fresh results should say so and exit cleanly ''' init_directory(tmpdir) init_directory(tmpdir, log) add_single_stale_result(tmpdir) p = create_parser() args = p.parse_args('-d {} -vvvv stats'.format(tmpdir).split()) conf = get_config(args, log_fn=log.debug) sbws.commands.stats.main(args, conf, log) captured = capsys.readouterr() lines = captured.out.strip().split('\n') lines = [l for l in log.test_get_logged_lines()] assert 'No fresh results' == lines[-1] def test_stats_fresh_result(tmpdir, capsys): def test_stats_fresh_result(tmpdir, capsys, log): ''' An initialized .sbws directory with a fresh error result should have some boring stats and exit cleanly ''' init_directory(tmpdir) init_directory(tmpdir, log) add_single_fresh_result(tmpdir) p = create_parser() args = p.parse_args( Loading @@ -134,42 +127,52 @@ def test_stats_fresh_result(tmpdir, capsys): sbws.commands.stats.main(args, conf, log) captured = capsys.readouterr() lines = captured.out.strip().split('\n') needed_lines = [ 'Read 1 lines from {}/{}/{}.txt'.format( tmpdir, 'datadir', date.fromtimestamp(time.time())), 'Keeping 1/1 results', needed_output_lines = [ '1 relays have recent results', 'Average 0.00 successful measurements per relay', '0 success results and 1 error results', ] for needed_line in needed_lines: for needed_line in needed_output_lines: assert needed_line in lines lines = [l for l in log.test_get_logged_lines()] needed_log_lines = [ 'Read 1 lines from {}/{}/{}.txt'.format( tmpdir, 'datadir', date.fromtimestamp(time.time())), 'Keeping 1/1 results', ] for needed_line in needed_log_lines: assert needed_line in lines def test_stats_fresh_results(tmpdir, capsys): def test_stats_fresh_results(tmpdir, capsys, log): ''' An initialized .sbws directory with a fresh error and fresh success should have some exciting stats and exit cleanly ''' init_directory(tmpdir) init_directory(tmpdir, log) add_two_fresh_results(tmpdir) p = create_parser() args = p.parse_args( '-d {} -vvvv stats --error-types'.format(tmpdir).split()) conf = get_config(args, log_fn=log.debug) sbws.commands.stats.main(args, conf, log) needed_output_lines = [ '1 relays have recent results', '1 success results and 1 error results', 'Average 1.00 successful measurements per relay', '1/2 (50.00%) results were error-misc', ] captured = capsys.readouterr() lines = captured.out.strip().split('\n') needed_lines = [ for needed_line in needed_output_lines: assert needed_line in lines lines = [l for l in log.test_get_logged_lines()] needed_log_lines = [ 'Read 2 lines from {}/{}/{}.txt'.format( tmpdir, 'datadir', date.fromtimestamp(time.time())), 'Keeping 2/2 results', '1 relays have recent results', '1 success results and 1 error results', 'Average 1.00 successful measurements per relay', 'Found a _ResultType.Error for the first time', 'Found a _ResultType.Success for the first time', '1/2 (50.00%) results were error-misc', ] for needed_line in needed_lines: for needed_line in needed_log_lines: assert needed_line in lines tests/common.py +12 −2 Original line number Diff line number Diff line class MockPastlyLogger: def __init__(self, *a, **kw): def __init__(self, *a, _do_print=False, **kw): self._logged_lines = [] self._do_print = _do_print pass def debug(self, *s): self._logged_lines.append(' '.join(str(_) for _ in s)) if self._do_print: print(*s) def info(self, *s): Loading @@ -16,3 +20,9 @@ class MockPastlyLogger: def error(self, *s): return self.warn(*s) def test_get_logged_lines(self, clear=True): for line in self._logged_lines: yield line if clear: self._logged_lines = [] tests/conftest.py 0 → 100644 +8 −0 Original line number Diff line number Diff line from tests.common import MockPastlyLogger import pytest @pytest.fixture(scope='module') def log(): pl = MockPastlyLogger() return pl Loading
tests/commands/test_stats.py +34 −31 Original line number Diff line number Diff line Loading @@ -8,10 +8,9 @@ import sbws.commands.stats from datetime import date import os import time from tests.common import MockPastlyLogger def init_directory(dname): def init_directory(dname, log): p = create_parser() args = p.parse_args('-d {} -vvvv init'.format(dname).split()) conf = get_config(args, log_fn=log.debug) Loading Loading @@ -63,10 +62,7 @@ def add_two_fresh_results(dname): fd.write('{}\n'.format(str(r2))) log = MockPastlyLogger() def test_stats_uninitted(tmpdir, capsys): def test_stats_uninitted(tmpdir, log): ''' An un-initialized .sbws directory should fail hard and exit immediately ''' Loading @@ -79,17 +75,16 @@ def test_stats_uninitted(tmpdir, capsys): assert e.code == 1 else: assert None, 'Should have failed' captured = capsys.readouterr() lines = captured.out.strip().split('\n') lines = [l for l in log.test_get_logged_lines()] assert 'Sbws isn\'t initialized. Try sbws init' == lines[-1] def test_stats_initted(tmpdir, capsys): def test_stats_initted(tmpdir, log): ''' An initialized but rather empty .sbws directory should fail about missing ~/.sbws/datadir ''' init_directory(tmpdir) init_directory(tmpdir, log) p = create_parser() args = p.parse_args('-d {} -vvvv stats'.format(tmpdir).split()) conf = get_config(args, log_fn=log.debug) Loading @@ -99,33 +94,31 @@ def test_stats_initted(tmpdir, capsys): assert e.code == 1 else: assert None, 'Should have failed' captured = capsys.readouterr() lines = captured.out.strip().split('\n') lines = [l for l in log.test_get_logged_lines()] assert '{}/datadir does not exist'.format(tmpdir) == lines[-1] def test_stats_stale_result(tmpdir, capsys): def test_stats_stale_result(tmpdir, log): ''' An initialized .sbws directory with no fresh results should say so and exit cleanly ''' init_directory(tmpdir) init_directory(tmpdir, log) add_single_stale_result(tmpdir) p = create_parser() args = p.parse_args('-d {} -vvvv stats'.format(tmpdir).split()) conf = get_config(args, log_fn=log.debug) sbws.commands.stats.main(args, conf, log) captured = capsys.readouterr() lines = captured.out.strip().split('\n') lines = [l for l in log.test_get_logged_lines()] assert 'No fresh results' == lines[-1] def test_stats_fresh_result(tmpdir, capsys): def test_stats_fresh_result(tmpdir, capsys, log): ''' An initialized .sbws directory with a fresh error result should have some boring stats and exit cleanly ''' init_directory(tmpdir) init_directory(tmpdir, log) add_single_fresh_result(tmpdir) p = create_parser() args = p.parse_args( Loading @@ -134,42 +127,52 @@ def test_stats_fresh_result(tmpdir, capsys): sbws.commands.stats.main(args, conf, log) captured = capsys.readouterr() lines = captured.out.strip().split('\n') needed_lines = [ 'Read 1 lines from {}/{}/{}.txt'.format( tmpdir, 'datadir', date.fromtimestamp(time.time())), 'Keeping 1/1 results', needed_output_lines = [ '1 relays have recent results', 'Average 0.00 successful measurements per relay', '0 success results and 1 error results', ] for needed_line in needed_lines: for needed_line in needed_output_lines: assert needed_line in lines lines = [l for l in log.test_get_logged_lines()] needed_log_lines = [ 'Read 1 lines from {}/{}/{}.txt'.format( tmpdir, 'datadir', date.fromtimestamp(time.time())), 'Keeping 1/1 results', ] for needed_line in needed_log_lines: assert needed_line in lines def test_stats_fresh_results(tmpdir, capsys): def test_stats_fresh_results(tmpdir, capsys, log): ''' An initialized .sbws directory with a fresh error and fresh success should have some exciting stats and exit cleanly ''' init_directory(tmpdir) init_directory(tmpdir, log) add_two_fresh_results(tmpdir) p = create_parser() args = p.parse_args( '-d {} -vvvv stats --error-types'.format(tmpdir).split()) conf = get_config(args, log_fn=log.debug) sbws.commands.stats.main(args, conf, log) needed_output_lines = [ '1 relays have recent results', '1 success results and 1 error results', 'Average 1.00 successful measurements per relay', '1/2 (50.00%) results were error-misc', ] captured = capsys.readouterr() lines = captured.out.strip().split('\n') needed_lines = [ for needed_line in needed_output_lines: assert needed_line in lines lines = [l for l in log.test_get_logged_lines()] needed_log_lines = [ 'Read 2 lines from {}/{}/{}.txt'.format( tmpdir, 'datadir', date.fromtimestamp(time.time())), 'Keeping 2/2 results', '1 relays have recent results', '1 success results and 1 error results', 'Average 1.00 successful measurements per relay', 'Found a _ResultType.Error for the first time', 'Found a _ResultType.Success for the first time', '1/2 (50.00%) results were error-misc', ] for needed_line in needed_lines: for needed_line in needed_log_lines: assert needed_line in lines
tests/common.py +12 −2 Original line number Diff line number Diff line class MockPastlyLogger: def __init__(self, *a, **kw): def __init__(self, *a, _do_print=False, **kw): self._logged_lines = [] self._do_print = _do_print pass def debug(self, *s): self._logged_lines.append(' '.join(str(_) for _ in s)) if self._do_print: print(*s) def info(self, *s): Loading @@ -16,3 +20,9 @@ class MockPastlyLogger: def error(self, *s): return self.warn(*s) def test_get_logged_lines(self, clear=True): for line in self._logged_lines: yield line if clear: self._logged_lines = []
tests/conftest.py 0 → 100644 +8 −0 Original line number Diff line number Diff line from tests.common import MockPastlyLogger import pytest @pytest.fixture(scope='module') def log(): pl = MockPastlyLogger() return pl