107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Tests for `openrun.setup.init_workspace` — idempotency, state reporting."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from openrun.config import Config
|
|
from openrun.setup import EXPECTED_TABLES, init_workspace
|
|
|
|
|
|
def _cfg(tmp_path: Path) -> Config:
|
|
"""A Config that points at an isolated tmp_path DB."""
|
|
return Config(db_path=tmp_path / "test.db")
|
|
|
|
|
|
def _empty_secrets(tmp_path: Path) -> Path:
|
|
"""A non-existent secrets dir under tmp_path — guarantees tests never read the real one."""
|
|
return tmp_path / ".secrets"
|
|
|
|
|
|
def test_first_run_creates_db(tmp_path: Path) -> None:
|
|
cfg = _cfg(tmp_path)
|
|
status = init_workspace(config=cfg, secrets_dir=_empty_secrets(tmp_path))
|
|
|
|
assert not status.db_existed_before
|
|
assert cfg.db_path.exists()
|
|
assert status.is_ready()
|
|
assert set(EXPECTED_TABLES).issubset(status.tables_present)
|
|
|
|
|
|
def test_second_run_is_idempotent(tmp_path: Path) -> None:
|
|
cfg = _cfg(tmp_path)
|
|
secrets = _empty_secrets(tmp_path)
|
|
|
|
first = init_workspace(config=cfg, secrets_dir=secrets)
|
|
second = init_workspace(config=cfg, secrets_dir=secrets)
|
|
|
|
assert not first.db_existed_before
|
|
assert second.db_existed_before
|
|
assert first.tables_present == second.tables_present
|
|
# Idempotent: re-running mutates nothing, so all counts stay zero.
|
|
assert all(n == 0 for n in second.table_counts.values())
|
|
|
|
|
|
def test_preserves_existing_data(tmp_path: Path) -> None:
|
|
cfg = _cfg(tmp_path)
|
|
secrets = _empty_secrets(tmp_path)
|
|
|
|
init_workspace(config=cfg, secrets_dir=secrets)
|
|
|
|
# Insert one row through a raw connection; init_workspace must not touch it.
|
|
conn = sqlite3.connect(cfg.db_path)
|
|
conn.execute(
|
|
"INSERT INTO activities (activity_id, raw, fetched_at) VALUES (1, '{}', 'now')"
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
status = init_workspace(config=cfg, secrets_dir=secrets)
|
|
assert status.db_existed_before
|
|
assert status.table_counts["activities"] == 1
|
|
|
|
|
|
def test_reports_auth_tokens(tmp_path: Path) -> None:
|
|
cfg = _cfg(tmp_path)
|
|
secrets = tmp_path / ".secrets"
|
|
|
|
missing = init_workspace(config=cfg, secrets_dir=secrets)
|
|
assert not missing.auth_tokens_present
|
|
|
|
secrets.mkdir()
|
|
(secrets / "oauth1_token.json").write_text("{}")
|
|
|
|
present = init_workspace(config=cfg, secrets_dir=secrets)
|
|
assert present.auth_tokens_present
|
|
|
|
|
|
def test_reports_sync_state(tmp_path: Path) -> None:
|
|
cfg = _cfg(tmp_path)
|
|
secrets = _empty_secrets(tmp_path)
|
|
|
|
init_workspace(config=cfg, secrets_dir=secrets)
|
|
conn = sqlite3.connect(cfg.db_path)
|
|
conn.execute(
|
|
"INSERT INTO sync_state (key, value, updated_at) VALUES (?, ?, datetime('now'))",
|
|
("last_sync_utc", "2026-05-18T10:30:00"),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
status = init_workspace(config=cfg, secrets_dir=secrets)
|
|
assert status.last_sync_utc == "2026-05-18T10:30:00"
|
|
assert status.last_ingest_utc is None
|
|
|
|
|
|
def test_status_serialises_to_json(tmp_path: Path) -> None:
|
|
"""`to_dict()` output must be JSON-encodable for `--json` CLI usage."""
|
|
import json
|
|
|
|
status = init_workspace(config=_cfg(tmp_path), secrets_dir=_empty_secrets(tmp_path))
|
|
blob = json.dumps(status.to_dict(), default=str)
|
|
parsed = json.loads(blob)
|
|
assert parsed["db_path"].endswith("test.db")
|
|
assert parsed["db_existed_before"] is False
|
|
assert "activities" in parsed["tables_present"]
|