160 lines
6.2 KiB
Python
160 lines
6.2 KiB
Python
"""Команды wiki и фоновая синхронизация (wiki-module, spec wiki «Синхронизация вики»)."""
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
from sdlc_hub.core import db, queue, settings
|
|
from sdlc_hub.core.errors import CommandError
|
|
from sdlc_hub.modules.jira import refresh
|
|
from sdlc_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}
|
|
|
|
|
|
# пакет D (#135): T8.3 — одна синхронизация на корень
|
|
SYNC = [{"type": "wiki_sync", "params": {"label": "AWPCA", "root": "DOC-1"}}]
|
|
|
|
|
|
def test_check_batch_refuses_second_sync_same_root(conn):
|
|
b = queue.create_batch(conn, "t", "u", SYNC, project="AWPCA")
|
|
with pytest.raises(CommandError, match=f"#{b['id']}"):
|
|
queue.create_batch(conn, "t", "u", SYNC, project="AWPCA")
|
|
|
|
|
|
def test_check_batch_other_root_allowed(conn):
|
|
queue.create_batch(conn, "t", "u", SYNC, project="AWPCA")
|
|
other = [{"type": "wiki_sync", "params": {"label": "AWPCA", "root": "DOC-2"}}]
|
|
assert queue.create_batch(conn, "t", "u", other, project="AWPCA")["count"] == 1
|
|
|
|
|
|
def test_retry_failed_sync_without_active_passes(conn):
|
|
b = queue.create_batch(conn, "t", "u", SYNC, project="AWPCA")
|
|
conn.execute("UPDATE command SET status = 'failed' WHERE batch_id = ?", (b["id"],))
|
|
assert queue.create_batch(conn, "t", "u", SYNC, project="AWPCA")["count"] == 1
|
|
|
|
|
|
def test_enqueue_sync_returns_active_same_root(conn):
|
|
_root(conn)
|
|
first = commands.enqueue_sync(conn, "AWPCA", "DOC-1", "u")
|
|
again = commands.enqueue_sync(conn, "AWPCA", "DOC-1", "u")
|
|
assert again == first and _count(conn) == 1
|