feat: add mock Stata runtime for CI tests

.github/workflows/main.yml

@@ -66,15 +66,15 @@ jobs:
 
       - name: Run tests
         shell: bash -l {0}
-        run: just test
+        run: just test-ci
 
       - name: Upload test coverage reports to Codecov with GitHub Action
         uses: codecov/codecov-action@e79a6962e0d4c0c17b229090214935d2e33f8354 # v6.0.1
 
       - name: Run tests with lowest resolution
         if: matrix.python-version == '3.10' && matrix.os == 'ubuntu-latest'
-        run: just test-lowest
+        run: just test-lowest-ci
 
       - name: Run tests with highest resolution
         if: matrix.python-version == '3.14' && matrix.os == 'ubuntu-latest'
-        run: just test-highest
+        run: just test-highest-ci

justfile

@@ -6,6 +6,10 @@ install:
 test:
     uv run --group test pytest --cov=src --cov=tests --cov-report=xml
 
+# Run tests in CI with the mock Stata executable installed.
+test-ci:
+    uv run --group test --group test-mock-stata pytest --cov=src --cov=tests --cov-report=xml
+
 # Run type checking
 typing:
     uv run --group typing --group test --isolated ty check
@@ -21,6 +25,14 @@ check: lint typing test
 test-lowest:
     uv run --group test --resolution lowest-direct pytest
 
+# Run tests with lowest dependency resolution in CI with the mock Stata executable installed.
+test-lowest-ci:
+    uv run --group test --group test-mock-stata --resolution lowest-direct pytest
+
 # Run tests with highest dependency resolution
 test-highest:
     uv run --group test --resolution highest pytest
+
+# Run tests with highest dependency resolution in CI with the mock Stata executable installed.
+test-highest-ci:
+    uv run --group test --group test-mock-stata --resolution highest pytest

packages/stata_mock/pyproject.toml

@@ -0,0 +1,29 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "stata-mock"
+version = "0.0.0"
+description = "Mock Stata executable for pytask-stata tests."
+requires-python = ">=3.10"
+
+[project.scripts]
+Stata = "stata_mock.cli:main"
+Stata-64 = "stata_mock.cli:main"
+Stata-ia = "stata_mock.cli:main"
+Stata64 = "stata_mock.cli:main"
+Stata64MP = "stata_mock.cli:main"
+Stata64SE = "stata_mock.cli:main"
+StataMP = "stata_mock.cli:main"
+StataMP-64 = "stata_mock.cli:main"
+StataMP-ia = "stata_mock.cli:main"
+StataSE = "stata_mock.cli:main"
+StataSE-64 = "stata_mock.cli:main"
+StataSE-ia = "stata_mock.cli:main"
+WMPSTATA = "stata_mock.cli:main"
+WSESTATA = "stata_mock.cli:main"
+WSTATA = "stata_mock.cli:main"
+stata = "stata_mock.cli:main"
+stata-mp = "stata_mock.cli:main"
+stata-se = "stata_mock.cli:main"

packages/stata_mock/src/stata_mock/__init__.py

@@ -0,0 +1 @@
+"""Mock Stata executable for tests."""

packages/stata_mock/src/stata_mock/cli.py

@@ -0,0 +1,107 @@
+"""Command line interface for the mock Stata executable."""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+INVALID_SYNTAX = 198
+UNKNOWN_COMMAND = 199
+MINIMUM_ARGUMENTS = 4
+
+
+def main() -> int:
+    """Run a tiny subset of Stata syntax for pytask-stata tests."""
+    parsed = _parse_invocation(sys.argv[1:])
+    if parsed is None:
+        return INVALID_SYNTAX
+
+    script, options, log = parsed
+    lines, error_code = _run_script(script, options)
+
+    lines.append(
+        f"r({error_code})" if error_code is not None else "end of mock do-file"
+    )
+    log.write_text("\n".join(lines) + "\n")
+    return 0
+
+
+def _parse_invocation(args: list[str]) -> tuple[Path, list[str], Path] | None:
+    if len(args) < MINIMUM_ARGUMENTS or args[:2] != ["-e", "do"]:
+        return None
+
+    script = Path(args[2]).resolve()
+    options = args[3:-1]
+    log_arg = args[-1]
+    log = (
+        script.with_suffix(".log")
+        if log_arg == "-"
+        else Path.cwd() / f"{log_arg.removeprefix('-')}.log"
+    )
+    return script, options, log
+
+
+def _run_script(script: Path, options: list[str]) -> tuple[list[str], int | None]:
+    lines = [f"running mock Stata for {script.name}"]
+    macros: dict[str, str] = {}
+
+    for raw_line in script.read_text().splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("*"):
+            continue
+
+        line = _expand_local_macros(line, macros)
+        lines.append(f". {line}")
+        error_code = _execute_line(line, macros, options)
+        if error_code is not None:
+            return lines, error_code
+
+    return lines, None
+
+
+def _execute_line(line: str, macros: dict[str, str], options: list[str]) -> int | None:
+    command, _, rest = line.partition(" ")
+    command = command.lower()
+    rest = rest.strip()
+
+    if command == "args":
+        macros.update(dict(zip(rest.split(), options, strict=False)))
+    elif command == "sysuse":
+        return None
+    elif command == "save":
+        return _save_dataset(rest)
+    elif command in {"error", "exit"}:
+        return int(rest.split()[0])
+    else:
+        return UNKNOWN_COMMAND
+
+    return None
+
+
+def _save_dataset(rest: str) -> int | None:
+    target = _parse_save_target(rest)
+    if not target:
+        return INVALID_SYNTAX
+
+    path = Path(target)
+    if path.suffix == "":
+        path = path.with_suffix(".dta")
+    if not path.is_absolute():
+        path = Path.cwd() / path
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("mock Stata dataset\n")
+    return None
+
+
+def _expand_local_macros(line: str, macros: dict[str, str]) -> str:
+    return re.sub(r"`([^']+)'", lambda match: macros.get(match.group(1), ""), line)
+
+
+def _parse_save_target(rest: str) -> str:
+    target = rest.split(",", maxsplit=1)[0].strip()
+    return target.strip("\"'")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pyproject.toml

@@ -38,8 +38,15 @@ test = [
     "pytest-cov>=5.0.0",
     "pytest-xdist>=3.6.1",
 ]
+test-mock-stata = ["stata-mock"]
 typing = ["pytask-parallel>=0.5.1", "ty>=0.0.8"]
 
+[tool.uv.sources]
+stata-mock = { workspace = true }
+
+[tool.uv.workspace]
+members = ["packages/stata_mock"]
+
 [tool.hatch.build.hooks.vcs]
 version-file = "src/pytask_stata/_version.py"
 

tests/conftest.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import os
 import shutil
 import sys
+import warnings
 from contextlib import contextmanager
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 import pytest
@@ -14,13 +17,51 @@
 if TYPE_CHECKING:
     from collections.abc import Callable
 
-needs_stata = pytest.mark.skipif(
-    next(
-        (executable for executable in STATA_COMMANDS if shutil.which(executable)), None
+needs_stata = pytest.mark.usefixtures("stata_runtime")
+
+
+@pytest.fixture(scope="session")
+def stata_runtime():
+    """Use mock Stata on CI and require a real Stata executable locally."""
+    executable_path = _find_stata_executable()
+    if os.environ.get("CI"):
+        if executable_path is None:
+            msg = (
+                "CI requires the stata-mock package to be installed so one of "
+                f"{STATA_COMMANDS} is available on PATH."
+            )
+            pytest.fail(msg)
+        warnings.warn(
+            f"Using mock Stata runtime because CI is set: {executable_path}.",
+            stacklevel=1,
+        )
+        yield
+        return
+
+    if executable_path is None or _is_mock_stata_executable(executable_path):
+        msg = (
+            "Stata-dependent tests require a real Stata executable on PATH when CI is "
+            "not set. Run the CI test recipe to use the stata-mock package."
+        )
+        pytest.fail(msg)
+
+    yield
+
+
+def _is_mock_stata_executable(executable_path: str) -> bool:
+    try:
+        Path(executable_path).resolve().relative_to(Path(sys.prefix).resolve())
+    except ValueError:
+        return False
+    else:
+        return True
+
+
+def _find_stata_executable() -> str | None:
+    return next(
+        (path for executable in STATA_COMMANDS if (path := shutil.which(executable))),
+        None,
     )
-    is None,
-    reason="Stata needs to be installed.",
-)
 
 
 class SysPathsSnapshot:

tests/test_execute.py

@@ -44,10 +44,10 @@ def test_pytask_execute_task_setup_raise_error(stata, platform, expectation):
 def test_run_do_file(runner, tmp_path):
     task_source = """
     import pytask
+    from pathlib import Path
 
     @pytask.mark.stata(script="script.do")
-    @pytask.mark.produces("auto.dta")
-    def task_run_do_file():
+    def task_run_do_file(produces=Path("auto.dta")):
         pass
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))
@@ -73,11 +73,12 @@ def task_run_do_file():
 def test_run_do_file_w_task_decorator(runner, tmp_path):
     task_source = """
     import pytask
+    from pathlib import Path
+    from pytask import task
 
-    @pytask.mark.task
+    @task
     @pytask.mark.stata(script="script.do")
-    @pytask.mark.produces("auto.dta")
-    def run_do_file():
+    def run_do_file(produces=Path("auto.dta")):
         pass
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))
@@ -130,10 +131,10 @@ def test_run_do_file_w_wrong_cmd_option(runner, tmp_path):
     """Apparently, Stata simply discards wrong cmd options."""
     task_source = """
     import pytask
+    from pathlib import Path
 
     @pytask.mark.stata(script="script.do", options="--wrong-flag")
-    @pytask.mark.produces("out.dta")
-    def task_run_do_file():
+    def task_run_do_file(produces=Path("out.dta")):
         pass
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))
@@ -158,8 +159,7 @@ def test_run_do_file_by_passing_path(runner, tmp_path):
     from pathlib import Path
 
     @pytask.mark.stata(script="script.do", options=Path(__file__).parent / "auto.dta")
-    @pytask.mark.produces("auto.dta")
-    def task_run_do_file():
+    def task_run_do_file(produces=Path("auto.dta")):
         pass
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))
@@ -175,15 +175,38 @@ def task_run_do_file():
     assert result.exit_code == ExitCode.OK
 
 
+@needs_stata
+def test_run_do_file_fails_with_stata_error(runner, tmp_path):
+    task_source = """
+    import pytask
+    from pathlib import Path
+
+    @pytask.mark.stata(script="script.do")
+    def task_run_do_file(produces=Path("out.dta")):
+        pass
+    """
+    tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))
+
+    do_file = """
+    error 601
+    """
+    tmp_path.joinpath("script.do").write_text(textwrap.dedent(do_file))
+
+    result = runner.invoke(cli, [tmp_path.as_posix()])
+
+    assert result.exit_code == ExitCode.FAILED
+    assert "r(601)" in result.output
+
+
 @needs_stata
 def test_run_do_file_fails_with_multiple_marks(runner, tmp_path):
     task_source = """
     import pytask
+    from pathlib import Path
 
     @pytask.mark.stata(script="script.do")
     @pytask.mark.stata(script="script.do")
-    @pytask.mark.produces("auto.dta")
-    def task_run_do_file():
+    def task_run_do_file(produces=Path("auto.dta")):
         pass
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))
@@ -199,10 +222,11 @@ def task_run_do_file():
 def test_with_task_without_path(runner, tmp_path):
     task_source = """
     import pytask
+    from pathlib import Path
     from pytask import task
 
     task_example = pytask.mark.stata(script="script.do")(
-        pytask.mark.produces("auto.dta")(task()(lambda x: None))
+        task(kwargs={"produces": Path("auto.dta")})(lambda produces: None)
     )
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(task_source))