test(coverage): add tests for saga_crypto, read_only, migrations (+25)

tests/test_shared/test_migrations_coverage.py

@@ -0,0 +1,84 @@
+"""Tests for shared/migrations.py — full coverage."""
+
+import asyncio
+import pytest
+from shared.migrations import MigrationManager
+
+
+@pytest.fixture
+def mm(tmp_path):
+    from shared.connection import AsyncConnectionManager
+
+    cm = AsyncConnectionManager(base_dir=str(tmp_path))
+    return MigrationManager(cm=cm)
+
+
+def test_get_current_version_empty(mm):
+    """get_current_version should return 0 for empty DB."""
+
+    async def t():
+        version = await mm.get_current_version()
+        assert version == 0
+
+    asyncio.run(t())
+
+
+def test_migrate_runs(mm):
+    """migrate should run without error."""
+
+    async def t():
+        result = await mm.migrate()
+        assert isinstance(result, dict)
+        assert "applied" in result
+        assert len(result["applied"]) > 0
+
+    asyncio.run(t())
+
+
+def test_migrate_idempotent(mm):
+    """Running migrate twice should not re-apply."""
+
+    async def t():
+        r1 = await mm.migrate()
+        r2 = await mm.migrate()
+        assert len(r2["applied"]) == 0  # No new migrations
+        assert r2["current_version"] == r1["new_version"]
+
+    asyncio.run(t())
+
+
+def test_get_pending(mm):
+    """get_pending should return pending migrations."""
+
+    async def t():
+        pending = await mm.get_pending()
+        assert isinstance(pending, list)
+        # After migrate, no pending
+        await mm.migrate()
+        pending_after = await mm.get_pending()
+        assert len(pending_after) == 0
+
+    asyncio.run(t())
+
+
+def test_migrate_returns_version_info(mm):
+    """migrate should return version info."""
+
+    async def t():
+        result = await mm.migrate()
+        assert "current_version" in result
+        assert "new_version" in result
+        assert result["new_version"] >= result["current_version"]
+
+    asyncio.run(t())
+
+
+def test_get_current_version_after_migrate(mm):
+    """get_current_version should return correct version after migrate."""
+
+    async def t():
+        await mm.migrate()
+        version = await mm.get_current_version()
+        assert version > 0
+
+    asyncio.run(t())

tests/test_shared/test_read_only_coverage.py

@@ -0,0 +1,122 @@
+"""Tests for shared/read_only.py — full coverage."""
+
+import sqlite3
+from shared.read_only import ReadOnlyReplica
+
+
+def _create_source_db(path):
+    """Helper to create a source database with test data."""
+    path.mkdir(parents=True, exist_ok=True)
+    db = path / "memory.db"
+    conn = sqlite3.connect(str(db))
+    conn.execute("CREATE TABLE test (id INTEGER, name TEXT)")
+    conn.execute("INSERT INTO test VALUES (1, 'alice')")
+    conn.execute("INSERT INTO test VALUES (2, 'bob')")
+    conn.commit()
+    conn.close()
+    return db
+
+
+def test_sync_creates_replica(tmp_path):
+    """sync should create a replica database."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    result = replica.sync()
+    assert result.get("memory.db") == 1
+    assert (tmp_path / "replica" / "memory.db").exists()
+
+
+def test_sync_returns_empty_when_no_source(tmp_path):
+    """sync should return empty when source doesn't exist."""
+    replica = ReadOnlyReplica(source_dir=str(tmp_path / "nonexistent"), replica_dir=str(tmp_path / "replica"))
+    result = replica.sync()
+    assert result == {}
+
+
+def test_get_conn_returns_readonly(tmp_path):
+    """get_conn should return a read-only connection."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    replica.sync()
+
+    conn = replica.get_conn()
+    assert conn is not None
+    # Should be able to read
+    cur = conn.execute("SELECT * FROM test")
+    rows = cur.fetchall()
+    assert len(rows) == 2
+
+
+def test_get_conn_falls_back_to_source(tmp_path):
+    """get_conn should fall back to source if replica doesn't exist."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    # Don't sync — replica doesn't exist
+
+    conn = replica.get_conn()
+    assert conn is not None
+    cur = conn.execute("SELECT * FROM test")
+    rows = cur.fetchall()
+    assert len(rows) == 2
+
+
+def test_is_ready_false_when_no_replica(tmp_path):
+    """is_ready should return False when replica doesn't exist."""
+    replica = ReadOnlyReplica(source_dir=str(tmp_path / "source"), replica_dir=str(tmp_path / "replica"))
+    assert replica.is_ready() is False
+
+
+def test_is_ready_true_after_sync(tmp_path):
+    """is_ready should return True after sync."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    replica.sync()
+    assert replica.is_ready() is True
+
+
+def test_start_stop_auto_sync(tmp_path):
+    """start_auto_sync/stop should manage the background thread."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    replica.start_auto_sync(interval_seconds=1)
+    assert replica._running is True
+    assert replica._thread is not None
+
+    replica.stop()
+    assert replica._running is False
+
+
+def test_start_auto_sync_idempotent(tmp_path):
+    """start_auto_sync should not create multiple threads."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    replica.start_auto_sync(interval_seconds=1)
+    thread1 = replica._thread
+    replica.start_auto_sync(interval_seconds=1)
+    thread2 = replica._thread
+    assert thread1 is thread2
+
+    replica.stop()
+
+
+def test_sync_updates_last_sync_time(tmp_path):
+    """sync should update _last_sync timestamp."""
+    src = tmp_path / "source"
+    _create_source_db(src)
+
+    replica = ReadOnlyReplica(source_dir=str(src), replica_dir=str(tmp_path / "replica"))
+    assert replica._last_sync == 0.0
+    replica.sync()
+    assert replica._last_sync > 0.0

tests/test_shared/test_saga_crypto_coverage.py

@@ -0,0 +1,106 @@
+"""Tests for shared/saga_crypto.py — full coverage."""
+
+import json
+import warnings
+import pytest
+from pathlib import Path
+from shared.saga_crypto import read_state, read_state_legacy_or_encrypted, write_state_atomic
+
+
+def test_write_state_atomic_creates_encrypted_file(tmp_path):
+    """write_state_atomic should create an encrypted file."""
+    path = tmp_path / "test.json"
+    state = {"key": "value", "nested": {"a": 1}}
+    write_state_atomic(path, state)
+    assert path.exists()
+    data = path.read_bytes()
+    assert len(data) > 0
+    # Should not be plain JSON (encrypted)
+    assert not data.startswith(b"{")
+
+
+def test_write_state_atomic_creates_parent_dirs(tmp_path):
+    """write_state_atomic should create parent directories."""
+    path = tmp_path / "deep" / "nested" / "dir" / "test.json"
+    write_state_atomic(path, {"key": "value"})
+    assert path.exists()
+
+
+def test_write_state_atomic_replaces_existing(tmp_path):
+    """write_state_atomic should replace existing file."""
+    path = tmp_path / "test.json"
+    write_state_atomic(path, {"old": True})
+    write_state_atomic(path, {"new": True})
+    assert path.exists()
+    # Should be re-encrypted
+    data = path.read_bytes()
+    assert len(data) > 0
+
+
+def test_read_state_reads_encrypted(tmp_path):
+    """read_state should read encrypted file."""
+    path = tmp_path / "test.json"
+    write_state_atomic(path, {"key": "value"})
+    loaded = read_state(path)
+    assert loaded == {"key": "value"}
+
+
+def test_read_state_raises_for_missing():
+    """read_state should raise FileNotFoundError for missing file."""
+    with pytest.raises(FileNotFoundError):
+        read_state(Path("/nonexistent/path.json"))
+
+
+def test_read_state_legacy_rotates_to_encrypted(tmp_path):
+    """read_state_legacy_or_encrypted should rotate legacy JSON to encrypted."""
+    path = tmp_path / "legacy.json"
+    # Write plain JSON (not encrypted)
+    path.write_text(json.dumps({"legacy": True}), encoding="utf-8")
+
+    with warnings.catch_warnings(record=True) as w:
+        warnings.simplefilter("always")
+        loaded = read_state_legacy_or_encrypted(path)
+        assert loaded == {"legacy": True}
+        # Should have warned about rotation
+        assert len(w) == 1
+        assert "rotating" in str(w[0].message).lower()
+
+    # File should now be encrypted
+    data = path.read_bytes()
+    assert not data.startswith(b"{")
+
+
+def test_read_state_legacy_reads_encrypted(tmp_path):
+    """read_state_legacy_or_encrypted should read already-encrypted file."""
+    path = tmp_path / "encrypted.json"
+    write_state_atomic(path, {"encrypted": True})
+
+    with warnings.catch_warnings(record=True) as w:
+        warnings.simplefilter("always")
+        loaded = read_state_legacy_or_encrypted(path)
+        assert loaded == {"encrypted": True}
+        # No warning for encrypted files
+        assert len(w) == 0
+
+
+def test_read_state_legacy_raises_for_missing():
+    """read_state_legacy_or_encrypted should raise FileNotFoundError."""
+    with pytest.raises(FileNotFoundError):
+        read_state_legacy_or_encrypted(Path("/nonexistent.json"))
+
+
+def test_write_read_roundtrip(tmp_path):
+    """Write then read should return same data."""
+    path = tmp_path / "roundtrip.json"
+    original = {"users": ["alice", "bob"], "count": 42, "nested": {"deep": True}}
+    write_state_atomic(path, original)
+    loaded = read_state(path)
+    assert loaded == original
+
+
+def test_write_state_atomic_chmod_error(tmp_path):
+    """write_state_atomic should handle chmod errors gracefully."""
+    path = tmp_path / "test.json"
+    # Should not raise even if chmod fails
+    write_state_atomic(path, {"key": "value"})
+    assert path.exists()