e07c8e76e9
- 76 new tests covering CLI commands, API routes, auth, runner, load modules — coverage 99% (169 remaining lines are infrastructure) - Enhanced /health endpoint with DuckDB, bib, pipeline service checks - Added `stack health` CLI command with graceful degradation - SnapshotManager for DuckDB data versioning and rollback - .woodpecker.yml CI pipeline: lint, test (99% gate), marimo check - Error handling audit: no bare except clauses, all exceptions logged Closes #31, #32, #33, #34, #35.
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
"""Tests for aco.lake.snapshot — data versioning."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
|
|
from aco.lake.snapshot import SnapshotManager
|
|
|
|
|
|
class TestSnapshotManager:
|
|
def test_create_and_list(self, tmp_path) -> None:
|
|
db_path = tmp_path / "test.duckdb"
|
|
con = duckdb.connect(str(db_path))
|
|
con.execute("CREATE TABLE t1 (a INT)")
|
|
con.execute("INSERT INTO t1 VALUES (1), (2), (3)")
|
|
con.close()
|
|
|
|
mgr = SnapshotManager(str(db_path))
|
|
snap_id = mgr.create(label="test snapshot")
|
|
assert snap_id
|
|
|
|
snaps = mgr.list_snapshots()
|
|
assert len(snaps) == 1
|
|
assert snaps[0]["label"] == "test snapshot"
|
|
assert snaps[0]["snapshot_id"] == snap_id
|
|
|
|
def test_rollback(self, tmp_path) -> None:
|
|
db_path = tmp_path / "test.duckdb"
|
|
|
|
# Create initial state
|
|
con = duckdb.connect(str(db_path))
|
|
con.execute("CREATE TABLE t1 (a INT)")
|
|
con.execute("INSERT INTO t1 VALUES (1), (2), (3)")
|
|
con.close()
|
|
|
|
mgr = SnapshotManager(str(db_path))
|
|
snap_id = mgr.create(label="before change")
|
|
|
|
# Modify the database
|
|
con = duckdb.connect(str(db_path))
|
|
con.execute("INSERT INTO t1 VALUES (4), (5)")
|
|
count_after = con.execute("SELECT COUNT(*) FROM t1").fetchone()[0]
|
|
con.close()
|
|
assert count_after == 5
|
|
|
|
# Rollback
|
|
assert mgr.rollback(snap_id) is True
|
|
|
|
# Verify rollback
|
|
con = duckdb.connect(str(db_path), read_only=True)
|
|
count_rolled = con.execute("SELECT COUNT(*) FROM t1").fetchone()[0]
|
|
con.close()
|
|
assert count_rolled == 3
|
|
|
|
def test_rollback_missing(self, tmp_path) -> None:
|
|
db_path = tmp_path / "test.duckdb"
|
|
duckdb.connect(str(db_path)).close()
|
|
|
|
mgr = SnapshotManager(str(db_path))
|
|
assert mgr.rollback("nonexistent") is False
|
|
|
|
def test_empty_list(self, tmp_path) -> None:
|
|
db_path = tmp_path / "test.duckdb"
|
|
duckdb.connect(str(db_path)).close()
|
|
|
|
mgr = SnapshotManager(str(db_path))
|
|
snaps = mgr.list_snapshots()
|
|
assert snaps == []
|
|
|
|
def test_git_sha(self, tmp_path) -> None:
|
|
db_path = tmp_path / "test.duckdb"
|
|
mgr = SnapshotManager(str(db_path))
|
|
sha = mgr._git_sha()
|
|
# Should return a short SHA or "unknown"
|
|
assert isinstance(sha, str)
|
|
assert len(sha) > 0
|