shoko/jigaido
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Address PR #20 feedback:

Browse Source 
- Removed PersonalStorage (redundant - RoomStorage handles both via room_id)
- Added ensure_room() and ensure_tracking() methods for explicit creation
- Added @runtime_checkable to Protocols for isinstance checks
- Added tests/test_ports.py with 11 unit tests for storage protocols
This commit is contained in:
shokollm
2026-04-02 23:57:49 +00:00
parent 5450d12400
commit 43603659de
3 changed files with 214 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
core/__init__.py
View File

@@ -8,7 +8,6 @@ from core.models import (
)
from core.ports import (
RoomStorage,
PersonalStorage,
TrackingStorage,
)
@@ -18,6 +17,5 @@ __all__ = [
"RoomData",
"TrackingData",
"RoomStorage",
"PersonalStorage",
"TrackingStorage",
]

57
core/ports.py
View File

@@ -1,14 +1,19 @@
"""Abstract storage interfaces (Ports) for JIGAIDO storage adapters."""
from typing import Protocol
from typing import Protocol, runtime_checkable
from core.models import Bounty, RoomData, TrackingData, TrackedBounty
@runtime_checkable
class RoomStorage(Protocol):
"""Storage port for room (group) bounties.
"""Storage port for room bounties.
Implement this protocol to provide room bounty storage capability.
A room is identified by room_id:
- Negative room_id: Telegram group (e.g., -1001)
- Positive room_id: DM/personal context (user's Telegram ID)
This single port handles both group and personal bounties.
"""
def load(self, room_id: int) -> RoomData | None:
@@ -19,8 +24,12 @@ class RoomStorage(Protocol):
"""Save all data for a room."""
...
def ensure_room(self, room_id: int) -> RoomData:
"""Ensure a room exists, creating it if necessary. Returns RoomData."""
...
def add_bounty(self, room_id: int, bounty: Bounty) -> None:
"""Add a new bounty to a room."""
"""Add a new bounty to a room. Creates room if it doesn't exist."""
...
def update_bounty(self, room_id: int, bounty: Bounty) -> None:
@@ -36,42 +45,12 @@ class RoomStorage(Protocol):
...
class PersonalStorage(Protocol):
"""Storage port for personal (DM) bounties.
Personal bounties are stored in RoomData with a positive room_id (user's ID).
This port provides the same operations as RoomStorage but for personal context.
"""
def load(self, user_id: int) -> RoomData | None:
"""Load personal bounty data for a user. Returns None if not found."""
...
def save(self, room_data: RoomData) -> None:
"""Save personal bounty data for a user."""
...
def add_bounty(self, user_id: int, bounty: Bounty) -> None:
"""Add a new bounty to a user's personal storage."""
...
def update_bounty(self, user_id: int, bounty: Bounty) -> None:
"""Update an existing bounty in personal storage."""
...
def delete_bounty(self, user_id: int, bounty_id: int) -> None:
"""Delete a bounty from personal storage."""
...
def get_bounty(self, user_id: int, bounty_id: int) -> Bounty | None:
"""Get a specific bounty from personal storage by ID."""
...
@runtime_checkable
class TrackingStorage(Protocol):
"""Storage port for tracking data.
Tracks which bounties a user is tracking in a specific room.
Use ensure_tracking() to create a new tracking entry before tracking bounties.
"""
def load(self, room_id: int, user_id: int) -> TrackingData | None:
@@ -82,8 +61,12 @@ class TrackingStorage(Protocol):
"""Save tracking data."""
...
def ensure_tracking(self, room_id: int, user_id: int) -> TrackingData:
"""Ensure tracking exists for user in room, creating if necessary. Returns TrackingData."""
...
def track_bounty(self, room_id: int, user_id: int, tracked: TrackedBounty) -> None:
"""Add a bounty to a user's tracking list."""
"""Add a bounty to a user's tracking list. Creates tracking if needed."""
...
def untrack_bounty(self, room_id: int, user_id: int, bounty_id: int) -> None:

194
tests/test_ports.py Normal file
View File

@@ -0,0 +1,194 @@
"""Tests for core/ports.py — storage interfaces."""
import pytest
from typing import Self
from core.models import Bounty, RoomData, TrackingData, TrackedBounty
from core.ports import RoomStorage, TrackingStorage
class MockRoomStorage:
"""Mock implementation of RoomStorage for testing."""
def __init__(self):
self._rooms: dict[int, RoomData] = {}
def load(self, room_id: int) -> RoomData | None:
return self._rooms.get(room_id)
def save(self, room_data: RoomData) -> None:
self._rooms[room_data.room_id] = room_data
def ensure_room(self, room_id: int) -> RoomData:
if room_id not in self._rooms:
self._rooms[room_id] = RoomData(room_id=room_id, bounties=[], next_id=1)
return self._rooms[room_id]
def add_bounty(self, room_id: int, bounty: Bounty) -> None:
if room_id not in self._rooms:
self.ensure_room(room_id)
self._rooms[room_id].bounties.append(bounty)
def update_bounty(self, room_id: int, bounty: Bounty) -> None:
if room_id in self._rooms:
for i, b in enumerate(self._rooms[room_id].bounties):
if b.id == bounty.id:
self._rooms[room_id].bounties[i] = bounty
break
def delete_bounty(self, room_id: int, bounty_id: int) -> None:
if room_id in self._rooms:
self._rooms[room_id].bounties = [
b for b in self._rooms[room_id].bounties if b.id != bounty_id
]
def get_bounty(self, room_id: int, bounty_id: int) -> Bounty | None:
if room_id in self._rooms:
for b in self._rooms[room_id].bounties:
if b.id == bounty_id:
return b
return None
class MockTrackingStorage:
"""Mock implementation of TrackingStorage for testing."""
def __init__(self):
self._tracking: dict[tuple[int, int], TrackingData] = {}
def load(self, room_id: int, user_id: int) -> TrackingData | None:
return self._tracking.get((room_id, user_id))
def save(self, tracking_data: TrackingData) -> None:
self._tracking[(tracking_data.room_id, tracking_data.user_id)] = tracking_data
def ensure_tracking(self, room_id: int, user_id: int) -> TrackingData:
key = (room_id, user_id)
if key not in self._tracking:
self._tracking[key] = TrackingData(
room_id=room_id, user_id=user_id, tracked=[]
)
return self._tracking[key]
def track_bounty(self, room_id: int, user_id: int, tracked: TrackedBounty) -> None:
key = (room_id, user_id)
if key not in self._tracking:
self.ensure_tracking(room_id, user_id)
self._tracking[key].tracked.append(tracked)
def untrack_bounty(self, room_id: int, user_id: int, bounty_id: int) -> None:
key = (room_id, user_id)
if key in self._tracking:
self._tracking[key].tracked = [
t for t in self._tracking[key].tracked if t.bounty_id != bounty_id
]
class TestRoomStorage:
def test_mock_implements_room_storage_protocol(self):
storage: RoomStorage = MockRoomStorage()
assert isinstance(storage, RoomStorage)
def test_ensure_room_creates_room_if_not_exists(self):
storage = MockRoomStorage()
room = storage.ensure_room(-1001)
assert room.room_id == -1001
assert room.bounties == []
assert room.next_id == 1
def test_ensure_room_returns_existing_room(self):
storage = MockRoomStorage()
room1 = storage.ensure_room(-1001)
room2 = storage.ensure_room(-1001)
assert room1 is room2
def test_add_bounty_creates_room_if_not_exists(self):
storage = MockRoomStorage()
bounty = Bounty(
id=1,
text="Test",
link=None,
due_date_ts=None,
created_at=0,
created_by_user_id=123,
)
storage.add_bounty(-1001, bounty)
room = storage.load(-1001)
assert room is not None
assert len(room.bounties) == 1
assert room.bounties[0].text == "Test"
def test_update_bounty(self):
storage = MockRoomStorage()
bounty = Bounty(
id=1,
text="Original",
link=None,
due_date_ts=None,
created_at=0,
created_by_user_id=123,
)
storage.add_bounty(-1001, bounty)
updated = Bounty(
id=1,
text="Updated",
link=None,
due_date_ts=None,
created_at=0,
created_by_user_id=123,
)
storage.update_bounty(-1001, updated)
result = storage.get_bounty(-1001, 1)
assert result is not None
assert result.text == "Updated"
def test_delete_bounty(self):
storage = MockRoomStorage()
bounty = Bounty(
id=1,
text="Test",
link=None,
due_date_ts=None,
created_at=0,
created_by_user_id=123,
)
storage.add_bounty(-1001, bounty)
storage.delete_bounty(-1001, 1)
assert storage.get_bounty(-1001, 1) is None
class TestTrackingStorage:
def test_mock_implements_tracking_storage_protocol(self):
storage: TrackingStorage = MockTrackingStorage()
assert isinstance(storage, TrackingStorage)
def test_ensure_tracking_creates_if_not_exists(self):
storage = MockTrackingStorage()
tracking = storage.ensure_tracking(-1001, 123456)
assert tracking.room_id == -1001
assert tracking.user_id == 123456
assert tracking.tracked == []
def test_ensure_tracking_returns_existing(self):
storage = MockTrackingStorage()
t1 = storage.ensure_tracking(-1001, 123456)
t2 = storage.ensure_tracking(-1001, 123456)
assert t1 is t2
def test_track_bounty_creates_tracking_if_not_exists(self):
storage = MockTrackingStorage()
tracked = TrackedBounty(bounty_id=5, created_at=0)
storage.track_bounty(-1001, 123456, tracked)
result = storage.load(-1001, 123456)
assert result is not None
assert len(result.tracked) == 1
assert result.tracked[0].bounty_id == 5
def test_untrack_bounty(self):
storage = MockTrackingStorage()
tracked = TrackedBounty(bounty_id=5, created_at=0)
storage.track_bounty(-1001, 123456, tracked)
storage.untrack_bounty(-1001, 123456, 5)
result = storage.load(-1001, 123456)
assert result is not None
assert len(result.tracked) == 0