"""books - Tools for loading the books"""

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

import datetime

import os

from pathlib import Path

from beancount import loader as bc_loader

from typing import (

    Any,

    Iterable,

    Iterator,

    Mapping,

    NamedTuple,

    Optional,

    Union,

)

from .beancount_types import (

    LoadResult,

)

PathLike = Union[str, Path]

Year = Union[int, datetime.date]

@contextlib.contextmanager

def workdir(path: PathLike) -> Iterator[Path]:

    old_dir = os.getcwd()

    os.chdir(path)

    try:

        yield Path(old_dir)

    finally:

        os.chdir(old_dir)

class FiscalYear(NamedTuple):

    month: int = 3

    day: int = 1

    def for_date(self, date: Optional[datetime.date]=None) -> int:

        if date is None:

            date = datetime.date.today()

        if (date.month, date.day) < self:

            return date.year - 1

        else:

            return date.year

    def range(self, from_fy: Year, to_fy: Optional[Year]=None) -> Iterable[int]:

        """Return a range of fiscal years

        Both arguments can be either a year (represented as an integer) or a

        date. Dates will be converted into a year by calling for_date() on

        them.

        If the first argument is negative or below 1000, it will be treated as

        an offset. You'll get a range of fiscal years between the second

        argument offset by this amount.

        If the second argument is omitted, it defaults to the current fiscal

        year.

        Note that unlike normal Python ranges, these ranges include the final

        fiscal year.

        Examples:

          range(2015)  # Iterate all fiscal years from 2015 to today, inclusive

          range(-1)  # Iterate the previous fiscal year and current fiscal year

        """

        if not isinstance(from_fy, int):

            from_fy = self.for_date(from_fy)

        if to_fy is None:

            to_fy = self.for_date()

        elif not isinstance(to_fy, int):

            to_fy = self.for_date(to_fy - datetime.timedelta(days=1))

        if from_fy < 1:

            from_fy += to_fy

        elif from_fy < 1000:

            from_fy, to_fy = to_fy, from_fy + to_fy

        return range(from_fy, to_fy + 1)

class Loader:

    """Load Beancount books organized by fiscal year"""

    def __init__(self,

                 books_root: Path,

                 fiscal_year: FiscalYear,

    ) -> None:

        """Set up a books loader

        Arguments:

        * books_root: A Path to a Beancount books checkout.

        * fiscal_year: A FiscalYear object, used to determine what books to

          load for a given date range.

        """

        self.books_root = books_root

        self.fiscal_year = fiscal_year

    def _iter_fy_books(self, fy_range: Iterable[int]) -> Iterator[Path]:

        for year in fy_range:

            if path.exists():

                yield path

                      from_fy: Year,

                      to_fy: Optional[Year]=None,

        This method generates a range of fiscal years by calling

        FiscalYear.range() with its first two arguments. It returns a string of

        Beancount directives to load the books from the first available fiscal

        year through the end of the range.

        """

        try:

        except StopIteration:

option "title" "Books from 2018"

include "../2018.beancount"

option "title" "Books from 2019"

include "../2019.beancount"

option "title" "Books from 2020"

include "../2020.beancount"

"""test_books_loader - Unit tests for books Loader class"""

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

import re

from datetime import date

from pathlib import Path

import pytest

from . import testutil

from conservancy_beancount import books

books_path = testutil.test_path('books')

@pytest.fixture(scope='module')

def conservancy_loader():

    return books.Loader(books_path, books.FiscalYear(3))

])

    assert not errors

    narrations = {getattr(entry, 'narration', None) for entry in entries}

def test_load_fy_range_empty(conservancy_loader):

    entries, errors, options_map = conservancy_loader.load_fy_range(2020, 2019)

    assert not errors

    assert not entries

import contextlib

import decimal

import operator

import os

import re

from pathlib import Path

import pytest

from . import testutil

from conservancy_beancount import config as config_mod

RT_AUTH_METHODS = frozenset(['basic', 'gssapi', 'rt'])

RT_ENV_KEYS = (

    'RTSERVER',

    'RTUSER',

    'RTPASSWD',

    'RTAUTH',

)

RT_ENV_CREDS = (

    'https://example.org/envrt',

    'envuser',

    'env  password',

    'gssapi',

)

RT_FILE_CREDS = (

    'https://example.org/filert',

    'fileuser',

    'file  password',

    'basic',

)

RT_GENERIC_CREDS = config_mod.RTCredentials(

    'https://example.org/genericrt',

    'genericuser',

    'generic password',

    None,

)

@pytest.fixture

def rt_environ():

    return dict(zip(RT_ENV_KEYS, RT_ENV_CREDS))

def _update_environ(updates):

    for key, value in updates.items():

        if value is None:

            os.environ.pop(key, None)

        else:

            os.environ[key] = str(value)

@contextlib.contextmanager

def update_environ(**kwargs):

    revert = {key: os.environ.get(key) for key in kwargs}

    _update_environ(kwargs)

    try:

        yield

    finally:

        _update_environ(revert)

@contextlib.contextmanager

def update_umask(mask):

    old_mask = os.umask(mask)

    try:

        yield old_mask

    finally:

        os.umask(old_mask)

def test_repository_from_environment():

    config = config_mod.Config()

    assert config.repository_path() == testutil.test_path('repository')

def test_no_repository():

    with update_environ(CONSERVANCY_REPOSITORY=None):

        config = config_mod.Config()

        assert config.repository_path() is None

def test_no_rt_credentials():

    with update_environ(HOME=testutil.TESTS_DIR):

        config = config_mod.Config()

        rt_credentials = config.rt_credentials()

    assert rt_credentials.server is None

    assert rt_credentials.user is None

    assert rt_credentials.passwd is None

    assert rt_credentials.auth == 'rt'

def test_rt_credentials_from_file():

    config = config_mod.Config()

    rt_credentials = config.rt_credentials()

    assert rt_credentials == RT_FILE_CREDS

def test_rt_credentials_from_environment(rt_environ):

    with update_environ(**rt_environ):

        config = config_mod.Config()

        rt_credentials = config.rt_credentials()

    assert rt_credentials == RT_ENV_CREDS

@pytest.mark.parametrize('index,drop_key', enumerate(RT_ENV_KEYS))

def test_rt_credentials_from_file_and_environment_mixed(rt_environ, index, drop_key):

    del rt_environ[drop_key]

    with update_environ(**rt_environ):

        config = config_mod.Config()

        rt_credentials = config.rt_credentials()

    expected = list(RT_ENV_CREDS)

    expected[index] = RT_FILE_CREDS[index]

    assert rt_credentials == tuple(expected)

def test_rt_credentials_from_all_sources_mixed(tmp_path):

    server = 'https://example.org/mixedrt'

    with (tmp_path / '.rtrc').open('w') as rtrc_file:

        print('user basemix', 'passwd mixed up', file=rtrc_file, sep='\n')

    with update_environ(HOME=tmp_path, RTSERVER=server, RTUSER='mixedup'):

        config = config_mod.Config()

        rt_credentials = config.rt_credentials()

    assert rt_credentials == (server, 'mixedup', 'mixed up', 'rt')

def check_rt_client_url(credentials, client):

    pattern = '^{}/?$'.format(re.escape(credentials[0].rstrip('/') + '/REST/1.0'))

    assert re.match(pattern, client.url)

@pytest.mark.parametrize('authmethod', RT_AUTH_METHODS)

def test_rt_client(authmethod):

    rt_credentials = RT_GENERIC_CREDS._replace(auth=authmethod)

    config = config_mod.Config()

    rt_client = config.rt_client(rt_credentials, testutil.RTClient)

    check_rt_client_url(RT_GENERIC_CREDS, rt_client)

    assert rt_client.auth_method == ('HTTPBasicAuth' if authmethod == 'basic' else 'login')

    assert rt_client.last_login == (

        RT_GENERIC_CREDS.user,

        RT_GENERIC_CREDS.passwd,

        True,

    )

def test_default_rt_client(rt_environ):

    with update_environ(**rt_environ):

        config = config_mod.Config()

        rt_client = config.rt_client(client=testutil.RTClient)

    check_rt_client_url(RT_ENV_CREDS, rt_client)

    assert rt_client.last_login[:-1] == RT_ENV_CREDS[1:3]

    assert rt_client.last_login[-1]

@pytest.mark.parametrize('authmethod', RT_AUTH_METHODS)

def test_rt_client_login_failure(authmethod):

    rt_credentials = RT_GENERIC_CREDS._replace(

        auth=authmethod,

        passwd='bad{}'.format(authmethod),

    )

    config = config_mod.Config()

    assert config.rt_client(rt_credentials, testutil.RTClient) is None

def test_no_rt_client_without_server():

    rt_credentials = RT_GENERIC_CREDS._replace(server=None, auth='rt')

    config = config_mod.Config()

    assert config.rt_client(rt_credentials, testutil.RTClient) is None

def test_rt_wrapper():

    config = config_mod.Config()

    rt = config.rt_wrapper(RT_GENERIC_CREDS._replace(auth='rt'), testutil.RTClient)

    assert rt.exists(1)

def test_rt_wrapper_default_creds():

    config = config_mod.Config()

    rt = config.rt_wrapper(None, testutil.RTClient)

    assert rt.rt.url.startswith(RT_FILE_CREDS[0])

def test_rt_wrapper_default_creds_from_environ(rt_environ):

    with update_environ(**rt_environ):

        config = config_mod.Config()

        rt = config.rt_wrapper(None, testutil.RTClient)

    assert rt.rt.url.startswith(RT_ENV_CREDS[0])

def test_rt_wrapper_no_creds():

    with update_environ(HOME=testutil.TESTS_DIR):

        config = config_mod.Config()

        assert config.rt_wrapper(None, testutil.RTClient) is None

def test_rt_wrapper_bad_creds():

    rt_credentials = RT_GENERIC_CREDS._replace(passwd='badpass', auth='rt')

    config = config_mod.Config()

    assert config.rt_wrapper(rt_credentials, testutil.RTClient) is None

def test_rt_wrapper_caches():

    rt_credentials = RT_GENERIC_CREDS._replace(auth='rt')

    config = config_mod.Config()

    rt1 = config.rt_wrapper(rt_credentials, testutil.RTClient)

    rt2 = config.rt_wrapper(rt_credentials, testutil.RTClient)

    assert rt1 is rt2

def test_rt_wrapper_caches_by_creds():

    config = config_mod.Config()

    rt1 = config.rt_wrapper(RT_GENERIC_CREDS._replace(auth='rt'), testutil.RTClient)

    rt2 = config.rt_wrapper(None, testutil.RTClient)

    assert rt1 is not rt2

def test_rt_wrapper_cache_responds_to_external_credential_changes(rt_environ):

    config = config_mod.Config()

    rt1 = config.rt_wrapper(None, testutil.RTClient)

    with update_environ(**rt_environ):

        rt2 = config.rt_wrapper(None, testutil.RTClient)

    assert rt1 is not rt2

def test_rt_wrapper_has_cache(tmp_path):

    with update_environ(XDG_CACHE_HOME=tmp_path), update_umask(0o002):

        config = config_mod.Config()

        rt = config.rt_wrapper(None, testutil.RTClient)

        rt.exists(1)

    expected = 'conservancy_beancount/{}@*.sqlite3'.format(RT_FILE_CREDS[1])

    actual = None

    for actual in tmp_path.glob(expected):

        assert not actual.stat().st_mode & 0o177

    assert actual is not None, "did not find any generated cache file"

def test_rt_wrapper_without_cache(tmp_path):

    tmp_path.chmod(0)

    with update_environ(XDG_CACHE_HOME=tmp_path):

        config = config_mod.Config()

        rt = config.rt_wrapper(None, testutil.RTClient)

    tmp_path.chmod(0o600)

    assert not any(tmp_path.iterdir())

def test_cache_mkdir(tmp_path):

    expected = tmp_path / 'TESTcache'

    with update_environ(XDG_CACHE_HOME=tmp_path):

        config = config_mod.Config()

        cache_path = config.cache_dir_path(expected.name)

    assert cache_path == tmp_path / 'TESTcache'

    assert cache_path.is_dir()

def test_cache_mkdir_parent(tmp_path):

    xdg_cache_dir = tmp_path / 'xdgcache'

    expected = xdg_cache_dir / 'conservancy_beancount'

    with update_environ(XDG_CACHE_HOME=xdg_cache_dir):

        config = config_mod.Config()

        cache_path = config.cache_dir_path(expected.name)

    assert cache_path == expected

    assert cache_path.is_dir()

def test_cache_mkdir_from_home(tmp_path):

    expected = tmp_path / '.cache' / 'TESTcache'

    with update_environ(HOME=tmp_path, XDG_CACHE_HOME=None):

        config = config_mod.Config()

        cache_path = config.cache_dir_path(expected.name)

    assert cache_path == expected

    assert cache_path.is_dir()

def test_cache_mkdir_exists_ok(tmp_path):

    expected = tmp_path / 'TESTcache'

    expected.mkdir()

    with update_environ(XDG_CACHE_HOME=tmp_path):

        config = config_mod.Config()

        cache_path = config.cache_dir_path(expected.name)

    assert cache_path == expected

def test_cache_path_conflict(tmp_path):

    extant_path = tmp_path / 'TESTcache'

    extant_path.touch()

    with update_environ(XDG_CACHE_HOME=tmp_path):

        config = config_mod.Config()

        cache_path = config.cache_dir_path(extant_path.name)

    assert cache_path is None

    assert extant_path.is_file()

def test_cache_path_parent_conflict(tmp_path):

    (tmp_path / '.cache').touch()

    with update_environ(HOME=tmp_path, XDG_CACHE_HOME=None):

        config = config_mod.Config()

        assert config.cache_dir_path('TESTcache') is None

def test_relative_xdg_cache_home_ignored(tmp_path):

    with update_environ(HOME=tmp_path,

                        XDG_CACHE_HOME='nonexistent/test/cache/directory/tree'):

        config = config_mod.Config()

        cache_dir_path = config.cache_dir_path('TESTcache')

    assert cache_dir_path == tmp_path / '.cache/TESTcache'

def test_default_payment_threshold():

    threshold = config_mod.Config().payment_threshold()

    assert isinstance(threshold, (int, decimal.Decimal))

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

    '15',

    ' +15',

    '15. ',

    '15.0',

    '15.00',

])

def test_payment_threshold(config_threshold):

    config = config_mod.Config()

    config.load_string(f'[Beancount]\npayment threshold = {config_threshold}\n')

    assert config.payment_threshold() == decimal.Decimal(15)

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

    None,

    '',

    'nonexistent/relative/path',

])

def test_config_file_path(config_path):

    expected = Path('~/.config/conservancy_beancount/config.ini').expanduser()

    with update_environ(XDG_CONFIG_HOME=config_path):

        config = config_mod.Config()

        assert config.config_file_path() == expected

def test_config_file_path_respects_xdg_config_home():

    with update_environ(XDG_CONFIG_HOME='/etc'):

        config = config_mod.Config()

        assert config.config_file_path() == Path('/etc/conservancy_beancount/config.ini')

def test_config_file_path_with_subdir():

    expected = testutil.test_path('userconfig/conftest/config.ini')

    config = config_mod.Config()

    assert config.config_file_path('conftest') == expected

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

    None,

    testutil.test_path('userconfig/conservancy_beancount/config.ini'),

])

def test_load_file(path):

    config = config_mod.Config()

    config.load_file(path)

    assert config.books_path() == Path('/test/conservancy_beancount')

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

    lambda path: None,

    operator.methodcaller('touch', 0o200),

])

def test_load_file_error(tmp_path, path_func):

    config_path = tmp_path / 'nonexistent.ini'

    path_func(config_path)

    config = config_mod.Config()

    with pytest.raises(OSError):

        config.load_file(config_path)

def test_no_books_path():

    config = config_mod.Config()

    assert config.books_path() is None

@pytest.mark.parametrize('value,month,day', [

    ('2', 2, 1),

    ('3 ', 3, 1),

    ('  4', 4, 1),

    (' 5 ', 5, 1),

    ('6 1', 6, 1),

    ('  06  03  ', 6, 3),

    ('6-05', 6, 5),

    ('06 - 10', 6, 10),

    ('6/15', 6, 15),

    ('06  /  20', 6, 20),

    ('10.25', 10, 25),

    (' 10 . 30 ', 10, 30),

])

def test_fiscal_year_begin(value, month, day):

    config = config_mod.Config()

    config.load_string(f'[Beancount]\nfiscal year begin = {value}\n')

    assert config.fiscal_year_begin() == (month, day)

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

    'text',

    '1900',

    '13',

    '010',

    '2 30',

    '4-31',

])

def test_bad_fiscal_year_begin(value):

    config = config_mod.Config()

    config.load_string(f'[Beancount]\nfiscal year begin = {value}\n')

    with pytest.raises(ValueError):

        config.fiscal_year_begin()

def test_default_fiscal_year_begin():

    config = config_mod.Config()

    actual = config.fiscal_year_begin()

    assert actual.month == 3

    assert actual.day == 1

def test_books_loader():

    books_path = testutil.test_path('books')

    config = config_mod.Config()

    config.load_string(f'[Beancount]\nbooks dir = {books_path}\n')

    loader = config.books_loader()

def test_books_loader_without_books():

    assert config_mod.Config().books_loader() is None

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

from decimal import Decimal

from pathlib import Path

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

    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,

            type_=bc_data.Posting, **meta):

    if cost is not None:

        cost = Cost(*cost)

    if not meta:

        meta = None

    return 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),