7 Commits

16 changed files with 725 additions and 73 deletions

View File

@ -13,7 +13,7 @@ sys.path.insert(0, os.path.abspath('../../src/'))
project = 'Mia! Accounting' project = 'Mia! Accounting'
copyright = '2023, imacat' copyright = '2023, imacat'
author = 'imacat' author = 'imacat'
release = '1.3.1' release = '1.3.2'
# -- General configuration --------------------------------------------------- # -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

View File

@ -17,7 +17,7 @@
[project] [project]
name = "mia-accounting" name = "mia-accounting"
version = "1.3.1" version = "1.3.2"
description = "A Flask accounting module." description = "A Flask accounting module."
readme = "README.rst" readme = "README.rst"
requires-python = ">=3.11" requires-python = ">=3.11"

View File

@ -20,11 +20,10 @@
from datetime import date from datetime import date
from flask import abort from flask import abort
from sqlalchemy.orm import selectinload
from werkzeug.routing import BaseConverter from werkzeug.routing import BaseConverter
from accounting import db from accounting import db
from accounting.models import JournalEntry, JournalEntryLineItem from accounting.models import JournalEntry
from accounting.utils.journal_entry_types import JournalEntryType from accounting.utils.journal_entry_types import JournalEntryType

View File

@ -129,5 +129,5 @@ def __update_file_rev_date(file: Path) -> None:
main.add_command(babel_extract) main.add_command(babel_extract)
main.add_command(babel_compile) main.add_command(babel_compile)
if __name__ == '__main__': if __name__ == "__main__":
main() main()

View File

@ -129,5 +129,5 @@ def __update_file_rev_date(file: Path) -> None:
main.add_command(babel_extract) main.add_command(babel_extract)
main.add_command(babel_compile) main.add_command(babel_compile)
if __name__ == '__main__': if __name__ == "__main__":
main() main()

306
tests/make-sample.py Executable file
View File

@ -0,0 +1,306 @@
#! env python3
# The Mia! Accounting Project.
# Author: imacat@mail.imacat.idv.tw (imacat), 2023/4/9
# Copyright (c) 2023 imacat.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""The sample data generation.
"""
from datetime import date, timedelta
import click
from testlib import Accounts, create_test_app, JournalEntryLineItemData, \
JournalEntryCurrencyData, JournalEntryData, \
BaseTestData
@click.command()
@click.argument("file")
def main(file) -> None:
"""Creates the sample data and output to a file."""
data: SampleData = SampleData(create_test_app(), "editor")
with open(file, "wt") as fp:
fp.write(data.json())
class SampleData(BaseTestData):
"""The sample data."""
def _init_data(self) -> None:
self.__add_recurring()
self.__add_offsets()
self.__add_meals()
def __add_recurring(self) -> None:
"""Adds the recurring data.
:return: None.
"""
self.__add_usd_recurring()
self.__add_twd_recurring()
def __add_usd_recurring(self) -> None:
"""Adds the recurring data in USD.
:return: None.
"""
today: date = date.today()
days: int
year: int
month: int
# Recurring in USD
j_date: date = date(today.year - 5, today.month, today.day)
j_date = j_date + timedelta(days=(4 - j_date.weekday()))
days = (today - j_date).days
while True:
if days < 0:
break
self.__add_journal_entry(
days, "USD", "2600",
Accounts.BANK, "Transfer", Accounts.SERVICE, "Payroll")
days = days - 1
if days < 0:
break
self.__add_journal_entry(
days, "USD", "1200",
Accounts.CASH, None, Accounts.BANK, "Withdraw")
days = days - 13
year = today.year - 5
month = today.month
while True:
month = month + 1
if month > 12:
year = year + 1
month = 1
days = (today - date(year, month, 1)).days
if days < 0:
break
self.__add_journal_entry(
days, "USD", "1800",
Accounts.RENT_EXPENSE, "Rent", Accounts.BANK, "Transfer")
def __add_twd_recurring(self) -> None:
"""Adds the recurring data in TWD.
:return: None.
"""
today: date = date.today()
year: int = today.year - 5
month: int = today.month
while True:
days: int = (today - date(year, month, 5)).days
if days < 0:
break
self.__add_journal_entry(
days, "TWD", "50000",
Accounts.BANK, "薪資轉帳", Accounts.SERVICE, "薪水")
days = days - 1
if days < 0:
break
self.__add_journal_entry(
days, "TWD", "25000",
Accounts.CASH, None, Accounts.BANK, "提款")
days = days - 4
if days < 0:
break
self.__add_journal_entry(
days, "TWD", "18000",
Accounts.RENT_EXPENSE, "房租", Accounts.BANK, "轉帳")
month = month + 1
if month > 12:
year = year + 1
month = 1
def __add_offsets(self) -> None:
"""Adds the offset data.
:return: None.
"""
days: int
year: int
month: int
description: str
line_item_or: JournalEntryLineItemData
line_item_of: JournalEntryLineItemData
# Full offset and unmatched in USD
description = "Speaking—Institute"
line_item_or = JournalEntryLineItemData(
Accounts.RECEIVABLE, description, "120")
self._add_journal_entry(JournalEntryData(
40, [JournalEntryCurrencyData(
"USD", [line_item_or], [JournalEntryLineItemData(
Accounts.SERVICE, description, "120")])]))
line_item_of = JournalEntryLineItemData(
Accounts.RECEIVABLE, description, "120",
original_line_item=line_item_or)
self._add_journal_entry(JournalEntryData(
5, [JournalEntryCurrencyData(
"USD", [JournalEntryLineItemData(
Accounts.BANK, description, "120")],
[line_item_of])]))
self.__add_journal_entry(
30, "USD", "120",
Accounts.BANK, description, Accounts.SERVICE, description)
# Partial offset in USD
line_item_or = JournalEntryLineItemData(
Accounts.PAYABLE, "Computer", "1600")
self._add_journal_entry(JournalEntryData(
60, [JournalEntryCurrencyData(
"USD", [JournalEntryLineItemData(
Accounts.MACHINERY, "Computer", "1600")],
[line_item_or])]))
line_item_of = JournalEntryLineItemData(
Accounts.PAYABLE, "Computer", "800",
original_line_item=line_item_or)
self._add_journal_entry(JournalEntryData(
35, [JournalEntryCurrencyData(
"USD", [line_item_of], [JournalEntryLineItemData(
Accounts.BANK, "Computer", "800")])]))
line_item_of = JournalEntryLineItemData(
Accounts.PAYABLE, "Computer", "400",
original_line_item=line_item_or)
self._add_journal_entry(JournalEntryData(
10, [JournalEntryCurrencyData(
"USD", [line_item_of], [JournalEntryLineItemData(
Accounts.CASH, "Computer", "400")])]))
# Full offset and unmatched in TWD
description = "演講費—母校"
line_item_or = JournalEntryLineItemData(
Accounts.RECEIVABLE, description, "3000")
self._add_journal_entry(JournalEntryData(
45, [JournalEntryCurrencyData(
"TWD", [line_item_or], [JournalEntryLineItemData(
Accounts.SERVICE, description, "3000")])]))
line_item_of = JournalEntryLineItemData(
Accounts.RECEIVABLE, description, "3000",
original_line_item=line_item_or)
self._add_journal_entry(JournalEntryData(
6, [JournalEntryCurrencyData(
"TWD", [JournalEntryLineItemData(
Accounts.BANK, description, "3000")],
[line_item_of])]))
self.__add_journal_entry(
25, "TWD", "3000",
Accounts.BANK, description, Accounts.SERVICE, description)
# Partial offset in TWD
line_item_or = JournalEntryLineItemData(
Accounts.PAYABLE, "手機", "30000")
self._add_journal_entry(JournalEntryData(
55, [JournalEntryCurrencyData(
"TWD", [JournalEntryLineItemData(
Accounts.MACHINERY, "手機", "30000")],
[line_item_or])]))
line_item_of = JournalEntryLineItemData(
Accounts.PAYABLE, "手機", "16000",
original_line_item=line_item_or)
self._add_journal_entry(JournalEntryData(
27, [JournalEntryCurrencyData(
"TWD", [line_item_of], [JournalEntryLineItemData(
Accounts.BANK, "手機", "16000")])]))
line_item_of = JournalEntryLineItemData(
Accounts.PAYABLE, "手機", "6000",
original_line_item=line_item_or)
self._add_journal_entry(JournalEntryData(
8, [JournalEntryCurrencyData(
"TWD", [line_item_of], [JournalEntryLineItemData(
Accounts.CASH, "手機", "6000")])]))
def __add_meals(self) -> None:
"""Adds the meal data.
:return: None.
"""
days = 60
while days >= 0:
# Meals in USD
if days % 4 == 2:
self.__add_journal_entry(
days, "USD", "2.9",
Accounts.MEAL, "Lunch—Coffee", Accounts.CASH, None)
else:
self.__add_journal_entry(
days, "USD", "3.9",
Accounts.MEAL, "Lunch—Coffee", Accounts.CASH, None)
if days % 15 == 3:
self.__add_journal_entry(
days, "USD", "5.45",
Accounts.MEAL, "Dinner—Pizza",
Accounts.PAYABLE, "Dinner—Pizza")
else:
self.__add_journal_entry(
days, "USD", "5.9",
Accounts.MEAL, "Dinner—Pasta", Accounts.CASH, None)
# Meals in TWD
if days % 5 == 3:
self.__add_journal_entry(
days, "TWD", "125",
Accounts.MEAL, "午餐—鄰家咖啡", Accounts.CASH, None)
else:
self.__add_journal_entry(
days, "TWD", "80",
Accounts.MEAL, "午餐—便當", Accounts.CASH, None)
if days % 15 == 3:
self.__add_journal_entry(
days, "TWD", "320",
Accounts.MEAL, "晚餐—牛排", Accounts.PAYABLE, "晚餐—牛排")
else:
self.__add_journal_entry(
days, "TWD", "100",
Accounts.MEAL, "晚餐—自助餐", Accounts.CASH, None)
days = days - 1
def __add_journal_entry(
self, days: int, currency: str, amount: str,
debit_account: str, debit_description: str | None,
credit_account: str, credit_description: str | None) -> None:
"""Adds a simple journal entry.
:param days: The number of days before today.
:param currency: The currency code.
:param amount: The amount.
:param debit_account: The debit account code.
:param debit_description: The debit description.
:param credit_account: The credit account code.
:param credit_description: The credit description.
:return: None.
"""
self._add_journal_entry(JournalEntryData(
days,
[JournalEntryCurrencyData(
currency,
[JournalEntryLineItemData(
debit_account, debit_description, amount)],
[JournalEntryLineItemData(
credit_account, credit_description, amount)])]))
if __name__ == "__main__":
main()

View File

@ -51,8 +51,8 @@ class OffsetTestCase(unittest.TestCase):
JournalEntryLineItem.query.delete() JournalEntryLineItem.query.delete()
self.client, self.csrf_token = get_client(self.app, "editor") self.client, self.csrf_token = get_client(self.app, "editor")
self.data: OffsetTestData = OffsetTestData( self.data: OffsetTestData = OffsetTestData(self.app, "editor")
self.app, self.client, self.csrf_token) self.data.populate()
def test_add_receivable_offset(self) -> None: def test_add_receivable_offset(self) -> None:
"""Tests to add the receivable offset. """Tests to add the receivable offset.

View File

@ -55,7 +55,7 @@ class ReportTestCase(unittest.TestCase):
:return: None. :return: None.
""" """
client, csrf_token = get_client(self.app, "nobody") client, csrf_token = get_client(self.app, "nobody")
ReportTestData(self.app, self.client, self.csrf_token) ReportTestData(self.app, "editor").populate()
response: httpx.Response response: httpx.Response
response = client.get(PREFIX) response = client.get(PREFIX)
@ -130,7 +130,7 @@ class ReportTestCase(unittest.TestCase):
:return: None. :return: None.
""" """
client, csrf_token = get_client(self.app, "viewer") client, csrf_token = get_client(self.app, "viewer")
ReportTestData(self.app, self.client, self.csrf_token) ReportTestData(self.app, "editor").populate()
response: httpx.Response response: httpx.Response
response = client.get(PREFIX) response = client.get(PREFIX)
@ -215,7 +215,7 @@ class ReportTestCase(unittest.TestCase):
:return: None. :return: None.
""" """
ReportTestData(self.app, self.client, self.csrf_token) ReportTestData(self.app, "editor").populate()
response: httpx.Response response: httpx.Response
response = self.client.get(PREFIX) response = self.client.get(PREFIX)

View File

@ -71,6 +71,9 @@ def create_app(is_testing: bool = False) -> Flask:
from . import auth from . import auth
auth.init_app(app) auth.init_app(app)
from . import reset
reset.init_app(app)
class UserUtilities(accounting.UserUtilityInterface[auth.User]): class UserUtilities(accounting.UserUtilityInterface[auth.User]):
def can_view(self) -> bool: def can_view(self) -> bool:

File diff suppressed because one or more lines are too long

153
tests/test_site/reset.py Normal file
View File

@ -0,0 +1,153 @@
# The Mia! Accounting Demonstration Website.
# Author: imacat@mail.imacat.idv.tw (imacat), 2023/4/12
# Copyright (c) 2023 imacat.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""The data reset for the Mia! Accounting demonstration website.
"""
import json
import typing as t
from datetime import date, timedelta
from decimal import Decimal
from pathlib import Path
import sqlalchemy as sa
from flask import Flask, Blueprint, url_for, flash, redirect, session, \
render_template
from flask_babel import lazy_gettext
from accounting.utils.cast import s
from . import db
from .auth import User, current_user
bp: Blueprint = Blueprint("reset", __name__, url_prefix="/")
@bp.get("reset", endpoint="reset-page")
def reset() -> str:
"""Resets the sample data.
:return: Redirection to the accounting application.
"""
return render_template("reset.html")
@bp.post("sample", endpoint="sample")
def reset_sample() -> redirect:
"""Resets the sample data.
:return: Redirection to the accounting application.
"""
__reset_database()
__populate_sample_data()
flash(s(lazy_gettext(
"The sample data are emptied and reset successfully.")), "success")
return redirect(url_for("accounting-report.default"))
@bp.post("reset", endpoint="clean-up")
def clean_up() -> redirect:
"""Clean-up the database data.
:return: Redirection to the accounting application.
"""
__reset_database()
db.session.commit()
flash(s(lazy_gettext("The database is emptied successfully.")), "success")
return redirect(url_for("accounting-report.default"))
def __populate_sample_data() -> None:
"""Populates the sample data.
:return: None.
"""
from accounting.models import Account, JournalEntry, JournalEntryLineItem
file: Path = Path(__file__).parent / "data" / "sample.json"
with open(file) as fp:
json_data = json.load(fp)
today: date = date.today()
user: User | None = current_user()
assert user is not None
def filter_journal_entry(data: list[t.Any]) -> dict[str, t.Any]:
"""Filters the journal entry data from JSON.
:param data: The journal entry data.
:return: The journal entry data from JSON.
"""
return {"id": data[0],
"date": today - timedelta(days=data[1]),
"no": data[2],
"note": data[3],
"created_by_id": user.id,
"updated_by_id": user.id}
def filter_line_item(data: list[t.Any]) -> dict[str, t.Any]:
"""Filters the journal entry line item data from JSON.
:param data: The journal entry line item data.
:return: The journal entry line item data from JSON.
"""
return {"id": data[0],
"journal_entry_id": data[1],
"original_line_item_id": data[2],
"is_debit": data[3],
"no": data[4],
"account_id": Account.find_by_code(data[5]).id,
"currency_code": data[6],
"description": data[7],
"amount": Decimal(data[8])}
db.session.execute(sa.insert(JournalEntry),
[filter_journal_entry(x) for x in json_data[0]])
db.session.execute(sa.insert(JournalEntryLineItem),
[filter_line_item(x) for x in json_data[1]])
db.session.commit()
def __reset_database() -> None:
"""Resets the database.
:return: None.
"""
from accounting.models import Currency, CurrencyL10n, BaseAccount, \
BaseAccountL10n, Account, AccountL10n, JournalEntry, \
JournalEntryLineItem
from accounting.base_account import init_base_accounts_command
from accounting.account import init_accounts_command
from accounting.currency import init_currencies_command
JournalEntryLineItem.query.delete()
JournalEntry.query.delete()
CurrencyL10n.query.delete()
Currency.query.delete()
AccountL10n.query.delete()
Account.query.delete()
BaseAccountL10n.query.delete()
BaseAccount.query.delete()
init_base_accounts_command()
init_accounts_command(session["user"])
init_currencies_command(session["user"])
def init_app(app: Flask) -> None:
"""Initialize the localization.
:param app: The Flask application.
:return: None.
"""
app.register_blueprint(bp)

View File

@ -72,6 +72,14 @@ First written: 2023/1/27
</button> </button>
</form> </form>
</li> </li>
{% if current_user().username == "admin" %}
<li>
<a class="dropdown-item {% if request.endpoint and request.endpoint.startswith("reset.") %} active {% endif %}" href="{{ url_for("reset.reset-page") }}">
<i class="fa-solid fa-rotate-right"></i>
{{ _("Reset") }}
</a>
</li>
{% endif %}
</ul> </ul>
</li> </li>
{% else %} {% else %}

View File

@ -0,0 +1,48 @@
{#
The Mia! Accounting Demonstration Website
reset.html: The reset page.
Copyright (c) 2023 imacat.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Author: imacat@mail.imacat.idv.tw (imacat)
First written: 2023/4/12
#}
{% extends "base.html" %}
{% block header %}{% block title %}{{ _("Reset Database") }}{% endblock %}{% endblock %}
{% block content %}
<p>{{ _("Warning: All the current accounting data will be deleted. This cannot be undone. Please backup your database first.") }}</p>
<p>{{ _("Database reset is provided by the live demonstration. This is not part of the Mia! Accounting project.") }}</p>
<form class="mb-2" action="{{ url_for("reset.clean-up") }}" method="post">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
{% if request.args.next %}
<input type="hidden" name="next" value="{{ request.args.next }}">
{% endif %}
<button class="btn btn-primary" type="submit">{{ _("Empty the Database") }}</button>
</form>
<form class="mb-2" action="{{ url_for("reset.sample") }}" method="post">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
{% if request.args.next %}
<input type="hidden" name="next" value="{{ request.args.next }}">
{% endif %}
<button class="btn btn-primary" type="submit">{{ _("Empty and reset the Sample Data") }}</button>
</form>
{% endblock %}

View File

@ -9,8 +9,8 @@ msgid ""
msgstr "" msgstr ""
"Project-Id-Version: mia-accounting-test-site 1.0.0\n" "Project-Id-Version: mia-accounting-test-site 1.0.0\n"
"Report-Msgid-Bugs-To: imacat@mail.imacat.idv.tw\n" "Report-Msgid-Bugs-To: imacat@mail.imacat.idv.tw\n"
"POT-Creation-Date: 2023-04-11 22:18+0800\n" "POT-Creation-Date: 2023-04-12 17:59+0800\n"
"PO-Revision-Date: 2023-04-11 22:18+0800\n" "PO-Revision-Date: 2023-04-12 18:00+0800\n"
"Last-Translator: imacat <imacat@mail.imacat.idv.tw>\n" "Last-Translator: imacat <imacat@mail.imacat.idv.tw>\n"
"Language: zh_Hant\n" "Language: zh_Hant\n"
"Language-Team: zh_Hant <imacat@mail.imacat.idv.tw>\n" "Language-Team: zh_Hant <imacat@mail.imacat.idv.tw>\n"
@ -20,6 +20,14 @@ msgstr ""
"Content-Transfer-Encoding: 8bit\n" "Content-Transfer-Encoding: 8bit\n"
"Generated-By: Babel 2.12.1\n" "Generated-By: Babel 2.12.1\n"
#: tests/test_site/reset.py:55
msgid "The sample data are emptied and reset successfully."
msgstr "範例資料已清空重設。"
#: tests/test_site/reset.py:68
msgid "The database is emptied successfully."
msgstr "資料庫已清空。"
#: tests/test_site/templates/base.html:23 #: tests/test_site/templates/base.html:23
msgid "en" msgid "en"
msgstr "zh-Hant" msgstr "zh-Hant"
@ -32,12 +40,16 @@ msgstr "首頁"
msgid "Log Out" msgid "Log Out"
msgstr "登出" msgstr "登出"
#: tests/test_site/templates/base.html:81 #: tests/test_site/templates/base.html:79
msgid "Reset"
msgstr "重設"
#: tests/test_site/templates/base.html:89
#: tests/test_site/templates/login.html:24 #: tests/test_site/templates/login.html:24
msgid "Log In" msgid "Log In"
msgstr "登入" msgstr "登入"
#: tests/test_site/templates/base.html:122 #: tests/test_site/templates/base.html:130
msgid "Error:" msgid "Error:"
msgstr "錯誤:" msgstr "錯誤:"
@ -77,3 +89,27 @@ msgstr "管理者"
msgid "Nobody" msgid "Nobody"
msgstr "沒有權限者" msgstr "沒有權限者"
#: tests/test_site/templates/reset.html:24
msgid "Reset Database"
msgstr "資料庫重設"
#: tests/test_site/templates/reset.html:28
msgid ""
"Warning: All the current accounting data will be deleted. This cannot be"
" undone. Please backup your database first."
msgstr "警告:現有資料會全部刪除,無法復原。請先備份您的資料。"
#: tests/test_site/templates/reset.html:30
msgid ""
"Database reset is provided by the live demonstration. This is not part "
"of the Mia! Accounting project."
msgstr "資料庫重設是示範站的功能,不是 Mia! Accounting 的功能。"
#: tests/test_site/templates/reset.html:37
msgid "Empty the Database"
msgstr "清空資料庫"
#: tests/test_site/templates/reset.html:45
msgid "Empty and reset the Sample Data"
msgstr "清空並重設範例資料"

View File

@ -54,7 +54,7 @@ class UnmatchedOffsetTestCase(unittest.TestCase):
:return: None. :return: None.
""" """
client, csrf_token = get_client(self.app, "nobody") client, csrf_token = get_client(self.app, "nobody")
DifferentTestData(self.app, self.client, self.csrf_token) DifferentTestData(self.app, "nobody").populate()
response: httpx.Response response: httpx.Response
response = client.get(PREFIX) response = client.get(PREFIX)
@ -73,7 +73,7 @@ class UnmatchedOffsetTestCase(unittest.TestCase):
:return: None. :return: None.
""" """
client, csrf_token = get_client(self.app, "viewer") client, csrf_token = get_client(self.app, "viewer")
DifferentTestData(self.app, self.client, self.csrf_token) DifferentTestData(self.app, "viewer").populate()
response: httpx.Response response: httpx.Response
response = client.get(PREFIX) response = client.get(PREFIX)
@ -91,7 +91,7 @@ class UnmatchedOffsetTestCase(unittest.TestCase):
:return: None. :return: None.
""" """
DifferentTestData(self.app, self.client, self.csrf_token) DifferentTestData(self.app, "editor").populate()
response: httpx.Response response: httpx.Response
response = self.client.get(PREFIX) response = self.client.get(PREFIX)
@ -132,8 +132,8 @@ class UnmatchedOffsetTestCase(unittest.TestCase):
""" """
from accounting.models import Account, JournalEntryLineItem from accounting.models import Account, JournalEntryLineItem
from accounting.utils.offset_matcher import OffsetMatcher from accounting.utils.offset_matcher import OffsetMatcher
data: DifferentTestData \ data: DifferentTestData = DifferentTestData(self.app, "editor")
= DifferentTestData(self.app, self.client, self.csrf_token) data.populate()
account: Account | None account: Account | None
line_item: JournalEntryLineItem | None line_item: JournalEntryLineItem | None
matcher: OffsetMatcher matcher: OffsetMatcher
@ -248,8 +248,8 @@ class UnmatchedOffsetTestCase(unittest.TestCase):
""" """
from accounting.models import Account, JournalEntryLineItem from accounting.models import Account, JournalEntryLineItem
from accounting.utils.offset_matcher import OffsetMatcher from accounting.utils.offset_matcher import OffsetMatcher
data: SameTestData \ data: SameTestData = SameTestData(self.app, "editor")
= SameTestData(self.app, self.client, self.csrf_token) data.populate()
account: Account | None account: Account | None
line_item: JournalEntryLineItem | None line_item: JournalEntryLineItem | None
matcher: OffsetMatcher matcher: OffsetMatcher
@ -483,14 +483,12 @@ class DifferentTestData(BaseTestData):
5, [JournalEntryCurrencyData( 5, [JournalEntryCurrencyData(
"USD", [self.l_p_of5d], [self.l_p_of5c])]) "USD", [self.l_p_of5d], [self.l_p_of5c])])
self._set_need_offset({Accounts.RECEIVABLE, Accounts.PAYABLE}, False)
self._add_journal_entry(self.j_r_of1) self._add_journal_entry(self.j_r_of1)
self._add_journal_entry(self.j_r_of2) self._add_journal_entry(self.j_r_of2)
self._add_journal_entry(self.j_r_of3) self._add_journal_entry(self.j_r_of3)
self._add_journal_entry(self.j_p_of1) self._add_journal_entry(self.j_p_of1)
self._add_journal_entry(self.j_p_of2) self._add_journal_entry(self.j_p_of2)
self._add_journal_entry(self.j_p_of3) self._add_journal_entry(self.j_p_of3)
self._set_need_offset({Accounts.RECEIVABLE, Accounts.PAYABLE}, True)
class SameTestData(BaseTestData): class SameTestData(BaseTestData):
@ -525,8 +523,6 @@ class SameTestData(BaseTestData):
self.l_p_or6d, self.l_p_or6c = self._add_simple_journal_entry( self.l_p_or6d, self.l_p_or6c = self._add_simple_journal_entry(
10, "USD", "Steak", "120", Accounts.MEAL, Accounts.PAYABLE) 10, "USD", "Steak", "120", Accounts.MEAL, Accounts.PAYABLE)
self._set_need_offset({Accounts.RECEIVABLE, Accounts.PAYABLE}, False)
# Receivable offset items # Receivable offset items
self.l_r_of1d, self.l_r_of1c = self._add_simple_journal_entry( self.l_r_of1d, self.l_r_of1c = self._add_simple_journal_entry(
65, "USD", "Noodles", "100", Accounts.CASH, Accounts.RECEIVABLE) 65, "USD", "Noodles", "100", Accounts.CASH, Accounts.RECEIVABLE)
@ -563,6 +559,5 @@ class SameTestData(BaseTestData):
self.l_p_of6d, self.l_p_of6c = self._add_simple_journal_entry( self.l_p_of6d, self.l_p_of6c = self._add_simple_journal_entry(
15, "USD", "Steak", "120", Accounts.PAYABLE, Accounts.CASH) 15, "USD", "Steak", "120", Accounts.PAYABLE, Accounts.CASH)
self._set_need_offset({Accounts.RECEIVABLE, Accounts.PAYABLE}, True)
self._add_journal_entry(j_r_of3) self._add_journal_entry(j_r_of3)
self._add_journal_entry(j_p_of3) self._add_journal_entry(j_p_of3)

View File

@ -19,17 +19,21 @@
""" """
from __future__ import annotations from __future__ import annotations
import json
import re import re
import typing as t import typing as t
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from datetime import date, timedelta from datetime import date, timedelta
from secrets import randbelow
from _decimal import Decimal from decimal import Decimal
import sqlalchemy as sa
import httpx import httpx
from flask import Flask, render_template_string from flask import Flask, render_template_string
from test_site import create_app, db from test_site import create_app, db
from test_site.auth import User
TEST_SERVER: str = "https://testserver" TEST_SERVER: str = "https://testserver"
"""The test server URI.""" """The test server URI."""
@ -44,6 +48,7 @@ class Accounts:
BANK: str = "1113-001" BANK: str = "1113-001"
NOTES_RECEIVABLE: str = "1131-001" NOTES_RECEIVABLE: str = "1131-001"
RECEIVABLE: str = "1141-001" RECEIVABLE: str = "1141-001"
MACHINERY: str = "1441-001"
PREPAID: str = "1258-001" PREPAID: str = "1258-001"
NOTES_PAYABLE: str = "2131-001" NOTES_PAYABLE: str = "2131-001"
PAYABLE: str = "2141-001" PAYABLE: str = "2141-001"
@ -163,7 +168,7 @@ def match_journal_entry_detail(location: str) -> int:
class JournalEntryLineItemData: class JournalEntryLineItemData:
"""The journal entry line item data.""" """The journal entry line item data."""
def __init__(self, account: str, description: str, amount: str, def __init__(self, account: str, description: str | None, amount: str,
original_line_item: JournalEntryLineItemData | None = None): original_line_item: JournalEntryLineItemData | None = None):
"""Constructs the journal entry line item data. """Constructs the journal entry line item data.
@ -178,7 +183,7 @@ class JournalEntryLineItemData:
self.original_line_item: JournalEntryLineItemData | None \ self.original_line_item: JournalEntryLineItemData | None \
= original_line_item = original_line_item
self.account: str = account self.account: str = account
self.description: str = description self.description: str | None = description
self.amount: Decimal = Decimal(amount) self.amount: Decimal = Decimal(amount)
def form(self, prefix: str, debit_credit: str, index: int, def form(self, prefix: str, debit_credit: str, index: int,
@ -294,17 +299,21 @@ class JournalEntryData:
class BaseTestData(ABC): class BaseTestData(ABC):
"""The base test data.""" """The base test data."""
def __init__(self, app: Flask, client: httpx.Client, csrf_token: str): def __init__(self, app: Flask, username: str):
"""Constructs the test data. """Constructs the test data.
:param app: The Flask application. :param app: The Flask application.
:param client: The client. :param username: The username.
:param csrf_token: The CSRF token.
""" """
self.app: Flask = app self.__app: Flask = app
self.client: httpx.Client = client with self.__app.app_context():
self.csrf_token: str = csrf_token current_user: User | None = User.query\
self._init_data() .filter(User.username == username).first()
assert current_user is not None
self.__current_user_id: int = current_user.id
self.__journal_entries: list[dict[str, t.Any]] = []
self.__line_items: list[dict[str, t.Any]] = []
self._init_data()
@abstractmethod @abstractmethod
def _init_data(self) -> None: def _init_data(self) -> None:
@ -313,6 +322,61 @@ class BaseTestData(ABC):
:return: None :return: None
""" """
def populate(self) -> None:
"""Populates the data into the database.
:return: None
"""
from accounting.models import JournalEntry, JournalEntryLineItem
with self.__app.app_context():
db.session.execute(sa.insert(JournalEntry), self.__journal_entries)
db.session.execute(sa.insert(JournalEntryLineItem),
self.__line_items)
db.session.commit()
def json(self) -> str:
"""Returns the data as JSON.
:return: The JSON string.
"""
from accounting.models import Account
today: date = date.today()
def filter_journal_entry(data: dict[str, t.Any]) -> list[t.Any]:
"""Filters the journal entry data for JSON encoding.
:param data: The journal entry data.
:return: The journal entry data for JSON encoding.
"""
data = data.copy()
data["date"] = (today - data["date"]).days
del data["created_by_id"]
del data["updated_by_id"]
return [data[x] for x in ["id", "date", "no", "note"]]
def filter_line_item(data: dict[str, t.Any]) -> list[t.Any]:
"""Filters the journal entry line item data for JSON encoding.
:param data: The journal entry line item data.
:return: The journal entry line item data for JSON encoding.
"""
data = data.copy()
with self.__app.app_context():
data["account_id"] \
= db.session.get(Account, data["account_id"]).code
data["amount"] = str(data["amount"])
if "original_line_item_id" not in data:
data["original_line_item_id"] = None
return [data[x] for x in ["id", "journal_entry_id",
"original_line_item_id", "is_debit",
"no", "account_id", "currency_code",
"description", "amount"]]
return json.dumps(
[[filter_journal_entry(x) for x in self.__journal_entries],
[filter_line_item(x) for x in self.__line_items]],
ensure_ascii=False, separators=(",", ":"))
@staticmethod @staticmethod
def _couple(description: str, amount: str, debit: str, credit: str) \ def _couple(description: str, amount: str, debit: str, credit: str) \
-> tuple[JournalEntryLineItemData, JournalEntryLineItemData]: -> tuple[JournalEntryLineItemData, JournalEntryLineItemData]:
@ -333,26 +397,82 @@ class BaseTestData(ABC):
:param journal_entry_data: The journal entry data. :param journal_entry_data: The journal entry data.
:return: None. :return: None.
""" """
from accounting.models import JournalEntry from accounting.models import Account
store_uri: str = "/accounting/journal-entries/store/transfer" existing_j_id: set[int] = {x["id"] for x in self.__journal_entries}
existing_l_id: set[int] = {x["id"] for x in self.__line_items}
journal_entry_data.id = self.__new_id(existing_j_id)
j_date: date = date.today() - timedelta(days=journal_entry_data.days)
self.__journal_entries.append(
{"id": journal_entry_data.id,
"date": j_date,
"no": self.__next_j_no(j_date),
"note": journal_entry_data.note,
"created_by_id": self.__current_user_id,
"updated_by_id": self.__current_user_id})
debit_no: int = 0
credit_no: int = 0
for currency in journal_entry_data.currencies:
for line_item in currency.debit:
account: Account | None \
= Account.find_by_code(line_item.account)
assert account is not None
debit_no = debit_no + 1
line_item.id = self.__new_id(existing_l_id)
data: dict[str, t.Any] \
= {"id": line_item.id,
"journal_entry_id": journal_entry_data.id,
"is_debit": True,
"no": debit_no,
"account_id": account.id,
"currency_code": currency.code,
"description": line_item.description,
"amount": line_item.amount}
if line_item.original_line_item is not None:
data["original_line_item_id"] \
= line_item.original_line_item.id
self.__line_items.append(data)
for line_item in currency.credit:
account: Account | None \
= Account.find_by_code(line_item.account)
assert account is not None
credit_no = credit_no + 1
line_item.id = self.__new_id(existing_l_id)
data: dict[str, t.Any] \
= {"id": line_item.id,
"journal_entry_id": journal_entry_data.id,
"is_debit": False,
"no": credit_no,
"account_id": account.id,
"currency_code": currency.code,
"description": line_item.description,
"amount": line_item.amount}
if line_item.original_line_item is not None:
data["original_line_item_id"] \
= line_item.original_line_item.id
self.__line_items.append(data)
response: httpx.Response = self.client.post( @staticmethod
store_uri, data=journal_entry_data.new_form(self.csrf_token)) def __new_id(existing_id: set[int]) -> int:
assert response.status_code == 302 """Generates and returns a new random unique ID.
journal_entry_id: int \
= match_journal_entry_detail(response.headers["Location"]) :param existing_id: The existing ID.
journal_entry_data.id = journal_entry_id :return: The newly-generated random unique ID.
with self.app.app_context(): """
journal_entry: JournalEntry | None \ while True:
= db.session.get(JournalEntry, journal_entry_id) obj_id: int = 100000000 + randbelow(900000000)
assert journal_entry is not None if obj_id not in existing_id:
for i in range(len(journal_entry.currencies)): existing_id.add(obj_id)
for j in range(len(journal_entry.currencies[i].debit)): return obj_id
journal_entry_data.currencies[i].debit[j].id \
= journal_entry.currencies[i].debit[j].id def __next_j_no(self, j_date: date) -> int:
for j in range(len(journal_entry.currencies[i].credit)): """Returns the next journal entry number in a day.
journal_entry_data.currencies[i].credit[j].id \
= journal_entry.currencies[i].credit[j].id :param j_date: The journal entry date.
:return: The next journal entry number.
"""
existing: set[int] = {x["no"] for x in self.__journal_entries
if x["date"] == j_date}
return 1 if len(existing) == 0 else max(existing) + 1
def _add_simple_journal_entry( def _add_simple_journal_entry(
self, days: int, currency: str, description: str, amount: str, self, days: int, currency: str, description: str, amount: str,
@ -374,20 +494,3 @@ class BaseTestData(ABC):
days, [JournalEntryCurrencyData( days, [JournalEntryCurrencyData(
currency, [debit_item], [credit_item])])) currency, [debit_item], [credit_item])]))
return debit_item, credit_item return debit_item, credit_item
def _set_need_offset(self, account_codes: set[str],
is_need_offset: bool) -> None:
"""Sets whether the line items in some accounts need offset.
:param account_codes: The account codes.
:param is_need_offset: True if the line items in the accounts need
offset, or False otherwise.
:return:
"""
from accounting.models import Account
with self.app.app_context():
for code in account_codes:
account: Account | None = Account.find_by_code(code)
assert account is not None
account.is_need_offset = is_need_offset
db.session.commit()