|
| 1 | +# Code shared by wasm testing in cockle (conftest_cockle.py) and JupyterLite (conftest_lite.py). |
| 2 | +import pathlib |
| 3 | +import re |
| 4 | +import subprocess |
| 5 | + |
| 6 | +import pytest |
| 7 | + |
| 8 | + |
| 9 | +class MockPath(pathlib.Path): |
| 10 | + def __init__(self, path: str = ""): |
| 11 | + super().__init__(path) |
| 12 | + |
| 13 | + def exists(self) -> bool: |
| 14 | + p = subprocess.run(["stat", str(self)]) |
| 15 | + return p.returncode == 0 |
| 16 | + |
| 17 | + def is_dir(self) -> bool: |
| 18 | + p = subprocess.run(["stat", "-c", "%F", str(self)], capture_output=True, text=True) |
| 19 | + return p.returncode == 0 and p.stdout.strip() == "directory" |
| 20 | + |
| 21 | + def is_file(self) -> bool: |
| 22 | + p = subprocess.run(["stat", "-c", "%F", str(self)], capture_output=True, text=True) |
| 23 | + return p.returncode == 0 and p.stdout.strip() == "regular file" |
| 24 | + |
| 25 | + def iterdir(self): |
| 26 | + p = subprocess.run( |
| 27 | + ["ls", str(self), "-a", "-1"], capture_output=True, text=True, check=True |
| 28 | + ) |
| 29 | + for f in filter(lambda f: f not in ["", ".", ".."], re.split(r"\r?\n", p.stdout)): |
| 30 | + yield MockPath(self / f) |
| 31 | + |
| 32 | + def mkdir(self, *, parents=False): |
| 33 | + args = [str(self)] |
| 34 | + if parents: |
| 35 | + args.append("-p") |
| 36 | + subprocess.run(["mkdir"] + args, capture_output=True, text=True, check=True) |
| 37 | + |
| 38 | + def read_bytes(self) -> bytes: |
| 39 | + raise RuntimeError("Not implemented") |
| 40 | + |
| 41 | + def read_text(self) -> str: |
| 42 | + p = subprocess.run(["cat", str(self)], capture_output=True, text=True, check=True) |
| 43 | + text = p.stdout |
| 44 | + text = text.removesuffix("\n") |
| 45 | + return text |
| 46 | + |
| 47 | + def write_bytes(self, data: bytes): |
| 48 | + # Convert binary data to a string where each element is backslash-escaped so that we can |
| 49 | + # write to file in cockle using `echo -e <backslash-escaped data>`. |
| 50 | + encoded_string = "".join(f"\\x{d:02x}" for d in data) |
| 51 | + cmd = ["echo", "-e", encoded_string, ">", str(self)] |
| 52 | + subprocess.run(cmd, capture_output=True, text=True, check=True) |
| 53 | + return len(data) |
| 54 | + |
| 55 | + def write_text(self, data: str): |
| 56 | + # Note that in general it is not valid to direct output of a subprocess.run call to a file, |
| 57 | + # but we get away with it here as the command arguments are passed straight through to |
| 58 | + # cockle without being checked. |
| 59 | + data = data.removesuffix("\n") |
| 60 | + cmd = ["echo", data, ">", str(self)] |
| 61 | + subprocess.run(cmd, capture_output=True, text=True, check=True) |
| 62 | + return len(data) |
| 63 | + |
| 64 | + def __truediv__(self, other): |
| 65 | + if isinstance(other, str): |
| 66 | + return MockPath(f"{self}/{other}") |
| 67 | + raise RuntimeError("MockPath.__truediv__ only supports strings") |
| 68 | + |
| 69 | + |
| 70 | +@pytest.fixture(scope="function") |
| 71 | +def tmp_path() -> MockPath: |
| 72 | + # Assumes only one tmp_path needed per test. |
| 73 | + path = MockPath("/drive/tmp0") |
| 74 | + subprocess.run(["mkdir", str(path)], check=True) |
| 75 | + assert path.exists() |
| 76 | + assert path.is_dir() |
| 77 | + return path |
| 78 | + |
| 79 | + |
| 80 | +def maybe_wrap_arg(s: str | MockPath) -> str: |
| 81 | + # An argument containing whitespace (including newlines) needs to be wrapped in quotes if |
| 82 | + # it is not already, due to how the command is passed to the shell as a single string. |
| 83 | + # Could do better here. |
| 84 | + s = str(s) |
| 85 | + if re.search(r"\s", s) and not s.endswith("'"): |
| 86 | + return "'" + s + "'" |
| 87 | + return s |
| 88 | + |
| 89 | + |
| 90 | +def os_chdir(dir: str): |
| 91 | + subprocess.run(["cd", str(dir)], capture_output=True, check=True, text=True) |
| 92 | + |
| 93 | + |
| 94 | +def os_getcwd(): |
| 95 | + return subprocess.run(["pwd"], capture_output=True, check=True, text=True).stdout.strip() |
| 96 | + |
| 97 | + |
| 98 | +def os_remove(file: str): |
| 99 | + return subprocess.run(["rm", str(file)], capture_output=True, check=True, text=True) |
| 100 | + |
| 101 | + |
| 102 | +def os_rename(src: str | MockPath, dst: str | MockPath): |
| 103 | + return subprocess.run(["mv", str(src), str(dst)], capture_output=True, check=True, text=True) |
| 104 | + |
| 105 | + |
| 106 | +def os_symlink(src: str | MockPath, dst: str | MockPath): |
| 107 | + return subprocess.run( |
| 108 | + ["ln", "-s", str(src), str(dst)], capture_output=True, check=True, text=True |
| 109 | + ) |
0 commit comments