117 lines
5.2 KiB
Python
117 lines
5.2 KiB
Python
"""Общие фикстуры тестов. Соединения SQLite, открытые тестом, закрываются после него — в тестах нет
|
||
ResourceWarning «unclosed database» (hub-hygiene-d, решение 14; M-9/O-4 #4)."""
|
||
import re
|
||
import threading
|
||
|
||
import pytest
|
||
|
||
from jira_hub.core import api, db, projects, registry, settings
|
||
from jira_hub.core.client import HubClient
|
||
from jira_hub.core.modules import CommandHooks, Module
|
||
from jira_hub.modules.jira import MODULE as JIRA_MODULE
|
||
from jira_hub.modules.jira import project_fields
|
||
from jira_hub.modules.jira import projects as jira_projects
|
||
|
||
|
||
@pytest.fixture
|
||
def open_db():
|
||
"""Фабрика соединений: open_db() — база в памяти, open_db(path) — файл. Все закрываются после теста."""
|
||
opened = []
|
||
|
||
def make(path=":memory:"):
|
||
conn = db.connect(path)
|
||
opened.append(conn)
|
||
return conn
|
||
|
||
yield make
|
||
for conn in opened:
|
||
conn.close()
|
||
|
||
|
||
@pytest.fixture
|
||
def add_project():
|
||
"""add_project(conn, label="AWPCA", name=None, board_id=4216, board_type="scrum") -> dict: завести проект."""
|
||
def make(conn, label="AWPCA", name=None, board_id=4216, board_type="scrum"):
|
||
if name is None:
|
||
name = jira_projects.LEGACY_NAME if label == jira_projects.LEGACY_LABEL else label
|
||
jira_projects.create(conn, label, name, board_id) # проект ядра и доска в jira_project
|
||
if board_id is not None and board_type is not None:
|
||
jira_projects.record_board_types(conn, {str(board_id): board_type})
|
||
return projects.get(conn, label) | project_fields.view(conn, [label])[label]
|
||
return make
|
||
|
||
|
||
def _stub_route(hub, conn, m, q, body):
|
||
return 200, {"agent": q["_agent"]}
|
||
|
||
|
||
def _stub_hooks() -> CommandHooks:
|
||
"""Заглушка хуков второго модуля: типы stub_cmd (только чтение), остальное — пустые функции."""
|
||
noop = lambda *a, **k: None # noqa: E731
|
||
return CommandHooks(
|
||
types=frozenset({"stub_cmd"}), read_types=frozenset({"stub_cmd"}), validate=lambda cmds: cmds,
|
||
check_batch=lambda conn, cmds, project, files_dir: cmds, tail=lambda cmds: None, pause=lambda conn, t: 0,
|
||
deliver_params=lambda conn, cmd, files_dir: cmd["params"], temp_definer=None, temp_keys_used=lambda c: [],
|
||
substitute=lambda t, p, mapping: p, drop_unresolved=lambda c, u: u, retry_params=lambda c: c["params"],
|
||
check_confirmed=lambda conn, c: (True, None), on_claim=noop, on_failed=noop,
|
||
check_result=lambda conn, c, status, result, files_dir: (status, None), on_result=noop, after_result=noop,
|
||
on_late_result=lambda *a: False)
|
||
|
||
|
||
@pytest.fixture
|
||
def stub_module(monkeypatch) -> Module:
|
||
"""Второй модуль с командами и своим маршрутом агента; реестр подменён на [jira, stub]. Маршрут добавлен и
|
||
в скомпилированную таблицу api (она собирается при импорте)."""
|
||
stub = Module(name="stub", commands=_stub_hooks(), agent_keys=("stub_key",), settings_defaults={"stub_key": 1},
|
||
routes=(("POST", r"^/agent/stub$", _stub_route, "agent"),))
|
||
monkeypatch.setattr(registry, "_cache", [JIRA_MODULE, stub])
|
||
route = stub.routes[0]
|
||
monkeypatch.setattr(api, "_COMPILED", api._COMPILED + [(route[0], re.compile(route[1]), route[2], route[3], "stub")])
|
||
return stub
|
||
|
||
|
||
@pytest.fixture
|
||
def conn_two_modules(stub_module, open_db):
|
||
"""База в памяти при двух модулях: строки agent_status и токены на каждый."""
|
||
conn = open_db()
|
||
settings.ensure_defaults(conn)
|
||
return conn
|
||
|
||
|
||
@pytest.fixture
|
||
def hub_server_two_modules(stub_module, tmp_path):
|
||
"""HTTP-хаб на порту 0 при двух модулях: {"hub", "base", "client", "agents": {модуль: HubClient}}."""
|
||
h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
|
||
h.claim_wait = 0
|
||
conn = h.conn()
|
||
tok = settings.tokens(conn)
|
||
conn.close()
|
||
srv = api.make_server(h, "127.0.0.1", 0)
|
||
th = threading.Thread(target=srv.serve_forever, daemon=True)
|
||
th.start()
|
||
base = f"http://127.0.0.1:{srv.server_port}"
|
||
yield {"hub": h, "base": base, "client": HubClient(base, tok["client"]),
|
||
"agents": {name: HubClient(base, token) for name, token in tok["agents"].items()}}
|
||
srv.shutdown()
|
||
srv.server_close()
|
||
th.join(5)
|
||
|
||
|
||
@pytest.fixture
|
||
def hub_server(tmp_path):
|
||
"""HTTP-хаб на порту 0 с одним модулем: {"hub", "base", "client", "agent"} (агент — токеном jira)."""
|
||
h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
|
||
h.claim_wait = 0
|
||
conn = h.conn()
|
||
tok = settings.tokens(conn)
|
||
conn.close()
|
||
srv = api.make_server(h, "127.0.0.1", 0)
|
||
th = threading.Thread(target=srv.serve_forever, daemon=True)
|
||
th.start()
|
||
base = f"http://127.0.0.1:{srv.server_port}"
|
||
yield {"hub": h, "base": base, "client": HubClient(base, tok["client"]),
|
||
"agent": HubClient(base, tok["agents"]["jira"])}
|
||
srv.shutdown()
|
||
srv.server_close()
|
||
th.join(5)
|