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_base_currency_commodity(self, book_basic):
cdty = Commodity(namespace="AMEX", mnemonic="APPLE", fullname="Apple", book=book_basic)
with pytest.raises(GnucashException):
cdty.base_currency
# should trigger creation of USD currency
cdty["quoted_currency"] = "USD"
assert cdty.base_currency.mnemonic == 'USD'
book_basic.flush()
assert cdty.base_currency == book_basic.currencies(mnemonic="USD")
cdty["quoted_currency"] = "EUR"
assert cdty.base_currency == book_basic.currencies(mnemonic="EUR")
def test_book_getters(self, new_book):
cur = new_book.currencies[0]
assert cur == new_book.get(Commodity, mnemonic=cur.mnemonic)
with pytest.raises(ValueError):
new_book.get(Commodity, mnemonic="FOO")
with pytest.raises(ValueError):
new_book.get(Commodity, mnemonic="CAD")
assert new_book.get(Commodity).all() == [cur]
assert new_book.accounts == []
assert new_book.transactions == []
assert new_book.commodities == [cur]
assert new_book.currencies == [cur]
assert new_book.prices == []
assert new_book.customers == []
assert new_book.vendors == []
def test_create_commodity(self, book_basic):
assert len(book_basic.commodities) == 2
cdty = Commodity(namespace="AMEX", mnemonic="APPLE", fullname="Apple", book=book_basic)
book_basic.flush()
assert len(book_basic.commodities) == 3
with pytest.raises(GnucashException):
cdty.base_currency
cdty["quoted_currency"] = "EUR"
assert cdty.base_currency == book_basic.commodities(mnemonic="EUR")
def query(self):
""" Main accounts query """
query = (
self.book.session.query(Account)
.join(Commodity)
.filter(Commodity.namespace != "template")
.filter(Account.type != AccountType.root.value)
)
return query
def get_aggregate(self, security: Commodity) -> SecurityAggregate:
""" Returns the aggregate for the entity """
assert security is not None
assert isinstance(security, Commodity)
return SecurityAggregate(self.book, security)
def query(self):
""" Main accounts query """
query = (
self.book.session.query(Account)
.join(Commodity)
.filter(Commodity.namespace != "template")
.filter(Account.type != AccountType.root.value)
)
return query
def __get_base_query(self):
""" Returns the base query which filters out data for all queries. """
query = (
self.book.session.query(Commodity)
.filter(Commodity.namespace != "CURRENCY",
Commodity.namespace != "template")
)
return query
def get_latest_price(self, security: Commodity) -> PriceModel:
""" Returns the latest available price for commodity """
assert isinstance(security, Commodity)
symbol = SecuritySymbol(security.namespace, security.mnemonic)
prices = PriceDbApplication()
result = prices.get_latest_price(symbol)
return result
Returns all income accounts for this security.
Income accounts are accounts not under Trading, expressed in currency, and
having the same name as the mnemonic.
They should be under Assets but this requires a recursive SQL query.
"""
# trading = self.book.trading_account(self.security)
# log(DEBUG, "trading account = %s, %s", trading.fullname, trading.guid)
# Example on how to self-link, i.e. parent account, using alias.
# parent_alias = aliased(Account)
# .join(parent_alias, Account.parent)
# parent_alias.parent_guid != trading.guid
query = (
self.book.session.query(Account)
.join(Commodity)
.filter(Account.name == self.security.mnemonic)
.filter(Commodity.namespace == "CURRENCY")
# .filter(Account.type != "TRADING")
.filter(Account.type == AccountType.income.value)
)
# generic.print_sql(query)
return query.all()