conftest_wasm.py

# Extra fixtures used for wasm testing, including some that override the default pytest fixtures.
from functools import partial
import os
import pathlib
from playwright.sync_api import Page
import pytest
import re
import subprocess
import time


# Only include particular test files when testing wasm.
# This can be removed when all tests support wasm.
def pytest_ignore_collect(collection_path: pathlib.Path) -> bool:
    return collection_path.name not in [
        "test_add.py",
        "test_branch.py",
        "test_checkout.py",
        "test_clone.py",
        "test_commit.py",
        "test_config.py",
        "test_diff.py",
        "test_fetch.py",
        "test_fixtures.py",
        "test_git.py",
        "test_init.py",
        "test_log.py",
        "test_merge.py",
        "test_mv.py",
        "test_push.py",
        "test_rebase.py",
        "test_remote.py",
        "test_reset.py",
        "test_revlist.py",
        "test_revparse.py",
        "test_rm.py",
        "test_stash.py",
        "test_status.py",
        "test_tag.py",
    ]


@pytest.fixture(scope="session", autouse=True)
def run_web_server():
    with open("serve.log", "w") as f:
        cwd = pathlib.Path(__file__).parent.parent / "wasm/test"
        proc = subprocess.Popen(["npm", "run", "serve"], stdout=f, stderr=f, cwd=cwd)
        # Wait a bit until server ready to receive connections.
        time.sleep(0.3)
        yield
        proc.terminate()


@pytest.fixture(scope="function", autouse=True)
def load_page(page: Page):
    # Load web page at start of every test.
    page.goto("http://localhost:8000")
    page.locator("#loaded").wait_for()


class MockPath(pathlib.Path):
    def __init__(self, path: str = ""):
        super().__init__(path)

    def exists(self) -> bool:
        p = subprocess.run(["stat", str(self)])
        return p.returncode == 0

    def is_dir(self) -> bool:
        p = subprocess.run(["stat", "-c", "%F", str(self)], capture_output=True, text=True)
        return p.returncode == 0 and p.stdout.strip() == "directory"

    def is_file(self) -> bool:
        p = subprocess.run(["stat", "-c", "%F", str(self)], capture_output=True, text=True)
        return p.returncode == 0 and p.stdout.strip() == "regular file"

    def iterdir(self):
        p = subprocess.run(
            ["ls", str(self), "-a", "-1"], capture_output=True, text=True, check=True
        )
        for f in filter(lambda f: f not in ["", ".", ".."], re.split(r"\r?\n", p.stdout)):
            yield MockPath(self / f)

    def mkdir(self, *, parents=False):
        args = [str(self)]
        if parents:
            args.append("-p")
        subprocess.run(["mkdir"] + args, capture_output=True, text=True, check=True)

    def read_bytes(self) -> bytes:
        raise RuntimeError("Not implemented")

    def read_text(self) -> str:
        p = subprocess.run(["cat", str(self)], capture_output=True, text=True, check=True)
        text = p.stdout
        if text.endswith("\n"):
            text = text[:-1]
        return text

    def write_bytes(self, data: bytes):
        # Convert binary data to a string where each element is backslash-escaped so that we can
        # write to file in cockle using `echo -e <backslash-escaped data>`.
        encoded_string = "".join(map(lambda d: f"\\x{d:02x}", data))
        cmd = ["echo", "-e", encoded_string, ">", str(self)]
        subprocess.run(cmd, capture_output=True, text=True, check=True)
        return len(data)

    def write_text(self, data: str):
        # Note that in general it is not valid to direct output of a subprocess.run call to a file,
        # but we get away with it here as the command arguments are passed straight through to
        # cockle without being checked.
        if data.endswith("\n"):
            data = data[:-1]
        cmd = ["echo", data, ">", str(self)]
        subprocess.run(cmd, capture_output=True, text=True, check=True)
        return len(data)

    def __truediv__(self, other):
        if isinstance(other, str):
            return MockPath(f"{self}/{other}")
        raise RuntimeError("MockPath.__truediv__ only supports strings")


def os_chdir(dir: str):
    subprocess.run(["cd", str(dir)], capture_output=True, check=True, text=True)


def os_getcwd():
    return subprocess.run(["pwd"], capture_output=True, check=True, text=True).stdout.strip()


def os_remove(file: str):
    return subprocess.run(["rm", str(file)], capture_output=True, check=True, text=True)


def os_rename(src: str | MockPath, dst: str | MockPath):
    return subprocess.run(["mv", str(src), str(dst)], capture_output=True, check=True, text=True)


def os_symlink(src: str | MockPath, dst: str | MockPath):
    return subprocess.run(
        ["ln", "-s", str(src), str(dst)], capture_output=True, check=True, text=True
    )


def subprocess_run(
    page: Page,
    cmd: list[str],
    *,
    capture_output: bool = False,
    check: bool = False,
    cwd: str | MockPath | None = None,
    input: str | None = None,
    text: bool | None = None,
) -> subprocess.CompletedProcess:
    shell_run = "async obj => await window.cockle.shellRun(obj.cmd, obj.input)"

    # Set cwd.
    if cwd is not None:
        proc = page.evaluate(shell_run, {"cmd": "pwd"})
        if proc["returncode"] != 0:
            raise RuntimeError("Error getting pwd")
        old_cwd = proc["stdout"].strip()
        if old_cwd == str(cwd):
            # cwd is already correct.
            cwd = None
        else:
            proc = page.evaluate(shell_run, {"cmd": f"cd {cwd}"})
            if proc["returncode"] != 0:
                raise RuntimeError(f"Error setting cwd to {cwd}")

    def maybe_wrap_arg(s: str | MockPath) -> str:
        # An argument containing spaces needs to be wrapped in quotes if it is not already, due
        # to how the command is passed to cockle as a single string.
        # Could do better here.
        s = str(s)
        if " " in s and not s.endswith("'"):
            return "'" + s + "'"
        return s

    shell_run_args = {"cmd": " ".join([maybe_wrap_arg(s) for s in cmd]), "input": input}
    proc = page.evaluate(shell_run, shell_run_args)

    # TypeScript object is auto converted to Python dict.
    # Want to return subprocess.CompletedProcess, consider namedtuple if this fails in future.
    returncode = proc["returncode"]
    stdout = proc["stdout"] if capture_output else ""
    stderr = proc["stderr"] if capture_output else ""
    if not text:
        stdout = stdout.encode("utf-8")
        stderr = stderr.encode("utf-8")

    # Reset cwd.
    if cwd is not None:
        proc = page.evaluate(shell_run, {"cmd": "cd " + old_cwd})
        if proc["returncode"] != 0:
            raise RuntimeError(f"Error setting cwd to {old_cwd}")

    if check and returncode != 0:
        raise subprocess.CalledProcessError(returncode, cmd, stdout, stderr)

    return subprocess.CompletedProcess(
        args=cmd, returncode=returncode, stdout=stdout, stderr=stderr
    )


@pytest.fixture(scope="function")
def tmp_path() -> MockPath:
    # Assumes only one tmp_path needed per test.
    path = MockPath("/drive/tmp0")
    subprocess.run(["mkdir", str(path)], check=True)
    assert path.exists()
    assert path.is_dir()
    return path


@pytest.fixture(scope="function", autouse=True)
def mock_subprocess_run(page: Page, monkeypatch):
    monkeypatch.setattr(subprocess, "run", partial(subprocess_run, page))
    monkeypatch.setattr(os, "chdir", os_chdir)
    monkeypatch.setattr(os, "getcwd", os_getcwd)
    monkeypatch.setattr(os, "remove", os_remove)
    monkeypatch.setattr(os, "rename", os_rename)
    monkeypatch.setattr(os, "symlink", os_symlink)