Coverage for dibbler / queries / current_penalty.py: 69%

1from datetime import datetime 

2 

3from sqlalchemy import BindParameter, bindparam, select 

4from sqlalchemy.orm import Session 

5 

6from dibbler.models import Transaction, TransactionType 

7from dibbler.models.Transaction import ( 

8 DEFAULT_PENALTY_MULTIPLIER_PERCENT, 

9 DEFAULT_PENALTY_THRESHOLD, 

10) 

11from dibbler.queries.query_helpers import until_filter 

12 

13 

14def current_penalty( 

15 sql_session: Session, 

16 until_time: BindParameter[datetime] | datetime | None = None, 

17 until_transaction: BindParameter[Transaction] | Transaction | None = None, 

18 until_inclusive: bool = True, 

19) -> tuple[int, int]: 

20 """ 

21 Get the current penalty settings (threshold and multiplier percentage) as of a given time or transaction. 

22 

23 Returns a tuple of `(penalty_threshold, penalty_multiplier_percentage)`. 

24 """ 

25 

26 if not (until_time is None or until_transaction is None): 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

27 raise ValueError("Cannot filter by both until_time and until_transaction.") 

28 

29 if isinstance(until_time, datetime): 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true

30 until_time = BindParameter("until_time", value=until_time) 

31 

32 if isinstance(until_transaction, Transaction): 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 if until_transaction.id is None: 

34 raise ValueError("until_transaction must be persisted in the database.") 

35 until_transaction_id = bindparam("until_transaction_id", value=until_transaction.id) 

36 else: 

37 until_transaction_id = None 

38 

39 result = sql_session.scalars( 

40 select(Transaction) 

41 .where( 

42 Transaction.type_ == TransactionType.ADJUST_PENALTY, 

43 until_filter( 

44 until_time=until_time, 

45 until_transaction_id=until_transaction_id, 

46 until_inclusive=until_inclusive, 

47 ), 

48 ) 

49 .order_by(Transaction.time.desc()) 

50 .limit(1) 

51 ).one_or_none() 

52 

53 if result is None: 

54 return DEFAULT_PENALTY_THRESHOLD, DEFAULT_PENALTY_MULTIPLIER_PERCENT 

55 

56 assert result.penalty_threshold is not None, "Penalty threshold must be set" 

57 assert result.penalty_multiplier_percent is not None, "Penalty multiplier percent must be set" 

58 

59 return result.penalty_threshold, result.penalty_multiplier_percent