Version of bisect for if an E2E test passes as long as at least one of multiple affected files is not compiled

bisect_for_multiple_files.py

import asyncio
import json
import pathlib
import random
import signal
import typing

"""

vite.config.ts has these lines:

function djb2Hash(str: string): number {
  let hash = 5381
  for (let i = 0; i < str.length; i++) {
    hash = (hash * 33) ^ str.charCodeAt(i)
  }
  return hash >>> 0 // Ensure positive integer
}

const NUMERATOR = 41
const DENOMINATOR = 256

"""


def write_constants(numerator: int, denominator: int) -> None:
    with open("vite_config/vite.config.ts") as f:
        lines = f.readlines()

    for i, line in enumerate(lines):
        if "const NUMERATOR" in line:
            lines[i] = f"const NUMERATOR = {numerator}\n"
        elif "const DENOMINATOR" in line:
            lines[i] = f"const DENOMINATOR = {denominator}\n"

    with open("vite_config/vite.config.ts", "w") as f:
        f.writelines(lines)


SUCCESS_MESSAGE = "✓ Table cells can be edited for every type"
FAILURE_MESSAGE = "Command failed with exit code"


async def forward_stream(
    input_stream: asyncio.StreamReader,
    output_stream: typing.IO[str],
) -> None:
    """Forward a subprocess stream to one of our output streams."""
    while not input_stream.at_eof():
        line_bytes = await input_stream.readline()
        line = line_bytes.decode()
        output_stream.write(line)
        output_stream.flush()


async def forward_and_parse_stream(
    input_stream: asyncio.StreamReader,
    output_stream: typing.IO[str],
) -> bool:
    """Process stderr and look for signs of success or failure.

    Return True if we see a line indicating successful startup.
    Return False if we see an error message.

    """
    success: typing.Optional[bool] = None

    while True:
        line_bytes = await input_stream.readline()
        line = line_bytes.decode()

        # Forward each line of the process's stdout to our stdout
        output_stream.write(line)
        output_stream.flush()

        # Look for messages which indicate success or failure
        if success is None:
            if SUCCESS_MESSAGE in line:
                success = True
            elif FAILURE_MESSAGE in line:
                success = False

        if success is not None:
            return success


async def run_build(
    stdout: typing.IO[str],
    stderr: typing.IO[str],
) -> bool:
    process = await asyncio.create_subprocess_exec(
        "/opt/homebrew/bin/pnpm",
        "cy:run",
        "--spec",
        "cypress/integration/components/table/tableCellsEditable.ts",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    assert process.stdout is not None
    assert process.stderr is not None

    success = await forward_and_parse_stream(process.stdout, stdout)
    stderr_forward = asyncio.create_task(forward_stream(process.stderr, stderr))

    process.kill()
    await process.wait()
    await stderr_forward
    return success


History = typing.List[typing.Tuple[typing.Set[str], bool]]


def calculate_suspects(files: typing.Set[str], history: History) -> typing.Set[str]:
    suspects: typing.Set[str] = set()
    for day_suspects, day_success in history:
        if not day_success:
            suspects = suspects.union(day_suspects)

    potential_suspects = files.difference(suspects)
    num_last_suspects = 0 if len(history) == 0 else len(history[-1][0])
    while len(suspects) - num_last_suspects < len(potential_suspects) // 10:
        new_suspect = random.choice(list(potential_suspects))
        potential_suspects.remove(new_suspect)
        suspects.add(new_suspect)
    return suspects


async def bisect():
    with open("files.txt") as f:
        files = set((line.rstrip() for line in f.readlines()))

    history: History = []
    with open("stdout.txt", "w") as stdout:
        with open("stderr.txt", "w") as stderr:
            while True:
                suspects = calculate_suspects(files, history)
                with open("suspects.txt", "w") as f:
                    f.write("\n".join(list(suspects)))

                pathlib.Path.touch("vite_config/vite.config.ts")

                print(f"jeff running: {len(suspects)} suspects...")
                success = await run_build(stdout, stderr)

                status_emoji = "✅" if success else "❌"
                print(f"jeff result: ({len(suspects)} suspects): {status_emoji}")

                history.append((suspects, success))

                json.dump(
                    [
                        {"success": day_success, "files": list(day_files)}
                        for (day_files, day_success) in history
                    ],
                    open("history.json", "w"),
                )

                if len(suspects) > 3000:
                    break


if __name__ == "__main__":
    asyncio.run(bisect())