forked from sass/tipibot
114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
"""Shared fixtures: an in-memory PocketBase stand-in wired into core.economy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from core import economy, pb_client # noqa: E402
|
|
|
|
|
|
class FakePocketBase:
|
|
"""In-memory stand-in for core.pb_client.
|
|
|
|
Mimics the behaviours the economy layer depends on:
|
|
- records are returned as deep copies (like JSON over REST)
|
|
- "field+" / "field-" body keys are atomic number modifiers
|
|
- fields not in `schema_fields` are silently dropped, like PocketBase
|
|
does for fields missing from the collection schema (schema_fields=None
|
|
keeps everything)
|
|
"""
|
|
|
|
def __init__(self, schema_fields: set[str] | None = None):
|
|
self.records: dict[str, dict] = {}
|
|
self.schema_fields = schema_fields
|
|
self._next_id = 0
|
|
|
|
def _filter(self, data: dict) -> dict:
|
|
if self.schema_fields is None:
|
|
return dict(data)
|
|
return {k: v for k, v in data.items() if k.rstrip("+-") in self.schema_fields}
|
|
|
|
def _apply(self, record: dict, data: dict) -> None:
|
|
for key, value in self._filter(data).items():
|
|
if key.endswith("+"):
|
|
record[key[:-1]] = record.get(key[:-1], 0) + value
|
|
elif key.endswith("-"):
|
|
record[key[:-1]] = record.get(key[:-1], 0) - value
|
|
else:
|
|
record[key] = value
|
|
|
|
async def get_record(self, user_id: str) -> dict | None:
|
|
await asyncio.sleep(0) # yield, so unserialized tasks would interleave
|
|
for record in self.records.values():
|
|
if record.get("user_id") == user_id:
|
|
return copy.deepcopy(record)
|
|
return None
|
|
|
|
async def create_record(self, record: dict) -> dict:
|
|
await asyncio.sleep(0)
|
|
self._next_id += 1
|
|
stored = self._filter(record)
|
|
stored["id"] = f"rec{self._next_id}"
|
|
stored["user_id"] = record.get("user_id", "")
|
|
self.records[stored["id"]] = stored
|
|
return copy.deepcopy(stored)
|
|
|
|
async def update_record(self, record_id: str, data: dict) -> dict:
|
|
await asyncio.sleep(0)
|
|
record = self.records[record_id]
|
|
self._apply(record, data)
|
|
return copy.deepcopy(record)
|
|
|
|
async def list_all_records(self, page_size: int = 500) -> list[dict]:
|
|
await asyncio.sleep(0)
|
|
return copy.deepcopy(list(self.records.values()))
|
|
|
|
async def count_records(self) -> int:
|
|
return len(self.records)
|
|
|
|
async def get_collection_fields(self) -> set[str]:
|
|
if self.schema_fields is None:
|
|
return set(economy._default_user()) | {"user_id"}
|
|
return set(self.schema_fields)
|
|
|
|
# -- test helpers -------------------------------------------------------
|
|
def record_for(self, user_id: int) -> dict:
|
|
for record in self.records.values():
|
|
if record.get("user_id") == str(user_id):
|
|
return record
|
|
raise KeyError(user_id)
|
|
|
|
|
|
def _install(monkeypatch, fake: FakePocketBase) -> FakePocketBase:
|
|
for name in ("get_record", "create_record", "update_record",
|
|
"list_all_records", "count_records", "get_collection_fields"):
|
|
monkeypatch.setattr(pb_client, name, getattr(fake, name))
|
|
monkeypatch.setattr(economy, "HOUSE_ID", None)
|
|
monkeypatch.setattr(economy, "_house_pb_id", None)
|
|
economy._user_locks.clear()
|
|
return fake
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_pb(monkeypatch) -> FakePocketBase:
|
|
return _install(monkeypatch, FakePocketBase())
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_pb_without_quest_fields(monkeypatch) -> FakePocketBase:
|
|
"""A fake whose collection schema predates the quest migration."""
|
|
schema = set(economy._default_user().keys()) | {"user_id"}
|
|
schema -= {"quest_daily", "quest_weekly"}
|
|
return _install(monkeypatch, FakePocketBase(schema_fields=schema))
|
|
|
|
|
|
def run(coro):
|
|
return asyncio.run(coro)
|