anteater/main.py

import os
import subprocess
import threading
from shutil import rmtree
from time import sleep
from typing import Optional

import click
from rich import print as rprint
from rich.live import Live
from rich.panel import Panel

from runbot import _commits_from_batch, _get_batches, Repo, bundles
from git import GitRepository

worktrees = {
    "17.0": "/home/hubert/src/17.0-proj",
    "18.0": "/home/hubert/src/18.0-proj",
}


def _run_odoo(worktree_root, test_tags: str, init: str) -> bool:
    venv_path = f"{worktree_root}/venv-bisect"
    rmtree(venv_path, ignore_errors=True)
    subprocess.check_call(
        ["uv", "venv", "-p", "3.12", venv_path],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    env = os.environ.copy()
    env["VIRTUAL_ENV"] = venv_path
    env["PYTHONUNBUFFERED"] = "1"
    subprocess.check_call(
        ["uv", "pip", "install", "-r", f"{worktree_root}/odoo/requirements.txt"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=env,
    )
    subprocess.check_call(
        ["uv", "pip", "install", "websocket-client", "mock", "dbfread"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=env,
    )
    process = subprocess.Popen(
        ["odoo", "--no-patch", "--drop", "-i", init, "--test-tags", test_tags],
        cwd=worktree_root,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    )

    limit = 10

    output_lines = []

    def read_stream(stream, container):
        for line in stream:
            container.append(line.rstrip())

    stdout_thread = threading.Thread(
        target=read_stream, args=(process.stdout, output_lines)
    )
    stdout_thread.start()

    with Live(refresh_per_second=10) as live:
        while process.poll() is None:
            display_lines = output_lines[-limit:]
            panel = Panel(
                "\n".join(display_lines), title="Command Output", border_style="green"
            )
            live.update(panel)
            sleep(0.1)

        stdout_thread.join()

        sleep(0.2)

        if process.returncode == 0:
            display_lines = output_lines[-limit:]
            extra = ""
            if len(output_lines) > limit:
                extra = f"\n[dim]... (showing last {limit} of {len(output_lines)} lines)[/dim]"
            panel = Panel(
                "\n".join(display_lines) + extra, title="Success", border_style="green"
            )
        else:
            full_output = "\n".join(output_lines)
            panel = Panel(
                f"[bold red]Command failed (exit {process.returncode})[/bold red]\n\n{full_output}",
                title="Failure",
                border_style="red",
            )
        live.update(panel, refresh=True)

    return process.returncode == 0


def _find_first_failing_commit(batches: list[int], test_failed):
    low = 0
    high = len(batches)

    while low < high:
        mid = (low + high) // 2
        if test_failed(batches[mid]):
            high = mid
        else:
            low = mid + 1

    if low < len(batches):
        return batches[low]
    raise ValueError("404")


def _bisect(version: str, test_tags: str, init: str) -> Optional[dict[Repo, str]]:
    bundle = bundles[version]
    worktree_root = worktrees[version]
    odoo_repository = GitRepository(f"{worktree_root}/odoo")
    enterprise_repository = GitRepository(f"{worktree_root}/enterprise")
    batches = list(reversed(_get_batches(bundle)))

    def test_failed(batch: int):
        commits = _commits_from_batch(batch)
        odoo_repository.checkout(commits["odoo"])
        enterprise_repository.checkout(commits["enterprise"])
        return not _run_odoo(worktree_root, test_tags, init)

    suspect = _find_first_failing_commit(batches, test_failed)
    if suspect:

        def print_batch(name, batch, commits):
            rprint(f"[green]{name}")
            rprint(commits)

        idx = batches.index(suspect)
        previous_batch = batches[idx - 1]
        commits = _commits_from_batch(suspect)
        previous_commits = _commits_from_batch(previous_batch)
        print_batch("suspect", suspect, commits)
        print_batch("previous", previous_batch, commits)

        rprint("[blue]odoo")
        rprint(odoo_repository.between(previous_commits["odoo"], commits["odoo"]))

        rprint("[blue]enterprise")
        rprint(
            enterprise_repository.between(
                previous_commits["enterprise"], commits["enterprise"]
            )
        )

        return None  # TODO

    return None


@click.command()
@click.option("--version", default="17.0", type=click.Choice(bundles.keys()))
@click.option("--test-tags")
@click.option("-i", "--init")
def main(version, test_tags, init):
    _bisect(version, test_tags, init)


if __name__ == "__main__":
    main()