307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""Tests for core/obsidian_sync.py — KIN-013."""
|
||
|
||
import sqlite3
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from core.db import init_db
|
||
from core.obsidian_sync import (
|
||
export_decisions_to_md,
|
||
parse_task_checkboxes,
|
||
sync_obsidian,
|
||
)
|
||
from core import models
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 0. Migration — obsidian_vault_path column must exist after init_db
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_migration_obsidian_vault_path_column_exists():
|
||
"""init_db создаёт или мигрирует колонку obsidian_vault_path в таблице projects."""
|
||
conn = init_db(db_path=":memory:")
|
||
cols = {r[1] for r in conn.execute("PRAGMA table_info(projects)").fetchall()}
|
||
conn.close()
|
||
assert "obsidian_vault_path" in cols
|
||
|
||
|
||
@pytest.fixture
|
||
def tmp_vault(tmp_path):
|
||
"""Returns a temporary vault root directory."""
|
||
return tmp_path / "vault"
|
||
|
||
|
||
@pytest.fixture
|
||
def db(tmp_path):
|
||
"""Returns an in-memory SQLite connection with schema + test data."""
|
||
db_path = tmp_path / "test.db"
|
||
conn = init_db(db_path)
|
||
models.create_project(conn, "proj1", "Test Project", "/tmp/proj1")
|
||
yield conn
|
||
conn.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. export creates files with correct frontmatter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_export_decisions_creates_md_files(tmp_vault):
|
||
decisions = [
|
||
{
|
||
"id": 42,
|
||
"project_id": "proj1",
|
||
"type": "gotcha",
|
||
"category": "testing",
|
||
"title": "Proxy через SSH не работает без ssh-agent",
|
||
"description": "При подключении через ProxyJump ssh-agent должен быть запущен.",
|
||
"tags": ["testing", "mock", "subprocess"],
|
||
"created_at": "2026-03-10T12:00:00",
|
||
}
|
||
]
|
||
tmp_vault.mkdir(parents=True)
|
||
created = export_decisions_to_md("proj1", decisions, tmp_vault)
|
||
|
||
assert len(created) == 1
|
||
md_file = created[0]
|
||
assert md_file.exists()
|
||
|
||
content = md_file.read_text(encoding="utf-8")
|
||
assert "kin_decision_id: 42" in content
|
||
assert "project: proj1" in content
|
||
assert "type: gotcha" in content
|
||
assert "category: testing" in content
|
||
assert "2026-03-10" in content
|
||
assert "# Proxy через SSH не работает без ssh-agent" in content
|
||
assert "При подключении через ProxyJump" in content
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. export is idempotent (overwrite, not duplicate)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_export_idempotent(tmp_vault):
|
||
decisions = [
|
||
{
|
||
"id": 1,
|
||
"project_id": "p",
|
||
"type": "decision",
|
||
"category": None,
|
||
"title": "Use SQLite",
|
||
"description": "SQLite is the source of truth.",
|
||
"tags": [],
|
||
"created_at": "2026-01-01",
|
||
}
|
||
]
|
||
tmp_vault.mkdir(parents=True)
|
||
|
||
export_decisions_to_md("p", decisions, tmp_vault)
|
||
export_decisions_to_md("p", decisions, tmp_vault)
|
||
|
||
out_dir = tmp_vault / "p" / "decisions"
|
||
files = list(out_dir.glob("*.md"))
|
||
assert len(files) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. parse_task_checkboxes — done checkbox
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_parse_task_checkboxes_done(tmp_vault):
|
||
tasks_dir = tmp_vault / "proj1" / "tasks"
|
||
tasks_dir.mkdir(parents=True)
|
||
(tasks_dir / "kanban.md").write_text(
|
||
"- [x] KIN-001 Implement login\n- [ ] KIN-002 Add tests\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
results = parse_task_checkboxes(tmp_vault, "proj1")
|
||
done_items = [r for r in results if r["task_id"] == "KIN-001"]
|
||
assert len(done_items) == 1
|
||
assert done_items[0]["done"] is True
|
||
assert done_items[0]["title"] == "Implement login"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. parse_task_checkboxes — pending checkbox
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_parse_task_checkboxes_pending(tmp_vault):
|
||
tasks_dir = tmp_vault / "proj1" / "tasks"
|
||
tasks_dir.mkdir(parents=True)
|
||
(tasks_dir / "kanban.md").write_text(
|
||
"- [ ] KIN-002 Add tests\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
results = parse_task_checkboxes(tmp_vault, "proj1")
|
||
pending = [r for r in results if r["task_id"] == "KIN-002"]
|
||
assert len(pending) == 1
|
||
assert pending[0]["done"] is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. parse_task_checkboxes — lines without task ID are skipped
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_parse_task_checkboxes_no_id(tmp_vault):
|
||
tasks_dir = tmp_vault / "proj1" / "tasks"
|
||
tasks_dir.mkdir(parents=True)
|
||
(tasks_dir / "notes.md").write_text(
|
||
"- [x] Some task without ID\n"
|
||
"- [ ] Another line without identifier\n"
|
||
"- [x] KIN-003 With ID\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
results = parse_task_checkboxes(tmp_vault, "proj1")
|
||
assert all(r["task_id"].startswith("KIN-") for r in results)
|
||
assert len(results) == 1
|
||
assert results[0]["task_id"] == "KIN-003"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. sync_obsidian updates task status when done=True
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_sync_updates_task_status(db, tmp_vault):
|
||
tmp_vault.mkdir(parents=True)
|
||
models.update_project(db, "proj1", obsidian_vault_path=str(tmp_vault))
|
||
|
||
task = models.create_task(db, "PROJ1-001", "proj1", "Do something", status="in_progress")
|
||
assert task["status"] == "in_progress"
|
||
|
||
# Write checkbox file
|
||
tasks_dir = tmp_vault / "proj1" / "tasks"
|
||
tasks_dir.mkdir(parents=True)
|
||
(tasks_dir / "sprint.md").write_text(
|
||
"- [x] PROJ1-001 Do something\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
result = sync_obsidian(db, "proj1")
|
||
|
||
assert result["tasks_updated"] == 1
|
||
assert not result["errors"]
|
||
updated = models.get_task(db, "PROJ1-001")
|
||
assert updated["status"] == "done"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7. sync_obsidian raises ValueError when vault_path not set
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_sync_no_vault_path(db):
|
||
# project exists but obsidian_vault_path is NULL
|
||
with pytest.raises(ValueError, match="obsidian_vault_path not set"):
|
||
sync_obsidian(db, "proj1")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 8. export — frontmatter обёрнут в разделители ---
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_export_frontmatter_has_yaml_delimiters(tmp_vault):
|
||
"""Экспортированный файл начинается с '---' и содержит закрывающий '---'."""
|
||
decisions = [
|
||
{
|
||
"id": 99,
|
||
"project_id": "p",
|
||
"type": "decision",
|
||
"category": None,
|
||
"title": "YAML Delimiter Test",
|
||
"description": "Verifying frontmatter delimiters.",
|
||
"tags": [],
|
||
"created_at": "2026-01-01",
|
||
}
|
||
]
|
||
tmp_vault.mkdir(parents=True)
|
||
created = export_decisions_to_md("p", decisions, tmp_vault)
|
||
content = created[0].read_text(encoding="utf-8")
|
||
|
||
assert content.startswith("---\n"), "Frontmatter должен начинаться с '---\\n'"
|
||
# первые --- открывают, вторые --- закрывают frontmatter
|
||
parts = content.split("---\n")
|
||
assert len(parts) >= 3, "Должно быть минимум два разделителя '---'"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 9. sync_obsidian — несуществующий vault_path → директория создаётся автоматически
|
||
# KIN-070: Регрессионный тест на автоматическое создание vault directory
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_kin070_sync_creates_missing_vault_directory(db, tmp_path):
|
||
"""KIN-070: Если vault_path не существует, sync автоматически создаёт директорию.
|
||
|
||
Проверяет что:
|
||
- Директория создаётся без ошибок
|
||
- sync_obsidian не падает с ошибкой
|
||
- Возвращаемый результат содержит errors=[]
|
||
"""
|
||
nonexistent = tmp_path / "ghost_vault"
|
||
models.update_project(db, "proj1", obsidian_vault_path=str(nonexistent))
|
||
|
||
result = sync_obsidian(db, "proj1")
|
||
|
||
assert result["errors"] == []
|
||
assert nonexistent.is_dir() # директория автоматически создана
|
||
assert result["exported_decisions"] == 0 # нет decisions в DB
|
||
assert result["tasks_updated"] == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 10. sync_obsidian + decisions: несуществующий vault + decisions в БД → export success
|
||
# KIN-070: Проверяет что decisions экспортируются когда vault создаётся автоматически
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_kin070_sync_creates_vault_and_exports_decisions(db, tmp_path):
|
||
"""KIN-070: sync экспортирует decisions и автоматически создаёт vault_path.
|
||
|
||
Проверяет что:
|
||
- vault директория создаётся автоматически
|
||
- decisions экспортируются в .md-файлы (exported_decisions > 0)
|
||
- errors == [] (нет ошибок)
|
||
"""
|
||
nonexistent = tmp_path / "missing_vault"
|
||
models.update_project(db, "proj1", obsidian_vault_path=str(nonexistent))
|
||
|
||
# Создаём decision в БД
|
||
models.add_decision(
|
||
db,
|
||
project_id="proj1",
|
||
type="decision",
|
||
title="Use SQLite for sync state",
|
||
description="SQLite will be the single source of truth.",
|
||
tags=["database", "sync"],
|
||
)
|
||
|
||
result = sync_obsidian(db, "proj1")
|
||
|
||
# Проверяем успешный экспорт
|
||
assert result["errors"] == []
|
||
assert nonexistent.is_dir() # директория создана
|
||
assert result["exported_decisions"] == 1 # одно decision экспортировано
|
||
assert result["tasks_updated"] == 0
|
||
|
||
# Проверяем что .md-файл создан в правильной директории
|
||
decisions_dir = nonexistent / "proj1" / "decisions"
|
||
assert decisions_dir.is_dir()
|
||
md_files = list(decisions_dir.glob("*.md"))
|
||
assert len(md_files) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 11. sync_obsidian — пустой vault → 0 экспортов, 0 обновлений, нет ошибок
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_sync_empty_vault_no_errors(db, tmp_vault):
|
||
"""Пустой vault (нет decisions, нет task-файлов) → exported=0, updated=0, errors=[]."""
|
||
tmp_vault.mkdir(parents=True)
|
||
models.update_project(db, "proj1", obsidian_vault_path=str(tmp_vault))
|
||
|
||
result = sync_obsidian(db, "proj1")
|
||
|
||
assert result["exported_decisions"] == 0
|
||
assert result["tasks_updated"] == 0
|
||
assert result["errors"] == []
|