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

 """cliutil - Utilities for CLI tools"""
 PKGNAME = 'conservancy_beancount'
 LICENSE = """
 Copyright © 2020  Brett Smith
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU Affero General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (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
 from . import data
 from . import filters
 from . import rtutil
 from typing import (
     cast,
     Any,
     BinaryIO,
     Callable,
     IO,
     Iterable,
     NamedTuple,
     NoReturn,
     Optional,
     Sequence,
     TextIO,
     Type,
     Union,
+)
 from .beancount_types import (
     MetaKey,
+)
 OutputFile = Union[int, IO]
 STDSTREAM_PATH = Path('-')
 VERSION = pkg_resources.require(PKGNAME)[0].version
 class ExceptHook:
     def __init__(self,
                  logger: Optional[logging.Logger]=None,
                  default_exitcode: int=3,
     ) -> None:
         if logger is None:
             logger = logging.getLogger()
         self.logger = logger
         self.default_exitcode = default_exitcode
     def __call__(self,
                  exc_type: Type[BaseException],
                  exc_value: BaseException,
                  exc_tb: types.TracebackType,
     ) -> NoReturn:
         exitcode = self.default_exitcode
         if isinstance(exc_value, KeyboardInterrupt):
             signal.signal(signal.SIGINT, signal.SIG_DFL)
             os.kill(0, signal.SIGINT)
             signal.pause()
         elif isinstance(exc_value, OSError):
             exitcode += 1
             msg = "I/O error: {e.filename}: {e.strerror}".format(e=exc_value)
         else:
             parts = [type(exc_value).__name__, *exc_value.args]
             msg = "internal " + ": ".join(parts)
         self.logger.critical(msg)
         self.logger.debug(
             ''.join(traceback.format_exception(exc_type, exc_value, exc_tb)),
+        )
         raise SystemExit(exitcode)
 class InfoAction(argparse.Action):
     def __call__(self,
                  parser: argparse.ArgumentParser,
                  namespace: argparse.Namespace,
                  values: Union[Sequence[Any], str, None]=None,
                  option_string: Optional[str]=None,
     ) -> NoReturn:
         if isinstance(self.const, str):
             info = self.const
             exitcode = 0
         else:
             info, exitcode = self.const
         print(info)
         raise SystemExit(exitcode)
 class LogLevel(enum.IntEnum):
     DEBUG = logging.DEBUG
     INFO = logging.INFO
     WARNING = logging.WARNING
     ERROR = logging.ERROR
     CRITICAL = logging.CRITICAL
     WARN = WARNING
     ERR = ERROR
     CRIT = CRITICAL
     @classmethod
     def from_arg(cls, arg: str) -> int:
         try:
             return cls[arg.upper()].value
         except KeyError:
             raise ValueError(f"unknown loglevel {arg!r}") from None
     @classmethod
     def choices(cls) -> Iterable[str]:
         for level in sorted(cls, key=operator.attrgetter('value')):
             yield level.name.lower()
 class SearchTerm(NamedTuple):
     """NamedTuple representing a user's metadata filter
     SearchTerm knows how to parse and store posting metadata filters provided
     by the user in `key=value` format. Reporting tools can use this to filter
     postings that match the user's criteria, to report on subsets of the books.
     Typical usage looks like::
       argument_parser.add_argument(
         'search_terms',
         type=SearchTerm.arg_parser(),
         …,
+      )
       args = argument_parser.parse_args(…)
       for query in args.search_terms:
         postings = query.filter_postings(postings)
     """
     meta_key: MetaKey
     pattern: str
     @classmethod
     def arg_parser(cls,
                    default_key: Optional[str]=None,
                    ticket_default_key: Optional[str]=None,
     ) -> Callable[[str], 'SearchTerm']:
         """Build a SearchTerm parser
         This method returns a function that can parse strings in ``key=value``
         format and return a corresponding SearchTerm.
         If you specify a default key, then strings that just specify a ``value``
         will be parsed as if they said ``default_key=value``. Otherwise,
         parsing strings without a metadata key will raise a ValueError.
         If you specify a default key ticket links, then values in the format
         ``number``, ``rt:number``, or ``rt://ticket/number`` will be parsed as
         if they said ``ticket_default_key=value``.
         """
         if ticket_default_key is None:
             ticket_default_key = default_key
         def parse_search_term(arg: str) -> 'SearchTerm':
             key: Optional[str] = None
             if re.match(r'^[a-z][-\w]*=', arg):
                 key, _, raw_link = arg.partition('=')
             else:
                 raw_link = arg
             rt_ids = rtutil.RT.parse(raw_link)
             if rt_ids is None:
                 rt_ids = rtutil.RT.parse('rt:' + raw_link)
             if rt_ids is None:
                 if key is None:
                     key = default_key
                 pattern = r'(?:^|\s){}(?:\s|$)'.format(re.escape(raw_link))
             else:
                 ticket_id, attachment_id = rt_ids
                 if key is None:
                     if attachment_id is None:
                         key = ticket_default_key
                     else:
                         key = default_key
                 pattern = rtutil.RT.metadata_regexp(
                     ticket_id,
                     attachment_id,
                     first_link_only=key == 'rt-id' and attachment_id is None,
+                )
             if key is None:
                 raise ValueError(f"invalid search term {arg!r}: no metadata key")
             return cls(key, pattern)
         return parse_search_term
     def filter_postings(self, postings: Iterable[data.Posting]) -> Iterable[data.Posting]:
         return filters.filter_meta_match(
             postings, self.meta_key, re.compile(self.pattern),
+        )
 def add_loglevel_argument(parser: argparse.ArgumentParser,
                           default: LogLevel=LogLevel.INFO) -> argparse.Action:
     return parser.add_argument(
         '--loglevel',
         metavar='LEVEL',
         default=default.value,
         type=LogLevel.from_arg,
         help="Show logs at this level and above."
         f" Specify one of {', '.join(LogLevel.choices())}."
         f" Default {default.name.lower()}.",
+    )
 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)``.
     """
     mode = f'{mode}b'
     if path is None or path == STDSTREAM_PATH:
         if isinstance(default, int):
             retval = open(default, mode)
         elif isinstance(default, TextIO):
             retval = default.buffer
         else:
             retval = default
     else:
         retval = path.open(mode)
     return cast(BinaryIO, retval)
 def text_output(path: Optional[Path]=None,
                 default: OutputFile=sys.stdout,
                 mode: str='w',
                 encoding: Optional[str]=None,
 ) -> TextIO:
     """Get a file-like object suitable for text output
     If ``path`` is ``None`` or ``-``, returns a file-like object backed by
     ``default``. If ``default`` is a file descriptor or binary IO object, this
     method returns a file-like object that writes to the same place.
     Otherwise, returns ``path.open(mode)``.
     """
     if path is None or path == STDSTREAM_PATH:
         if isinstance(default, int):
             retval = open(default, mode, encoding=encoding)
         elif isinstance(default, BinaryIO):
             retval = io.TextIOWrapper(default, encoding=encoding)
         else:
             retval = default
     else:
         retval = path.open(mode, encoding=encoding)
     return cast(TextIO, retval)

conservancy_beancount/reports/accrual.py

➞

Show inline comments

@@ ... / @@ -474,284 +474,282 @@ class AgingReport(BaseReport): @@
+                )
             if group and not group.is_zero():
                 rows.append(group)
         rows.sort(key=lambda related: (
             related.account,
             related[0].meta.date,
             ('\0'.join(related.entities())
              if related.entity is related.INCONSISTENT
              else related.entity),
         ))
         self.ods.write(rows)
         self.ods.save_file(self.out_bin)
 class BalanceReport(BaseReport):
     def _report(self, posts: AccrualPostings, index: int) -> Iterable[str]:
         posts = posts.since_last_nonzero()
         date_s = posts[0].meta.date.strftime('%Y-%m-%d')
         if index:
             yield ""
         yield f"{posts.invoice}:"
         yield f"  {posts.balance_at_cost()} outstanding since {date_s}"
 class OutgoingReport(BaseReport):
     def __init__(self, rt_wrapper: rtutil.RT, out_file: TextIO) -> None:
         super().__init__(out_file)
         self.rt_wrapper = rt_wrapper
         self.rt_client = rt_wrapper.rt
     def _primary_rt_id(self, posts: AccrualPostings) -> rtutil.TicketAttachmentIds:
         rt_id = posts.rt_id
         if rt_id is None:
             raise ValueError("no rt-id links found")
         elif isinstance(rt_id, Sentinel):
             raise ValueError("multiple rt-id links found")
         parsed = rtutil.RT.parse(rt_id)
         if parsed is None:
             raise ValueError("rt-id is not a valid RT reference")
         else:
             return parsed
     def _report(self, posts: AccrualPostings, index: int) -> Iterable[str]:
         posts = posts.since_last_nonzero()
         try:
             ticket_id, _ = self._primary_rt_id(posts)
             ticket = self.rt_client.get_ticket(ticket_id)
             # Note we only use this when ticket is None.
             errmsg = f"ticket {ticket_id} not found"
         except (ValueError, rt.RtError) as error:
             ticket = None
             errmsg = error.args[0]
         if ticket is None:
             self.logger.error(
                 "can't generate outgoings report for %s because no RT ticket available: %s",
                 posts.invoice, errmsg,
+            )
             return
         try:
             rt_requestor = self.rt_client.get_user(ticket['Requestors'][0])
         except (IndexError, rt.RtError):
             rt_requestor = None
         if rt_requestor is None:
             requestor = ''
             requestor_name = ''
         else:
             requestor_name = (
                 rt_requestor.get('RealName')
                 or ticket.get('CF.{payment-to}')
                 or ''
+            )
             requestor = f'{requestor_name} <{rt_requestor["EmailAddress"]}>'.strip()
         balance_s = posts.end_balance.format(None)
         raw_balance = -posts.balance()
         if raw_balance != posts.end_balance:
             balance_s = f'{raw_balance} ({balance_s})'
         contract_links = list(posts.all_meta_links('contract'))
         if contract_links:
             contract_s = ' , '.join(self.rt_wrapper.iter_urls(
                 contract_links, missing_fmt='<BROKEN RT LINK: {}>',
             ))
         else:
             contract_s = "NO CONTRACT GOVERNS THIS TRANSACTION"
         projects = [v for v in posts.meta_values('project')
                     if isinstance(v, str)]
         yield "PAYMENT FOR APPROVAL:"
         yield f"REQUESTOR: {requestor}"
         yield f"TOTAL TO PAY: {balance_s}"
         yield f"AGREEMENT: {contract_s}"
         yield f"PAYMENT TO: {ticket.get('CF.{payment-to}') or requestor_name}"
         yield f"PAYMENT METHOD: {ticket.get('CF.{payment-method}', '')}"
         yield f"PROJECT: {', '.join(projects)}"
         yield "\nBEANCOUNT ENTRIES:\n"
         last_txn: Optional[Transaction] = None
         for post in posts:
             txn = post.meta.txn
             if txn is not last_txn:
                 last_txn = txn
                 txn = self.rt_wrapper.txn_with_urls(txn, '{}')
                 yield bc_printer.format_entry(txn)
 class ReportType(enum.Enum):
     AGING = AgingReport
     BALANCE = BalanceReport
     OUTGOING = OutgoingReport
     AGE = AGING
     BAL = BALANCE
     OUT = OUTGOING
     OUTGOINGS = OUTGOING
     @classmethod
     def by_name(cls, name: str) -> 'ReportType':
         try:
             return cls[name.upper()]
         except KeyError:
             raise ValueError(f"unknown report type {name!r}") from None
 class ReturnFlag(enum.IntFlag):
     LOAD_ERRORS = 1
     # 2 was used in the past, it can probably be reclaimed.
     REPORT_ERRORS = 4
     NOTHING_TO_REPORT = 8
 def filter_search(postings: Iterable[data.Posting],
                   search_terms: Iterable[cliutil.SearchTerm],
 ) -> Iterable[data.Posting]:
     accounts = tuple(AccrualAccount.account_names())
     postings = (post for post in postings if post.account.is_under(*accounts))
     for query in search_terms:
         postings = query.filter_postings(postings)
     return postings
 def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(prog=PROGNAME)
     cliutil.add_version_argument(parser)
     parser.add_argument(
         '--report-type', '-t',
         metavar='NAME',
         type=ReportType.by_name,
         help="""The type of report to generate, one of `aging`, `balance`, or
 `outgoing`. If not specified, the default is `aging` when no search terms are
 given, `outgoing` for search terms that return a single outstanding payable,
 and `balance` any other time.
 """)
     parser.add_argument(
         '--since',
         metavar='YEAR',
         type=int,
         default=-1,
         help="""How far back to search the books for related transactions.
 You can either specify a fiscal year, or a negative offset from the current
 fiscal year, to start loading entries from. The default is -1 (start from the
 previous fiscal year).
 """)
     parser.add_argument(
         '--output-file', '-O',
         metavar='PATH',
         type=Path,
         help="""Write the report to this file, or stdout when PATH is `-`.
 The default is stdout for the balance and outgoing reports, and a generated
 filename for other reports.
 """)
     cliutil.add_loglevel_argument(parser)
     parser.add_argument(
         'search_terms',
         metavar='FILTER',
         type=cliutil.SearchTerm.arg_parser('invoice', 'rt-id'),
         nargs=argparse.ZERO_OR_MORE,
         help="""Report on accruals that match this criteria. The format is
 NAME=TERM. TERM is a link or word that must exist in a posting's NAME
 metadata to match. A single ticket number is a shortcut for
 `rt-id=rt:NUMBER`. Any other link, including an RT attachment link in
 `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)
     filters.remove_opening_balance_txn(entries)
     for error in load_errors:
         bc_printer.print_error(error, file=stderr)
         returncode |= ReturnFlag.LOAD_ERRORS
     postings = list(filter_search(
         data.Posting.from_entries(entries), args.search_terms,
     ))
     if not postings:
         logger.warning("no matching entries found to report")
         returncode |= ReturnFlag.NOTHING_TO_REPORT
     # groups is a mapping of metadata value strings to AccrualPostings.
     # The keys are basically arbitrary, the report classes don't rely on them,
     # but they do help symbolize what's being grouped.
     # For the outgoing approval report, groups maps rt-id link strings to
     # associated accruals.
     # For all other reports, groups starts by grouping postings together by
     # invoice link string, then uses AccrualReport.make_consistent() to split
     # out groups that need it.
     groups: PostGroups
     if args.report_type is None or args.report_type is ReportType.OUTGOING:
         groups = dict(AccrualPostings.group_by_first_meta_link(postings, 'rt-id'))
         if (args.report_type is None
             and len(groups) == 1
             and all(group.accrual_type is AccrualAccount.PAYABLE
                     and not group.is_paid()
                     and key  # Make sure we have a usable rt-id
                     for key, group in groups.items())
         ):
             args.report_type = ReportType.OUTGOING
     if args.report_type is not ReportType.OUTGOING:
         groups = {
             key: group
             for _, source in AccrualPostings.group_by_first_meta_link(postings, 'invoice')
             for key, group in source.make_consistent()
+        }
     if args.report_type is not ReportType.AGING:
         groups = {
             key: posts for key, posts in groups.items() if not posts.is_paid()
         } or groups
     del postings
     report: Optional[BaseReport] = None
     output_path: Optional[Path] = None
     if args.report_type is ReportType.AGING:
         rt_wrapper = config.rt_wrapper()
         if rt_wrapper is None:
             logger.error("unable to generate aging report: RT client is required")
         else:
             now = datetime.datetime.now()
             if args.output_file is None:
                 args.output_file = Path(now.strftime('AgingReport_%Y-%m-%d_%H:%M.ods'))
                 logger.info("Writing report to %s", args.output_file)
             out_bin = cliutil.bytes_output(args.output_file, stdout)
             report = AgingReport(rt_wrapper, out_bin)
     elif args.report_type is ReportType.OUTGOING:
         rt_wrapper = config.rt_wrapper()
         if rt_wrapper is None:
             logger.error("unable to generate outgoing report: RT client is required")
         else:
             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

 #!/usr/bin/env python3
 from setuptools import setup
 setup(
     name='conservancy_beancount',
     description="Plugin, library, and reports for reading Conservancy's books",
     version='1.1.11',
     author='Software Freedom Conservancy',
     author_email='info@sfconservancy.org',
     license='GNU AGPLv3+',
     install_requires=[
         'babel>=2.6',  # Debian:python3-babel
         'beancount>=2.2',  # Debian:beancount
         # 1.4.1 crashes when trying to save some documents.
         'odfpy>=1.4.0,!=1.4.1',  # Debian:python3-odf
         'PyYAML>=3.0',  # Debian:python3-yaml
         'regex',  # Debian:python3-regex
         'rt>=2.0',
     ],
     setup_requires=[
         'pytest-mypy',
         'pytest-runner',  # Debian:python3-pytest-runner
     ],
     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

 """Test CLI utilities"""
 # Copyright © 2020  Brett Smith
+#
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (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 errno
 import io
 import inspect
 import logging
 import os
 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):
         try:
             return type(self)(self._stack, self._index + 1)
         except IndexError:
             return None
 @pytest.fixture(scope='module')
 def argparser():
     parser = argparse.ArgumentParser(prog='test_cliutil')
     cliutil.add_loglevel_argument(parser)
     cliutil.add_version_argument(parser)
     return parser
 @pytest.mark.parametrize('path_name', FILE_NAMES)
 def test_bytes_output_path(path_name, tmp_path):
     path = tmp_path / path_name
     stream = io.BytesIO()
     actual = cliutil.bytes_output(path, stream)
     assert actual is not stream
     assert str(actual.name) == str(path)
     assert 'w' in actual.mode
     assert 'b' in actual.mode
 @pytest.mark.parametrize('path', STREAM_PATHS)
 def test_bytes_output_stream(path):
     stream = io.BytesIO()
     actual = cliutil.bytes_output(path, stream)
     assert actual is stream
 @pytest.mark.parametrize('func_name', [
     'bytes_output',
     'text_output',
 ])
 def test_default_output(func_name):
     actual = getattr(cliutil, func_name)()
     assert actual.fileno() == sys.stdout.fileno()
 @pytest.mark.parametrize('path_name', FILE_NAMES)
 def test_text_output_path(path_name, tmp_path):
     path = tmp_path / path_name
     stream = io.StringIO()
     actual = cliutil.text_output(path, stream)
     assert actual is not stream
     assert str(actual.name) == str(path)
     assert 'w' in actual.mode
 @pytest.mark.parametrize('path', STREAM_PATHS)
 def test_text_output_stream(path):
     stream = io.StringIO()
     actual = cliutil.text_output(path, stream)
     assert actual is stream
 @pytest.mark.parametrize('errnum', [
     errno.EACCES,
     errno.EPERM,
     errno.ENOENT,
 ])
 def test_excepthook_oserror(errnum, caplog):
     error = OSError(errnum, os.strerror(errnum), 'TestFilename')
     with pytest.raises(SystemExit) as exc_check:
         cliutil.ExceptHook()(type(error), error, None)
     assert exc_check.value.args[0] == 4
     assert caplog.records
     for log in caplog.records:
         assert log.levelname == 'CRITICAL'
         assert log.message == f"I/O error: {error.filename}: {error.strerror}"
 @pytest.mark.parametrize('exc_type', [
     AttributeError,
     RuntimeError,
     ValueError,
 ])
 def test_excepthook_bug(exc_type, caplog):
     error = exc_type("test message")
     with pytest.raises(SystemExit) as exc_check:
         cliutil.ExceptHook()(exc_type, error, None)
     assert exc_check.value.args[0] == 3
     assert caplog.records
     for log in caplog.records:
         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())
     assert re.match(r'^test_cliutil version \d+\.\d+\.\d+', next(lines, "<EOF>"))

tests/test_reports_accrual.py

➞

Show inline comments

@@ ... / @@ -401,309 +401,313 @@ def test_make_consistent_across_entity(acct_name): @@
         assert len(posts) == 1
         entities = posts.entities()
         assert next(entities, None) == posts.entity
         assert next(entities, None) is None
         assert posts.entity in key
 @pytest.mark.parametrize('acct_name', ACCOUNTS)
 def test_make_consistent_entity_differs_accrual_payment(acct_name):
     invoice = 'Invoices/DifferPay.pdf'
     txn = testutil.Transaction(postings=[
         # Depending on the account, the order of the accrual and payment might
         # be swapped here, but that shouldn't matter.
         (acct_name, 125, {'invoice': invoice, 'entity': 'Positive'}),
         (acct_name, -125, {'invoice': invoice, 'entity': 'Negative'}),
     ])
     related = accrual.AccrualPostings(data.Posting.from_txn(txn))
     consistent = related.make_consistent()
     _, actual = next(consistent)
     assert actual is related
     assert next(consistent, None) is None
 def check_output(output, expect_patterns):
     output.seek(0)
     testutil.check_lines_match(iter(output), expect_patterns)
 def run_outgoing(rt_id, postings, rt_client=None):
     if rt_client is None:
         rt_client = RTClient()
     rt_wrapper = rtutil.RT(rt_client)
     if not isinstance(postings, core.RelatedPostings):
         postings = accruals_by_meta(postings, rt_id, 'rt-id', wrap_type=accrual.AccrualPostings)
     output = io.StringIO()
     report = accrual.OutgoingReport(rt_wrapper, output)
     report.run({rt_id: postings})
     return output
 @pytest.mark.parametrize('invoice,expected', [
     ('rt:505/5050', "Zero balance outstanding since 2010-05-05"),
     ('rt:510/5100', "Zero balance outstanding since 2010-05-10"),
     ('rt:510/6100', "-280.00 USD outstanding since 2010-06-10"),
     ('rt://ticket/515/attachments/5150', "1,500.00 USD outstanding since 2010-05-15",),
 ])
 def test_balance_report(accrual_postings, invoice, expected, caplog):
     related = accruals_by_meta(accrual_postings, invoice, wrap_type=accrual.AccrualPostings)
     output = io.StringIO()
     report = accrual.BalanceReport(output)
     report.run({invoice: related})
     assert not caplog.records
     check_output(output, [invoice, expected])
 def test_outgoing_report(accrual_postings, caplog):
     output = run_outgoing('rt:510', accrual_postings)
     rt_url = RTClient.DEFAULT_URL[:-9]
     rt_id_url = rf'\b{re.escape(f"{rt_url}Ticket/Display.html?id=510")}\b'
     contract_url = rf'\b{re.escape(f"{rt_url}Ticket/Attachment/4000/4000/contract.pdf")}\b'
     assert not caplog.records
     check_output(output, [
         r'^PAYMENT FOR APPROVAL:$',
         r'^REQUESTOR: Mx\. 510 <mx510@example\.org>$',
         r'^TOTAL TO PAY: \$280\.00$',
         fr'^AGREEMENT: {contract_url}',
         r'^PAYMENT TO: Hon\. Mx\. 510$',
         r'^PAYMENT METHOD: payment method 510$',
         r'^BEANCOUNT ENTRIES:$',
         # For each transaction, check for the date line, a metadata, and the
         # Expenses posting.
         r'^\s*2010-06-10\s',
         fr'^\s+rt-id: "{rt_id_url}"$',
         r'^\s+Expenses:Services:Legal\s+220\.00 USD$',
         r'^\s*2010-06-12\s',
         fr'^\s+contract: "{contract_url}"$',
         r'^\s+Expenses:FilingFees\s+60\.00 USD$',
     ])
 def test_outgoing_report_custom_field_fallbacks(accrual_postings, caplog):
     rt_client = RTClient(want_cfs=False)
     output = run_outgoing('rt:510', accrual_postings, rt_client)
     assert not caplog.records
     check_output(output, [
         r'^PAYMENT FOR APPROVAL:$',
         r'^REQUESTOR: <mx510@example\.org>$',
         r'^PAYMENT TO:\s*$',
         r'^PAYMENT METHOD:\s*$',
     ])
 def test_outgoing_report_fx_amounts(accrual_postings, caplog):
     output = run_outgoing('rt:520 rt:525', accrual_postings)
     assert not caplog.records
     check_output(output, [
         r'^PAYMENT FOR APPROVAL:$',
         r'^REQUESTOR: Mx\. 520 <mx520@example\.org>$',
         r'^TOTAL TO PAY: 1,000\.00 EUR \(\$1,100.00\)$',
     ])
 def test_outgoing_report_multi_invoice(accrual_postings, caplog):
     output = run_outgoing('rt:310', accrual_postings)
     assert not caplog.records
     check_output(output, [
         r'^PAYMENT FOR APPROVAL:$',
         r'^REQUESTOR: Mx\. 310 <mx310@example\.org>$',
         r'^TOTAL TO PAY: \$420.00$',
     ])
 def test_outgoing_report_without_rt_id(accrual_postings, caplog):
     invoice = 'rt://ticket/515/attachments/5150'
     related = accruals_by_meta(
         accrual_postings, invoice, wrap_type=accrual.AccrualPostings,
+    )
     output = run_outgoing(None, related)
     assert caplog.records
     log = caplog.records[0]
     assert log.message.startswith(
         f"can't generate outgoings report for {invoice} because no RT ticket available:",
+    )
     assert not output.getvalue()
 def run_aging_report(postings, today=None):
     if today is None:
         today = datetime.date.today()
     postings = (
         post for post in postings
         if post.account.is_under('Assets:Receivable', 'Liabilities:Payable')
+    )
     groups = {
         key: group
         for _, related in accrual.AccrualPostings.group_by_meta(postings, 'invoice')
         for key, group in related.make_consistent()
+    }
     output = io.BytesIO()
     rt_wrapper = rtutil.RT(RTClient())
     report = accrual.AgingReport(rt_wrapper, output, today)
     report.run(groups)
     return output
 def test_aging_report(accrual_postings):
     output = run_aging_report(accrual_postings)
     check_aging_ods(output)
 @pytest.mark.parametrize('date,recv_end,pay_end', [
     # Both these dates are chosen for their off-by-one potential:
     # the first is exactly 30 days after the 2010-06-10 payable;
     # the second is exactly 60 days after the 2010-05-15 receivable.
     (datetime.date(2010, 7, 10), 1, 5),
     (datetime.date(2010, 7, 14), 2, 5),
 ])
 def test_aging_report_date_cutoffs(accrual_postings, date, recv_end, pay_end):
     expect_recv = AGING_AR[:recv_end]
     expect_pay = AGING_AP[:pay_end]
     if 10 <= date.day < 12:
         # Take the 60 USD posting out of the invoice 510/6100 payable.
         expect_pay[-1] = expect_pay[-1]._replace(
             at_cost=testutil.Amount(expect_pay[-1].at_cost.number - 60),
+        )
     output = run_aging_report(accrual_postings, date)
     check_aging_ods(output, date, expect_recv, expect_pay)
 def test_aging_report_entity_consistency(accrual_postings):
     output = run_aging_report((
         post for post in accrual_postings
         if post.meta.get('rt-id') == 'rt:480'
         and post.units.number < 0
     ))
     check_aging_ods(output, None, [], [
         AgingRow.make_simple('2010-04-15', 'MultiPartyA', 125, 'rt:480/4800'),
         AgingRow.make_simple('2010-04-15', 'MultiPartyB', 125, 'rt:480/4800'),
     ])
 def test_aging_report_does_not_include_too_recent_postings(accrual_postings):
     # This date is after the Q3 posting, but too soon after for that to be
     # included in the aging report.
     date = datetime.date(2010, 10, 1)
     output = run_aging_report((
         post for post in accrual_postings
         if post.meta.get('rt-id') == 'rt:470'
     ), date)
     # Date+amount are both from the Q2 posting only.
     check_aging_ods(output, date, [
         AgingRow.make_simple('2010-06-15', 'GrantCo', 5500, 'rt:470/4700',
                              project='Development Grant'),
     ], [])
 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)
 @pytest.mark.parametrize('arglist,expect_invoice', [
     (['40'], 'rt:40/400'),
     (['44/440'], 'rt:44/440'),
 ])
 def test_output_payments_when_only_match(arglist, expect_invoice):
     retcode, output, errors = run_main(arglist)
     assert not errors.getvalue()
     assert retcode == 0
     check_output(output, [
         rf'^{re.escape(expect_invoice)}:$',
         r' outstanding since ',
     ])
 @pytest.mark.parametrize('arglist,expect_amount', [
     (['310'], 420),
     (['310/3120'], 220),
     (['entity=Vendor'], 420),
 ])
 def test_main_outgoing_report(arglist, expect_amount):
     retcode, output, errors = run_main(arglist)
     assert not errors.getvalue()
     assert retcode == 0
     rt_url = RTClient.DEFAULT_URL[:-9]
     rt_id_url = re.escape(f'<{rt_url}Ticket/Display.html?id=310>')
     contract_url = re.escape(f'<{rt_url}Ticket/Attachment/3120/3120/VendorContract.pdf>')
     check_output(output, [
         r'^REQUESTOR: Mx\. 310 <mx310@example\.org>$',
         rf'^TOTAL TO PAY: \${expect_amount}\.00$',
         r'^\s*2010-04-30\s',
         r'^\s+Expenses:Travel\s+220 USD$',
     ])
 @pytest.mark.parametrize('arglist', [
     ['-t', 'balance'],
     ['515/5150'],
     ['entity=MatchingProgram'],
 ])
 def test_main_balance_report(arglist):
     retcode, output, errors = run_main(arglist)
     assert not errors.getvalue()
     assert retcode == 0
     check_output(output, [
         r'\brt://ticket/515/attachments/5150:$',
         r'^\s+1,500\.00 USD outstanding since 2010-05-15$',
     ])
 def test_main_balance_report_because_no_rt_id():
     invoice = 'Invoices/2010StateRegistration.pdf'
     retcode, output, errors = run_main([invoice])
     assert not errors.getvalue()
     assert retcode == 0
     check_output(output, [
         rf'\b{re.escape(invoice)}:$',
         r'^\s+-50\.00 USD outstanding since 2010-06-20$',
     ])
 @pytest.mark.parametrize('arglist', [
     [],
     ['-t', 'aging', 'entity=Lawyer'],
 ])
 def test_main_aging_report(tmp_path, arglist):
     if arglist:
         recv_rows = [row for row in AGING_AR if 'Lawyer' in row.entity]
         pay_rows = [row for row in AGING_AP if 'Lawyer' in row.entity]
     else:
         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

 """Mock Beancount objects for testing"""
 # Copyright © 2020  Brett Smith
+#
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (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 datetime
 import itertools
 import re
 import beancount.core.amount as bc_amount
 import beancount.core.data as bc_data
 import beancount.loader as bc_loader
 import odf.element
 import odf.opendocument
 import odf.table
 from decimal import Decimal
 from pathlib import Path
 from typing import Any, Optional, NamedTuple
 from conservancy_beancount import books, rtutil
 EXTREME_FUTURE_DATE = datetime.date(datetime.MAXYEAR, 12, 30)
 FUTURE_DATE = datetime.date.today() + datetime.timedelta(days=365 * 99)
 FY_START_DATE = datetime.date(2020, 3, 1)
 FY_MID_DATE = datetime.date(2020, 9, 1)
 PAST_DATE = datetime.date(2000, 1, 1)
 TESTS_DIR = Path(__file__).parent
 def _ods_cell_value_type(cell):
     assert cell.tagName == 'table:table-cell'
     return cell.getAttribute('valuetype')
 def _ods_cell_value(cell):
     value_type = cell.getAttribute('valuetype')
     if value_type == 'currency' or value_type == 'float':
         return Decimal(cell.getAttribute('value'))
     elif value_type == 'date':
         return datetime.datetime.strptime(
             cell.getAttribute('datevalue'), '%Y-%m-%d',
         ).date()
     else:
         return cell.getAttribute('value')
 def _ods_elem_text(elem):
     if isinstance(elem, odf.element.Text):
         return elem.data
     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):
     stop = 0
     for seq in value_seqs:
         try:
             stop = max(stop, len(seq))
         except TypeError:
             pass
     return itertools.islice(
         zip(*(itertools.cycle(seq) for seq in value_seqs)),
         stop,
+    )
 def date_seq(date=FY_MID_DATE, step=1):
     while True:
         yield date
         date += datetime.timedelta(days=step)
 def parse_date(s, fmt='%Y-%m-%d'):
     return datetime.datetime.strptime(s, fmt).date()
 def test_path(s):
     if s is None:
         return s
     s = Path(s)
     if not s.is_absolute():
         s = TESTS_DIR / s
     return s
 def Amount(number, currency='USD'):
     return bc_amount.Amount(Decimal(number), currency)
 def Cost(number, currency='USD', date=FY_MID_DATE, label=None):
     return bc_data.Cost(Decimal(number), currency, date, label)
 def Posting(account, number,
             currency='USD', cost=None, price=None, flag=None,
             _post_type=bc_data.Posting, _meta_type=None, **meta):
     if cost is not None:
         cost = Cost(*cost)
     if not meta:
         meta = None
     elif _meta_type:
         meta = _meta_type(meta)
     return _post_type(
         account,
         Amount(number, currency),
         cost,
         price,
         flag,
         meta,
+    )
 def Transaction(date=FY_MID_DATE, flag='*', payee=None,
                 narration='', tags=None, links=None, postings=(),
                 **meta):
     if isinstance(date, str):
         date = parse_date(date)
     meta.setdefault('filename', '<test>')
     meta.setdefault('lineno', 0)
     real_postings = []
     for post in postings:
         try:
             post.account
         except AttributeError:
             if isinstance(post[-1], dict):
                 args = post[:-1]
                 kwargs = post[-1]
             else:
                 args = post
                 kwargs = {}
             post = Posting(*args, **kwargs)
         real_postings.append(post)
     return bc_data.Transaction(
         meta,
         date,
         flag,
         payee,
         narration,
         set(tags or ''),
         set(links or ''),
         real_postings,
+    )
 LINK_METADATA_STRINGS = {
     'Invoices/304321.pdf',
     'rt:123/456',
     'rt://ticket/234',
+}
 NON_LINK_METADATA_STRINGS = {
     '',
     ' ',
     '     ',
+}
 NON_STRING_METADATA_VALUES = [
     Decimal(5),
     FY_MID_DATE,
     Amount(50),
     Amount(500, None),
+]
 OPENING_EQUITY_ACCOUNTS = itertools.cycle([
     'Equity:Funds:Unrestricted',
     'Equity:Funds:Restricted',
     'Equity:OpeningBalance',
 ])
 class ODSCell:
     @classmethod
     def from_row(cls, row):
         return row.getElementsByType(odf.table.TableCell)
     @classmethod
     def from_sheet(cls, spreadsheet):
         for row in spreadsheet.getElementsByType(odf.table.TableRow):
             yield list(cls.from_row(row))
     @classmethod
     def from_ods_file(cls, path):
         ods = odf.opendocument.load(path)
         return cls.from_sheet(ods.spreadsheet)
 def OpeningBalance(acct=None, **txn_meta):
     if acct is None:
         acct = next(OPENING_EQUITY_ACCOUNTS)
     return Transaction(**txn_meta, postings=[
         ('Assets:Receivable:Accounts', 100),
         ('Assets:Receivable:Loans', 200),
         ('Liabilities:Payable:Accounts', -15),
         ('Liabilities:Payable:Vacation', -25),
         (acct, -260),
     ])
 class TestBooksLoader(books.Loader):
     def __init__(self, source):
         self.source = source
     def load_all(self, from_year=None):
         return bc_loader.load_file(self.source)
     def load_fy_range(self, from_fy, to_fy=None):
         return self.load_all()
 class TestConfig:
     def __init__(self, *,
                  books_path=None,
                  payment_threshold=0,
                  repo_path=None,
                  rt_client=None,
     ):
         if books_path is None:
             self._books_loader = None
         else:
             self._books_loader = TestBooksLoader(books_path)
         self._payment_threshold = Decimal(payment_threshold)
         self.repo_path = test_path(repo_path)
         self._rt_client = rt_client
         if rt_client is None:
             self._rt_wrapper = None
         else:
             self._rt_wrapper = rtutil.RT(rt_client)
     def books_loader(self):
         return self._books_loader
     def config_file_path(self):
         return test_path('userconfig/conservancy_beancount/config.ini')
     def payment_threshold(self):
         return self._payment_threshold
     def repository_path(self):
         return self.repo_path
     def rt_client(self):
         return self._rt_client
     def rt_wrapper(self):

0 comments (0 inline, 0 general)