LogalDeveloper/Crabstero
0
Fork 0
Code Issues Actions Packages Releases 11 Activity

Modernized codebase with NamedTuples, StrEnum, override decorators, slots, and other idiomatic improvements.
CI / Formatting (push) Successful in 4s
Details
CI / Linting (push) Successful in 5s
Details
CI / Tests (push) Successful in 21s
Details
CI / Type Checking (push) Successful in 10s
Details
CI / Spelling (push) Successful in 5s
Details

Browse Source
This commit is contained in:
LogalDeveloper
2026-03-22 20:02:31 -04:00
parent a23d252f27
commit 23cc721612
12 changed files with 155 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_database.py
+50 -32
View File
@@ -22,7 +22,7 @@ from typing import TYPE_CHECKING
import pytest
from crabstero.database import Database
from crabstero.database import ChannelImage, Database, StartWord, Transition
if TYPE_CHECKING:
from pathlib import Path
@@ -58,21 +58,21 @@ class TestAddMarkovData:
async def test_stores_start_word(self, db: Database) -> None:
"""Inserted start word can be retrieved by channel."""
await db.add_markov_data([(1, 100, "Hello")], [])
await db.add_markov_data([StartWord(1, 100, "Hello")], [])
result = await db.get_random_start_word(1)
assert result == "Hello"
async def test_stores_transition(self, db: Database) -> None:
"""Inserted transition can be retrieved by channel and word."""
await db.add_markov_data([], [(1, 100, "Hello", "world.")])
await db.add_markov_data([], [Transition(1, 100, "Hello", "world.")])
result = await db.get_random_next_word(1, "Hello")
assert result == "world."
async def test_stores_both_in_single_call(self, db: Database) -> None:
"""Start words and transitions are stored in a single call."""
await db.add_markov_data(
[(1, 100, "Hello")],
[(1, 100, "Hello", "world.")],
[StartWord(1, 100, "Hello")],
[Transition(1, 100, "Hello", "world.")],
)
assert await db.get_random_start_word(1) == "Hello"
assert await db.get_random_next_word(1, "Hello") == "world."
@@ -98,8 +98,8 @@ class TestAddMarkovData:
await db.add_markov_data(
[],
[
(1, 100, "Hello", "beautiful"),
(1, 100, "Hello", completing_word),
Transition(1, 100, "Hello", "beautiful"),
Transition(1, 100, "Hello", completing_word),
],
)
for _ in range(100):
@@ -108,7 +108,7 @@ class TestAddMarkovData:
async def test_completing_returns_none_without_match(self, db: Database) -> None:
"""Returns None when no transitions end with sentence punctuation."""
await db.add_markov_data([], [(1, 100, "Hello", "beautiful")])
await db.add_markov_data([], [Transition(1, 100, "Hello", "beautiful")])
result = await db.get_random_completing_next_word(1, "Hello")
assert result is None
@@ -127,7 +127,7 @@ class TestImages:
async def test_add_and_retrieve(self, db: Database) -> None:
"""Inserted image URL can be retrieved by channel."""
await db.add_images([(1, 100, "https://example.com/cat.png")])
await db.add_images([ChannelImage(1, 100, "https://example.com/cat.png")])
result = await db.get_random_image(1)
assert result == "https://example.com/cat.png"
@@ -186,48 +186,63 @@ class TestRemoveMarkovData:
async def test_removes_one_start_word(self, db: Database) -> None:
"""Removes exactly one matching start word row."""
await db.add_markov_data([(1, 100, "Hello"), (1, 100, "Hello")], [])
await db.remove_markov_data([(1, 100, "Hello")], [])
await db.add_markov_data(
[StartWord(1, 100, "Hello"), StartWord(1, 100, "Hello")], []
)
await db.remove_markov_data([StartWord(1, 100, "Hello")], [])
# One copy should remain.
assert await db.get_random_start_word(1) == "Hello"
async def test_removes_one_transition(self, db: Database) -> None:
"""Removes exactly one matching transition row."""
await db.add_markov_data(
[], [(1, 100, "Hello", "world."), (1, 100, "Hello", "world.")]
[],
[
Transition(1, 100, "Hello", "world."),
Transition(1, 100, "Hello", "world."),
],
)
await db.remove_markov_data([], [(1, 100, "Hello", "world.")])
await db.remove_markov_data([], [Transition(1, 100, "Hello", "world.")])
assert await db.get_random_next_word(1, "Hello") == "world."
async def test_removes_last_start_word(self, db: Database) -> None:
"""Removing the only start word leaves the table empty for that channel."""
await db.add_markov_data([(1, 100, "Hello")], [])
await db.remove_markov_data([(1, 100, "Hello")], [])
await db.add_markov_data([StartWord(1, 100, "Hello")], [])
await db.remove_markov_data([StartWord(1, 100, "Hello")], [])
assert await db.get_random_start_word(1) is None
async def test_removes_last_transition(self, db: Database) -> None:
"""Removing the only transition leaves no next word."""
await db.add_markov_data([], [(1, 100, "Hello", "world.")])
await db.remove_markov_data([], [(1, 100, "Hello", "world.")])
await db.add_markov_data([], [Transition(1, 100, "Hello", "world.")])
await db.remove_markov_data([], [Transition(1, 100, "Hello", "world.")])
assert await db.get_random_next_word(1, "Hello") is None
async def test_no_match_is_noop(self, db: Database) -> None:
"""Removing a non-existent row does not raise."""
await db.remove_markov_data([(1, 100, "nope")], [(1, 100, "nope", "nah")])
await db.remove_markov_data(
[StartWord(1, 100, "nope")],
[Transition(1, 100, "nope", "nah")],
)
async def test_start_word_removal_scoped_to_channel(self, db: Database) -> None:
"""Removing a start word in one channel leaves another channel intact."""
await db.add_markov_data([(1, 100, "Hello"), (2, 200, "Hello")], [])
await db.remove_markov_data([(1, 100, "Hello")], [])
await db.add_markov_data(
[StartWord(1, 100, "Hello"), StartWord(2, 200, "Hello")], []
)
await db.remove_markov_data([StartWord(1, 100, "Hello")], [])
assert await db.get_random_start_word(1) is None
assert await db.get_random_start_word(2) == "Hello"
async def test_transition_removal_scoped_to_channel(self, db: Database) -> None:
"""Removing a transition in one channel leaves another channel intact."""
await db.add_markov_data(
[], [(1, 100, "Hello", "world."), (2, 200, "Hello", "world.")]
[],
[
Transition(1, 100, "Hello", "world."),
Transition(2, 200, "Hello", "world."),
],
)
await db.remove_markov_data([], [(1, 100, "Hello", "world.")])
await db.remove_markov_data([], [Transition(1, 100, "Hello", "world.")])
assert await db.get_random_next_word(1, "Hello") is None
assert await db.get_random_next_word(2, "Hello") == "world."
@@ -243,35 +258,35 @@ class TestRemoveImages:
"""Removes exactly one matching image row."""
await db.add_images(
[
(1, 100, "https://example.com/a.png"),
(1, 100, "https://example.com/a.png"),
ChannelImage(1, 100, "https://example.com/a.png"),
ChannelImage(1, 100, "https://example.com/a.png"),
]
)
await db.remove_images([(1, 100, "https://example.com/a.png")])
await db.remove_images([ChannelImage(1, 100, "https://example.com/a.png")])
# One copy should remain.
assert await db.get_random_image(1) == "https://example.com/a.png"
async def test_removes_last_image(self, db: Database) -> None:
"""Removing the only image leaves none for that channel."""
await db.add_images([(1, 100, "https://example.com/a.png")])
await db.remove_images([(1, 100, "https://example.com/a.png")])
await db.add_images([ChannelImage(1, 100, "https://example.com/a.png")])
await db.remove_images([ChannelImage(1, 100, "https://example.com/a.png")])
assert await db.get_random_image(1) is None
async def test_image_removal_scoped_to_channel(self, db: Database) -> None:
"""Removing an image in one channel leaves another channel intact."""
await db.add_images(
[
(1, 100, "https://example.com/a.png"),
(2, 200, "https://example.com/a.png"),
ChannelImage(1, 100, "https://example.com/a.png"),
ChannelImage(2, 200, "https://example.com/a.png"),
]
)
await db.remove_images([(1, 100, "https://example.com/a.png")])
await db.remove_images([ChannelImage(1, 100, "https://example.com/a.png")])
assert await db.get_random_image(1) is None
assert await db.get_random_image(2) == "https://example.com/a.png"
async def test_no_match_is_noop(self, db: Database) -> None:
"""Removing a non-existent image does not raise."""
await db.remove_images([(1, 100, "https://example.com/nope.png")])
await db.remove_images([ChannelImage(1, 100, "https://example.com/nope.png")])
async def test_empty_list_is_noop(self, db: Database) -> None:
"""Empty list does not error."""
@@ -285,7 +300,10 @@ class TestWriteDurability:
"""Data written via add_markov_data is durable after close/reopen."""
db_path = str(tmp_path / "durability.db")
db = await Database.connect(db_path)
await db.add_markov_data([(1, 100, "Hello")], [(1, 100, "Hello", "world.")])
await db.add_markov_data(
[StartWord(1, 100, "Hello")],
[Transition(1, 100, "Hello", "world.")],
)
await db.close()
db2 = await Database.connect(db_path)
tests/test_markov.py
+22 -16
View File
@@ -22,6 +22,7 @@ from typing import TYPE_CHECKING
import pytest
from crabstero.database import StartWord, Transition
from crabstero.markov import (
DEFAULT_SENTENCE_END,
_ingest_sentence,
@@ -211,16 +212,16 @@ class TestGenerate:
# continuing ("E") and completing ("end.") transition, so
# get_random_completing_next_word deterministically picks "end.".
await db.add_markov_data(
[(1, 100, "A")],
[StartWord(1, 100, "A")],
[
(1, 100, "A", "B"),
(1, 100, "B", "C"),
(1, 100, "C", "D"),
(1, 100, "D", "E"),
(1, 100, "D", "F"),
(1, 100, "D", "G"),
(1, 100, "D", "H"),
(1, 100, "D", "end."),
Transition(1, 100, "A", "B"),
Transition(1, 100, "B", "C"),
Transition(1, 100, "C", "D"),
Transition(1, 100, "D", "E"),
Transition(1, 100, "D", "F"),
Transition(1, 100, "D", "G"),
Transition(1, 100, "D", "H"),
Transition(1, 100, "D", "end."),
],
)
@@ -230,14 +231,14 @@ class TestGenerate:
async def test_start_word_already_ends_sentence(self, db: Database) -> None:
"""Generation stops immediately when the start word is sentence-ending."""
await db.add_markov_data([(1, 100, "Yes.")], [])
await db.add_markov_data([StartWord(1, 100, "Yes.")], [])
result = await generate(db, 1)
assert result == "Yes."
async def test_chain_dead_end(self, db: Database) -> None:
"""Generation stops when no next word exists (dead-end chain)."""
await db.add_markov_data([(1, 100, "Hello")], [])
await db.add_markov_data([StartWord(1, 100, "Hello")], [])
# "Hello" has no transitions, so the loop breaks immediately.
result = await generate(db, 1)
@@ -247,10 +248,10 @@ class TestGenerate:
"""Section sign at the truncation boundary is stripped."""
# Build a chain: "A" -> "B" -> "C§".
await db.add_markov_data(
[(1, 100, "A")],
[StartWord(1, 100, "A")],
[
(1, 100, "A", "B"),
(1, 100, "B", f"C{DEFAULT_SENTENCE_END}"),
Transition(1, 100, "A", "B"),
Transition(1, 100, "B", f"C{DEFAULT_SENTENCE_END}"),
],
)
@@ -264,14 +265,19 @@ class TestGenerate:
"""After soft_limit, falls back when no completing word exists."""
# "A" -> "B" (no completing transition). Past soft_limit,
# get_random_completing_next_word returns None, falls back to "B".
await db.add_markov_data([(1, 100, "A")], [(1, 100, "A", "B")])
await db.add_markov_data(
[StartWord(1, 100, "A")], [Transition(1, 100, "A", "B")]
)
result = await generate(db, 1, soft_limit=1, hard_limit=1000)
assert result == "A B"
async def test_hard_limit_truncates_mid_word(self, db: Database) -> None:
"""Hard limit slices output even when it falls inside a word."""
await db.add_markov_data([(1, 100, "AB")], [(1, 100, "AB", "CDEF")])
await db.add_markov_data(
[StartWord(1, 100, "AB")],
[Transition(1, 100, "AB", "CDEF")],
)
# "AB CDEF" is 7 chars; hard_limit=5 truncates to "AB CD".
result = await generate(db, 1, soft_limit=100, hard_limit=5)
tests/test_uningest.py
+4 -3
View File
@@ -19,6 +19,7 @@ from typing import TYPE_CHECKING
import pytest
from crabstero.cache import CachedMessage, IngestCache
from crabstero.database import ChannelImage
from crabstero.markov import ingest, uningest
if TYPE_CHECKING:
@@ -50,8 +51,8 @@ class TestIngestUningestCycle:
async def test_image_round_trip(self, db: Database) -> None:
"""Ingest and uningest an image leaves the database clean."""
await db.add_images([(1, 100, "https://example.com/cat.png")])
await db.remove_images([(1, 100, "https://example.com/cat.png")])
await db.add_images([ChannelImage(1, 100, "https://example.com/cat.png")])
await db.remove_images([ChannelImage(1, 100, "https://example.com/cat.png")])
assert await db.get_random_image(1) is None
@@ -140,7 +141,7 @@ class TestUningestRestoresState:
) -> None:
"""Ingest then uningest preserves unrelated pre-existing data exactly."""
await ingest(db, 99, 200, "Pre-existing data stays safe.")
await db.add_images([(99, 200, "https://example.com/existing.png")])
await db.add_images([ChannelImage(99, 200, "https://example.com/existing.png")])
before = await _snapshot(db)
await ingest(db, 1, 100, text)