Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_sync_order(self):
ledger = Ledger(os.path.join('fixtures', 'empty.lgr'))
sync = OfxSynchronizer(ledger)
with open(os.path.join('fixtures', 'checking_order.ofx'), 'rb') as ofx_file:
ofx = OfxParser.parse(ofx_file)
txns = sync.filter(
ofx.account.statement.transactions,
ofx.account.account_id)
self.assertTrue(txns[0].date < txns[1].date and
txns[1].date < txns[2].date)
def test_write(self):
with open_file('fidelity.ofx') as f:
ofx = OfxParser.parse(f)
self.assertEqual(str(ofx), "")
def test_unknownaccount(self):
with open(os.path.join('fixtures', 'checking.ofx'), 'rb') as ofx_file:
ofx = OfxParser.parse(ofx_file)
converter = OfxConverter(account=ofx.account, name="Foo",
unknownaccount='Expenses:Unknown')
self.assertEqualLedgerPosting(
converter.convert(
ofx.account.statement.transactions[0]).format(),
"""2011/03/31 DIVIDEND EARNED FOR PERIOD OF 03/01/2011 THROUGH 03/31/2011 ANNUAL PERCENTAGE YIELD EARNED IS 0.05%
Foo $0.01
def download_parsed(self, days=60):
"""Downloaded OFX response parsed by :py:meth:`OfxParser.parse`
:param days: Number of days to look back at
:type days: integer
:rtype: :py:class:`ofxparser.Ofx`
"""
if IS_PYTHON_2:
return OfxParser.parse(
self.download(days=days)
)
else:
return OfxParser.parse(
BytesIO(self.download(days=days).read().encode())
)
def import_transactions(ofx_path):
transactions = []
with open(ofx_path) as ofx_file:
logger.info('Opening ofx file %s', ofx_path)
try:
ofx = OfxParser.parse(ofx_file)
for transaction in ofx.account.statement.transactions:
try:
transaction_time = transaction.date
transactions.append(ImportStatement(
notes=transaction.payee,
book_date=transaction_time.date(),
transaction_date=transaction_time.date(),
amount=transaction.amount
))
except ValueError:
logger.error('Cannot import transaction: {}'.format(transaction))
pass
except ValueError:
logger.error('Failed to import all transactions! Wrong file format?')
return transactions
def accounts(self):
"""Ask the bank for the known :py:class:`ofxclient.Account` list.
:rtype: list of :py:class:`ofxclient.Account` objects
"""
from ofxclient.account import Account
client = self.client()
query = client.account_list_query()
resp = client.post(query)
resp_handle = StringIO(resp)
if IS_PYTHON_2:
parsed = OfxParser.parse(resp_handle)
else:
parsed = OfxParser.parse(BytesIO(resp_handle.read().encode()))
return [Account.from_ofxparse(a, institution=self)
for a in parsed.accounts]
days = 7
last_txns_len = 0
while (True):
logging.debug(
"Downloading %d days of transactions for %s (max_days=%d)." % (
days, acct.description, max_days))
raw = acct.download(days=days)
if raw.read() == 'Server error occured. Received HttpStatusCode of 400':
raise Exception(
"Error connecting to account %s" %
(acct.description))
raw.seek(0)
ofx = None
try:
ofx = OfxParser.parse(raw)
except OfxParserException as ex:
if ex.message == 'The ofx file is empty!':
return (ofx, [])
else:
raise ex
if ofx.signon is not None:
if ofx.signon.severity == 'ERROR':
raise Exception(
"Error returned from server for %s: %s" %
(acct.description, ofx.signon.message))
if not(hasattr(ofx, 'account')):
# some banks return this for no txns
if (days >= max_days):
logging.debug("Hit max days.")
# return None to let the caller know that we don't
# even have account info