Implement balance monitoring and notification for Selectel API integration
- Added support for balance tracking using `SELECTEL_API_TOKEN` in the configuration. - Introduced new balance notification logic in the local watchdog, alerting users on balance changes based on defined thresholds. - Updated documentation to include instructions for setting up balance notifications and the required environment variables. - Enhanced the `ready` and `session` modules to initialize balance state and handle notifications during GPU operations. - Refactored the CLI and related components to support the new balance monitoring features, ensuring a seamless user experience.
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
from gpu_rent.balance import (
|
||||
BalanceWatchState,
|
||||
balance_rub_from_payload,
|
||||
plan_balance_notices,
|
||||
spend_steps,
|
||||
)
|
||||
|
||||
|
||||
def test_balance_kopecks():
|
||||
payload = {
|
||||
"data": {
|
||||
"billings": [
|
||||
{"balances_values_sum": 123456}, # 1234.56 ₽
|
||||
]
|
||||
}
|
||||
}
|
||||
assert balance_rub_from_payload(payload) == 1234.56
|
||||
|
||||
|
||||
def test_spend_steps_200():
|
||||
assert spend_steps(1000, 850, 200) == 0
|
||||
assert spend_steps(1000, 800, 200) == 1
|
||||
assert spend_steps(1000, 600, 200) == 2
|
||||
assert spend_steps(1000, 1100, 200) == 0 # top-up direction
|
||||
|
||||
|
||||
def test_plan_notifies_each_step():
|
||||
state = BalanceWatchState(baseline_rub=1000.0)
|
||||
state, msgs = plan_balance_notices(state, 750.0, step_rub=200.0)
|
||||
assert state.last_notified_step == 1
|
||||
assert len(msgs) == 1
|
||||
assert "200" in msgs[0]
|
||||
state, msgs2 = plan_balance_notices(state, 550.0, step_rub=200.0)
|
||||
assert state.last_notified_step == 2
|
||||
assert len(msgs2) == 1
|
||||
# no re-notify same step
|
||||
state, msgs3 = plan_balance_notices(state, 520.0, step_rub=200.0)
|
||||
assert msgs3 == []
|
||||
assert state.last_notified_step == 2
|
||||
|
||||
|
||||
def test_plan_topup_resets():
|
||||
state = BalanceWatchState(baseline_rub=500.0, last_notified_step=2)
|
||||
state, msgs = plan_balance_notices(state, 900.0, step_rub=200.0)
|
||||
assert state.baseline_rub == 900.0
|
||||
assert state.last_notified_step == 0
|
||||
assert any("пополнен" in m for m in msgs)
|
||||
|
||||
|
||||
def test_plan_low_once():
|
||||
state = BalanceWatchState(baseline_rub=1000.0)
|
||||
state, msgs = plan_balance_notices(state, 150.0, step_rub=200.0, low_rub=200.0)
|
||||
assert state.low_notified
|
||||
assert any("низкий" in m for m in msgs)
|
||||
state, msgs2 = plan_balance_notices(state, 100.0, step_rub=200.0, low_rub=200.0)
|
||||
assert not any("низкий" in m for m in msgs2)
|
||||
Reference in New Issue
Block a user