Browse Source

Add tox (tests and linting) and fix types

master
Jakub Valenta 9 months ago
parent
commit
ed473b1901

+ 5
- 3
.gitignore View File

@@ -1,7 +1,9 @@
*.pyc
*.egg-info
.mypy_cache
.pytest_cache
.tox
__pycache__
build
dist
*.DS_Store
*.egg-info
tests/_*
reports

+ 10
- 0
Makefile View File

@@ -0,0 +1,10 @@
.PHONY: test lint help

test: ## Run unit tests and linting
tox

lint: ## Run linting
tox -e lint isort

help: # https://gist.github.com/jhermsmeier/2d831eb8ad2fb0803091
@grep -E '^[a-zA-Z0-9_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-16s\033[0m %s\n", $$1, $$2}'

+ 1
- 0
account-statement View File

@@ -1,5 +1,6 @@
#!/usr/bin/env python

import account_statement.account_statement

if __name__ == '__main__':
account_statement.account_statement.main()

+ 0
- 1
account_statement/__init__.py View File

@@ -1 +0,0 @@
from .account_statement import *

+ 9
- 10
account_statement/account_statement.py View File

@@ -3,9 +3,7 @@ import json
import operator
import os.path
import re
from collections import defaultdict
from copy import deepcopy
from typing import Any, Callable, Dict, Iterable, Iterator, List
from typing import Any, Callable, Dict, Iterable, Iterator, List, Sequence

import listio
import yaml
@@ -16,15 +14,16 @@ Secrets = List[Dict[str, Any]]

# https://bugs.python.org/issue1519638#msg155982
REGEX_CENSOR_CODES = re.compile(
'((\d{1,2}\.\d{1,2}\.\d{4})|(\d{4,}\.\d{2})|(dne \d{4}))|\d{4,}')
r'((\d{1,2}\.\d{1,2}\.\d{4})|(\d{4,}\.\d{2})|(dne \d{4}))|\d{4,}')
REGEX_CENSOR_NAMES = re.compile(
'^[A-Z]\S+\s+[A-Z]\S+(\s+[A-Z]\S+)?$|'
'^[A-Z]{2,}\s+[A-Z]{2,}(\s+[A-Z]{2,})?$')
r'^[A-Z]\S+\s+[A-Z]\S+(\s+[A-Z]\S+)?$|'
r'^[A-Z]{2,}\s+[A-Z]{2,}(\s+[A-Z]{2,})?$')


def censor(transactions: Iterable[Transaction], fn: Callable,
fields: List[str]=['info', 'message', 'spec', 'comment'],
verbose: bool=False) -> Iterator[Transaction]:
def censor(transactions: Iterable[Transaction],
fn: Callable,
fields: Sequence[str] = ('info', 'message', 'spec', 'comment'),
verbose: bool = False) -> Iterator[Transaction]:
for tx in transactions:
censored = tx
for field in fields:
@@ -53,7 +52,7 @@ def censor_names(value: str) -> str:
return value


def censor_list(value: str, excl: List[str]=None) -> str:
def censor_list(value: str, excl: Sequence[str] = ()) -> str:
if not value:
return value
if value in excl:

+ 29
- 27
account_statement/backends/fio.py View File

@@ -1,13 +1,12 @@
# coding: utf-8

import datetime
import json
from typing import Callable, Dict, Iterator, TypeVar
from typing import Any, Dict, Iterator, Optional

import requests

from ..common import Transaction, read_cached_data, write_data_to_cache

from account_statement.common import (
Transaction, read_cached_data, write_data_to_cache,
)

FioTransaction = Dict[str, Dict[str, str]]

@@ -28,40 +27,44 @@ def fetch_data(token: str,
return r.text


T = TypeVar('T', str, int, float)


def get_fio_tx_prop(fio_tx: FioTransaction, name: str,
transform: Callable[[str], T]=None) -> T:
def get_fio_tx_prop(fio_tx: FioTransaction, name: str) -> Optional[str]:
for _, prop in fio_tx.items():
if prop and prop['name'] == name:
value = prop['value']
if value is not None and transform:
return transform(value)
return value
return prop['value']
return None


def safe_int(val: Any) -> Optional[int]:
if val is None:
return val
return int(val)


def parse_data(data: str) -> Iterator[Transaction]:
fio_res = json.loads(data)
fio_statement = fio_res['accountStatement']
currency = fio_statement['info']['currency']
balance = float(fio_statement['info']['openingBalance'])
for fio_tx in fio_statement['transactionList']['transaction']:
amount = get_fio_tx_prop(fio_tx, 'Objem', float)
amount_str = get_fio_tx_prop(fio_tx, 'Objem')
if amount_str is None:
raise ValueError(f'No amount found for {fio_tx}')
amount = float(amount_str)
balance = balance + amount
date_str = get_fio_tx_prop(fio_tx, 'Datum')
if date_str is None:
raise ValueError(f'No date found for {fio_tx}')
dt = datetime.datetime.strptime(date_str, '%Y-%m-%d%z')
yield Transaction(
date=datetime.datetime.strptime(
get_fio_tx_prop(fio_tx, 'Datum'), '%Y-%m-%d%z'
).strftime('%Y-%m-%d'),
date=dt.strftime('%Y-%m-%d'),
amount=amount,
currency=get_fio_tx_prop(fio_tx, 'Měna'),
account=get_fio_tx_prop(fio_tx, 'Protiúčet'),
account_bank=get_fio_tx_prop(fio_tx, 'Kód banky', int),
account_bank=safe_int(get_fio_tx_prop(fio_tx, 'Kód banky')),
bank_name=get_fio_tx_prop(fio_tx, 'Název banky'),
ks=get_fio_tx_prop(fio_tx, 'KS', int),
vs=get_fio_tx_prop(fio_tx, 'VS', int),
ss=get_fio_tx_prop(fio_tx, 'SS', int),
ks=safe_int(get_fio_tx_prop(fio_tx, 'KS')),
vs=safe_int(get_fio_tx_prop(fio_tx, 'VS')),
ss=safe_int(get_fio_tx_prop(fio_tx, 'SS')),
info=get_fio_tx_prop(fio_tx, 'Uživatelská identifikace'),
message=get_fio_tx_prop(fio_tx, 'Zpráva pro příjemce'),
type=get_fio_tx_prop(fio_tx, 'Typ'),
@@ -76,7 +79,7 @@ def parse_data(data: str) -> Iterator[Transaction]:
def read_account_statement(
token: str, path_cache_dir: str,
date_from: datetime.date,
date_to: datetime.date=None) -> Iterator[Transaction]:
date_to: Optional[datetime.date] = None) -> Iterator[Transaction]:
if date_to is None:
date_to = datetime.datetime.now().date()
data_cached = read_cached_data(path_cache_dir, date_from, date_to)
@@ -85,7 +88,6 @@ def read_account_statement(
data = data_cached
else:
data = fetch_data(token, date_from, date_to)
if not data:
return None
write_data_to_cache(path_cache_dir, date_from, date_to, data)
return parse_data(data)
if data:
write_data_to_cache(path_cache_dir, date_from, date_to, data)
yield from parse_data(data)

+ 14
- 11
account_statement/backends/hbci.py View File

@@ -1,17 +1,18 @@
import datetime
import json
from typing import Dict, Iterable, Iterator

import attr
from typing import Dict, Iterable, Iterator, Optional

import aqbanking
import attr

from ..common import Transaction, read_cached_data, write_data_to_cache
from account_statement.common import (
Transaction, read_cached_data, write_data_to_cache,
)

AqTransaction = Dict[str, str]


def get_aq_tx_prop(aq_tx: AqTransaction, prop: str) -> str:
def get_aq_tx_prop(aq_tx: AqTransaction, prop: str) -> Optional[str]:
val = aq_tx[prop]
if val == 'None':
return None
@@ -87,9 +88,12 @@ def fetch_data(account_no: str,

def parse_data(data: str, currency: str) -> Iterator[Transaction]:
for aq_tx in json.loads(data):
amount_str = get_aq_tx_prop(aq_tx, 'value')
if amount_str is None:
raise ValueError(f'No amount found for {aq_tx}')
yield Transaction(
date=get_aq_tx_prop(aq_tx, 'date'),
amount=float(get_aq_tx_prop(aq_tx, 'value')),
amount=float(amount_str),
currency=get_aq_tx_prop(aq_tx, 'currency'),
account=get_aq_tx_prop(aq_tx, 'remoteAccount'),
account_bank=get_aq_tx_prop(aq_tx, 'remoteIban'),
@@ -127,7 +131,7 @@ def read_account_statement(
currency: str,
path_cache_dir: str,
date_from: datetime.date,
date_to: datetime.date=None) -> Iterator[Transaction]:
date_to: Optional[datetime.date] = None) -> Iterator[Transaction]:
if date_to is None:
date_to = datetime.datetime.now().date()
data_cached = read_cached_data(path_cache_dir, date_from, date_to)
@@ -136,7 +140,6 @@ def read_account_statement(
data = data_cached
else:
data = fetch_data(account_no, password, date_from, date_to)
if not data:
return None
write_data_to_cache(path_cache_dir, date_from, date_to, data)
return calc_balance(parse_data(data, currency))
if data:
write_data_to_cache(path_cache_dir, date_from, date_to, data)
yield from calc_balance(parse_data(data, currency))

+ 2
- 1
account_statement/common.py View File

@@ -1,6 +1,7 @@
import datetime
import json
import os
from typing import Optional

import attr

@@ -40,7 +41,7 @@ def _format_cache_key(date_from: datetime.date, date_to: datetime.date) -> str:


def read_cached_data(path_cache_dir: str, date_from: datetime.date,
date_to: datetime.date) -> str:
date_to: datetime.date) -> Optional[str]:
cache_key = _format_cache_key(date_from, date_to)
path_cache_file = os.path.join(path_cache_dir, cache_key)
print('Reading cache file "{}".'.format(path_cache_file))

+ 3
- 1
setup.py View File

@@ -1,7 +1,8 @@
from setuptools import setup, find_packages
from codecs import open
from os import path

from setuptools import find_packages, setup

here = path.abspath(path.dirname(__file__))

with open(path.join(here, 'README.md'), encoding='utf-8') as f:
@@ -28,6 +29,7 @@ setup(
'Topic :: Artistic Software',
'License :: OSI Approved :: Apache Software License',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.6',
],

keywords='',

+ 10
- 16
tests/test_account_statement.py View File

@@ -1,13 +1,11 @@
#!/usr/bin/env python
from unittest import TestCase

import unittest
from account_statement.account_statement import (
censor_codes, censor_list, censor_names,
)

from account_statement import censor_codes
from account_statement import censor_names
from account_statement import censor_list


class Test(unittest.TestCase):
class Test(TestCase):

def test_fio_censor_codes(self):
self.assertEqual(
@@ -16,23 +14,19 @@ class Test(unittest.TestCase):
)

def test_fio_censor_names(self):
with open('_test_censor_names_true.txt', 'r') as f:
with open('_test_censor_names_true.txt') as f:
for line in f:
self.assertEqual(censor_names(line.strip()), '')
with open('_test_censor_names_false.txt', 'r') as f:
with open('_test_censor_names_false.txt') as f:
for line in f:
self.assertEqual(censor_names(line.strip()), line.strip())

def test_fio_censor_list(self):
with open('_test_censor_names_false.txt', 'r') as f:
with open('_test_censor_names_false.txt') as f:
excl = [x.strip() for x in f.readlines()]
with open('_test_censor_names_false.txt', 'r') as f:
with open('_test_censor_names_false.txt') as f:
for line in f:
self.assertEqual(censor_list(line.strip(), excl), '')
with open('_test_censor_names_true.txt', 'r') as f:
with open('_test_censor_names_true.txt') as f:
for line in f:
self.assertEqual(censor_list(line.strip(), excl), line.strip())


if __name__ == '__main__':
unittest.main()

+ 32
- 0
tox.ini View File

@@ -0,0 +1,32 @@
[tox]
envlist = py37, lint, isort
skip_missing_interpreters = True
skipsdist = True

[testenv]
deps = pytest
commands = py.test --junitxml=reports/junit-{envname}.xml

[testenv:lint]
deps =
flake8
mypy
isort
commands =
flake8 account_statement
mypy account_statement --ignore-missing-imports

[testenv:isort]
deps = isort
commands = isort -c -rc .

[isort]
combine_as_imports = true
default_section = THIRDPARTY
include_trailing_comma = true
known_first_party = account_statement
line_length = 79
multi_line_output = 5
not_skip = __init__.py
atomic = true
skip = .tox, venv

Loading…
Cancel
Save