Projects/dibbler
10
2
Fork 0
Code Issues 21 Pull Requests 1 Actions Activity

fixup! WIP
All checks were successful
Run tests / run-tests (push) Successful in 1m23s
Details

Browse Source
This commit is contained in:
Oystein Kristoffer Tveit
2025-12-10 15:38:32 +09:00
parent a4a22e6565
commit e9ce51b97b
16 changed files with 518 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
dibbler/lib/query_helpers.py
View File

@@ -3,6 +3,7 @@ from sqlalchemy import BindParameter, literal
T = TypeVar("T")
def const(value: T) -> BindParameter[T]:
"""
Create a constant SQL literal bind parameter.
@@ -13,6 +14,7 @@ def const(value: T) -> BindParameter[T]:
return literal(value, literal_execute=True)
CONST_ZERO: BindParameter[int] = const(0)
CONST_ONE: BindParameter[int] = const(1)
CONST_TRUE: BindParameter[bool] = const(True)

9
dibbler/queries/adjust_interest.py
View File

@@ -1,6 +1,6 @@
from sqlalchemy.orm import Session
from dibbler.models import Transaction
from dibbler.models import Transaction, User
# TODO: this type of transaction should be password protected.
# the password can be set as a string literal in the config file.
@@ -8,15 +8,18 @@ from dibbler.models import Transaction
def adjust_interest(
sql_session: Session,
user_id: int,
user: User,
new_interest: int,
message: str | None = None,
) -> None:
if new_interest < 0:
raise ValueError("Interest rate cannot be negative")
if user.id is None:
raise ValueError("User must be persisted in the database.")
transaction = Transaction.adjust_interest(
user_id=user_id,
user_id=user.id,
interest_rate_percent=new_interest,
message=message,
)

9
dibbler/queries/adjust_penalty.py
View File

@@ -1,6 +1,6 @@
from sqlalchemy.orm import Session
from dibbler.models import Transaction
from dibbler.models import Transaction, User
from dibbler.queries.current_penalty import current_penalty
# TODO: this type of transaction should be password protected.
@@ -9,7 +9,7 @@ from dibbler.queries.current_penalty import current_penalty
def adjust_penalty(
sql_session: Session,
user_id: int,
user: User,
new_penalty: int | None = None,
new_penalty_multiplier: int | None = None,
message: str | None = None,
@@ -20,6 +20,9 @@ def adjust_penalty(
if new_penalty_multiplier is not None and new_penalty_multiplier < 100:
raise ValueError("Penalty multiplier cannot be less than 100%")
if user.id is None:
raise ValueError("User must be persisted in the database.")
if new_penalty is None or new_penalty_multiplier is None:
existing_penalty, existing_penalty_multiplier = current_penalty(sql_session)
if new_penalty is None:
@@ -28,7 +31,7 @@ def adjust_penalty(
new_penalty_multiplier = existing_penalty_multiplier
transaction = Transaction.adjust_penalty(
user_id=user_id,
user_id=user.id,
penalty_threshold=new_penalty,
penalty_multiplier_percent=new_penalty_multiplier,
message=message,

3
dibbler/queries/current_interest.py
View File

@@ -5,6 +5,9 @@ from dibbler.models import Transaction, TransactionType
from dibbler.models.Transaction import DEFAULT_INTEREST_RATE_PERCENTAGE
# TODO: add until transaction parameter
# TODO: add until datetime parameter
def current_interest(sql_session: Session) -> int:
result = sql_session.scalars(
select(Transaction)

3
dibbler/queries/current_penalty.py
View File

@@ -8,6 +8,9 @@ from dibbler.models.Transaction import (
)
# TODO: add until transaction parameter
# TODO: add until datetime parameter
def current_penalty(sql_session: Session) -> tuple[int, int]:
result = sql_session.scalars(
select(Transaction)

6
dibbler/queries/joint_buy_product.py
View File

@@ -22,9 +22,15 @@ def joint_buy_product(
Create buy product transactions for multiple users at once.
"""
if product.id is None:
raise ValueError("Product must be persisted in the database.")
if instigator not in users:
raise ValueError("Instigator must be in the list of users buying the product.")
if any(user.id is None for user in users):
raise ValueError("All users must be persisted in the database.")
if product_count <= 0:
raise ValueError("Product count must be positive.")

15
dibbler/queries/product_owners.py
View File

@@ -8,12 +8,11 @@ from sqlalchemy import (
bindparam,
case,
func,
literal,
select,
)
from sqlalchemy.orm import Session
from dibbler.lib.query_helpers import CONST_NONE, CONST_ONE, CONST_TRUE, CONST_ZERO, const
from dibbler.lib.query_helpers import CONST_NONE, CONST_ONE, CONST_TRUE, CONST_ZERO
from dibbler.models import (
Product,
Transaction,
@@ -168,6 +167,7 @@ class ProductOwnersLogEntry:
user: User | None
products_left_to_account_for: int
# TODO: add until datetime parameter
def product_owners_log(
sql_session: Session,
@@ -181,6 +181,12 @@ def product_owners_log(
If 'until' is given, only transactions up to that time are considered.
"""
if product.id is None:
raise ValueError("Product must be persisted in the database.")
if until is not None and until.id is None:
raise ValueError("'until' transaction must be persisted in the database.")
recursive_cte = _product_owners_query(
product_id=product.id,
use_cache=use_cache,
@@ -222,6 +228,8 @@ def product_owners_log(
]
# TODO: add until transaction parameter
def product_owners(
sql_session: Session,
product: Product,
@@ -234,6 +242,9 @@ def product_owners(
If 'until' is given, only transactions up to that time are considered.
"""
if product.id is None:
raise ValueError("Product must be persisted in the database.")
recursive_cte = _product_owners_query(
product_id=product.id,
use_cache=use_cache,

19
dibbler/queries/product_price.py
View File

@@ -10,12 +10,11 @@ from sqlalchemy import (
case,
cast,
func,
literal,
select,
)
from sqlalchemy.orm import Session
from dibbler.lib.query_helpers import CONST_NONE, CONST_ONE, CONST_TRUE, CONST_ZERO, const
from dibbler.lib.query_helpers import CONST_NONE, CONST_ONE, CONST_TRUE, CONST_ZERO
from dibbler.models import (
Product,
Transaction,
@@ -171,6 +170,8 @@ class ProductPriceLogEntry:
product_count: int
# TODO: add until datetime parameter
def product_price_log(
sql_session: Session,
product: Product,
@@ -181,6 +182,12 @@ def product_price_log(
Calculates the price of a product and returns a log of the price changes.
"""
if product.id is None:
raise ValueError("Product must be persisted in the database.")
if until is not None and until.id is None:
raise ValueError("'until' transaction must be persisted in the database.")
recursive_cte = _product_price_query(
product.id,
use_cache=use_cache,
@@ -217,6 +224,8 @@ def product_price_log(
]
# TODO: add until datetime parameter
def product_price(
sql_session: Session,
product: Product,
@@ -228,6 +237,12 @@ def product_price(
Calculates the price of a product.
"""
if product.id is None:
raise ValueError("Product must be persisted in the database.")
if until is not None and until.id is None:
raise ValueError("'until' transaction must be persisted in the database.")
recursive_cte = _product_price_query(
product.id,
use_cache=use_cache,

5
dibbler/queries/product_stock.py
View File

@@ -78,6 +78,8 @@ def _product_stock_query(
return query
# TODO: add until transaction parameter
def product_stock(
sql_session: Session,
product: Product,
@@ -90,6 +92,9 @@ def product_stock(
If 'until' is given, only transactions up to that time are considered.
"""
if product.id is None:
raise ValueError("Product must be persisted in the database.")
query = _product_stock_query(
product_id=product.id,
use_cache=use_cache,

6
dibbler/queries/transaction_log.py
View File

@@ -38,6 +38,12 @@ def transaction_log(
if not (user is None or product is None):
raise ValueError("Cannot filter by both user and product.")
if user is not None and user.id is None:
raise ValueError("User must be persisted in the database.")
if product is not None and product.id is None:
raise ValueError("Product must be persisted in the database.")
if not (after_time is None or after_transaction_id is None):
raise ValueError("Cannot filter by both from_time and from_transaction_id.")

13
dibbler/queries/user_balance.py
View File

@@ -11,7 +11,6 @@ from sqlalchemy import (
cast,
column,
func,
literal,
or_,
select,
)
@@ -260,6 +259,7 @@ class UserBalanceLogEntry:
# return self.transaction.type_ == TransactionType.BUY_PRODUCT and prev?
# TODO: add until datetime parameter
def user_balance_log(
sql_session: Session,
@@ -273,6 +273,12 @@ def user_balance_log(
If 'until' is given, only transactions up to that time are considered.
"""
if user.id is None:
raise ValueError("User must be persisted in the database.")
if until is not None and until.id is None:
raise ValueError("'until' transaction must be persisted in the database.")
recursive_cte = _user_balance_query(
user.id,
use_cache=use_cache,
@@ -313,6 +319,8 @@ def user_balance_log(
]
# TODO: add until datetime parameter
def user_balance(
sql_session: Session,
user: User,
@@ -325,6 +333,9 @@ def user_balance(
If 'until' is given, only transactions up to that time are considered.
"""
if user.id is None:
raise ValueError("User must be persisted in the database.")
recursive_cte = _user_balance_query(
user.id,
use_cache=use_cache,

6
tests/queries/test_adjust_interest.py
View File

@@ -20,7 +20,7 @@ def test_adjust_interest_no_history(sql_session: Session) -> None:
adjust_interest(
sql_session,
user_id=user.id,
user=user,
new_interest=3,
message="Setting initial interest rate",
)
@@ -50,7 +50,7 @@ def test_adjust_interest_existing_history(sql_session: Session) -> None:
adjust_interest(
sql_session,
user_id=user.id,
user=user,
new_interest=2,
message="Adjusting interest rate",
)
@@ -66,7 +66,7 @@ def test_adjust_interest_negative_failure(sql_session: Session) -> None:
with pytest.raises(ValueError, match="Interest rate cannot be negative"):
adjust_interest(
sql_session,
user_id=user.id,
user=user,
new_interest=-1,
message="Attempting to set negative interest rate",
)

14
tests/queries/test_adjust_penalty.py
View File

@@ -24,7 +24,7 @@ def test_adjust_penalty_no_history(sql_session: Session) -> None:
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty=-200,
message="Setting initial interest rate",
)
@@ -41,7 +41,7 @@ def test_adjust_penalty_multiplier_no_history(sql_session: Session) -> None:
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty_multiplier=125,
message="Setting initial interest rate",
)
@@ -58,7 +58,7 @@ def test_adjust_penalty_multiplier_less_than_100_fail(sql_session: Session) -> N
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty_multiplier=100,
message="Setting initial interest rate",
)
@@ -71,7 +71,7 @@ def test_adjust_penalty_multiplier_less_than_100_fail(sql_session: Session) -> N
with pytest.raises(ValueError, match="Penalty multiplier cannot be less than 100%"):
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty_multiplier=99,
message="Setting initial interest rate",
)
@@ -97,7 +97,7 @@ def test_adjust_penalty_existing_history(sql_session: Session) -> None:
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty=-250,
message="Adjusting penalty threshold",
)
@@ -127,7 +127,7 @@ def test_adjust_penalty_multiplier_existing_history(sql_session: Session) -> Non
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty_multiplier=130,
message="Adjusting penalty multiplier",
)
@@ -141,7 +141,7 @@ def test_adjust_penalty_and_multiplier(sql_session: Session) -> None:
adjust_penalty(
sql_session,
user_id=user.id,
user=user,
new_penalty=-300,
new_penalty_multiplier=150,
message="Setting both penalty and multiplier",

174
tests/queries/test_joint_buy_product.py
View File

@@ -1,26 +1,174 @@
from datetime import datetime
import pytest
from sqlalchemy.orm import Session
@pytest.mark.skip(reason="Not yet implemented")
def test_joint_buy_product_missing_product(sql_session: Session) -> None: ...
from dibbler.models import Product, Transaction, User
from dibbler.queries import joint_buy_product
@pytest.mark.skip(reason="Not yet implemented")
def test_joint_buy_product_missing_user(sql_session: Session) -> None: ...
def insert_test_data(sql_session: Session) -> tuple[User, User, User, Product]:
user1 = User("Test User 1")
user2 = User("Test User 2")
user3 = User("Test User 3")
product = Product("1234567890123", "Test Product")
sql_session.add_all([user1, user2, user3, product])
sql_session.commit()
transactions = [
Transaction.add_product(
user_id=user1.id,
product_id=product.id,
amount=30,
per_product=10,
product_count=3,
time=datetime(2024, 1, 1, 10, 0, 0),
)
]
sql_session.add_all(transactions)
sql_session.commit()
return user1, user2, user3, product
@pytest.mark.skip(reason="Not yet implemented")
def test_joint_buy_product_out_of_stock(sql_session: Session) -> None: ...
def test_joint_buy_product_missing_product(sql_session: Session) -> None:
user = User("Test User 1")
sql_session.add(user)
sql_session.commit()
product = Product("1234567890123", "Test Product")
with pytest.raises(ValueError):
joint_buy_product(
sql_session,
instigator=user,
users=[user],
product=product,
product_count=1,
)
@pytest.mark.skip(reason="Not yet implemented")
def test_joint_buy_product(sql_session: Session) -> None: ...
def test_joint_buy_product_missing_user(sql_session: Session) -> None:
user = User("Test User 1")
product = Product("1234567890123", "Test Product")
sql_session.add(product)
sql_session.commit()
with pytest.raises(ValueError):
joint_buy_product(
sql_session,
instigator=user,
users=[user],
product=product,
product_count=1,
)
@pytest.mark.skip(reason="Not yet implemented")
def test_joint_buy_product_duplicate_user(sql_session: Session) -> None: ...
def test_joint_buy_product_invalid_product_count(sql_session: Session) -> None:
user, _, _, product = insert_test_data(sql_session)
with pytest.raises(ValueError):
joint_buy_product(
sql_session,
instigator=user,
users=[user],
product=product,
product_count=0,
)
with pytest.raises(ValueError):
joint_buy_product(
sql_session,
instigator=user,
users=[user],
product=product,
product_count=-1,
)
@pytest.mark.skip(reason="Not yet implemented")
def test_joint_buy_product_non_involved_instigator(sql_session: Session) -> None: ...
def test_joint_single_user(sql_session: Session) -> None:
user, _, _, product = insert_test_data(sql_session)
joint_buy_product(
sql_session,
instigator=user,
users=[user],
product=product,
product_count=1,
)
def test_joint_buy_product(sql_session: Session) -> None:
user, user2, user3, product = insert_test_data(sql_session)
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2, user3],
product=product,
product_count=1,
)
def test_joint_buy_product_more_than_in_stock(sql_session: Session) -> None:
user, user2, user3, product = insert_test_data(sql_session)
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2, user3],
product=product,
product_count=5,
)
def test_joint_buy_product_out_of_stock(sql_session: Session) -> None:
user, user2, user3, product = insert_test_data(sql_session)
transactions = [
Transaction.buy_product(
user_id=user.id,
product_id=product.id,
product_count=3,
time=datetime(2024, 1, 2, 10, 0, 0),
)
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2, user3],
product=product,
product_count=10,
)
def test_joint_buy_product_duplicate_user(sql_session: Session) -> None:
user, user2, _, product = insert_test_data(sql_session)
joint_buy_product(
sql_session,
instigator=user,
users=[user, user, user2],
product=product,
product_count=1,
)
def test_joint_buy_product_non_involved_instigator(sql_session: Session) -> None:
user, user2, user3, product = insert_test_data(sql_session)
with pytest.raises(ValueError):
joint_buy_product(
sql_session,
instigator=user,
users=[user2, user3],
product=product,
product_count=1,
)

28
tests/queries/test_product_stock.py
View File

@@ -1,7 +1,6 @@
from datetime import datetime
import pytest
from sqlalchemy import select
from sqlalchemy.orm import Session
@@ -215,9 +214,28 @@ def test_product_stock_joint_transaction(sql_session: Session) -> None:
assert product_stock(sql_session, product) == 5 - 3
@pytest.mark.skip(reason="Not yet implemented")
def test_product_stock_until(sql_session: Session) -> None: ...
def test_product_stock_until(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 12, 0, 0),
amount=10,
per_product=10,
user_id=user.id,
product_id=product.id,
product_count=1,
),
Transaction.add_product(
time=datetime(2023, 10, 2, 12, 0, 0),
amount=20,
per_product=10,
user_id=user.id,
product_id=product.id,
product_count=2,
),
]
sql_session.add_all(transactions)
sql_session.commit()
@pytest.mark.skip(reason="Not yet implemented")
def test_product_stock_throw_away(sql_session: Session) -> None: ...
assert product_stock(sql_session, product, until=datetime(2023, 10, 1, 23, 59, 59)) == 1

285
tests/queries/test_user_balance.py
View File

@@ -6,25 +6,27 @@ import pytest
from sqlalchemy.orm import Session
from dibbler.models import Product, Transaction, User
from dibbler.queries import user_balance, user_balance_log
from dibbler.models.Transaction import DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE
from dibbler.queries import joint_buy_product, user_balance, user_balance_log
# TODO: see if we can use pytest_runtest_makereport to print the "user_balance_log"s
# only on failures instead of inlining it in every test function
def insert_test_data(sql_session: Session) -> tuple[User, Product]:
def insert_test_data(sql_session: Session) -> tuple[User, User, User, Product]:
user = User("Test User")
user2 = User("Test User 2")
user3 = User("Test User 3")
product = Product("1234567890123", "Test Product")
sql_session.add(user)
sql_session.add(product)
sql_session.add_all([user, user2, user3, product])
sql_session.commit()
return user, product
return user, user2, user3, product
def test_user_balance_no_transactions(sql_session: Session) -> None:
user, _ = insert_test_data(sql_session)
user, *_ = insert_test_data(sql_session)
pprint(user_balance_log(sql_session, user))
@@ -34,7 +36,7 @@ def test_user_balance_no_transactions(sql_session: Session) -> None:
def test_user_balance_basic_history(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.adjust_balance(
@@ -63,11 +65,7 @@ def test_user_balance_basic_history(sql_session: Session) -> None:
def test_user_balance_with_transfers(sql_session: Session) -> None:
user1, product = insert_test_data(sql_session)
user2 = User("Test User 2")
sql_session.add(user2)
sql_session.commit()
user1, user2, _, product = insert_test_data(sql_session)
transactions = [
Transaction.adjust_balance(
@@ -108,7 +106,7 @@ def test_user_balance_complex_history(sql_session: Session) -> None: ...
def test_user_balance_penalty(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
@@ -137,11 +135,13 @@ def test_user_balance_penalty(sql_session: Session) -> None:
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == 27 - 200 - (27 * 2)
assert user_balance(sql_session, user) == 27 - 200 - (
27 * (DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE // 100)
)
def test_user_balance_changing_penalty(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
@@ -184,11 +184,13 @@ def test_user_balance_changing_penalty(sql_session: Session) -> None:
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == 27 - 200 - (27 * 2) - (27 * 3)
assert user_balance(sql_session, user) == 27 - 200 - (
27 * (DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE // 100)
) - (27 * 3)
def test_user_balance_interest(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
@@ -220,7 +222,7 @@ def test_user_balance_interest(sql_session: Session) -> None:
def test_user_balance_changing_interest(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
@@ -265,7 +267,7 @@ def test_user_balance_changing_interest(sql_session: Session) -> None:
def test_user_balance_penalty_interest_combined(sql_session: Session) -> None:
user, product = insert_test_data(sql_session)
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
@@ -300,36 +302,239 @@ def test_user_balance_penalty_interest_combined(sql_session: Session) -> None:
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (27 - 200 - math.ceil(27 * 2 * 1.1))
assert user_balance(sql_session, user) == (
27 - 200 - math.ceil(27 * (DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE // 100) * 1.1)
)
def test_user_balance_joint_transaction_single_user(sql_session: Session) -> None:
user, _, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 10, 0, 0),
user_id=user.id,
product_id=product.id,
amount=27 * 3,
per_product=27,
product_count=3,
),
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user],
product=product,
product_count=2,
)
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (27 * 3) - (27 * 2)
def test_user_balance_joint_transactions_multiple_users(sql_session: Session) -> None:
user, user2, user3, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 10, 0, 0),
user_id=user.id,
product_id=product.id,
amount=27 * 3,
per_product=27,
product_count=3,
),
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2, user3],
product=product,
product_count=2,
)
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (27 * 3) - math.ceil((27 * 2) / 3)
def test_user_balance_joint_transactions_multiple_times_self(sql_session: Session) -> None:
user, user2, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 10, 0, 0),
user_id=user.id,
product_id=product.id,
amount=27 * 3,
per_product=27,
product_count=3,
),
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user, user2],
product=product,
product_count=2,
)
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (27 * 3) - math.ceil((27 * 2) * (2 / 3))
def test_user_balance_joint_transactions_interest(sql_session: Session) -> None:
user, user2, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 10, 0, 0),
user_id=user.id,
product_id=product.id,
amount=27 * 3,
per_product=27,
product_count=3,
),
Transaction.adjust_interest(
time=datetime(2023, 10, 1, 11, 0, 0),
user_id=user.id,
interest_rate_percent=110,
),
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2],
product=product,
product_count=2,
)
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (27 * 3) - math.ceil(math.ceil(27 * 2 / 2) * 1.1)
def test_user_balance_joint_transactions_changing_interest(sql_session: Session) -> None:
user, user2, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 10, 0, 0),
user_id=user.id,
product_id=product.id,
amount=27 * 4,
per_product=27,
product_count=4,
),
# Pays 1.1x the price
Transaction.adjust_interest(
time=datetime(2023, 10, 1, 11, 0, 0),
user_id=user.id,
interest_rate_percent=110,
),
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2],
product=product,
product_count=2,
)
transactions = [
# Pays 1.2x the price
Transaction.adjust_interest(
time=datetime(2023, 10, 1, 12, 0, 0),
user_id=user.id,
interest_rate_percent=120,
)
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2],
product=product,
product_count=1,
)
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (
(27 * 4) - math.ceil(math.ceil(27 * 2 / 2) * 1.1) - math.ceil(math.ceil(27 * 1 / 2) * 1.2)
)
def test_user_balance_joint_transactions_penalty(sql_session: Session) -> None:
user, user2, _, product = insert_test_data(sql_session)
transactions = [
Transaction.add_product(
time=datetime(2023, 10, 1, 10, 0, 0),
user_id=user.id,
product_id=product.id,
amount=27 * 3,
per_product=27,
product_count=3,
),
Transaction.adjust_balance(
time=datetime(2023, 10, 1, 11, 0, 0),
user_id=user.id,
amount=-200,
),
]
sql_session.add_all(transactions)
sql_session.commit()
joint_buy_product(
sql_session,
instigator=user,
users=[user, user2],
product=product,
product_count=2,
)
pprint(user_balance_log(sql_session, user))
assert user_balance(sql_session, user) == (
(27 * 3)
- 200
- math.ceil(math.ceil(27 * 2 / 2) * (DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE // 100))
)
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_joint_transactions(sql_session: Session): ...
def test_user_balance_joint_transactions_changing_penalty(sql_session: Session) -> None: ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_joint_transactions_interest(sql_session: Session): ...
def test_user_balance_joint_transactions_penalty_interest_combined(
sql_session: Session,
) -> None: ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_joint_transactions_changing_interest(sql_session: Session): ...
def test_user_balance_until(sql_session: Session) -> None: ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_joint_transactions_penalty(sql_session: Session): ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_joint_transactions_changing_penalty(sql_session: Session): ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_joint_transactions_penalty_interest_combined(sql_session: Session): ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_until(sql_session: Session): ...
@pytest.mark.skip(reason="Not yet implemented")
def test_user_balance_throw_away_products(sql_session: Session): ...
def test_user_balance_throw_away_products(sql_session: Session) -> None: ...