test: cover QueueStore/AuditStore guard branches and supervise bundle spec
Add 8 tests covering the branches that were keeping diff-coverage below 90%: explicit db_path constructor arg, early-return guards when the DB file is absent, _chmod OSError swallowing in both store classes, and the supervise volume/env/daemon path in _bundle_launch_spec. Diff-coverage rises from 89.2% to 94.6% (176/186 changed lines).
This commit is contained in:
@@ -422,6 +422,14 @@ class TestBundleLaunchSpec(unittest.TestCase):
|
|||||||
spec.environment,
|
spec.environment,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def test_supervise_adds_daemon_volume_and_env(self):
|
||||||
|
from bot_bottle.supervise import DB_PATH_IN_CONTAINER
|
||||||
|
plan = _plan(supervise=True)
|
||||||
|
spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
|
||||||
|
self.assertIn("supervise", spec.daemons_csv)
|
||||||
|
self.assertIn(f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}", spec.environment)
|
||||||
|
self.assertIn(("/tmp/bot-bottle.db", DB_PATH_IN_CONTAINER, False), spec.volumes)
|
||||||
|
|
||||||
def test_canary_env_visible_to_smolvm_guest(self):
|
def test_canary_env_visible_to_smolvm_guest(self):
|
||||||
plan = _plan(canary=True)
|
plan = _plan(canary=True)
|
||||||
with patch.object(
|
with patch.object(
|
||||||
|
|||||||
@@ -7,9 +7,12 @@ from __future__ import annotations
|
|||||||
import tempfile
|
import tempfile
|
||||||
import time
|
import time
|
||||||
import unittest
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
from bot_bottle import supervise
|
from bot_bottle import supervise
|
||||||
|
from bot_bottle.audit_store import AuditStore
|
||||||
|
from bot_bottle.queue_store import QueueStore
|
||||||
from bot_bottle.supervise import (
|
from bot_bottle.supervise import (
|
||||||
AuditEntry,
|
AuditEntry,
|
||||||
Proposal,
|
Proposal,
|
||||||
@@ -132,5 +135,56 @@ class TestReadAuditEntries(unittest.TestCase):
|
|||||||
self.assertEqual([], entries)
|
self.assertEqual([], entries)
|
||||||
|
|
||||||
|
|
||||||
|
class TestStoreGuardBranches(unittest.TestCase):
|
||||||
|
"""Direct QueueStore / AuditStore construction and early-return guard branches."""
|
||||||
|
|
||||||
|
def test_queue_store_explicit_db_path(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "q.db"
|
||||||
|
store = QueueStore("key", db_path=db)
|
||||||
|
self.assertTrue(db.is_file())
|
||||||
|
self.assertEqual(db, store.db_path)
|
||||||
|
|
||||||
|
def test_queue_store_missing_db_list_pending_returns_empty(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "q.db"
|
||||||
|
store = QueueStore("key", db_path=db)
|
||||||
|
db.unlink()
|
||||||
|
self.assertEqual([], store.list_pending_proposals())
|
||||||
|
|
||||||
|
def test_queue_store_missing_db_list_all_returns_empty(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "q.db"
|
||||||
|
store = QueueStore("key", db_path=db)
|
||||||
|
db.unlink()
|
||||||
|
self.assertEqual([], store.list_all_pending_proposals())
|
||||||
|
|
||||||
|
def test_queue_store_missing_db_archive_is_noop(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "q.db"
|
||||||
|
store = QueueStore("key", db_path=db)
|
||||||
|
db.unlink()
|
||||||
|
store.archive_proposal("anything") # must not raise
|
||||||
|
|
||||||
|
def test_queue_store_chmod_oserror_is_swallowed(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "q.db"
|
||||||
|
with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
|
||||||
|
QueueStore("key", db_path=db) # must not raise
|
||||||
|
|
||||||
|
def test_audit_store_missing_db_read_returns_empty(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "a.db"
|
||||||
|
store = AuditStore(db_path=db)
|
||||||
|
db.unlink()
|
||||||
|
self.assertEqual([], store.read_audit_entries("egress", "slug"))
|
||||||
|
|
||||||
|
def test_audit_store_chmod_oserror_is_swallowed(self):
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
db = Path(d) / "a.db"
|
||||||
|
with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
|
||||||
|
AuditStore(db_path=db) # must not raise
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user