Harbor/tests/helpers.py

"""Shared test fixtures: temporary projects under R:\\Workspace\\Data\\openseas-chat\\test-tmp (never C:)."""
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import unittest
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from harbor import projects  # noqa: E402
from harbor.core import Harbor  # noqa: E402

TMP_ROOT = Path(os.environ.get("HARBOR_TEST_TMP", r"R:\Workspace\Data\openseas-chat\test-tmp"))
LEGACY_LINE = re.compile(r'^\[([^\]]+)\]\s+([A-Za-z]+)\s*(?:→|->)\s*([A-Za-z]+):\s?(.*)$')
ROSTER = [{"name": n, "role": r} for n, r in (("Claude", "Integration"), ("Codex", "Decompilation"),
                                              ("DeepSeek", "Decompilation"), ("OpenCode", "Decompilation"),
                                              ("Kimi", "Review"))] + [{"name": "User", "kind": "human"}]
PROJECT_TOML = """name = "{name}"
title = "Test {name}"
database = '{db}'
mirror = '{mirror}'
coordinators = ["Claude"]
reviewers = ["Claude"]

[identities]
"worktree-deepseek" = "DeepSeek"

[attention]
ack_overdue = 30
item_stale = 120
agent_quiet = 60
""" + "".join(f'\n[[agents]]\nname = "{a["name"]}"\n' + (f'kind = "{a["kind"]}"\n' if a.get("kind") else "")
              for a in ROSTER)


class Clock:
    def __init__(self, t=1_790_000_000.0):
        self.t = t

    def __call__(self):
        return self.t


def temp_dir():
    TMP_ROOT.mkdir(parents=True, exist_ok=True)
    fence = TMP_ROOT / ".harbor.toml"                   # tests never join a project marked above this folder
    if not fence.exists():
        fence.write_text("# Test folders belong to no Harbor project above this one.\nboundary = true\n", encoding="utf-8")
    return Path(tempfile.mkdtemp(dir=TMP_ROOT))


def make_home(folder, names=("test",)):
    """A Harbor home with registered projects; returns {name: Project}."""
    out = {}
    for name in names:
        pdir = folder / "projects"
        pdir.mkdir(parents=True, exist_ok=True)
        mirror = folder / name / "agents-chat.md"
        mirror.parent.mkdir(parents=True, exist_ok=True)
        mirror.write_text("# chat\n\n---\n", encoding="utf-8")
        (pdir / f"{name}.toml").write_text(PROJECT_TOML.format(name=name, db=folder / name / "harbor.db",
                                                               mirror=mirror), encoding="utf-8")
        out[name] = projects.load_named(name, home=folder)
    (folder / "config.toml").write_text(f'default_project = "{names[0]}"\n', encoding="utf-8")
    return out


class HarborCase(unittest.TestCase):
    def setUp(self):
        self.dir = temp_dir()
        self.project = make_home(self.dir)["test"]
        self.mirror = self.project.mirror
        self.clock = Clock()
        self.h = Harbor(self.project, clock=self.clock)

    def tearDown(self):
        self.h.close()
        shutil.rmtree(self.dir, ignore_errors=True)


def env_for(home, **extra):
    env = {k: v for k, v in os.environ.items() if not k.startswith("HARBOR_")}
    env.update(HARBOR_HOME=str(home), PYTHONDONTWRITEBYTECODE="1", HARBOR_NO_TOASTS="1", **extra)
    return env


def run_cli(home, *args, cwd=None, stdin=None, **env):
    return subprocess.run([sys.executable, "-B", "-m", "harbor", *args], cwd=cwd or home, env=env_for(home, **env),
                          input=stdin, capture_output=True, text=True, encoding="utf-8", timeout=60,
                          **({} if cwd else {}))


def run_module(home, *args, cwd=None, stdin=None, **env):
    environment = env_for(home, **env)
    environment["PYTHONPATH"] = str(ROOT)
    return subprocess.run([sys.executable, "-B", "-m", "harbor", *args], cwd=cwd or home, env=environment,
                          input=stdin, capture_output=True, text=True, encoding="utf-8", timeout=60)


def http(port, path, body=None, headers=None, browser=True, token=None):
    h = {}
    if body is not None:
        h["Content-Type"] = "application/json"
        if browser:
            h.update({"Sec-Fetch-Site": "same-origin", "X-Harbor-Token": token or "",
                      "Origin": f"http://127.0.0.1:{port}"})
    h.update(headers or {})
    data = json.dumps(body).encode("utf-8") if body is not None else None
    request = urllib.request.Request(f"http://127.0.0.1:{port}{path}", data=data, headers=h)
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            raw = response.read()
            kind = response.headers.get("Content-Type", "")
            return response.status, (json.loads(raw or b"null") if "json" in kind else raw.decode("utf-8", "replace"))
    except urllib.error.HTTPError as error:
        return error.code, json.loads(error.read() or b"null")


def wait_for_port(port, seconds=15):
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        try:
            urllib.request.urlopen(f"http://127.0.0.1:{port}/participants", timeout=1).read()
            return True
        except OSError:
            time.sleep(0.1)
    return False