NPO-Accounting/conservancy_beancount Changeset - cd578289c463

Changeset - cd578289c463

Parent rev.

Child rev.

[Not reviewed]

0 6 0

Brett Smith - 4 years ago 2020-06-12 19:08:08
brettcsmith@brettcsmith.org

cliutil: Add make_entry_point() function.

This provides better logging setup, reduces the amount of boilerplate in
main, and replaces is_main_script().

6 files changed with 60 insertions and 43 deletions:

conservancy_beancount/cliutil.py

conservancy_beancount/reports/accrual.py

setup.py

tests/test_cliutil.py

tests/test_reports_accrual.py

tests/testutil.py

0 comments (0 inline, 0 general)

conservancy_beancount/cliutil.py

➞

Show inline comments

@@ ... @@
 (at your option) any later version.
 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU Affero General Public License for more details.
 You should have received a copy of the GNU Affero General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>."""
 import argparse
 import enum
 import inspect
 import io
 import logging
 import operator
 import os
 import pkg_resources
 import re
 import signal
 import sys
 import traceback
 import types
 from pathlib import Path
@@ ... @@
+    )
 def add_version_argument(parser: argparse.ArgumentParser) -> argparse.Action:
     progname = parser.prog or sys.argv[0]
     return parser.add_argument(
         '--version', '--copyright', '--license',
         action=InfoAction,
         nargs=0,
         const=f"{progname} version {VERSION}\n{LICENSE}",
         help="Show program version and license information",
+    )
 def is_main_script(prog_name: str) -> bool:
     """Return true if the caller is the "main" program."""
     stack = iter(inspect.stack(context=False))
     next(stack)  # Discard the frame for calling this function
     caller_filename = next(stack).filename
     return all(frame.filename == caller_filename
                or Path(frame.filename).stem == prog_name
                for frame in stack)
 def make_entry_point(mod_name: str, prog_name: str=sys.argv[0]) -> Callable[[], int]:
     """Create an entry_point function for a tool
     The returned function is suitable for use as an entry_point in setup.py.
     It sets up the root logger and excepthook, then calls the module's main
     function.
     """
     def entry_point():  # type:ignore
         prog_mod = sys.modules[mod_name]
         setup_logger()
         prog_mod.logger = logging.getLogger(prog_name)
         sys.excepthook = ExceptHook(prog_mod.logger)
         return prog_mod.main()
     return entry_point
 def setup_logger(logger: Union[str, logging.Logger]='',
                  loglevel: int=logging.INFO,
                  stream: TextIO=sys.stderr,
                  fmt: str='%(name)s: %(levelname)s: %(message)s',
 ) -> logging.Logger:
     """Set up a logger with a StreamHandler with the given format"""
     if isinstance(logger, str):
         logger = logging.getLogger(logger)
     formatter = logging.Formatter(fmt)
     handler = logging.StreamHandler(stream)
     handler.setFormatter(formatter)
     logger.addHandler(handler)
     logger.setLevel(loglevel)
     return logger
 def set_loglevel(logger: logging.Logger, loglevel: int=logging.INFO) -> None:
     """Set the loglevel for a tool or module
     If the given logger is not under a hierarchy, this function sets the
     loglevel for the root logger, along with some specific levels for libraries
     used by reporting tools. Otherwise, it's the same as
     ``logger.setLevel(loglevel)``.
     """
     if '.' not in logger.name:
         logger = logging.getLogger()
         if loglevel <= logging.DEBUG:
             # At the debug level, the rt module logs the full body of every
             # request and response. That's too much.
             logging.getLogger('rt.rt').setLevel(logging.INFO)
     logger.setLevel(loglevel)
 def bytes_output(path: Optional[Path]=None,
                  default: OutputFile=sys.stdout,
                  mode: str='w',
 ) -> BinaryIO:
     """Get a file-like object suitable for binary output
     If ``path`` is ``None`` or ``-``, returns a file-like object backed by
     ``default``. If ``default`` is a file descriptor or text IO object, this
     method returns a file-like object that writes to the same place.
     Otherwise, returns ``path.open(mode)``.
     """

conservancy_beancount/reports/accrual.py

➞

Show inline comments

@@ ... @@
 `TIK/ATT` format, is a shortcut for `invoice=LINK`.
 """)
     args = parser.parse_args(arglist)
     if args.report_type is None and not args.search_terms:
         args.report_type = ReportType.AGING
     return args
 def main(arglist: Optional[Sequence[str]]=None,
          stdout: TextIO=sys.stdout,
          stderr: TextIO=sys.stderr,
          config: Optional[configmod.Config]=None,
 ) -> int:
     if cliutil.is_main_script(PROGNAME):
         global logger
         logger = logging.getLogger(PROGNAME)
         sys.excepthook = cliutil.ExceptHook(logger)
     args = parse_arguments(arglist)
-    cliutil.setup_logger(logger, args.loglevel, stderr)
+    cliutil.set_loglevel(logger, args.loglevel)
     if config is None:
         config = configmod.Config()
         config.load_file()
     returncode = 0
     books_loader = config.books_loader()
     if books_loader is None:
         entries, load_errors, _ = books.Loader.load_none(config.config_file_path())
     elif args.report_type is ReportType.AGING:
         entries, load_errors, _ = books_loader.load_all()
     else:
         entries, load_errors, _ = books_loader.load_all(args.since)
@@ ... / @@ -744,14 +740,16 @@ def main(arglist: Optional[Sequence[str]]=None, @@
             out_file = cliutil.text_output(args.output_file, stdout)
             report = OutgoingReport(rt_wrapper, out_file)
     else:
         out_file = cliutil.text_output(args.output_file, stdout)
         report = BalanceReport(out_file)
     if report is None:
         returncode |= ReturnFlag.REPORT_ERRORS
     else:
         report.run(groups)
     return 0 if returncode == 0 else 16 + returncode
 entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 if __name__ == '__main__':
-    exit(main())
+    exit(entry_point())

setup.py

➞

Show inline comments

@@ ... / @@ -26,16 +26,16 @@ setup( @@
     tests_require=[
         'mypy>=0.770',  # Debian:python3-mypy
         'pytest',  # Debian:python3-pytest
     ],
     packages=[
         'conservancy_beancount',
         'conservancy_beancount.plugin',
         'conservancy_beancount.reports',
     ],
     entry_points={
         'console_scripts': [
-            'accrual-report = conservancy_beancount.reports.accrual:main',
+            'accrual-report = conservancy_beancount.reports.accrual:entry_point',
         ],
     },
+)

tests/test_cliutil.py

➞

Show inline comments

@@ ... / @@ -24,29 +24,24 @@ import re @@
 import sys
 import traceback
 import pytest
 from pathlib import Path
 from conservancy_beancount import cliutil
 FILE_NAMES = ['-foobar', '-foo.bin']
 STREAM_PATHS = [None, Path('-')]
 class AlwaysEqual:
     def __eq__(self, other):
         return True
 class MockTraceback:
     def __init__(self, stack=None, index=0):
         if stack is None:
             stack = inspect.stack(context=False)
         self._stack = stack
         self._index = index
         frame_record = self._stack[self._index]
         self.tb_frame = frame_record.frame
         self.tb_lineno = frame_record.lineno
     @property
     def tb_next(self):
@@ ... / @@ -132,54 +127,46 @@ def test_excepthook_bug(exc_type, caplog): @@
         assert log.levelname == 'CRITICAL'
         assert log.message == f"internal {exc_type.__name__}: {error.args[0]}"
 def test_excepthook_traceback(caplog):
     error = KeyError('test')
     args = (type(error), error, MockTraceback())
     caplog.set_level(logging.DEBUG)
     with pytest.raises(SystemExit) as exc_check:
         cliutil.ExceptHook()(*args)
     assert caplog.records
     assert caplog.records[-1].message == ''.join(traceback.format_exception(*args))
 @pytest.mark.parametrize('prog_name,expected', [
     ('', False),
     (AlwaysEqual(), True),
 ])
 def test_is_main_script(prog_name, expected):
     assert cliutil.is_main_script(prog_name) == expected
 @pytest.mark.parametrize('arg,expected', [
     ('debug', logging.DEBUG),
     ('info', logging.INFO),
     ('warning', logging.WARNING),
     ('warn', logging.WARNING),
     ('error', logging.ERROR),
     ('err', logging.ERROR),
     ('critical', logging.CRITICAL),
     ('crit', logging.CRITICAL),
 ])
 def test_loglevel_argument(argparser, arg, expected):
     for method in ['lower', 'title', 'upper']:
         args = argparser.parse_args(['--loglevel', getattr(arg, method)()])
         assert args.loglevel is expected
 def test_setup_logger():
     stream = io.StringIO()
     logger = cliutil.setup_logger(
-        'test_cliutil', logging.INFO, stream, '%(name)s %(levelname)s: %(message)s',
         'test_cliutil', stream, '%(name)s %(levelname)s: %(message)s',
+    )
     logger.debug("test debug")
     logger.info("test info")
     assert stream.getvalue() == "test_cliutil INFO: test info\n"
     logger.critical("test crit")
     assert stream.getvalue() == "test_cliutil CRITICAL: test crit\n"
 @pytest.mark.parametrize('arg', [
     '--license',
     '--version',
     '--copyright',
 ])
 def test_version_argument(argparser, capsys, arg):
     with pytest.raises(SystemExit) as exc_check:
         args = argparser.parse_args(['--version'])
     assert exc_check.value.args[0] == 0
     stdout, _ = capsys.readouterr()
     lines = iter(stdout.splitlines())

tests/test_reports_accrual.py

➞

Show inline comments

@@ ... @@
 def run_main(arglist, config=None):
     if config is None:
         config = testutil.TestConfig(
             books_path=testutil.test_path('books/accruals.beancount'),
             rt_client=RTClient(),
+        )
     output = io.StringIO()
     errors = io.StringIO()
     retcode = accrual.main(arglist, output, errors, config)
     return retcode, output, errors
-def check_main_fails(arglist, config, error_flags, error_patterns):
 def check_main_fails(arglist, config, error_flags):
     retcode, output, errors = run_main(arglist, config)
     assert retcode > 16
     assert (retcode - 16) & error_flags
     check_output(errors, error_patterns)
     assert not output.getvalue()
     errors.seek(0)
     return errors
 @pytest.mark.parametrize('arglist', [
     ['--report-type=balance', 'entity=EarlyBird'],
     ['--report-type=outgoing', 'entity=EarlyBird'],
 ])
 def test_output_excludes_payments(arglist):
     retcode, output, errors = run_main(arglist)
     assert not errors.getvalue()
     assert retcode == 0
     output.seek(0)
     for line in output:
         assert not re.match(r'\brt:4\d\b', line)
@@ ... / @@ -677,33 +678,36 @@ def test_main_aging_report(tmp_path, arglist): @@
         recv_rows = AGING_AR
         pay_rows = AGING_AP
     output_path = tmp_path / 'AgingReport.ods'
     arglist.insert(0, f'--output-file={output_path}')
     retcode, output, errors = run_main(arglist)
     assert not errors.getvalue()
     assert retcode == 0
     assert not output.getvalue()
     with output_path.open('rb') as ods_file:
         check_aging_ods(ods_file, None, recv_rows, pay_rows)
 def test_main_no_books():
     check_main_fails([], testutil.TestConfig(), 1 | 8, [
     errors = check_main_fails([], testutil.TestConfig(), 1 | 8)
     testutil.check_lines_match(iter(errors), [
         r':[01]: +no books to load in configuration\b',
     ])
 @pytest.mark.parametrize('arglist', [
     ['499'],
     ['505/99999'],
     ['entity=NonExistent'],
 ])
 def test_main_no_matches(arglist):
     check_main_fails(arglist, None, 8, [
         r': WARNING: no matching entries found to report$',
 def test_main_no_matches(arglist, caplog):
     check_main_fails(arglist, None, 8)
     testutil.check_logs_match(caplog, [
         ('WARNING', 'no matching entries found to report'),
     ])
 def test_main_no_rt():
+def test_main_no_rt(caplog):
     config = testutil.TestConfig(
         books_path=testutil.test_path('books/accruals.beancount'),
+    )
     check_main_fails(['-t', 'out'], config, 4, [
         r': ERROR: unable to generate outgoing report: RT client is required\b',
     check_main_fails(['-t', 'out'], config, 4)
     testutil.check_logs_match(caplog, [
         ('ERROR', 'unable to generate outgoing report: RT client is required'),
     ])

tests/testutil.py

➞

Show inline comments

@@ ... / @@ -60,24 +60,32 @@ def _ods_elem_text(elem): @@
     else:
         return '\0'.join(_ods_elem_text(child) for child in elem.childNodes)
 odf.element.Element.value_type = property(_ods_cell_value_type)
 odf.element.Element.value = property(_ods_cell_value)
 odf.element.Element.text = property(_ods_elem_text)
 def check_lines_match(lines, expect_patterns, source='output'):
     for pattern in expect_patterns:
         assert any(re.search(pattern, line) for line in lines), \
             f"{pattern!r} not found in {source}"
 def check_logs_match(caplog, expected):
     records = iter(caplog.records)
     for exp_level, exp_msg in expected:
         exp_level = exp_level.upper()
         assert any(
             log.levelname == exp_level and log.message == exp_msg for log in records
         ), f"{exp_level} log {exp_msg!r} not found"
 def check_post_meta(txn, *expected_meta, default=None):
     assert len(txn.postings) == len(expected_meta)
     for post, expected in zip(txn.postings, expected_meta):
         if not expected:
             assert not post.meta
         else:
             actual = None if post.meta is None else {
                 key: post.meta.get(key, default) for key in expected
+            }
             assert actual == expected
 def combine_values(*value_seqs):

0 comments (0 inline, 0 general)