test: cover _daemon_reachable timeout path and DbStore.is_migrated
test / integration-firecracker (pull_request) Successful in 15s
tracker-policy-pr / check-pr (pull_request) Successful in 14s
test / integration-docker (pull_request) Successful in 28s
test / unit (pull_request) Successful in 37s
lint / lint (push) Successful in 43s
test / coverage (pull_request) Successful in 34s
test / integration-firecracker (pull_request) Successful in 15s
tracker-policy-pr / check-pr (pull_request) Successful in 14s
test / integration-docker (pull_request) Successful in 28s
test / unit (pull_request) Successful in 37s
lint / lint (push) Successful in 43s
test / coverage (pull_request) Successful in 34s
Two diff-coverage gaps on the ci-kvm-runner branch: 1. bot_bottle/backend/docker/setup.py: the try/except TimeoutExpired block added in a prior commit had no tests reaching the subprocess path. Add two tests to TestDockerSetupStatus: one for the success path (subprocess returns 0) and one for the TimeoutExpired fallback. 2. bot_bottle/db_store.py: the _connection() context manager change in is_migrated() was never exercised by unit tests (all callers mock is_migrated() directly). Add test_db_store.py covering the absent-DB, missing-schema-table, migrated, and behind-schema cases.
This commit is contained in:
@@ -215,6 +215,18 @@ class TestDockerSetupStatus(unittest.TestCase):
|
|||||||
with patch.object(dk.shutil, "which", return_value=None):
|
with patch.object(dk.shutil, "which", return_value=None):
|
||||||
self.assertFalse(dk._daemon_reachable())
|
self.assertFalse(dk._daemon_reachable())
|
||||||
|
|
||||||
|
def test_daemon_reachable_true_when_daemon_responds(self):
|
||||||
|
with patch.object(dk.shutil, "which", return_value="/usr/bin/docker"), \
|
||||||
|
patch.object(dk.subprocess, "run",
|
||||||
|
return_value=subprocess.CompletedProcess([], 0)):
|
||||||
|
self.assertTrue(dk._daemon_reachable())
|
||||||
|
|
||||||
|
def test_daemon_reachable_false_on_timeout(self):
|
||||||
|
with patch.object(dk.shutil, "which", return_value="/usr/bin/docker"), \
|
||||||
|
patch.object(dk.subprocess, "run",
|
||||||
|
side_effect=subprocess.TimeoutExpired(["docker", "info"], 5)):
|
||||||
|
self.assertFalse(dk._daemon_reachable())
|
||||||
|
|
||||||
def test_status_reports_missing_docker(self):
|
def test_status_reports_missing_docker(self):
|
||||||
with patch.object(dk.shutil, "which", return_value=None):
|
with patch.object(dk.shutil, "which", return_value=None):
|
||||||
rc, out = _cap(dk.status)
|
rc, out = _cap(dk.status)
|
||||||
|
|||||||
@@ -0,0 +1,58 @@
|
|||||||
|
"""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.db_store import DbStore
|
||||||
|
from bot_bottle.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()
|
||||||
Reference in New Issue
Block a user