Changeset - 9ae974009b13
Parent rev.
Child rev.
[Not reviewed]
0 2 0
Brett Smith - 4 years ago 2020-06-27 20:31:27
brettcsmith@brettcsmith.org
fund: Add outstanding balances to text fund report.
2 files changed with 60 insertions and 33 deletions:
conservancy_beancount/reports/fund.py
13
3
tests/test_reports_fund.py
47
30
0 comments (0 inline, 0 general)
conservancy_beancount/reports/fund.py
➞
Show inline comments
...
 
@@ -35,192 +35,203 @@ Query a specific restricted fund and get a quick balance on the terminal::
 
# 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 collections


			




	
	
	
		
 
import datetime


			




	
	
	
		
 
import enum


			




	
	
	
		
 
import locale


			




	
	
	
		
 
import logging


			




	
	
	
		
 
import sys


			




	
	
	
		
 

			




	
	
	
		
 
from typing import (


			




	
	
	
		
 
    Dict,


			




	
	
	
		
 
    Iterable,


			




	
	
	
		
 
    Iterator,


			




	
	
	
		
 
    Mapping,


			




	
	
	
		
 
    Optional,


			




	
	
	
		
 
    Sequence,


			




	
	
	
		
 
    TextIO,


			




	
	
	
		
 
    Tuple,


			




	
	
	
		
 
)


			




	
	
	
		
 
from ..beancount_types import (


			




	
	
	
		
 
    MetaValue,


			




	
	
	
		
 
)


			




	
	
	
		
 

			




	
	
	
		
 
from pathlib import Path


			




	
	
	
		
 

			




	
	
	
		
 
import odf.table  # type:ignore[import]


			




	
	
	
		
 

			




	
	
	
		
 
from beancount.parser import printer as bc_printer


			




	
	
	
		
 

			




	
	
	
		
 
from . import core


			




	
	
	
		
 
from .. import books


			




	
	
	
		
 
from .. import cliutil


			




	
	
	
		
 
from .. import config as configmod


			




	
	
	
		
 
from .. import data


			




	
	
	
		
 

			




	
	
	
		
 
AccountsMap = Mapping[data.Account, core.PeriodPostings]


			




	
	
	
		
 
FundPosts = Tuple[MetaValue, AccountsMap]


			




	
	
	
		
 

			




	
	
	
		
 
BALANCE_ACCOUNTS = ['Equity', 'Income', 'Expenses']


			




	
	
	
		
 
EQUITY_ACCOUNTS = ['Equity', 'Income', 'Expenses']


			




	
	
	
		
 
INFO_ACCOUNTS = [


			




	
	
	
		
 
    'Assets:Receivable',


			




	
	
	
		
 
    'Assets:Prepaid',


			




	
	
	
		
 
    'Liabilities:Payable',


			




	
	
	
		
 
    'Liabilities:UnearnedIncome',


			




	
	
	
		
 
]


			




	
	
	
		
 
PROGNAME = 'fund-report'


			




	
	
	
		
 
UNRESTRICTED_FUND = 'Conservancy'


			




	
	
	
		
 
logger = logging.getLogger('conservancy_beancount.reports.fund')


			




	
	
	
		
 

			




	
	
	
		
 
class ODSReport(core.BaseODS[FundPosts, None]):


			




	
	
	
		
 
    def __init__(self, start_date: datetime.date, stop_date: datetime.date) -> None:


			




	
	
	
		
 
        super().__init__()


			




	
	
	
		
 
        self.start_date = start_date


			




	
	
	
		
 
        self.stop_date = stop_date


			




	
	
	
		
 
        self.unrestricted: AccountsMap = {}


			




	
	
	
		
 

			




	
	
	
		
 
    def section_key(self, row: FundPosts) -> None:


			




	
	
	
		
 
        return None


			




	
	
	
		
 

			




	
	
	
		
 
    def start_spreadsheet(self) -> None:


			




	
	
	
		
 
        self.use_sheet("Fund Report")


			




	
	
	
		
 
        for width in [2.5, 1.5, 1.2, 1.2, 1.2, 1.5]:


			




	
	
	
		
 
            col_style = self.column_style(width)


			




	
	
	
		
 
            self.sheet.addElement(odf.table.TableColumn(stylename=col_style))


			




	
	
	
		
 
        center_bold = self.merge_styles(self.style_centertext, self.style_bold)


			




	
	
	
		
 
        self.add_row(


			




	
	
	
		
 
            self.string_cell(


			




	
	
	
		
 
                "Fund", stylename=self.merge_styles(self.style_endtext, self.style_bold),


			




	
	
	
		
 
            ),


			




	
	
	
		
 
            self.multiline_cell(["Balance as of", self.start_date.isoformat()],


			




	
	
	
		
 
                                stylename=center_bold),


			




	
	
	
		
 
            self.string_cell("Income", stylename=center_bold),


			




	
	
	
		
 
            self.string_cell("Expenses", stylename=center_bold),


			




	
	
	
		
 
            self.multiline_cell(["Realized", "Gain/Loss"], stylename=center_bold),


			




	
	
	
		
 
            self.multiline_cell(["Balance as of", self.stop_date.isoformat()],


			




	
	
	
		
 
                                stylename=center_bold),


			




	
	
	
		
 
        )


			




	
	
	
		
 
        self.lock_first_row()


			




	
	
	
		
 
        self.add_row()


			




	
	
	
		
 
        self.add_row(self.string_cell(


			




	
	
	
		
 
            f"Fund Report From {self.start_date.isoformat()} To {self.stop_date.isoformat()}",


			




	
	
	
		
 
            stylename=center_bold,


			




	
	
	
		
 
            numbercolumnsspanned=6,


			




	
	
	
		
 
        ))


			




	
	
	
		
 
        self.add_row()


			




	
	
	
		
 

			




	
	
	
		
 
    def _row_balances(self, accounts_map: AccountsMap) -> Iterable[core.Balance]:


			




	
	
	
		
 
        acct_order = ['Income', 'Expenses', 'Equity']


			




	
	
	
		
 
        key_order = [core.OPENING_BALANCE_NAME, *acct_order, core.ENDING_BALANCE_NAME]


			




	
	
	
		
 
        balances: Dict[str, core.Balance] = {key: core.MutableBalance() for key in key_order}


			




	
	
	
		
 
        for acct_s, balance in core.account_balances(accounts_map, acct_order):


			




	
	
	
		
 
            if acct_s in balances:


			




	
	
	
		
 
                balances[acct_s] = balance


			




	
	
	
		
 
            else:


			




	
	
	
		
 
                acct_root, _, _ = acct_s.partition(':')


			




	
	
	
		
 
                balances[acct_root] += balance


			




	
	
	
		
 
        for key in key_order:


			




	
	
	
		
 
            if key == 'Expenses':


			




	
	
	
		
 
                yield balances[key]


			




	
	
	
		
 
            else:


			




	
	
	
		
 
                yield -balances[key]


			




	
	
	
		
 

			




	
	
	
		
 
    def write_row(self, row: FundPosts) -> None:


			




	
	
	
		
 
        fund, accounts_map = row


			




	
	
	
		
 
        if fund == UNRESTRICTED_FUND:


			




	
	
	
		
 
            assert not self.unrestricted


			




	
	
	
		
 
            self.unrestricted = accounts_map


			




	
	
	
		
 
            return


			




	
	
	
		
 
        self.add_row(


			




	
	
	
		
 
            self.string_cell(fund, stylename=self.style_endtext),


			




	
	
	
		
 
            *(self.balance_cell(bal) for bal in self._row_balances(accounts_map)),


			




	
	
	
		
 
        )


			




	
	
	
		
 

			




	
	
	
		
 
    def write(self, rows: Iterable[FundPosts]) -> None:


			




	
	
	
		
 
        super().write(rows)


			




	
	
	
		
 
        if self.unrestricted:


			




	
	
	
		
 
            self.add_row()


			




	
	
	
		
 
            self.write_row(("Unrestricted", self.unrestricted))


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TextReport:


			




	
	
	
		
 
    def __init__(self,


			




	
	
	
		
 
                 start_date: datetime.date,


			




	
	
	
		
 
                 stop_date: datetime.date,


			




	
	
	
		
 
                 out_file: TextIO) -> None:


			




	
	
	
		
 
        self.start_date = start_date


			




	
	
	
		
 
        self.stop_date = stop_date


			




	
	
	
		
 
        self.out_file = out_file


			




	
	
	
		
 

			




	
	
	
		
 
    def _account_balances(self,


			




	
	
	
		
 
                          fund: str,


			




	
	
	
		
 
                          account_map: AccountsMap,


			




	
	
	
		
 
    ) -> Iterator[Tuple[str, Sequence[str]]]:


			




	
	
	
		
 
        total_fmt = f'{fund} balance as of {{}}'


			




	
	
	
		
 
        for acct_s, balance in core.account_balances(account_map, BALANCE_ACCOUNTS):


			




	
	
	
		
 
        for acct_s, balance in core.account_balances(account_map, EQUITY_ACCOUNTS):


			




	
	
	
		
 
            if acct_s is core.OPENING_BALANCE_NAME:


			




	
	
	
		
 
                acct_s = total_fmt.format(self.start_date.isoformat())


			




	
	
	
		
 
            elif acct_s is core.ENDING_BALANCE_NAME:


			




	
	
	
		
 
                acct_s = total_fmt.format(self.stop_date.isoformat())


			




	
	
	
		
 
            yield acct_s, (-balance).format(None, sep='\0').split('\0')


			




	
	
	
		
 
        for _, account in core.sort_and_filter_accounts(account_map, INFO_ACCOUNTS):


			




	
	
	
		
 
            balance = account_map[account].stop_bal


			




	
	
	
		
 
            if not balance.is_zero():


			




	
	
	
		
 
                balance = core.normalize_amount_func(account)(balance)


			




	
	
	
		
 
                yield account, balance.format(None, sep='\0').split('\0')


			




	
	
	
		
 

			




	
	
	
		
 
    def write(self, rows: Iterable[FundPosts]) -> None:


			




	
	
	
		
 
        output = [


			




	
	
	
		
 
            line


			




	
	
	
		
 
            for fund, account_map in rows


			




	
	
	
		
 
            for line in self._account_balances(fund, account_map)


			




	
	
	
		
 
        ]


			




	
	
	
		
 
        acct_width = max(len(acct_s) for acct_s, _ in output) + 2


			




	
	
	
		
 
        bal_width = max(len(s) for _, bal_s in output for s in bal_s)


			




	
	
	
		
 
        bal_width = max(bal_width, 8)


			




	
	
	
		
 
        line_fmt = f'{{:>{acct_width}}}  {{:>{bal_width}}}'


			




	
	
	
		
 
        print(line_fmt.replace('{:>', '{:^').format("ACCOUNT", "BALANCE"),


			




	
	
	
		
 
              file=self.out_file)


			




	
	
	
		
 
        fund_start = f' balance as of {self.start_date.isoformat()}'


			




	
	
	
		
 
        for acct_s, bal_seq in output:


			




	
	
	
		
 
            if acct_s.endswith(fund_start):


			




	
	
	
		
 
                print(line_fmt.format('―' * acct_width, '―' * bal_width),


			




	
	
	
		
 
                      file=self.out_file)


			




	
	
	
		
 
            bal_iter = iter(bal_seq)


			




	
	
	
		
 
            print(line_fmt.format(acct_s, next(bal_iter)), file=self.out_file)


			




	
	
	
		
 
            for bal_s in bal_iter:


			




	
	
	
		
 
                print(line_fmt.format('', bal_s), file=self.out_file)


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class ReportType(enum.Enum):


			




	
	
	
		
 
    TEXT = TextReport


			




	
	
	
		
 
    ODS = ODSReport


			




	
	
	
		
 
    TXT = TEXT


			




	
	
	
		
 
    SPREADSHEET = ODS


			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod


			




	
	
	
		
 
    def from_arg(cls, s: str) -> 'ReportType':


			




	
	
	
		
 
        try:


			




	
	
	
		
 
            return cls[s.upper()]


			




	
	
	
		
 
        except KeyError:


			




	
	
	
		
 
            raise ValueError(f"no report type matches {s!r}") from None


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class ReturnFlag(enum.IntFlag):


			




	
	
	
		
 
    LOAD_ERRORS = 1


			




	
	
	
		
 
    NOTHING_TO_REPORT = 8


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:


			




	
	
	
		
 
    parser = argparse.ArgumentParser(prog=PROGNAME)


			




	
	
	
		
 
    cliutil.add_version_argument(parser)


			




	
	
	
		
 
    parser.add_argument(


			




	
	
	
		
 
        '--begin', '--start', '-b',


			




	...
	
		
 
@@ -274,86 +285,85 @@ metadata to match. A single ticket number is a shortcut for


			




	
	
	
		
 
        else:


			




	
	
	
		
 
            args.report_type = ReportType.ODS


			




	
	
	
		
 
    return args


			




	
	
	
		
 

			




	
	
	
		
 
def diff_year(date: datetime.date, diff: int) -> datetime.date:


			




	
	
	
		
 
    new_year = date.year + diff


			




	
	
	
		
 
    try:


			




	
	
	
		
 
        return date.replace(year=new_year)


			




	
	
	
		
 
    except ValueError:


			




	
	
	
		
 
        # The original date is Feb 29, which doesn't exist in the new year.


			




	
	
	
		
 
        if diff < 0:


			




	
	
	
		
 
            return datetime.date(new_year, 2, 28)


			




	
	
	
		
 
        else:


			




	
	
	
		
 
            return datetime.date(new_year, 3, 1)


			




	
	
	
		
 

			




	
	
	
		
 
def main(arglist: Optional[Sequence[str]]=None,


			




	
	
	
		
 
         stdout: TextIO=sys.stdout,


			




	
	
	
		
 
         stderr: TextIO=sys.stderr,


			




	
	
	
		
 
         config: Optional[configmod.Config]=None,


			




	
	
	
		
 
) -> int:


			




	
	
	
		
 
    args = parse_arguments(arglist)


			




	
	
	
		
 
    cliutil.set_loglevel(logger, args.loglevel)


			




	
	
	
		
 
    if config is None:


			




	
	
	
		
 
        config = configmod.Config()


			




	
	
	
		
 
        config.load_file()


			




	
	
	
		
 

			




	
	
	
		
 
    if args.stop_date is None:


			




	
	
	
		
 
        if args.start_date is None:


			




	
	
	
		
 
            args.stop_date = datetime.date.today()


			




	
	
	
		
 
        else:


			




	
	
	
		
 
            args.stop_date = diff_year(args.start_date, 1)


			




	
	
	
		
 
    if args.start_date is None:


			




	
	
	
		
 
        args.start_date = diff_year(args.stop_date, -1)


			




	
	
	
		
 

			




	
	
	
		
 
    returncode = 0


			




	
	
	
		
 
    books_loader = config.books_loader()


			




	
	
	
		
 
    if books_loader is None:


			




	
	
	
		
 
        entries, load_errors, _ = books.Loader.load_none(config.config_file_path())


			




	
	
	
		
 
    else:


			




	
	
	
		
 
        entries, load_errors, _ = books_loader.load_fy_range(args.start_date, args.stop_date)


			




	
	
	
		
 
    for error in load_errors:


			




	
	
	
		
 
        bc_printer.print_error(error, file=stderr)


			




	
	
	
		
 
        returncode |= ReturnFlag.LOAD_ERRORS


			




	
	
	
		
 

			




	
	
	
		
 
    postings = (


			




	
	
	
		
 
        post


			




	
	
	
		
 
        for post in data.Posting.from_entries(entries)


			




	
	
	
		
 
        if post.meta.date < args.stop_date


			




	
	
	
		
 
        and post.account.is_under(*BALANCE_ACCOUNTS)


			




	
	
	
		
 
    )


			




	
	
	
		
 
    for search_term in args.search_terms:


			




	
	
	
		
 
        postings = search_term.filter_postings(postings)


			




	
	
	
		
 
    fund_postings = {


			




	
	
	
		
 
        key: related


			




	
	
	
		
 
        for key, related in core.RelatedPostings.group_by_meta(postings, 'project')


			




	
	
	
		
 
        if isinstance(key, str)


			




	
	
	
		
 
    }


			




	
	
	
		
 
    period_cls = core.PeriodPostings.with_start_date(args.start_date)


			




	
	
	
		
 
    fund_map = collections.OrderedDict(


			




	
	
	
		
 
        (fund, dict(period_cls.group_by_account(fund_postings[fund])))


			




	
	
	
		
 
        for fund in sorted(fund_postings, key=lambda s: locale.strxfrm(s.casefold()))


			




	
	
	
		
 
    )


			




	
	
	
		
 
    if not fund_map:


			




	
	
	
		
 
        logger.warning("no matching postings found to report")


			




	
	
	
		
 
        returncode |= ReturnFlag.NOTHING_TO_REPORT


			




	
	
	
		
 
    elif args.report_type is ReportType.TEXT:


			




	
	
	
		
 
        out_file = cliutil.text_output(args.output_file, stdout)


			




	
	
	
		
 
        report = TextReport(args.start_date, args.stop_date, out_file)


			




	
	
	
		
 
        report.write(fund_map.items())


			




	
	
	
		
 
    else:


			




	
	
	
		
 
        ods_report = ODSReport(args.start_date, args.stop_date)


			




	
	
	
		
 
        ods_report.write(fund_map.items())


			




	
	
	
		
 
        if args.output_file is None:


			




	
	
	
		
 
            out_dir_path = config.repository_path() or Path()


			




	
	
	
		
 
            args.output_file = out_dir_path / 'FundReport_{}_{}.ods'.format(


			




	
	
	
		
 
                args.start_date.isoformat(), args.stop_date.isoformat(),


			




	
	
	
		
 
            )


			




	
	
	
		
 
            logger.info("Writing report to %s", args.output_file)


			




	
	
	
		
 
        ods_file = cliutil.bytes_output(args.output_file, stdout)


			




	
	
	
		
 
        ods_report.save_file(ods_file)


			




	
	
	
		
 
    return 0 if returncode == 0 else 16 + returncode


			




	
	
	
		
 

			




	
	
	
		
 
entry_point = cliutil.make_entry_point(__name__, PROGNAME)


			




	
	
	
		
 

			




	
	
	
		
 
if __name__ == '__main__':


			




	
	
	
		
 
    exit(entry_point())


			




        

    


    
    

    

        

                

                    tests/test_reports_fund.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
"""test_reports_fund.py - Unit tests for fund report"""


			




	
	
	
		
 
# 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 collections


			




	
	
	
		
 
import copy


			




	
	
	
		
 
import datetime


			




	
	
	
		
 
import io


			




	
	
	
		
 
import itertools


			




	
	
	
		
 

			




	
	
	
		
 
import pytest


			




	
	
	
		
 

			




	
	
	
		
 
from . import testutil


			




	
	
	
		
 

			




	
	
	
		
 
import babel.numbers


			




	
	
	
		
 
import odf.opendocument


			




	
	
	
		
 
import odf.table


			




	
	
	
		
 

			




	
	
	
		
 
from beancount import loader as bc_loader


			




	
	
	
		
 
from conservancy_beancount import data


			




	
	
	
		
 
from conservancy_beancount.reports import core


			




	
	
	
		
 
from conservancy_beancount.reports import fund


			




	
	
	
		
 

			




	
	
	
		
 
from decimal import Decimal


			




	
	
	
		
 

			




	
	
	
		
 
_ledger_load = bc_loader.load_file(testutil.test_path('books/fund.beancount'))


			




	
	
	
		
 
START_DATE = datetime.date(2018, 3, 1)


			




	
	
	
		
 
MID_DATE = datetime.date(2019, 3, 1)


			




	
	
	
		
 
STOP_DATE = datetime.date(2020, 3, 1)


			




	
	
	
		
 

			




	
	
	
		
 
EQUITY_ROOT_ACCOUNTS = ('Expenses:', 'Equity:', 'Income:')


			




	
	
	
		
 

			




	
	
	
		
 
OPENING_BALANCES = {


			




	
	
	
		
 
    'Alpha': 3000,


			




	
	
	
		
 
    'Bravo': 2000,


			




	
	
	
		
 
    'Charlie': 1000,


			




	
	
	
		
 
    'Conservancy': 4000,


			




	
	
	
		
 
}


			




	
	
	
		
 

			




	
	
	
		
 
BALANCES_BY_YEAR = {


			




	
	
	
		
 
    ('Conservancy', 2018): [


			




	
	
	
		
 
        ('Income:Other', 40),


			




	
	
	
		
 
        ('Expenses:Other', -4),


			




	
	
	
		
 
        ('Assets:Receivable:Accounts', 40),


			




	
	
	
		
 
        ('Liabilities:Payable:Accounts', 4),


			




	
	
	
		
 
    ],


			




	
	
	
		
 
    ('Conservancy', 2019): [


			




	
	
	
		
 
        ('Income:Other', 42),


			




	
	
	
		
 
        ('Expenses:Other', Decimal('-4.20')),


			




	
	
	
		
 
        ('Equity:Realized:CurrencyConversion', Decimal('6.20')),


			




	
	
	
		
 
        ('Assets:Receivable:Accounts', -40),


			




	
	
	
		
 
        ('Liabilities:Payable:Accounts', -4),


			




	
	
	
		
 
    ],


			




	
	
	
		
 
    ('Alpha', 2018): [


			




	
	
	
		
 
        ('Income:Other', 60),


			




	
	
	
		
 
        ('Liabilities:UnearnedIncome', 30),


			




	
	
	
		
 
        ('Assets:Prepaid:Expenses', 20),


			




	
	
	
		
 
    ],


			




	
	
	
		
 
    ('Alpha', 2019): [


			




	
	
	
		
 
        ('Income:Other', 30),


			




	
	
	
		
 
        ('Expenses:Other', -26),


			




	
	
	
		
 
        ('Assets:Prepaid:Expenses', -20),


			




	
	
	
		
 
        ('Liabilities:UnearnedIncome', -30),


			




	
	
	
		
 
    ],


			




	
	
	
		
 
    ('Bravo', 2018): [


			




	
	
	
		
 
        ('Expenses:Other', -20),


			




	
	
	
		
 
    ],


			




	
	
	
		
 
    ('Bravo', 2019): [


			




	
	
	
		
 
        ('Income:Other', 200),


			




	
	
	
		
 
    ],


			




	
	
	
		
 
}


			




	
	
	
		
 

			




	
	
	
		
 
@pytest.fixture


			




	
	
	
		
 
def fund_entries():


			




	
	
	
		
 
    return copy.deepcopy(_ledger_load[0])


			




	
	
	
		
 

			




	
	
	
		
 
def fund_postings(entries, project, stop_date):


			




	
	
	
		
 
    return (


			




	
	
	
		
 
        post for post in data.Posting.from_entries(entries)


			




	
	
	
		
 
        if post.meta.date < stop_date


			




	
	
	
		
 
        and post.account.is_under('Equity', 'Income', 'Expenses')


			




	
	
	
		
 
        and post.meta.get('project') == project


			




	
	
	
		
 
    )


			




	
	
	
		
 

			




	
	
	
		
 
def split_text_lines(output):


			




	
	
	
		
 
    for line in output:


			




	
	
	
		
 
        account, amount = line.rsplit(None, 1)


			




	
	
	
		
 
        yield account.strip(), amount


			




	
	
	
		
 

			




	
	
	
		
 
def format_amount(amount, currency='USD'):


			




	
	
	
		
 
    return babel.numbers.format_currency(


			




	
	
	
		
 
        amount, currency, format_type='accounting',


			




	
	
	
		
 
    )


			




	
	
	
		
 

			




	
	
	
		
 
def check_text_balances(actual, expected, *expect_accounts):


			




	
	
	
		
 
    balance = Decimal()


			




	
	
	
		
 
    for expect_account in expect_accounts:


			




	
	
	
		
 
        expect_amount = expected[expect_account]


			




	
	
	
		
 
        if expect_amount:


			




	
	
	
		
 
            actual_account, actual_amount = next(actual)


			




	
	
	
		
 
            assert actual_account == expect_account


			




	
	
	
		
 
            assert actual_amount == format_amount(expect_amount)


			




	
	
	
		
 
            balance += expect_amount


			




	
	
	
		
 
    return balance


			




	
	
	
		
 

			




	
	
	
		
 
def check_text_report(output, project, start_date, stop_date):


			




	
	
	
		
 
    _, _, project = project.rpartition('=')


			




	
	
	
		
 
    balance_amount = Decimal(OPENING_BALANCES[project])


			




	
	
	
		
 
    expected = collections.defaultdict(Decimal)


			




	
	
	
		
 
    for year in range(2018, stop_date.year):


			




	
	
	
		
 
        try:


			




	
	
	
		
 
            amounts = BALANCES_BY_YEAR[(project, year)]


			




	
	
	
		
 
        except KeyError:


			




	
	
	
		
 
            pass


			




	
	
	
		
 
        else:


			




	
	
	
		
 
            for account, amount in amounts:


			




	
	
	
		
 
                if year < start_date.year:


			




	
	
	
		
 
                if year < start_date.year and account.startswith(EQUITY_ROOT_ACCOUNTS):


			




	
	
	
		
 
                    balance_amount += amount


			




	
	
	
		
 
                else:


			




	
	
	
		
 
                    expected[account] += amount


			




	
	
	
		
 
    expected.default_factory = None


			




	
	
	
		
 
    actual = split_text_lines(output)


			




	
	
	
		
 
    next(actual); next(actual)  # Discard headers


			




	
	
	
		
 
    open_acct, open_amt = next(actual)


			




	
	
	
		
 
    assert open_acct == "{} balance as of {}".format(


			




	
	
	
		
 
        project, start_date.isoformat(),


			




	
	
	
		
 
    )


			




	
	
	
		
 
    assert open_amt == format_amount(balance_amount)


			




	
	
	
		
 
    for expect_account in [


			




	
	
	
		
 
            'Equity:Realized:CurrencyConversion',


			




	
	
	
		
 
            'Income:Other',


			




	
	
	
		
 
            'Expenses:Other',


			




	
	
	
		
 
    ]:


			




	
	
	
		
 
        try:


			




	
	
	
		
 
            expect_amount = expected[expect_account]


			




	
	
	
		
 
        except KeyError:


			




	
	
	
		
 
            continue


			




	
	
	
		
 
        else:


			




	
	
	
		
 
            actual_account, actual_amount = next(actual)


			




	
	
	
		
 
            assert actual_account == expect_account


			




	
	
	
		
 
            assert actual_amount == format_amount(expect_amount)


			




	
	
	
		
 
            balance_amount += expect_amount


			




	
	
	
		
 
    balance_amount += check_text_balances(


			




	
	
	
		
 
        actual, expected,


			




	
	
	
		
 
        'Equity:Realized:CurrencyConversion',


			




	
	
	
		
 
        'Income:Other',


			




	
	
	
		
 
        'Expenses:Other',


			




	
	
	
		
 
    )


			




	
	
	
		
 
    end_acct, end_amt = next(actual)


			




	
	
	
		
 
    assert end_acct == "{} balance as of {}".format(


			




	
	
	
		
 
        project, stop_date.isoformat(),


			




	
	
	
		
 
    )


			




	
	
	
		
 
    assert end_amt == format_amount(balance_amount)


			




	
	
	
		
 
    balance_amount += check_text_balances(


			




	
	
	
		
 
        actual, expected,


			




	
	
	
		
 
        'Assets:Receivable:Accounts',


			




	
	
	
		
 
        'Assets:Prepaid:Expenses',


			




	
	
	
		
 
        'Liabilities:Payable:Accounts',


			




	
	
	
		
 
        'Liabilities:UnearnedIncome',


			




	
	
	
		
 
    )


			




	
	
	
		
 
    assert next(actual, None) is None


			




	
	
	
		
 

			




	
	
	
		
 
def check_ods_report(ods, start_date, stop_date):


			




	
	
	
		
 
    account_bals = collections.OrderedDict((key, {


			




	
	
	
		
 
        'opening': Decimal(amount),


			




	
	
	
		
 
        'Income': Decimal(0),


			




	
	
	
		
 
        'Expenses': Decimal(0),


			




	
	
	
		
 
        'Equity': Decimal(0),


			




	
	
	
		
 
        'Equity:Realized': Decimal(0),


			




	
	
	
		
 
        'Assets:Receivable': Decimal(0),


			




	
	
	
		
 
        'Assets:Prepaid': Decimal(0),


			




	
	
	
		
 
        'Liabilities:Payable': Decimal(0),


			




	
	
	
		
 
        'Liabilities': Decimal(0),  # UnearnedIncome


			




	
	
	
		
 
    }) for key, amount in sorted(OPENING_BALANCES.items()))


			




	
	
	
		
 
    for fund, year in itertools.product(account_bals, range(2018, stop_date.year)):


			




	
	
	
		
 
        try:


			




	
	
	
		
 
            amounts = BALANCES_BY_YEAR[(fund, year)]


			




	
	
	
		
 
        except KeyError:


			




	
	
	
		
 
            pass


			




	
	
	
		
 
        else:


			




	
	
	
		
 
            for account, amount in amounts:


			




	
	
	
		
 
                if year < start_date.year:


			




	
	
	
		
 
                if year < start_date.year and account.startswith(EQUITY_ROOT_ACCOUNTS):


			




	
	
	
		
 
                    acct_key = 'opening'


			




	
	
	
		
 
                else:


			




	
	
	
		
 
                    acct_key, _, _ = account.partition(':')


			




	
	
	
		
 
                    acct_key, _, _ = account.rpartition(':')


			




	
	
	
		
 
                account_bals[fund][acct_key] += amount


			




	
	
	
		
 
    account_bals['Unrestricted'] = account_bals.pop('Conservancy')


			




	
	
	
		
 
    for row in ods.getElementsByType(odf.table.TableRow):


			




	
	
	
		
 
        cells = iter(testutil.ODSCell.from_row(row))


			




	
	
	
		
 
        try:


			




	
	
	
		
 
            fund = next(cells).firstChild.text


			




	
	
	
		
 
        except (AttributeError, StopIteration):


			




	
	
	
		
 
            fund = None


			




	
	
	
		
 
        if fund in account_bals:


			




	
	
	
		
 
            balances = account_bals.pop(fund)


			




	
	
	
		
 
            assert next(cells).value == balances['opening']


			




	
	
	
		
 
            assert next(cells).value == balances['Income']


			




	
	
	
		
 
            assert next(cells).value == -balances['Expenses']


			




	
	
	
		
 
            if balances['Equity']:


			




	
	
	
		
 
                assert next(cells).value == balances['Equity']


			




	
	
	
		
 
            if balances['Equity:Realized']:


			




	
	
	
		
 
                assert next(cells).value == balances['Equity:Realized']


			




	
	
	
		
 
            else:


			




	
	
	
		
 
                assert not next(cells).value


			




	
	
	
		
 
            assert next(cells).value == sum(balances.values())


			




	
	
	
		
 
            assert next(cells).value == sum(balances[key] for key in [


			




	
	
	
		
 
                'opening', 'Income', 'Expenses', 'Equity:Realized',


			




	
	
	
		
 
            ])


			




	
	
	
		
 
    assert not account_bals, "did not see all funds in report"


			




	
	
	
		
 

			




	
	
	
		
 
def run_main(out_type, arglist, config=None):


			




	
	
	
		
 
    if config is None:


			




	
	
	
		
 
        config = testutil.TestConfig(


			




	
	
	
		
 
            books_path=testutil.test_path('books/fund.beancount'),


			




	
	
	
		
 
        )


			




	
	
	
		
 
    arglist.insert(0, '--output-file=-')


			




	
	
	
		
 
    output = out_type()


			




	
	
	
		
 
    errors = io.StringIO()


			




	
	
	
		
 
    retcode = fund.main(arglist, output, errors, config)


			




	
	
	
		
 
    output.seek(0)


			




	
	
	
		
 
    return retcode, output, errors


			




	
	
	
		
 

			




	
	
	
		
 
@pytest.mark.parametrize('project,start_date,stop_date', [


			




	
	
	
		
 
    ('Conservancy', START_DATE, STOP_DATE),


			




	
	
	
		
 
    ('project=Conservancy', MID_DATE, STOP_DATE),


			




	
	
	
		
 
    ('Conservancy', START_DATE, MID_DATE),


			




	
	
	
		
 
    ('Alpha', START_DATE, STOP_DATE),


			




	
	
	
		
 
    ('project=Alpha', MID_DATE, STOP_DATE),


			




	
	
	
		
 
    ('Alpha', START_DATE, MID_DATE),


			




	
	
	
		
 
    ('Bravo', START_DATE, STOP_DATE),


			




	
	
	
		
 
    ('project=Bravo', MID_DATE, STOP_DATE),


			




	
	
	
		
 
    ('Bravo', START_DATE, MID_DATE),


			




	
	
	
		
 
    ('project=Charlie', START_DATE, STOP_DATE),


			




	
	
	
		
 
])


			




	
	
	
		
 
def test_text_report(project, start_date, stop_date):


			




	
	
	
		
 
    retcode, output, errors = run_main(io.StringIO, [


			




	
	
	
		
 
        '-b', start_date.isoformat(), '-e', stop_date.isoformat(), project,


			




	
	
	
		
 
    ])


			




	
	
	
		
 
    assert not errors.getvalue()


			




	
	
	
		
 
    assert retcode == 0


			




	
	
	
		
 
    check_text_report(output, project, start_date, stop_date)


			




	
	
	
		
 

			




	
	
	
		
 
@pytest.mark.parametrize('start_date,stop_date', [


			




	
	
	
		
 
    (START_DATE, STOP_DATE),


			




	
	
	
		
 
    (MID_DATE, STOP_DATE),


			




	
	
	
		
 
    (START_DATE, MID_DATE),


			




	
	
	
		
 
])


			




	
	
	
		
 
def test_ods_report(start_date, stop_date):


			




	
	
	
		
 
    retcode, output, errors = run_main(io.BytesIO, [


			




	
	
	
		
 
        '--begin', start_date.isoformat(), '--end', stop_date.isoformat(),


			




	
	
	
		
 
    ])


			




	
	
	
		
 
    assert not errors.getvalue()


			




	
	
	
		
 
    assert retcode == 0


			




	
	
	
		
 
    ods = odf.opendocument.load(output)


			




	
	
	
		
 
    check_ods_report(ods, start_date, stop_date)


			




	
	
	
		
 

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







    


    

  


    








    
        Server instance: kapok.sfconservancy.org-23738
    
    
        This site is powered by
            Kallithea 0.6.3,
        which is
        © 2010–2020 by various authors & licensed under GPLv3.
            – Support
        – Privacy Policy