"""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 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,

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 setup_logger(logger: Union[str, logging.Logger]='',

                 stream: TextIO=sys.stderr,

                 fmt: str='%(name)s: %(levelname)s: %(message)s',

) -> logging.Logger:

    if isinstance(logger, str):

        logger = logging.getLogger(logger)

    formatter = logging.Formatter(fmt)

    handler = logging.StreamHandler(stream)

    handler.setFormatter(formatter)

    logger.addHandler(handler)

    return logger

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)

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:

    args = parse_arguments(arglist)

    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.

        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

if __name__ == '__main__':

        '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': [

        ],

    },

)

# 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 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('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('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(

    )

@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>"))

    # 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

    retcode, output, errors = run_main(arglist, config)

    assert retcode > 16

    assert (retcode - 16) & error_flags

    assert not output.getvalue()

@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 ',

        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():

        r':[01]: +no books to load in configuration\b',

    ])

@pytest.mark.parametrize('arglist', [

    ['499'],

    ['505/99999'],

    ['entity=NonExistent'],

])

    ])

    config = testutil.TestConfig(

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

    )

    ])

    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_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):