Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 17 additions & 6 deletions dashboard/plugin_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -942,12 +942,23 @@ def discover_kanban_boards() -> List[Dict[str, Any]]:
def _kanban_conn(path: Path) -> Optional[sqlite3.Connection]:
if not path.exists():
return None
try:
con = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=1.5)
con.row_factory = sqlite3.Row
return con
except Exception:
return None
# Some Kanban SQLite files can be opened in `mode=ro` but fail on the
# first read when SQLite tries to resolve journal/WAL sidecars. Verify the
# connection before returning it and fall back to immutable read-only mode.
for uri in (f"file:{path}?mode=ro", f"file:{path}?mode=ro&immutable=1"):
con: Optional[sqlite3.Connection] = None
try:
con = sqlite3.connect(uri, uri=True, timeout=1.5)
con.row_factory = sqlite3.Row
con.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchone()
return con
except Exception:
if con is not None:
try:
con.close()
except Exception:
pass
return None


def profile_config(profile_home: Path) -> Dict[str, Any]:
Expand Down
14 changes: 14 additions & 0 deletions tests/test_privacy_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,20 @@ def test_cron_ids_are_public_refs_when_local_labels_are_hidden(self):
self.assertIn("cron:", payload[0]["id"])
self.assertEqual(payload[0]["id"], payload[0]["job_id"])

def test_kanban_conn_reads_existing_board_db(self):
db = self.home / "kanban.db"
con = sqlite3.connect(db)
con.execute("CREATE TABLE tasks (id TEXT, status TEXT)")
con.commit()
con.close()

ro = plugin_api._kanban_conn(db)

self.assertIsNotNone(ro)
assert ro is not None
self.assertEqual(ro.execute("SELECT COUNT(*) FROM tasks").fetchone()[0], 0)
ro.close()

def test_kanban_worker_run_and_pid_ids_are_public_refs_when_local_labels_are_hidden(self):
db = self.home / "kanban.db"
con = sqlite3.connect(db)
Expand Down
Loading