feat: Replace SQLite with per-user JSON storage (fixes #2)
- Add storage.py with load_user(), save_user(), next_bounty_id() - Rewrite commands.py to use JSON storage (simplified) - Remove db.py, schema.sql, cron.py, test_db.py - Update SPEC.md to reflect new architecture - Admin model removed (anyone can add, creator only can edit/delete) - No reminders in v1
This commit is contained in:
shokollm
@@ -1,298 +0,0 @@
|
||||
"""Tests for db.py"""
|
||||
|
||||
import time
|
||||
import pytest
|
||||
import db as _db
|
||||
|
||||
|
||||
class TestUsers:
|
||||
def test_upsert_user_creates_new(self):
|
||||
uid = _db.upsert_user(123, "alice")
|
||||
assert uid > 0
|
||||
row = _db.get_user_by_telegram_id(123)
|
||||
assert row is not None
|
||||
assert row["telegram_user_id"] == 123
|
||||
assert row["username"] == "alice"
|
||||
|
||||
def test_upsert_user_updates_username(self):
|
||||
# Two upserts to the same telegram_user_id: second one updates the username.
|
||||
# Returns the same id both times (idempotent).
|
||||
uid1 = _db.upsert_user(123, "alice")
|
||||
uid2 = _db.upsert_user(123, "alice_updated")
|
||||
assert uid1 == uid2
|
||||
row = _db.get_user_by_telegram_id(123)
|
||||
assert row["username"] == "alice_updated"
|
||||
|
||||
def test_get_user_by_telegram_id_not_found(self):
|
||||
row = _db.get_user_by_telegram_id(999999)
|
||||
assert row is None
|
||||
|
||||
|
||||
class TestGroups:
|
||||
def test_upsert_group_creates_new(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
assert gid > 0
|
||||
row = _db.get_group(-100123)
|
||||
assert row is not None
|
||||
assert row["telegram_chat_id"] == -100123
|
||||
assert row["creator_user_id"] == uid
|
||||
|
||||
def test_upsert_group_idempotent(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid1 = _db.upsert_group(-100123, uid)
|
||||
gid2 = _db.upsert_group(-100123, uid)
|
||||
assert gid1 == gid2
|
||||
|
||||
def test_get_group_creator_user_id(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
_db.upsert_group(-100123, uid)
|
||||
assert _db.get_group_creator_user_id(_db.get_group(-100123)["id"]) == uid
|
||||
|
||||
def test_get_group_not_found(self, fresh_db):
|
||||
row = _db.get_group(-999999)
|
||||
assert row is None
|
||||
|
||||
|
||||
class TestGroupAdmins:
|
||||
def test_add_remove_is_admin(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
|
||||
assert not _db.is_group_admin(gid, uid)
|
||||
added = _db.add_group_admin(gid, uid)
|
||||
assert added is True
|
||||
assert _db.is_group_admin(gid, uid) is True
|
||||
|
||||
# Adding again returns False (already admin)
|
||||
assert _db.add_group_admin(gid, uid) is False
|
||||
|
||||
removed = _db.remove_group_admin(gid, uid)
|
||||
assert removed is True
|
||||
assert _db.is_group_admin(gid, uid) is False
|
||||
|
||||
def test_remove_nonexistent_admin_returns_false(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
assert _db.remove_group_admin(gid, uid) is False
|
||||
|
||||
def test_is_group_creator(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
other = _db.upsert_user(2, "other")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
_db.add_group_admin(gid, uid)
|
||||
|
||||
assert _db.is_group_creator(gid, uid) is True
|
||||
assert _db.is_group_creator(gid, other) is False
|
||||
|
||||
def test_get_user_by_username(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
row = _db.get_user_by_username("alice")
|
||||
assert row is not None
|
||||
assert row["id"] == uid
|
||||
assert _db.get_user_by_username("nobody") is None
|
||||
|
||||
|
||||
class TestBounties:
|
||||
def test_add_bounty_group(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
_db.add_group_admin(gid, uid)
|
||||
|
||||
bid = _db.add_bounty(
|
||||
group_id=gid,
|
||||
created_by_user_id=uid,
|
||||
informed_by_username="bob",
|
||||
text="Fix bug",
|
||||
link="https://github.com/bob/repo",
|
||||
due_date_ts=int(time.time()) + 86400,
|
||||
)
|
||||
assert bid > 0
|
||||
bounty = _db.get_bounty(bid)
|
||||
assert bounty["text"] == "Fix bug"
|
||||
assert bounty["link"] == "https://github.com/bob/repo"
|
||||
assert bounty["informed_by_username"] == "bob"
|
||||
assert bounty["group_id"] == gid
|
||||
|
||||
def test_add_bounty_personal(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
bid = _db.add_bounty(
|
||||
group_id=None,
|
||||
created_by_user_id=uid,
|
||||
informed_by_username="alice",
|
||||
text="Personal reminder",
|
||||
link=None,
|
||||
due_date_ts=None,
|
||||
)
|
||||
assert bid > 0
|
||||
bounty = _db.get_bounty(bid)
|
||||
assert bounty["group_id"] is None
|
||||
assert bounty["text"] == "Personal reminder"
|
||||
|
||||
def test_add_bounty_duplicate_link_rejected(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
|
||||
_db.add_bounty(gid, uid, "user1", "text1", "https://example.com", None)
|
||||
with pytest.raises(ValueError, match="Link already exists"):
|
||||
_db.add_bounty(gid, uid, "user2", "text2", "https://example.com", None)
|
||||
|
||||
def test_add_bounty_null_link_allows_multiples(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
|
||||
bid1 = _db.add_bounty(gid, uid, "user1", "text only 1", None, None)
|
||||
bid2 = _db.add_bounty(gid, uid, "user2", "text only 2", None, None)
|
||||
assert bid1 != bid2
|
||||
|
||||
def test_get_group_bounties(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
_db.add_group_admin(gid, uid)
|
||||
|
||||
_db.add_bounty(gid, uid, "user", "bounty1", None, None)
|
||||
_db.add_bounty(gid, uid, "user", "bounty2", None, None)
|
||||
|
||||
bounties = _db.get_group_bounties(gid)
|
||||
assert len(bounties) == 2
|
||||
|
||||
def test_get_user_personal_bounties(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
_db.add_bounty(None, uid, "alice", "personal1", None, None)
|
||||
_db.add_bounty(None, uid, "alice", "personal2", None, None)
|
||||
|
||||
# Group bounty should not appear
|
||||
other = _db.upsert_user(2, "bob")
|
||||
gid = _db.upsert_group(-100, other)
|
||||
_db.add_bounty(gid, other, "bob", "group bounty", None, None)
|
||||
|
||||
personal = _db.get_user_personal_bounties(uid)
|
||||
assert len(personal) == 2
|
||||
|
||||
def test_update_bounty(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
bid = _db.add_bounty(gid, uid, "user", "old text", None, None)
|
||||
|
||||
_db.update_bounty(bid, "new text", None, None)
|
||||
updated = _db.get_bounty(bid)
|
||||
assert updated["text"] == "new text"
|
||||
|
||||
def test_update_bounty_duplicate_link_rejected(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
_db.add_bounty(gid, uid, "user1", "bounty1", "https://a.com", None)
|
||||
bid2 = _db.add_bounty(gid, uid, "user2", "bounty2", None, None)
|
||||
|
||||
with pytest.raises(ValueError, match="Link already exists"):
|
||||
_db.update_bounty(bid2, None, "https://a.com", None)
|
||||
|
||||
def test_delete_bounty(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "creator")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
bid = _db.add_bounty(gid, uid, "user", "to delete", None, None)
|
||||
|
||||
assert _db.delete_bounty(bid) is True
|
||||
assert _db.get_bounty(bid) is None
|
||||
# Deleting again returns False
|
||||
assert _db.delete_bounty(bid) is False
|
||||
|
||||
|
||||
class TestTracking:
|
||||
def test_track_untrack_is_tracking(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
uid2 = _db.upsert_user(2, "bob")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
bid = _db.add_bounty(gid, uid, "alice", "task", None, None)
|
||||
|
||||
assert _db.track_bounty(uid, bid) is True
|
||||
assert _db.is_tracking(uid, bid) is True
|
||||
# Track again → False (already tracking)
|
||||
assert _db.track_bounty(uid, bid) is False
|
||||
|
||||
# Other user tracking same bounty
|
||||
assert _db.track_bounty(uid2, bid) is True
|
||||
assert _db.is_tracking(uid2, bid) is True
|
||||
|
||||
# Untrack
|
||||
assert _db.untrack_bounty(uid, bid) is True
|
||||
assert _db.is_tracking(uid, bid) is False
|
||||
assert _db.is_tracking(uid2, bid) is True # other user still tracking
|
||||
# Untrack again → False
|
||||
assert _db.untrack_bounty(uid, bid) is False
|
||||
|
||||
def test_get_user_tracked_bounties_in_group(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
bid1 = _db.add_bounty(gid, uid, "alice", "task1", None, None)
|
||||
bid2 = _db.add_bounty(gid, uid, "alice", "task2", None, None)
|
||||
|
||||
# Different group bounty
|
||||
other_gid = _db.upsert_group(-100124, uid)
|
||||
bid3 = _db.add_bounty(other_gid, uid, "alice", "other group task", None, None)
|
||||
|
||||
_db.track_bounty(uid, bid1)
|
||||
_db.track_bounty(uid, bid3)
|
||||
|
||||
tracked = _db.get_user_tracked_bounties_in_group(uid, gid)
|
||||
assert len(tracked) == 1
|
||||
assert tracked[0]["id"] == bid1
|
||||
|
||||
def test_get_user_tracked_bounties_personal(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
bid1 = _db.add_bounty(None, uid, "alice", "personal1", None, None)
|
||||
bid2 = _db.add_bounty(None, uid, "alice", "personal2", None, None)
|
||||
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
bid3 = _db.add_bounty(gid, uid, "alice", "group task", None, None)
|
||||
|
||||
_db.track_bounty(uid, bid1)
|
||||
_db.track_bounty(uid, bid3)
|
||||
|
||||
tracked = _db.get_user_tracked_bounties_personal(uid)
|
||||
assert len(tracked) == 1
|
||||
assert tracked[0]["id"] == bid1
|
||||
|
||||
|
||||
class TestReminders:
|
||||
def test_get_bounties_due_soon(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
|
||||
now = int(time.time())
|
||||
# Due in 3 days (< 7 days)
|
||||
bid_soon = _db.add_bounty(gid, uid, "alice", "soon", None, now + 3 * 86400)
|
||||
# Due in 10 days (> 7 days)
|
||||
_db.add_bounty(gid, uid, "alice", "later", None, now + 10 * 86400)
|
||||
# No due date
|
||||
bid_no_date = _db.add_bounty(gid, uid, "alice", "no date", None, None)
|
||||
|
||||
_db.track_bounty(uid, bid_soon)
|
||||
_db.track_bounty(uid, bid_no_date)
|
||||
|
||||
due = _db.get_bounties_due_soon(uid, days=7)
|
||||
assert len(due) == 1
|
||||
assert due[0]["id"] == bid_soon
|
||||
|
||||
def test_reminder_log_prevents_duplicate_reminders(self, fresh_db):
|
||||
uid = _db.upsert_user(1, "alice")
|
||||
gid = _db.upsert_group(-100123, uid)
|
||||
now = int(time.time())
|
||||
bid = _db.add_bounty(gid, uid, "alice", "task", None, now + 2 * 86400)
|
||||
_db.track_bounty(uid, bid)
|
||||
|
||||
due1 = _db.get_bounties_due_soon(uid, days=7)
|
||||
assert len(due1) == 1
|
||||
|
||||
# Log that we reminded
|
||||
_db.log_reminder(uid, bid)
|
||||
|
||||
# Should not appear again
|
||||
due2 = _db.get_bounties_due_soon(uid, days=7)
|
||||
assert len(due2) == 0
|
||||
|
||||
def test_get_all_user_ids(self, fresh_db):
|
||||
_db.upsert_user(1, "alice")
|
||||
_db.upsert_user(2, "bob")
|
||||
ids = _db.get_all_user_ids()
|
||||
assert sorted(ids) == [1, 2]
|
||||
Reference in New Issue
Block a user