4ea287fbdf
Каркас модуля wiki: схема (wiki_project, wiki_page, wiki_attachment, wiki_sync_stage), настройки, поле проекта wiki_root с хуками создания и правки, команды wiki_sync и wiki_fetch, фоновая синхронизация. Зависимость markdownify>=1.2.3. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
131 lines
4.9 KiB
Python
131 lines
4.9 KiB
Python
"""Команды wiki и фоновая синхронизация (wiki-module, spec wiki «Синхронизация вики»)."""
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
from jira_hub.core import db, queue, settings
|
|
from jira_hub.core.errors import CommandError
|
|
from jira_hub.modules.jira import refresh
|
|
from jira_hub.modules.wiki import commands, settings as wiki_settings, tick
|
|
|
|
|
|
@pytest.fixture
|
|
def conn(open_db, add_project):
|
|
c = open_db()
|
|
settings.ensure_defaults(c)
|
|
add_project(c, "AWPCA")
|
|
return c
|
|
|
|
|
|
def _root(conn, synced_hours=None, failed_hours=None):
|
|
ago = lambda h: None if h is None else (db.utcnow() - timedelta(hours=h)).isoformat() # noqa: E731
|
|
conn.execute("INSERT OR REPLACE INTO wiki_project(label, root_code, synced_at, failed_at) VALUES ('AWPCA', 'DOC-1', ?, ?)",
|
|
(ago(synced_hours), ago(failed_hours)))
|
|
|
|
|
|
def _count(conn):
|
|
return conn.execute("SELECT COUNT(*) AS n FROM batch WHERE project = 'AWPCA'").fetchone()["n"]
|
|
|
|
|
|
def test_auto_sync_tick(conn):
|
|
_root(conn, synced_hours=25)
|
|
tick.auto_sync_tick(None, conn)
|
|
assert _count(conn) == 1
|
|
batch = queue.list_batches(conn, project="AWPCA")[0]
|
|
assert batch["title"] == "Вики AWPCA" and batch["author"] == "hub"
|
|
tick.auto_sync_tick(None, conn) # пакет в очереди: второй не ставится
|
|
assert _count(conn) == 1
|
|
|
|
|
|
def test_auto_sync_tick_recent(conn):
|
|
_root(conn, synced_hours=23)
|
|
tick.auto_sync_tick(None, conn)
|
|
assert _count(conn) == 0
|
|
|
|
|
|
def test_auto_sync_tick_failed_resets_clock(conn):
|
|
_root(conn, synced_hours=30, failed_hours=1) # свежая неудача откладывает следующую попытку
|
|
tick.auto_sync_tick(None, conn)
|
|
assert _count(conn) == 0
|
|
_root(conn, synced_hours=30, failed_hours=26)
|
|
tick.auto_sync_tick(None, conn)
|
|
assert _count(conn) == 1
|
|
|
|
|
|
def test_auto_sync_tick_never_synced_without_batch(conn):
|
|
_root(conn) # корень задан, пакета нет
|
|
tick.auto_sync_tick(None, conn)
|
|
assert _count(conn) == 1
|
|
|
|
|
|
def test_auto_sync_tick_disabled(conn):
|
|
_root(conn, synced_hours=25)
|
|
settings.set_(conn, "wiki_auto_sync_hours", 0)
|
|
tick.auto_sync_tick(None, conn)
|
|
assert _count(conn) == 0
|
|
|
|
|
|
def test_tick_label():
|
|
assert tick.auto_sync_tick.label == "синхронизация вики"
|
|
|
|
|
|
def test_active_sync(conn):
|
|
_root(conn, synced_hours=1)
|
|
assert commands.active_sync(conn, "AWPCA") is None
|
|
batch = commands.enqueue_sync(conn, "AWPCA", "DOC-1", "u")
|
|
assert commands.active_sync(conn, "AWPCA")["id"] == batch["id"]
|
|
queue.cancel_batch(conn, batch["id"])
|
|
assert commands.active_sync(conn, "AWPCA") is None
|
|
|
|
|
|
def test_auto_refresh_ignores_wiki_batches(conn):
|
|
_root(conn, synced_hours=1)
|
|
conn.execute("UPDATE agent_status SET last_seen_at = ? WHERE agent = 'jira'", (db.now(),))
|
|
commands.enqueue_sync(conn, "AWPCA", "DOC-1", "u")
|
|
batch = refresh.maybe_auto_refresh(conn)
|
|
assert batch is not None
|
|
|
|
|
|
def test_check_batch_requires_own_project(conn):
|
|
cmd = [{"type": "wiki_sync", "params": {"label": "AWPCA", "root": "DOC-1"}}]
|
|
with pytest.raises(CommandError):
|
|
queue.create_batch(conn, "t", "u", cmd, project="OTHER")
|
|
with pytest.raises(CommandError):
|
|
queue.create_batch(conn, "t", "u", cmd)
|
|
assert queue.create_batch(conn, "t", "u", cmd, project="AWPCA")["count"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize("params", [{"label": "AWPCA"}, {"label": "AWPCA", "root": "страница"},
|
|
{"label": "AWPCA", "root": "DOC-1", "x": 1}])
|
|
def test_validate_wiki_sync(params):
|
|
with pytest.raises(CommandError):
|
|
commands.validate([{"type": "wiki_sync", "params": params}])
|
|
|
|
|
|
def test_validate_wiki_fetch():
|
|
assert commands.validate([{"type": "wiki_fetch", "params": {"attachment_id": 7, "url": "https://e/f"}}])
|
|
with pytest.raises(CommandError):
|
|
commands.validate([{"type": "wiki_fetch", "params": {"attachment_id": "7", "url": "https://e/f"}}])
|
|
|
|
|
|
def test_read_types_have_no_tail_or_pause(conn):
|
|
assert queue.create_batch(conn, "t", "u", [{"type": "wiki_fetch", "params": {"attachment_id": 1, "url": "u"}}])["count"] == 1
|
|
assert queue.estimate(conn, ["wiki_sync", "wiki_fetch"]) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("key,bad,good", [("wiki_pacing", [-1, 11, "x", True], [0, 0.2, 10]),
|
|
("wiki_auto_sync_hours", [-1, 1.5, True, "1"], [0, 24]),
|
|
("wiki_min_agent_version", [-1, 1.5, "1"], [0, 3])])
|
|
def test_validate_settings(key, bad, good):
|
|
for v in bad:
|
|
with pytest.raises(ValueError):
|
|
settings.validate(key, v)
|
|
for v in good:
|
|
settings.validate(key, v)
|
|
assert wiki_settings.DEFAULTS[key] is not None
|
|
|
|
|
|
def test_wiki_agent_gets_own_keys(conn):
|
|
out = settings.for_agent(conn, "wiki")
|
|
assert out == {"wiki_pacing": 0.2, "settings_version": 1}
|