120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from gibby.models import AccountRecord, StateFile, UsageSnapshot, UsageWindow
|
|
from gibby.store import JsonStateStore
|
|
|
|
|
|
def test_store_writes_minimal_accounts_schema(tmp_path) -> None:
|
|
store = JsonStateStore(tmp_path / "accounts.json")
|
|
store.save(
|
|
StateFile(
|
|
active_account="acc@example.com",
|
|
accounts=[
|
|
AccountRecord(
|
|
email="acc@example.com",
|
|
access_token="tok",
|
|
refresh_token="ref",
|
|
token_refresh_at=2000,
|
|
usage=UsageSnapshot(
|
|
checked_at=1000,
|
|
primary_window=UsageWindow(used_percent=70, reset_at=1300),
|
|
secondary_window=UsageWindow(used_percent=20, reset_at=4600),
|
|
),
|
|
usage_checked_at=1000,
|
|
disabled=False,
|
|
)
|
|
],
|
|
)
|
|
)
|
|
|
|
payload = json.loads((tmp_path / "accounts.json").read_text())
|
|
|
|
assert payload == {
|
|
"active_account": "acc@example.com",
|
|
"accounts": [
|
|
{
|
|
"email": "acc@example.com",
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"token_refresh_at": 2000,
|
|
"usage": {
|
|
"primary": {"used_percent": 70, "reset_at": 1300},
|
|
"secondary": {"used_percent": 20, "reset_at": 4600},
|
|
"limit_reached": False,
|
|
"allowed": True,
|
|
},
|
|
"usage_checked_at": 1000,
|
|
"disabled": False,
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def test_store_load_reconstructs_account_state(tmp_path) -> None:
|
|
path = tmp_path / "accounts.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"active_account": "acc@example.com",
|
|
"accounts": [
|
|
{
|
|
"email": "acc@example.com",
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"token_refresh_at": 2000,
|
|
"usage": {
|
|
"primary": {"used_percent": 80, "reset_at": 1300},
|
|
"secondary": {"used_percent": 15, "reset_at": 4600},
|
|
"limit_reached": True,
|
|
"allowed": False,
|
|
},
|
|
"usage_checked_at": 1000,
|
|
"disabled": True,
|
|
}
|
|
],
|
|
}
|
|
)
|
|
)
|
|
|
|
state = JsonStateStore(path).load()
|
|
|
|
assert state.active_account == "acc@example.com"
|
|
assert state.accounts[0].email == "acc@example.com"
|
|
assert state.accounts[0].token_refresh_at == 2000
|
|
assert state.accounts[0].usage is not None
|
|
assert state.accounts[0].usage.primary_window is not None
|
|
assert state.accounts[0].usage.primary_window.used_percent == 80
|
|
assert state.accounts[0].usage.limit_reached is True
|
|
assert state.accounts[0].usage.allowed is False
|
|
assert state.accounts[0].disabled is True
|
|
|
|
|
|
def test_append_failed_account_writes_failed_json_shape(tmp_path) -> None:
|
|
store = JsonStateStore(tmp_path / "accounts.json")
|
|
store.append_failed_account(
|
|
AccountRecord(
|
|
email="failed@example.com",
|
|
access_token="tok",
|
|
refresh_token="ref",
|
|
token_refresh_at=2000,
|
|
disabled=False,
|
|
)
|
|
)
|
|
|
|
payload = json.loads((tmp_path / "failed.json").read_text())
|
|
|
|
assert payload == {
|
|
"accounts": [
|
|
{
|
|
"email": "failed@example.com",
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"token_refresh_at": 2000,
|
|
"usage": None,
|
|
"usage_checked_at": None,
|
|
"disabled": False,
|
|
}
|
|
]
|
|
}
|