diff --git a/DESIGN.md b/DESIGN.md index ae8ec58..6024547 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -203,6 +203,16 @@ The sub-test's own pass/fail result is intentionally not propagated. The interpreter and entry point used to spawn the sub-instance are picked automatically by `_testium_launch_cmd()` based on how the parent was started (AppImage → `$APPIMAGE`; Flatpak → `flatpak run`; PyInstaller → the frozen binary; source/wheel → `[sys.executable, abspath(sys.argv[0])]`). The user cannot override either via the YAML — selecting a different testium binary or Python from a sub-test was removed because it was either ill-defined (bundle modes have no separable Python) or could mismatch the parent's environment in surprising ways. +### `pytest` item +`src/testium/interpreter/test_items/test_item_pytest.py` — the pytest analogue of the `unittest` item: runs a user pytest file and surfaces every collected test as a child item (one PASS/FAIL/SKIP per test, with duration + failure message in the report). + +Unlike `unittest` (which runs in-process), pytest runs in a **subprocess on the host interpreter** (`bins.python_bin()`), like `py_func`/`lua_func` — so the user's pytest install and test dependencies live on the host and the item works across every packaging channel (incl. Flatpak via the same staging used by `py_func`). + +- A stdlib-only pytest plugin (`_PLUGIN_SOURCE`, written to a temp dir and loaded with `-p`) streams sentinel-prefixed lines back over the subprocess stdout: `__TESTIUM_PYTEST_COLLECTED__` (node-id list, at collection), `__TESTIUM_PYTEST_START__` / `__TESTIUM_PYTEST_RESULT__` (per test). The parent parses them live; non-sentinel lines are forwarded to the log. +- `load()` runs `pytest --collect-only` once to build the child tree; `execute()` runs the enabled node-ids once and maps results back by node-id. +- pytest is invoked with `--capture=no` (so plugin sentinels + test prints reach our pipe), `-o addopts=` (neutralise user addopts — xdist/cov would break the per-test hook parsing), `-p no:cacheprovider`. `stop_on_failure` → `-x`; disabled children → NORUN without running. +- Params: `test_file` (required), `test_method` (optional list of function names, matched against the node-id function segment with the parametrisation suffix stripped). Registered as `cst.TYPE_PYTEST` / `TYPE_PYTEST_STEP`, loaded via the same self-loading branch as `unittest` in `test_set.load_test_recursively`. + ### Report exporters & plugins `src/testium/interpreter/test_report/test_report.py` — `_EXPORTER_REGISTRY` dict maps a format name (cmd key in the YAML `report.export`) to a lazy loader. Built-ins: `text`, `json`, `junit` (needs `junit_xml`), `html` (needs `lxml`). `sqlite` is the storage layer, no-op as an export. @@ -302,6 +312,7 @@ The `testium_assist` editor extension is a thin LSP client that spawns `testium Both Flatpak and AppImage export `TESTIUM_VERSION` from a launcher (Flatpak: launcher script in `org.testium.Testium.yaml`; AppImage: `runtime.env` in `AppImageBuilder.yml`). `get_testium_version()` checks `/.flatpak-info` / `APPIMAGE` and reads `TESTIUM_VERSION` rather than relying on package metadata or repo introspection. ## Recent fixes / notable changes +- `pytest` item: pytest analogue of `unittest`, but runs on the **host interpreter in a subprocess** (`bins.python_bin()`, like `py_func`) so it works across every packaging channel. A stdlib-only pytest plugin streams collected node-ids + per-test results back over stdout via sentinels; each test becomes a child item with its own PASS/FAIL/SKIP, duration and failure message. Params: `test_file`, `test_method`. Validation item: `test/validation/items/pytest/` (the validation venv now pip-installs `pytest`). See "### `pytest` item". - Subprocess RPC startup handshake: the `py_func`/`lua_func`/`eval_proc` worker now picks its own port (`bind 0`), announces it on stdout (`__TESTIUM_RPC_PORT__=`), and the parent connects only after reading it. Fixes intermittent Windows `failed to connect : timeout` and the matching `wait_ready()` hang; removes the reserve/close/rebind race and `SO_REUSEADDR`. See "Subprocess RPC startup handshake". - `build_all.sh`: builds the four heavy channels in parallel (serial prep for the shared venv + wheel), results in completion order, Ctrl+C kills the whole job tree; `--ram` puts the build scratch on tmpfs (`/dev/shm`) + skips UPX for fast builds on USB/SD storage (Flatpak excluded — rofiles-fuse can't mount tmpfs). See the "Building all channels" section. - LSP across packaging channels: `testium lsp` (and the `testium_assist` editor extension that spawns it) now works from source, wheel, PyInstaller, Flatpak and AppImage. Two enablers — (1) action items declare a class-level `ACTIONS = {key: class}` registry (like `PARAMS`), so `lsp/schema.py` builds the full schema from class attributes with no `inspect.getsource`/AST (which broke under frozen PyInstaller); (2) the `[lsp]` extra (pygls) is wired into every full-app channel. `test/validation/lsp_check.py`, run by `run.sh` before the suite, asserts per-channel that `schema` keeps its actions and `lsp` answers `initialize`. See the matching architecture sections. diff --git a/src/testium/interpreter/test_items/test_item_pytest.py b/src/testium/interpreter/test_items/test_item_pytest.py new file mode 100644 index 0000000..695fef9 --- /dev/null +++ b/src/testium/interpreter/test_items/test_item_pytest.py @@ -0,0 +1,379 @@ +"""``pytest`` test item. + +Runs a user pytest file and surfaces every collected test as a child item +(one PASS / FAIL / SKIP per test, with duration and failure message in the +report) — the pytest analogue of the ``unittest`` item. + +Unlike ``unittest`` (which runs in-process), pytest runs in a **subprocess on +the host interpreter** (``bins.python_bin()``), exactly like ``py_func`` / +``lua_func``. This keeps the user's pytest install and test dependencies on +the host (visible across every packaging channel — source, wheel, PyInstaller, +Flatpak, AppImage) instead of requiring them inside the bundled interpreter. + +A tiny stdlib-only pytest plugin (written to a temp dir and loaded with +``-p``) streams the collected node ids and per-test results back over the +subprocess stdout as sentinel-prefixed lines, which the parent parses live. +""" + +import os +import json +import shutil +import atexit +import tempfile +import threading +import queue +import subprocess + +import api.testium as tm +from runtime.tum_except import ETUMFileError, ETUMRuntimeError +from interpreter.test_items.test_item import TestItem, test_run, test_data +from interpreter.test_items.test_result import TestResult, TestValue +from interpreter.utils.constants import TestItemType as cst +from interpreter.utils.param_decl import Param, ParamSet, LIST +from interpreter.utils.paths import no_window_kwargs +from interpreter.utils import bins + + +# Sentinels streamed by the in-subprocess plugin (see _PLUGIN_SOURCE). Kept in +# sync with the plugin source below. +_SENT_COLLECTED = "__TESTIUM_PYTEST_COLLECTED__" +_SENT_START = "__TESTIUM_PYTEST_START__" +_SENT_RESULT = "__TESTIUM_PYTEST_RESULT__" + +_PLUGIN_MODULE = "_testium_pytest_plugin" + +# stdlib-only pytest plugin executed inside the host subprocess. It must not +# import anything from testium. It emits one sentinel line per event so the +# parent can rebuild the test tree (collection) and per-test results (run) +# without parsing pytest's human output or a JUnit XML. +_PLUGIN_SOURCE = '''\ +import sys +import json + +_SENT_COLLECTED = "__TESTIUM_PYTEST_COLLECTED__" +_SENT_START = "__TESTIUM_PYTEST_START__" +_SENT_RESULT = "__TESTIUM_PYTEST_RESULT__" + +_reports = {} + + +def _emit(payload): + # Leading newline guarantees the sentinel starts its own line even if a + # test printed without a trailing newline (pytest runs with --capture=no). + sys.stdout.write("\\n" + payload + "\\n") + sys.stdout.flush() + + +def pytest_collection_modifyitems(session, config, items): + _emit(_SENT_COLLECTED + json.dumps([it.nodeid for it in items])) + + +def pytest_runtest_logstart(nodeid, location): + _emit(_SENT_START + nodeid) + + +def pytest_runtest_logreport(report): + _reports.setdefault(report.nodeid, {})[report.when] = report + + +def _skip_reason(report): + lr = report.longrepr + if isinstance(lr, tuple) and len(lr) == 3: + return str(lr[2]) + return report.longreprtext or "" + + +def pytest_runtest_logfinish(nodeid, location): + phases = _reports.pop(nodeid, {}) + setup = phases.get("setup") + call = phases.get("call") + teardown = phases.get("teardown") + + duration = 0.0 + for rep in (setup, call, teardown): + if rep is not None: + duration += getattr(rep, "duration", 0.0) or 0.0 + + outcome = "pass" + message = "" + if setup is not None and setup.failed: + outcome, message = "fail", setup.longreprtext + elif setup is not None and setup.skipped: + outcome, message = "skip", _skip_reason(setup) + elif call is not None: + if call.failed: + outcome, message = "fail", call.longreprtext + elif call.skipped: + outcome, message = "skip", _skip_reason(call) + else: + outcome = "pass" + if teardown is not None and teardown.failed and outcome == "pass": + outcome, message = "fail", teardown.longreprtext + + _emit(_SENT_RESULT + json.dumps({ + "nodeid": nodeid, + "outcome": outcome, + "message": message, + "duration": duration, + })) +''' + + +class TestItemPytestElement(TestItem): + """One collected pytest test (leaf child of a pytest file item).""" + + def __init__(self, name, parent=None, status_queue=None, filename=""): + super().__init__(None, parent, status_queue, filename=filename) + self.is_container = False + self._name = name + self._type = cst.TYPE_PYTEST_STEP + self.banner = "" + self.footer = "" + self._nodeid = "" + self._reported_done = False + + +class TestItemPytestFile(TestItem): + + PARAMS = ParamSet( + Param("test_file", required=True, + doc="Path to the pytest test file."), + Param("test_method", kind=LIST, + doc="Optional list of test function names to restrict the run " + "to (matched against the function part of each node id, " + "parametrisation suffix stripped). When empty, every " + "collected test in the file is run."), + ) + + def __init__(self, dict_item, parent=None, status_queue=None, filename=""): + self._name = cst.TYPE_PYTEST.item_name + super().__init__(dict_item, parent, status_queue, filename=filename) + self.is_container = True + self._type = cst.TYPE_PYTEST + self._fileName = self._prms.getParam('test_file', required=True, processed=True) + self._testDir = '' + self._test_methods = self._prms.getParamAll('test_method', processed=True) + self._cwd = "" + self._plugin_dir = "" + + def setTestDir(self, dir): + self._testDir = dir + + # ---- subprocess plumbing ------------------------------------------------- + + def _write_plugin(self): + # In Flatpak the host process can only read /tmp (shared), so stage the + # plugin there; outside Flatpak the default temp dir is fine. + d = tempfile.mkdtemp(prefix="testium_pytest_", + dir="/tmp" if bins._in_flatpak() else None) + with open(os.path.join(d, _PLUGIN_MODULE + ".py"), "w") as f: + f.write(_PLUGIN_SOURCE) + atexit.register(shutil.rmtree, d, ignore_errors=True) + return d + + def _pytest_popen(self, args): + pbin = bins.python_bin() + if not pbin: + raise ETUMRuntimeError("No valid Python 3 interpreter found") + + env = os.environ.copy() + bins.apply_host_libs(env) + env.pop("PYTHONUSERBASE", None) + env["PYTHONPATH"] = self._plugin_dir + os.pathsep + env.get("PYTHONPATH", "") + + cmd_args = [ + "-m", "pytest", + "--capture=no", # let plugin sentinels + test prints reach our pipe + "-o", "addopts=", # neutralise user addopts (xdist/cov break parsing) + "-p", "no:cacheprovider", + "-p", _PLUGIN_MODULE, + *args, + ] + + if bins._in_flatpak(): + host_env = {k: env[k] for k in ("PYTHONPATH", "PATH") if env.get(k)} + params = bins.flatpak_host_spawn( + pbin, cmd_args, host_cwd=self._cwd, extra_env=host_env) + popen_kwargs = {} + else: + params = [pbin, *cmd_args] + popen_kwargs = {"env": env, "cwd": self._cwd} + + return subprocess.Popen( + params, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + encoding="utf-8", + errors="replace", + restore_signals=False, + **no_window_kwargs(), + **popen_kwargs, + ) + + # ---- loading (collection) ------------------------------------------------ + + def _collect(self): + proc = self._pytest_popen(["--collect-only", "-q", self._fileName]) + nodeids = [] + output = [] + for line in proc.stdout: + line = line.rstrip("\n") + if line.startswith(_SENT_COLLECTED): + try: + nodeids = json.loads(line[len(_SENT_COLLECTED):]) + except ValueError: + pass + elif line != "": + output.append(line) + proc.wait() + return nodeids, "\n".join(output) + + def load(self): + ret = {} + if self._fileName == '': + raise ETUMFileError('A file name is expected but got "None"') + + if not os.path.isabs(self._fileName): + self._fileName = os.path.normpath(os.path.join(self._testDir, self._fileName)) + if not os.path.isfile(self._fileName): + raise ETUMFileError('File "%s" is not found' % (self._fileName)) + + self._cwd = os.path.dirname(self._fileName) or "." + self._plugin_dir = self._write_plugin() + + nodeids, output = self._collect() + if not nodeids: + raise ETUMFileError( + 'No pytest test collected from "%s".\n%s' % (self._fileName, output)) + + if self._test_methods: + present = {nid.split("::")[-1].split("[")[0] for nid in nodeids} + for m in self._test_methods: + if m not in present: + raise ETUMFileError( + 'Test function "%s" is not found in "%s"' % (m, self._fileName)) + wanted = set(self._test_methods) + nodeids = [nid for nid in nodeids + if nid.split("::")[-1].split("[")[0] in wanted] + + for nid in nodeids: + disp = nid.split("::", 1)[1] if "::" in nid else nid + item = TestItemPytestElement(disp, self) + item._nodeid = nid + ret.update(test_data(item, {})) + + return ret + + # ---- execution (run) ----------------------------------------------------- + + def _finish_child(self, child, value, message=""): + if child._reported_done: + return + if getattr(child, "t0", None) is None: + child.t0 = tm.timestamp() + self.status_queue.put( + {'id': child.id(), 'status': 'started', 'timestamp': child.t0}) + child.duration = tm.timestamp() - child.t0 + res = TestResult(child, value, message) + res.test_id = child.id() + res.sendStatus(self.status_queue) + self.status_queue.put( + {'id': child.id(), 'status': 'finished', 'duration': child.duration}) + self.report.addTest(child, res) + child._reported_done = True + + def _stream_results(self, proc, by_nodeid): + overall = TestValue.SUCCESS + outq = queue.Queue() + + def reader(): + for line in proc.stdout: + outq.put(line) + outq.put(None) + + t = threading.Thread(target=reader, daemon=True) + t.start() + + while True: + try: + line = outq.get(timeout=0.1) + except queue.Empty: + if self.isStopped(): + try: + proc.terminate() + except Exception: + pass + break + continue + if line is None: + break + line = line.rstrip("\n") + + if line.startswith(_SENT_COLLECTED): + # pytest re-collects at the start of the run; the node list was + # already consumed at load time, so drop it here. + continue + elif line.startswith(_SENT_START): + child = by_nodeid.get(line[len(_SENT_START):]) + if child is not None and getattr(child, "t0", None) is None: + child.t0 = tm.timestamp() + self.status_queue.put( + {'id': child.id(), 'status': 'started', 'timestamp': child.t0}) + elif line.startswith(_SENT_RESULT): + try: + rec = json.loads(line[len(_SENT_RESULT):]) + except ValueError: + continue + child = by_nodeid.get(rec.get("nodeid")) + if child is None: + continue + value = { + "pass": TestValue.SUCCESS, + "fail": TestValue.FAILURE, + "skip": TestValue.NORUN, + }.get(rec.get("outcome"), TestValue.FAILURE) + self._finish_child(child, value, rec.get("message", "")) + if value == TestValue.FAILURE: + overall = TestValue.FAILURE + elif line != "": + print(line) + + proc.wait() + return overall + + @test_run + def execute(self): + by_nodeid = {} + enabled_nodeids = [] + for i in range(self.childCount()): + c = self.child(i) + c.t0 = None + c._reported_done = False + by_nodeid[c._nodeid] = c + if c.enabled: + enabled_nodeids.append(c._nodeid) + else: + self._finish_child(c, TestValue.NORUN, "test disabled") + + overall = TestValue.SUCCESS + if enabled_nodeids and not self.isStopped(): + args = list(enabled_nodeids) + if self._stop_on_failure: + args.append("-x") + proc = self._pytest_popen(args) + overall = self._stream_results(proc, by_nodeid) + + # Any enabled test that produced no result (crash, -x stop, user stop) + # is reported as NORUN so the tree stays consistent. + for i in range(self.childCount()): + c = self.child(i) + if c.enabled and not c._reported_done: + self._finish_child(c, TestValue.NORUN, "not executed") + + if self.isStopped(): + self.result.set(TestValue.NORUN, 'pytest execution aborted on user request') + else: + self.result.set(overall, 'pytest ' + str(overall)) diff --git a/src/testium/interpreter/test_set.py b/src/testium/interpreter/test_set.py index 5c5e8e4..1b30813 100644 --- a/src/testium/interpreter/test_set.py +++ b/src/testium/interpreter/test_set.py @@ -532,7 +532,7 @@ class TestSet: item.is_folded = is_folded child = {} # case where the test item loads itself its descendants - if it == cst_type.TYPE_UNITTEST: + if it in (cst_type.TYPE_UNITTEST, cst_type.TYPE_PYTEST): item.setTestDir(test_dir) child = item.load() elif issubclass(it.item_class, TestItemActions): diff --git a/src/testium/interpreter/utils/constants.py b/src/testium/interpreter/utils/constants.py index 2f198ab..8d640c3 100644 --- a/src/testium/interpreter/utils/constants.py +++ b/src/testium/interpreter/utils/constants.py @@ -10,6 +10,8 @@ class TestItemEnum(): class TestItemType(Enum): TYPE_UNITTEST = TestItemEnum("unittest", "unittest") TYPE_UNITTEST_STEP = TestItemEnum("unittest_step", "unittest step") + TYPE_PYTEST = TestItemEnum("pytest", "pytest") + TYPE_PYTEST_STEP = TestItemEnum("pytest_step", "pytest step") TYPE_CONSOLE = TestItemEnum("console", "Console") TYPE_CONSOLE_ACTION = TestItemEnum("console_action", "Console action") TYPE_CYCLE = TestItemEnum("loop", "Cycle") diff --git a/src/testium/interpreter/utils/test_init.py b/src/testium/interpreter/utils/test_init.py index 8c0d483..f5bb5bc 100644 --- a/src/testium/interpreter/utils/test_init.py +++ b/src/testium/interpreter/utils/test_init.py @@ -25,6 +25,7 @@ from interpreter.utils.version import ( from interpreter.test_items.test_item import TestItem from interpreter.test_items.test_item_sleep import TestItemSleep from interpreter.test_items.test_item_unittest import TestItemUnittestFile +from interpreter.test_items.test_item_pytest import TestItemPytestFile from interpreter.test_items.test_item_cycle import TestItemCycle from interpreter.test_items.test_item_runtime_plot import TestItemPlot from interpreter.test_items.test_item_group import TestItemGroup @@ -69,6 +70,7 @@ def _constants_init(): cst.TYPE_RUN.item_class = TestItemRun cst.TYPE_SLEEP.item_class = TestItemSleep cst.TYPE_UNITTEST.item_class = TestItemUnittestFile + cst.TYPE_PYTEST.item_class = TestItemPytestFile cst.TYPE_VALUE_DLG.item_class = TestItemValueDialog cst.TYPE_PARALLEL.item_class = TestItemParallel cst.TYPE_PARALLEL_BRANCH.item_class = TestItemParallelBranch diff --git a/src/testium/main_win/test_tree_items/test_tree_item.py b/src/testium/main_win/test_tree_items/test_tree_item.py index f9e4ed0..f5ba65e 100644 --- a/src/testium/main_win/test_tree_items/test_tree_item.py +++ b/src/testium/main_win/test_tree_items/test_tree_item.py @@ -12,6 +12,8 @@ from api.testium import print_warn _ITEM_CONFIG = { "unittest": {"icon": "folder.png", "icon_on": "folder-open.png", "expanded": True, "no_breakpoint": True}, "unittest step": {"icon": "document.png", "no_breakpoint": True}, + "pytest": {"icon": "python.png", "icon_on": "folder-open.png", "expanded": True, "no_breakpoint": True}, + "pytest step": {"icon": "document.png", "no_breakpoint": True}, "Console": {"icon": "terminal.png", "unfoldable": False}, "Console action": {"icon": "terminal.png"}, "Cycle": {"icon": "cycle.png", "expanded": True}, diff --git a/test/validation/items/pytest/param.yaml b/test/validation/items/pytest/param.yaml new file mode 100644 index 0000000..0af0f7f --- /dev/null +++ b/test/validation/items/pytest/param.yaml @@ -0,0 +1 @@ +no_param: Null diff --git a/test/validation/items/pytest/test.tum b/test/validation/items/pytest/test.tum new file mode 100644 index 0000000..253ca73 --- /dev/null +++ b/test/validation/items/pytest/test.tum @@ -0,0 +1,17 @@ +- pytest: + name: Pytest item + test_file: {{include_directory}}/test_cases.py + key: $(test)_PASS + test_method: + - test_01_pass + - test_02_pass + - test_05_param + +- pytest: + name: Pytest item + test_file: {{include_directory}}/test_cases.py + key: $(test)_FAIL + test_method: + - test_01_pass + - test_03_fail + - test_04_skip diff --git a/test/validation/items/pytest/test_cases.py b/test/validation/items/pytest/test_cases.py new file mode 100644 index 0000000..fa200d3 --- /dev/null +++ b/test/validation/items/pytest/test_cases.py @@ -0,0 +1,28 @@ +import pytest + + +def test_01_pass(): + ''' Test 01 passes ''' + assert 1 + 1 == 2 + + +def test_02_pass(): + ''' Test 02 passes ''' + assert "a" in "abc" + + +def test_03_fail(): + ''' Test 03 fails on purpose ''' + assert 1 == 2, "deliberate failure" + + +@pytest.mark.skip(reason="skipped on purpose") +def test_04_skip(): + ''' Test 04 is skipped ''' + assert False + + +@pytest.mark.parametrize("n", [1, 2]) +def test_05_param(n): + ''' Test 05 is parametrised, both cases pass ''' + assert n < 3 diff --git a/test/validation/run.sh b/test/validation/run.sh index 4a34687..789b2a0 100755 --- a/test/validation/run.sh +++ b/test/validation/run.sh @@ -21,7 +21,7 @@ # even Flatpak reaches it via flatpak-spawn --host. The validation venv # is created with --system-site-packages so existing system packages # (PySide6, lxml, ...) stay visible, then junit-xml is pip-installed -# for post_execution.py. +# for post_execution.py and pytest for the `pytest` item. # # The report file is suffixed with the mode (e.g. validation-flatpak.sqlite) # so consecutive runs in different modes don't overwrite each other. @@ -73,7 +73,7 @@ if [ ! -d "$VENV_DIR" ]; then echo "Creating validation venv at $VENV_DIR" python3 -m venv --system-site-packages "$VENV_DIR" "$VENV_DIR/bin/pip" install --quiet --upgrade pip - "$VENV_DIR/bin/pip" install --quiet junit-xml + "$VENV_DIR/bin/pip" install --quiet junit-xml pytest fi VENV_PYTHON="$VENV_DIR/bin/python3"