sdlc-hub/tests/test_relay.py

import http.client
import json
import os
import shutil
import threading
import time
from datetime import datetime, timedelta, timezone

import pytest

from sdlc_hub.core import api, relay, settings
from sdlc_hub.core.client import HubClient


@pytest.fixture
def env(tmp_path):
    h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
    h.claim_wait = 0
    conn = h.conn()
    root = tmp_path / "bridge"
    root.mkdir()
    settings.set_(conn, "relay_dir", str(root))
    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()
    r = relay.Relay(h, srv.server_port)
    client = HubClient(f"http://127.0.0.1:{srv.server_port}", tok["client"])
    yield {"hub": h, "root": root, "relay": r, "agent": "Bearer " + tok["agents"]["jira"], "client": client,
           "port": srv.server_port}
    srv.shutdown()
    srv.server_close()
    th.join(5)


def deadline(seconds):
    return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat()


def put_request(root, rid, method, path, auth, body=None, seconds=60, **extra):
    req = {"id": rid, "method": method, "path": path, "body": body, "authorization": auth,
           "deadline": deadline(seconds), **extra}
    relay.write_atomic(root / "req", f"{rid}.json", req)


def response(root, rid):
    p = root / "resp" / f"{rid}.json"
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None


def run_once(r):
    r.step()
    r.join()


def test_request_roundtrip(env):
    r, root = env["relay"], env["root"]
    r.step()                                            # подготовка: req/, resp/, work/
    put_request(root, "a1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    resp = response(root, "a1")
    assert resp["status"] == 200 and resp["id"] == "a1"
    assert json.loads(resp["body"])["project_key"] == "AURORA"
    assert not list((root / "work").iterdir()) and not list((root / "req").iterdir())


def test_status_passes_through(env):
    r, root = env["relay"], env["root"]
    r.step()
    put_request(root, "u1", "GET", "/issues", env["agent"])             # агентский токен на клиентский маршрут
    put_request(root, "u2", "GET", "/agent/commands/999", env["agent"])
    run_once(r)
    assert response(root, "u1")["status"] == 401
    assert response(root, "u2")["status"] == 404


def test_expired_and_short_claim_dropped(env):
    r, root, c = env["relay"], env["root"], env["client"]
    b = c.post("/batches", {"title": "t", "author": "test",
                            "commands": [{"type": "ensure_fields", "params": {"key": "A-1", "duedate": "2026-10-23"}}]})
    r.step()
    put_request(root, "c1", "POST", "/agent/claim", env["agent"], body={}, seconds=20)
    put_request(root, "s1", "GET", "/agent/settings", env["agent"], seconds=-1)
    run_once(r)
    assert response(root, "c1") is None and response(root, "s1") is None
    statuses = [x["status"] for x in c.get(f"/batches/{b['id']}")["commands"]]
    assert statuses[0] == "queued"
    assert not list((root / "work").iterdir())


def test_claim_with_enough_time_is_forwarded(env):
    r, root = env["relay"], env["root"]
    r.step()
    put_request(root, "c2", "POST", "/agent/claim", env["agent"], body={"agent_version": 2}, seconds=50)
    run_once(r)
    assert response(root, "c2")["status"] == 200


def test_long_claim_does_not_block_other_requests(env):
    env["hub"].claim_wait = 3
    r, root = env["relay"], env["root"]
    r.step()
    put_request(root, "c3", "POST", "/agent/claim", env["agent"], body={}, seconds=60)
    r.step()
    put_request(root, "s3", "GET", "/agent/settings", env["agent"])
    t0 = time.monotonic()
    while response(root, "s3") is None and time.monotonic() - t0 < 2.5:
        r.step()
        time.sleep(0.05)
    assert response(root, "s3")["status"] == 200
    assert response(root, "c3") is None                 # claim ещё ждёт
    r.join()
    assert response(root, "c3")["status"] == 200


def test_unreachable_hub_gives_502(env):
    r = relay.Relay(env["hub"], 1)                      # на порту 1 никто не слушает
    root = env["root"]
    r.step()
    put_request(root, "p1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    resp = response(root, "p1")
    assert resp["status"] == 502 and "хаб недоступен" in resp["body"]


def test_dotfiles_ignored_and_bad_json_dropped(env, monkeypatch, capsys):
    monkeypatch.setattr(relay, "BAD_JSON_WAIT", 0.2)
    r, root = env["relay"], env["root"]
    r.step()
    (root / "req" / ".x1.json.tmp").write_text("{}", encoding="utf-8")
    (root / "req" / "bad.json").write_text("{не json", encoding="utf-8")
    run_once(r)
    assert (root / "req" / ".x1.json.tmp").exists()
    assert response(root, "bad") is None and not list((root / "work").iterdir())
    assert "нечитаемый запрос bad.json" in capsys.readouterr().out


def test_bad_request_fields_get_400(env):
    r, root = env["relay"], env["root"]
    r.step()
    put_request(root, "b1", "GET", "agent/settings", env["agent"])                 # путь без /
    put_request(root, "b2", "DELETE", "/agent/settings", env["agent"])             # метод не из списка
    put_request(root, "b3", "GET", "/agent/settings", "Bearer x\r\nX-Evil: 1")     # перевод строки в заголовке
    run_once(r)
    for rid in ("b1", "b2", "b3"):
        assert response(root, rid)["status"] == 400, rid


def test_missing_deadline_is_dropped(env):
    r, root = env["relay"], env["root"]
    r.step()
    relay.write_atomic(root / "req", "d1.json", {"id": "d1", "method": "GET", "path": "/agent/settings",
                                                 "authorization": env["agent"]})
    relay.write_atomic(root / "req", "d2.json", {"id": "d2", "method": "GET", "path": "/agent/settings",
                                                 "authorization": env["agent"], "deadline": "завтра"})
    run_once(r)
    assert response(root, "d1") is None and response(root, "d2") is None


def test_vanished_request_is_skipped(env, capsys):
    r, root = env["relay"], env["root"]
    r.step()
    put_request(root, "v1", "GET", "/agent/settings", env["agent"])
    orig = os.rename

    def vanish(src, dst):
        os.remove(src)                                  # агент удалил свой запрос раньше реле
        return orig(src, dst)

    os.rename = vanish
    try:
        run_once(r)
    finally:
        os.rename = orig
    assert response(root, "v1") is None
    assert "Traceback" not in capsys.readouterr().err


def test_work_cleared_on_prepare(env, capsys):
    r, root = env["relay"], env["root"]
    (root / "work").mkdir()
    (root / "work" / "old.json").write_text("{}", encoding="utf-8")
    r.step()
    assert not (root / "work" / "old.json").exists()
    assert "old.json" in capsys.readouterr().out


def test_missing_dir_warns_once_then_works(env, capsys):
    r, root = env["relay"], env["root"]
    gone = root.parent / "bridge-later"
    conn = env["hub"].conn()
    settings.set_(conn, "relay_dir", str(gone))
    conn.close()
    for _ in range(5):
        r.step()
    out = capsys.readouterr().out
    assert out.count("нет — жду") == 1
    gone.mkdir()
    r.step()
    assert (gone / "req").is_dir() and (gone / "resp").is_dir() and (gone / "work").is_dir()
    put_request(gone, "m1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    assert response(gone, "m1")["status"] == 200


def test_disabled_and_reenabled_without_restart(env, monkeypatch):
    r, root = env["relay"], env["root"]
    conn = env["hub"].conn()
    settings.set_(conn, "relay_dir", None)
    conn.close()
    r.step()
    assert r.root is None and not (root / "req").exists()
    monkeypatch.setattr(relay, "SETTINGS_EVERY", 0)
    conn = env["hub"].conn()
    settings.set_(conn, "relay_dir", str(root))
    conn.close()
    r.step()
    assert r.root == root and (root / "req").is_dir()


def test_cleanup_by_age(env):
    r, root = env["relay"], env["root"]
    r.step()
    old, fresh = root / "resp" / "old.json", root / "resp" / "fresh.json"
    tmp_old = root / "resp" / ".t.json.tmp"
    for p in (old, fresh, tmp_old):
        p.write_text("{}", encoding="utf-8")
    now = time.time()
    os.utime(old, (now - 700, now - 700))
    os.utime(tmp_old, (now - 700, now - 700))
    stale_req = root / "req" / ".q.json.tmp"
    stale_req.write_text("{}", encoding="utf-8")
    os.utime(stale_req, (now - 700, now - 700))
    r.cleanup(now)
    assert not old.exists() and not tmp_old.exists() and not stale_req.exists()
    assert fresh.exists()


def test_log_line(env, capsys):
    r, root = env["relay"], env["root"]
    r.step()
    put_request(root, "l1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    assert "relay GET /agent/settings 200" in capsys.readouterr().out


def test_serve_runs_relay(tmp_path):
    h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
    root = tmp_path / "bridge"
    root.mkdir()
    conn = h.conn()
    settings.set_(conn, "relay_dir", str(root))
    auth = "Bearer " + settings.tokens(conn)["agents"]["jira"]
    conn.close()
    srv, stop = api.start(h, "127.0.0.1", 0)              # реле стучится на тот же хост, что слушает HTTP
    th = threading.Thread(target=srv.serve_forever, daemon=True)
    th.start()
    try:
        t0 = time.monotonic()
        while not (root / "req").is_dir() and time.monotonic() - t0 < 3:
            time.sleep(0.05)
        put_request(root, "e1", "GET", "/agent/settings", auth)
        while response(root, "e1") is None and time.monotonic() - t0 < 5:
            time.sleep(0.05)
        assert response(root, "e1")["status"] == 200
    finally:
        stop.set()
        srv.shutdown()
        srv.server_close()
        th.join(5)


def test_permission_denied_warns_once_then_works(env, capsys):
    r, root = env["relay"], env["root"]
    locked = root.parent / "locked"
    target = locked / "bridge"
    target.mkdir(parents=True)
    conn = env["hub"].conn()
    settings.set_(conn, "relay_dir", str(target))
    conn.close()
    locked.chmod(0)                                     # так выглядит отказ доступа (TCC) у процесса launchd
    try:
        for _ in range(5):
            r.step()
    finally:
        locked.chmod(0o755)
    out, err = capsys.readouterr()
    assert out.count("нет доступа к папке") == 1
    assert "Traceback" not in err
    r.step()
    put_request(target, "p1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    assert response(target, "p1")["status"] == 200


def test_removed_req_dir_recovers(env, capsys):
    r, root = env["relay"], env["root"]
    r.step()
    (root / "req").rmdir()                              # папку моста почистили при живом реле
    for _ in range(3):
        r.step()
    assert (root / "req").is_dir()
    assert "Traceback" not in capsys.readouterr().err
    put_request(root, "q1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    assert response(root, "q1")["status"] == 200


def test_removed_work_dir_recovers(env, capsys):
    r, root = env["relay"], env["root"]
    r.step()
    (root / "work").rmdir()
    put_request(root, "w1", "GET", "/agent/settings", env["agent"])
    for _ in range(3):
        r.step()
    r.join()
    assert response(root, "w1")["status"] == 200
    assert "нет доступа к папке" in capsys.readouterr().out


def test_removed_resp_dir_recreated_before_forward(env):
    r, root = env["relay"], env["root"]
    r.step()
    (root / "resp").rmdir()
    put_request(root, "r1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    assert response(root, "r1")["status"] == 200


def test_protocol_error_gives_502(env, monkeypatch):
    r, root = env["relay"], env["root"]
    r.step()

    def broken(*a, **k):
        raise http.client.BadStatusLine("мусор")
    monkeypatch.setattr("urllib.request.urlopen", broken)
    put_request(root, "b1", "GET", "/agent/settings", env["agent"])
    run_once(r)
    resp = response(root, "b1")
    assert resp["status"] == 502 and "хаб недоступен" in resp["body"]


def test_reprepare_keeps_work(env, capsys):
    r, root = env["relay"], env["root"]
    r.step()                                            # первая подготовка
    (root / "work" / "busy.json").write_text("{}", encoding="utf-8")
    (root / "req").rmdir()                              # пропал подкаталог — реле подготовит папку заново
    for _ in range(3):
        r.step()
    assert (root / "req").is_dir() and (root / "work" / "busy.json").exists()
    out = capsys.readouterr().out
    assert "удалён незавершённый запрос" not in out and "подготовлена заново" in out


def test_vanished_tree_keeps_work(env, capsys):
    r, root = env["relay"], env["root"]
    r.step()
    (root / "work" / "busy.json").write_text("{}", encoding="utf-8")
    capsys.readouterr()
    shutil.rmtree(root)                                 # папка пропала целиком и появилась снова
    root.mkdir()
    (root / "work").mkdir()
    (root / "work" / "busy.json").write_text("{}", encoding="utf-8")   # взятый запрос, который ещё обрабатывается
    for _ in range(3):
        r.step()
    assert (root / "work" / "busy.json").exists() and (root / "req").is_dir()
    out = capsys.readouterr().out
    assert "удалён незавершённый запрос" not in out and "подготовлена заново" in out    # minor-3 #16


def test_each_dir_cleaned_once(env, tmp_path, capsys):
    r, root, h = env["relay"], env["root"], env["hub"]
    other = tmp_path / "other"
    (other / "work").mkdir(parents=True)
    (other / "work" / "old.json").write_text("{}", encoding="utf-8")
    r.step()                                            # root вычищен
    conn = h.conn()
    settings.set_(conn, "relay_dir", str(other))
    r._settings_at = None
    r.step()
    assert not (other / "work" / "old.json").exists()   # новая папка чистится
    settings.set_(conn, "relay_dir", str(root))
    r._settings_at = None
    (root / "work" / "busy.json").write_text("{}", encoding="utf-8")
    r.step()
    conn.close()
    assert (root / "work" / "busy.json").exists()       # возврат A->B->A: A повторно не чистится


def test_relay_targets_serve_host(tmp_path, monkeypatch):
    seen = []

    class Spy(relay.Relay):
        def __init__(self, hub, port, host="127.0.0.1", poll=0.1):
            seen.append(host)
            super().__init__(hub, port, host, poll)

    monkeypatch.setattr(relay, "Relay", Spy)
    h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
    for host in ("0.0.0.0", "localhost"):
        srv, stop = api.start(h, host, 0)
        stop.set()
        srv.server_close()
    assert seen == ["127.0.0.1", "localhost"]          # конкретный адрес serve — туда же ходит реле (M-1 #8)


def test_symlink_to_same_dir_not_cleaned_again(env, tmp_path, capsys):
    """minor-1 #16, решение 12: relay_dir сменили на симлинк к той же папке — work/ не чистится второй раз."""
    r, root, h = env["relay"], env["root"], env["hub"]
    r.step()                                            # root вычищен
    link = tmp_path / "bridge-link"
    link.symlink_to(root, target_is_directory=True)
    (root / "work" / "busy.json").write_text("{}", encoding="utf-8")   # взятый запрос ещё обрабатывается
    conn = h.conn()
    settings.set_(conn, "relay_dir", str(link))
    conn.close()
    r._settings_at = None
    capsys.readouterr()
    r.step()
    out = capsys.readouterr().out
    assert (root / "work" / "busy.json").exists()
    assert "удалён незавершённый запрос" not in out and "подготовлена заново" in out