didericis
d7a58e52fd
lint / lint (push) Successful in 54s
test / integration-docker (pull_request) Successful in 38s
test / unit (pull_request) Successful in 45s
test / integration-firecracker (pull_request) Successful in 3m31s
test / coverage (pull_request) Successful in 18s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Failing after 6s
Move the DbStore base (db_store), the shared schema-migration base (migrations / TableMigrations), the concrete stores (audit_store, config_store, queue_store), and StoreManager out of the package root into a bot_bottle/store/ package, so persistence lives in one place rather than scattered across the root. Callers use direct submodule imports (from bot_bottle.store.store_manager import StoreManager). Inside the moved modules the relative imports point back up to their non-store siblings (..paths, ..supervise_types) while co-moved siblings stay package-local; the flat-import fallbacks are untouched. The orchestrator's own stores (config_store, secret_store, registry) stay in orchestrator/ and repoint to the moved DbStore / TableMigrations. No behavior change; full unit suite green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
"""Unit: DbStore._connection() context manager and is_migrated()."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from bot_bottle.store.db_store import DbStore
|
|
from bot_bottle.store.migrations import TableMigrations
|
|
|
|
|
|
def _store(tmp: Path) -> DbStore:
|
|
migrations = TableMigrations("test", ["CREATE TABLE items (id INTEGER PRIMARY KEY)"])
|
|
return DbStore(tmp / "test.db", migrations)
|
|
|
|
|
|
class TestDbStoreIsMigrated(unittest.TestCase):
|
|
def test_returns_false_when_db_absent(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
store = _store(Path(d))
|
|
self.assertFalse(store.is_migrated())
|
|
|
|
def test_returns_false_when_schema_versions_missing(self):
|
|
# DB file exists but has no schema_versions table → OperationalError → False.
|
|
with tempfile.TemporaryDirectory() as d:
|
|
store = _store(Path(d))
|
|
conn = sqlite3.connect(store.db_path)
|
|
conn.close()
|
|
self.assertFalse(store.is_migrated())
|
|
|
|
def test_returns_true_after_migrate(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
store = _store(Path(d))
|
|
store.migrate()
|
|
self.assertTrue(store.is_migrated())
|
|
|
|
def test_returns_false_when_behind(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
migrations = TableMigrations(
|
|
"test",
|
|
[
|
|
"CREATE TABLE items (id INTEGER PRIMARY KEY)",
|
|
"ALTER TABLE items ADD COLUMN name TEXT",
|
|
],
|
|
)
|
|
store = DbStore(Path(d) / "test.db", migrations)
|
|
# Apply only the first migration manually.
|
|
conn = sqlite3.connect(store.db_path)
|
|
with conn:
|
|
TableMigrations("test", [migrations.migrations[0]]).apply(conn)
|
|
conn.close()
|
|
self.assertFalse(store.is_migrated())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|