38 Commits

Author SHA1 Message Date
9db0f89522 removed reference to "terminal" mode. Corrected errors in doc generation. 2026-05-01 08:20:29 +02:00
f38a24190d Adopt EUPL-1.2 licence for the project
Add LICENSE (full EUPL-1.2 text + project copyright), CONTRIBUTING.md
with the inbound = outbound rule, and declare the licence in
pyproject.toml and README.md.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-01 07:58:22 +02:00
b16494ef6d Preserve test tree check and fold state across reload
Reload (re-loading the same .tum file) was rebuilding the GUI tree
from scratch, resetting every checkbox and unfolding everything.
Snapshot the user's selection and fold state via the existing
getCheckList/restoreCheckList and getFoldList/restoreFoldList methods
(already used for session persistence through prefs), so a same-file
reload keeps both as well. A change in the total item count (file
edited between loads) skips the restore.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-30 23:51:36 +02:00
b175ff4189 Add distinct icon for parallel branch tree items
A parallel branch now displays a single-arrow icon (parallel_branch.png)
distinct from the parallel container's three-arrow icon, making the
tree hierarchy easier to read at a glance.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-30 23:41:31 +02:00
d66a46736f Hide branch steps in parallel item F1 panel
The 'branches' list shown in the F1 panel previously contained each
branch's full 'steps', duplicating what is already visible in the test
tree. Strip 'steps' inside each branch dict while keeping 'branches'
itself so per-branch attributes (name, wait_for, condition, ...) stay
visible.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-30 23:35:43 +02:00
1b2d427ced Add parallel test item with thread-aware stdout routing
The parallel item runs branches concurrently with sync:all or sync:any
policy and optional per-branch wait_for synchronization. Each branch
runs in its own daemon thread and produces a clean per-item entry in
the SQLite report; the live output is prefixed [<branch_name>] so
concurrent branches stay readable.

Supporting changes:
- StdoutProxy (lib/stdout_redirect.py): thread-aware sys.stdout/stderr
  with per-thread capture buffers and per-branch live-output prefix.
  Adds writeln() for Python 3.14 unittest compatibility.
- TestItemContainer: shared base extracted from Group/Cycle for the
  sequential children execution pattern.
- TestItemSleep: interruptible loop polling _is_stopped so sync:any
  can cancel slow branches quickly.
- TestReport: thread-safe SQLite (check_same_thread=False + lock).

Also drops the unused -m/--terminal mode and its module.

Validation: 11 scenarios in test/validation/items/parallel covering
sync:all/any, no_fail, wait_for + timeout, conditions, multi-branch,
nested parallel, parallel inside loop, real branch failure.

Documentation: new parallel_test_item.rst added to the manual.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-30 23:23:31 +02:00
be540cd304 text mode effort finished on batch. 2026-04-27 17:20:15 +02:00
476b59c6f7 icons 2026-04-27 17:19:14 +02:00
bcafbfae18 Changed the "run" icon by testium. 2026-04-27 12:04:19 +02:00
e56a1f72c8 requirements changed for doc in readme. 2026-04-27 08:07:53 +02:00
83411482b2 Rename unittest_file item to unittest
- constants.py: TYPE_UNITTEST_FILE → TYPE_UNITTEST, cmd "unittest_file" → "unittest"
- All Python files updated: test_item_unittest.py, test_set.py, test_init.py,
  terminal.py, report_export_txt.py, test_tree_item.py
- All .tum files updated (examples, validation, doc)
- Sphinx doc: unittest_file_test_item.rst → unittest_test_item.rst,
  all references updated

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-27 08:05:40 +02:00
a28e644621 doc: use PASS/FAIL terminology in run item doc
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-27 07:59:54 +02:00
4a4a70b5f6 pdf doc updated 2026-04-27 07:56:43 +02:00
06c4cc62c6 doc: update run item documentation
Clarify result semantics (SUCCESS on launch, not on sub-test result),
batch vs GUI mode behaviour, and clean up attribute descriptions.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-27 07:53:26 +02:00
60dbcf0252 Fix run item and batch mode robustness
- run item: rename tum_fime→tum, remove stdout=PIPE (deadlock with
  spawn), support batch mode (-b), SUCCESS on any completed subprocess
  regardless of sub-test result
- batch.py: fix control("loaded") deadlock via daemon thread + Event +
  is_alive() polling; fix premature finish on gd_update messages;
  propagate success flag from finished message; guard control("close")
- process.py: include success flag in send_finished message
- py_process/lua_process: add stdout/stderr=DEVNULL to Popen
- test_run.py: fix finished detection ("id" in m and m["id"] is None)
- testium_win.py: track run_exit_code, SIGABRT handler, clean exit
- __init__.py: sys.exit with batch success flag
- Add run item validation tests and CLAUDE.md documentation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-27 07:49:16 +02:00
a3e449cc7d in batch mode, the dialog allways return FAIL, except if auto_result is defined (validation only).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 11:52:52 +02:00
95107117fa Color is automatically adapted to the theme of the console.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 11:50:28 +02:00
88cc410eed fix of blocking of the text output in batch mode.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 11:49:52 +02:00
fa7f8cef7c Text mode upgrade (to be fixed). 2026-04-26 09:20:39 +02:00
5a065128be Fix lua delgd test: use sentinel default instead of nil comparison
cjson decodes JSON null as cjson.null, not Lua nil, so == nil always fails.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:25:30 +02:00
b7b930aab1 Add validation tests for OS, get_main_dir, timestamp, timestamp_as_sec helper functions
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:21:56 +02:00
609ca57202 Add delgd validation test for py_func and lua_func
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:18:31 +02:00
d26b60435b Add auto_result param to dialog items for automated validation
Each dialog test item now accepts an optional auto_result parameter
(ok/cancel/yes/no) and auto_value for text dialogs. When set, the dialog
window opens, stays visible 2 seconds, then closes automatically with the
specified result — allowing the validation suite to run without manual
interaction.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:07:07 +02:00
de143b6cc3 Fix EOFError crash when dialog subprocess exits without sending a result
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:06:55 +02:00
d955ae81f9 Added variables list in "F1" dialog. They are modifiable. To be tested. 2026-04-20 23:27:39 +02:00
2cd3aa3305 updated doc + param 2026-04-20 22:55:09 +02:00
276d485905 Add store_result common attribute to test items
Allows any test item to store its result (or PASS/FAIL status when result
is None) into a named global variable, available to subsequent items via
$(variable_name). store_result runs after expected_result but before
no_fail so the real outcome is always captured.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 22:26:47 +02:00
95912dd3e1 Fix 'process_result must fail' test missing expected_result
Without expected_result, a False process_result value does not fail the
test. Adding expected_result: True makes the comparison fail as intended.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 21:43:33 +02:00
6d1fb6a6bc Add JSON-RPC echo server for validation suite
Replaces the external jrpces binary dependency with a self-contained
Python script. The server supports TCP (newline-delimited JSON, port 4321)
and UDP (port 4323), handles JSON-RPC 1.0 and 2.0, and implements:
  - echo(*args) -> [args, {}]
  - unknown methods -> error {code: -32000, message: "function not found"}

test.tum is updated to launch jrpc_echo_server.py via python3 and wait
for the "ready" readiness message before running tests.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 21:34:43 +02:00
2cc42e9065 Fix 15-second close delay after dialog tests
Dialog subprocesses were forked from TestProcess, inheriting its
multiprocessing Queue objects and their process-shared POSIX semaphores
(_wlock). If a fork happened while the feeder thread held _wlock, the
child exited without releasing it, permanently blocking the feeder
thread on the next wacquire() and stalling Python's atexit _finalize_join
— causing test_proc.join() (no timeout) to hang the app for ~15 seconds.

Fix: use multiprocessing.get_context('spawn') for dialog subprocesses so
they start with a clean interpreter and inherit no semaphores or Queue
state. Also add a terminate/kill fallback timeout to test_proc.join() as
a safety net, and fix the missing return in JsonRpcConnection.is_alive().

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 21:14:33 +02:00
2b7678c39e Fix dialog subprocesses: centralize Qt env setup and extract base class
- Add dialog_env.py service: forces QT_QPA_PLATFORM=xcb on Linux so Qt
  doesn't crash under Wayland in spawned subprocesses
- Use QMessageBox instance (instead of static methods) for msg/question
  dialogs so WindowStaysOnTopHint can be set, making them visible
- Add TestItemDialogBase with _run_dialog/_run_dialog_with_result/_cleanup_process,
  removing duplicated subprocess launch/poll/terminate logic from all 7 dialog items
- Reduce terminate() join timeout from 2s to 0.2s across all dialog items

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-19 20:01:56 +02:00
c72176d029 Improve loading error messages with item context and hierarchy path
Add item_load_context() context manager to tum_except.py that enriches
ETUMSyntaxError with the item type, name, and parent path instead of
replacing the original message with a vague 'missing or wrong parameter'.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-19 10:05:40 +02:00
617f599f86 Fix context subprocess leak and document py_func.tm helpers
- process.py: stop context_id engines in the inner finally block, before
  restore_gd() wipes _py_func_contexts/_lua_func_contexts from the global
  dict — engines were previously orphaned after every test run
- py_func/tm.py: add user-facing docstrings to gd/setgd/delgd; remove
  internal JSON-serialization details from the docs
- helper_lib.rst: auto-generate global variable helpers from py_func.tm
  (the actual subprocess API) instead of globdict
- conf.py: add src/ to Sphinx sys.path so py_func.tm is importable
- py_func_test_item.rst: simplify context sharing section, remove
  JSON-serializable/non-serializable distinction for end users
- Regenerated PDF manual

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:27:31 +02:00
d92f518e1e Add context_id to py_func and lua_func for shared persistent subprocess
- py_func and lua_func items accept a context_id parameter; items sharing
  the same id reuse the same subprocess for the duration of the test run
- Subprocess-side tm.setgd/tm.gd use a local fallback dict for non-JSON-
  serializable values (py_func only); serializable values reach the main
  process global dict and are accessible from any test item or subprocess
- Shared subprocess engines are cleaned up in process.py finally block
- LuaProcessBase gains is_alive() (was missing, broke all lua_func items)
- Validation tests cover serializable sharing across different context ids,
  non-serializable sharing within the same context_id, and cross-item access
- RST documentation updated for both py_func and lua_func items

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:02:36 +02:00
49354b8664 Merge branch 'feature/reload-without-restart'
Reload test without restarting testium (closes #18)
2026-04-18 15:24:11 +02:00
7383820aba Reload test without restarting testium (closes #18)
Replace os.execv restart in actionRefresh with file_manager.reload(),
leveraging the subprocess architecture so py_func modules are freshly
imported on each reload. Add a modal progress dialog with step labels
during loading. Fix checkbox reappearing on breakpoint with
show_checkboxes OFF.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 15:23:53 +02:00
67c879ab10 Fix checkboxes reappearing when setting a breakpoint with show_checkboxes OFF
itemChanged fires for any data change (including icon updates in the pause
column), causing on_testChecked to inadvertently restore CheckStateRole via
synchronizeEnabledState. Guard against non-checkbox column changes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 14:55:18 +02:00
aa72e349c6 Merge branch 'refactor/ui-modularization'
- Extract TestControllerService service layer over TestSetController
- Split MainWindow into TestRunner and TestFileManager coordinators
- Merge 21 QTestTreeItem subclasses into a single factory
- Replace _test_started/_test_paused booleans with TestState enum
2026-04-18 14:46:06 +02:00
125 changed files with 5165 additions and 1618 deletions

169
CLAUDE.md Normal file
View File

@@ -0,0 +1,169 @@
# Testium — Claude Context
## What is testium
Testium is a test sequencer/runner written in Python. It executes YAML-based test scripts ("`.tum`" files) and supports two execution modes:
- **GUI mode** (default, no flag): PySide6 Qt application (`src/testium/main_win/`)
- **Batch mode** (`-b` / `--batch-execution`): headless, non-interactive, runs tests and exits
Run from repo root: `./run.sh` (Linux) or `run.bat` / `run.ps1` (Windows).
Direct invocation: `python3 -m src/testium [-b] <test_file.tum>`
## Architecture
### Entry point
`src/testium/__init__.py` — parses CLI args, dispatches to the two modes.
`multiprocessing.set_start_method('spawn')` is called early (required for Linux dialog subprocesses).
### Core execution
- `src/testium/interpreter/process.py``TestProcess(multiprocessing.Process)`: runs the test in a child process. Stdout is redirected via a `StringQueue` → pipe → parent thread (`capture_stdout`) that writes to real stdout.
- `src/testium/interpreter/batch.py``Batch`: parent-side orchestrator for `-b` mode. Creates the `msg_queue`, starts `TestProcess`, waits for the "finished" signal.
- `src/testium/interpreter/test_set.py``TestSet`: builds and executes the tree of test items.
- `src/testium/interpreter/test_items/test_item*.py` — one file per test item type (check, cycle, group, let, unittest, py_func, lua_func, console, git, dialogs, report, parallel, …).
### Communication channels (parent ↔ child process)
- `msg_queue` (`multiprocessing.Queue`): carries status messages from child to parent.
- Item status: `{"id": <non-None>, "name": ..., "status": "started"|"finished", ...}`
- Global dict updates: `{"type": "gd_update"|"gd_delete", "key": ..., "value": ...}`**no "id" key**
- Process finished: `{"id": None, "name": "test_process", "status": "finished"}` — id key present but `None`
- `tst_ctrl` (`TestSetController`): sends control commands (execute, stop, pause, close, …) from parent to child.
- stdout pipe (`multiprocessing.Pipe`): streams test output from child back to parent's `capture_stdout` thread.
### Stdout pipeline (batch mode)
```
test item print()
→ sys.stdout (StringQueue, in child)
→ send_stdout thread (child) → pipe → capture_stdout thread (parent)
→ print() → sys.stdout (TermLog wrapping real stdout, in parent)
→ terminal
```
### Global dictionary
`src/testium/interpreter/utils/globdict.py` — shared state accessible from test scripts via `tm.gd()` / `tm.setgd()`. When `set_update_queue()` is active (during test execution), every `setgd`/`delgd` on a non-`_`-prefixed key pushes a message to `msg_queue`.
### Coloring (`-o` disables it)
`src/testium/interpreter/utils/termlog.py``TermLog` wraps stdout with colorama-based line coloring (PASS=green, FAIL=red, WARN=yellow, …). Applied in parent process for batch mode. Auto-detects light/dark terminal background via (in order): `COLORFGBG` env var, OSC 11 query, default dark.
### Dialog items in batch mode
All dialog items (`dialog_image`, `dialog_question`, `dialog_references`, `dialog_value`, `dialog_message`, `dialog_choices`, `dialog_note`) follow this rule in non-interactive text mode (`-b`):
- `auto_result` defined in the `.tum` → result controlled by it (`ok`/`yes` → SUCCESS, `cancel`/`no` → FAIL)
- `auto_result` absent → FAIL with `"Dialog not supported in batch mode"`
- `sleep dialog: true` → exception: just sleeps normally, no GUI, no failure
`auto_result` (and `auto_value` for value/note dialogs) is intended for the validation test suite (`test/validation/`) only.
### `parallel` item
`src/testium/interpreter/test_items/test_item_parallel.py` — runs multiple branches concurrently.
```yaml
- parallel:
name: My parallel block
sync: all # all: wait for all; any: stop as soon as one finishes
no_fail: true # (optional) don't propagate branch failures to parent
branches:
- name: Branch A
wait_for: # (optional) poll condition before starting
condition: <| expr |>
timeout: 10
steps:
- ...
- name: Branch B
steps:
- ...
```
- `TestItemParallel(TestItemContainer)`: mutates `dict_item["steps"]` to inject synthetic `parallel_branch` items so `load_test_recursively` loads branches normally as children.
- `TestItemParallelBranch(TestItemContainer)`: container for one branch. `wait_for` polls every 0.1s up to `timeout` seconds before running steps.
- `sync: any` calls `_stop_branch_recursively()` on all other branches when one *actually runs* (SUCCESS/FAILURE). A `NORUN` branch (disabled, condition not met) never wins the race.
- Each branch runs in a daemon thread; the parent waits with `.join()`.
- Branches stopped late (e.g. user disabled them in the GUI, or another sync:any branch already won) go through the normal `branch.stop() + branch.execute()` path so they always produce a clean DB entry via `addTest()`.
- Exceptions raised in a branch's `execute()` are caught by `run_branch`, logged to stdout, and converted to a `FAILURE` result so they never disappear silently.
- `sync: all` ignores `NORUN` branches when computing success (matches Group/Cycle semantics): only an actual `FAILURE` fails the parallel.
- `TestItemSleep` is interruptible (polls `self._is_stopped` in a loop) so `sync: any` can stop slow branches quickly. `py_func` and `console` items are not interruptible; their full duration is observed before the branch returns.
### `TestItemContainer` base class
`src/testium/interpreter/test_items/test_item_container.py` — shared base for Group, Cycle, Parallel, and ParallelBranch. Provides `_run_children_sequentially()` which handles stop-on-failure, `executedOnStop` items, and returns `(TestResult, stopped_bool)`.
### Report threading
`src/testium/interpreter/test_report/test_report.py` — SQLite report with thread-safe writes:
- `sqlite3.connect(..., check_same_thread=False)`
- `self._lock = threading.Lock()` guards the SQLite `INSERT` only.
- Per-item log capture (`stdio_redir.read()`) is naturally race-free thanks to per-thread buffers (see `StdoutProxy`).
### Thread-aware stdout (`StdoutProxy`)
`src/lib/stdout_redirect.py` — when `log_stored: True`, `intercept()` installs a `StdoutProxy` as `sys.stdout`/`sys.stderr` instead of a single shared `StringQueue`. The proxy:
- Holds one `StringQueue` per thread (registered via `register_thread(buffer=...)`). The main thread uses a default buffer; each parallel branch's thread registers its own at start and unregisters at end. `stdio_redir.read()` reads the calling thread's buffer → `addTest()` of an item running in branch X reads X's clean, non-interleaved output.
- For the live stream (terminal in batch / GUI panel), prefixes every line emitted from a branch's thread with `[<branch_name>] ` so concurrent branches stay readable.
- Exposes `write` / `writeln` / `flush` (Python 3.14's `unittest` calls `stream.writeln()` directly without `_WritelnDecorator`).
## Key files
| Path | Role |
|------|------|
| `src/testium/__init__.py` | CLI entry, mode dispatch |
| `src/testium/interpreter/batch.py` | `-b` mode orchestrator |
| `src/testium/interpreter/process.py` | Child test process |
| `src/testium/interpreter/test_set.py` | Test tree builder/executor |
| `src/testium/interpreter/test_items/test_item_container.py` | Base class for container items |
| `src/testium/interpreter/test_items/test_item_parallel.py` | `parallel` and `parallel_branch` items |
| `src/testium/interpreter/utils/globdict.py` | Global variable dict |
| `src/testium/interpreter/utils/termlog.py` | Terminal color output |
| `src/lib/stdout_redirect.py` | `StdioRedirect` singleton (`stdio_redir`) |
| `src/lib/string_queue.py` | Thread-safe string buffer used for stdout redirection |
| `src/testium/libs/testium.py` | Public API for test scripts (`tm.*`) |
## GUI icons (main_win)
Icons live in `src/testium/main_win/resources/` with three theme variants:
| Folder | Theme index | Usage |
|--------|-------------|-------|
| `color/` | 0 (default) | Coloured icons |
| `black/` | 1 | Black silhouette on transparent |
| `white/` | 2 | White silhouette on transparent (LA mode) |
Icons are **64×64 PNG**. Black variants: RGBA with RGB=`(0,0,0)`, alpha varies. White variants: LA with luminance=`255`, alpha varies.
The mapping item-type → icon filename is in `_ITEM_CONFIG` (`src/testium/main_win/test_tree_items/test_tree_item.py`). At runtime, `icon_prefix()` returns `:/color`, `:/black`, or `:/white` (Qt resource prefix) based on the user preference.
All icons must be declared in `src/testium/main_win/resources/testium_core_win.qrc` (one entry per theme section). After any QRC change, regenerate the compiled resource file:
```
cd src/testium/main_win/resources
pyside6-rcc testium_core_win.qrc -o testium_core_win_rc.py
```
Icons are assigned once when the test file is loaded (not updated live on theme change — a file reload is required).
### `run` item
`src/testium/interpreter/test_items/test_item_run.py` — launches a `.tum` file in a new testium instance (`-b` in batch mode, `-r` in GUI mode). Result:
- **SUCCESS** if the sub-instance launched and ran to completion (exit code is ignored)
- **FAILURE** if the file is not found, `wait_for_exec` is set without `start_time`/`end_time`, the time window was not reached, or any other launch error
The sub-test's own pass/fail result is intentionally not propagated.
## Recent fixes (branch `parallel_execution`)
- `test_item_parallel.py`: new `parallel` item with `sync: all|any`, `wait_for`, daemon threads, `_stop_branch_recursively()`. Each branch thread registers a per-thread stdout buffer with `stdio_redir.register_thread(...)` so its log capture and live-output prefix work in isolation.
- `test_item_container.py`: new `TestItemContainer` base class extracted from Group/Cycle patterns
- `test_item_sleep.py`: interruptible loop (checks `self._is_stopped`) instead of blocking `time.sleep()` so `sync: any` can stop slow branches quickly
- `stdout_redirect.py`: rewrote `intercept()` to install a `StdoutProxy` (thread-aware: per-thread capture buffers + branch-prefixed live output). Adds `writeln()` for Python 3.14 unittest compatibility.
- `test_report.py`: `check_same_thread=False` + lock around the SQLite `INSERT` for parallel branch concurrency. Log capture itself is race-free thanks to per-thread buffers.
- `__init__.py`: removed `-m`/`--terminal` mode
- `terminal.py`: deleted
## Recent fixes (branch `text_no_pyside`)
- `batch.py`: premature loop exit when `gd_update` messages (no `"id"` key) were mistaken for the "finished" signal — fix: `"id" in m and m["id"] is None`
- `batch.py`: `control("loaded")` deadlock if `TestProcess` crashed before `cmd_th` started — fix: daemon thread + `threading.Event` + `is_alive()` polling
- `termlog.py`: `COLOR_DEFAULT = Fore.WHITE` invisible on light terminals; added auto-detection + light palette. Also fixed `write()` residue accumulation bug (`s[pos:]``s[pos+1:]`).
- Dialog items: `auto_result`/`auto_value` now used in non-interactive text mode; dialogs without `auto_result` FAIL immediately in batch mode.
- `run` item: removed `stdout=PIPE` (caused deadlock with `multiprocessing` spawn); simplified result to SUCCESS on any completed subprocess.
## Validation tests
Located in `test/validation/`. Run with `-b` flag:
```
./run.sh -b -l mon_log.log -- test/validation/main.tum
```
Parallel item tests: `test/validation/items/parallel/test.tum`
## Dependencies
See `src/requirements.txt`. Key ones: `pyside6`, `pyyaml`, `jinja2`, `colorama`, `gitpython`, `pexpect`, `matplotlib`.

66
CONTRIBUTING.md Normal file
View File

@@ -0,0 +1,66 @@
# Contributing to testium
Thank you for your interest in contributing to testium.
## License of contributions
testium is licensed under the **European Union Public Licence v. 1.2 (EUPL-1.2)**
see the [LICENSE](LICENSE) file at the repository root.
By submitting a contribution to this project (pull request, patch, issue
attachment, or any other form of code, documentation or media), you agree
that your contribution is licensed to the project and to the public under the
**same EUPL-1.2** terms (or any later version of the EUPL approved by the
European Commission), and you certify that:
- you are the author of the contribution, or you have the right to submit it
under the EUPL-1.2;
- to the best of your knowledge, the contribution does not infringe any
third-party intellectual-property rights;
- the contribution may be redistributed by the project under the EUPL-1.2 and
any compatible licence listed in the EUPL-1.2 Appendix.
This is the **inbound = outbound** rule: contributions come in under the same
licence the project ships under.
You retain copyright on your contribution. The project does **not** ask you
to sign a CLA or assign your copyright.
## SPDX header in new source files
When creating a new source file, please include the following header at the
top of the file (adjust the comment marker to the file's language):
```python
# SPDX-License-Identifier: EUPL-1.2
# Copyright (c) <year> <your name>
```
For existing files, keep the header that is already there.
## How to contribute
1. Open an issue describing the change you want to make (bug, feature, doc).
2. Fork the repository, create a topic branch.
3. Commit with a clear message (one logical change per commit).
4. Make sure the validation suite still passes:
```
./run.sh -b -l mon_log.log -- test/validation/main.tum
```
5. Open a pull request against `main`.
## Coding conventions
- Python ≥ 3.11
- Follow existing style in the file you are modifying
- Add or update tests in `test/validation/` for new test items or behaviours
- Update `CLAUDE.md` and the Sphinx manual for user-visible changes
## Reporting security issues
Please do **not** report security vulnerabilities through public GitHub
issues. Instead, send an email to the project maintainer directly.
## Questions
Open a GitHub Discussion or an issue tagged `question`.

315
LICENSE Normal file
View File

@@ -0,0 +1,315 @@
Copyright (c) 2025-2026 François Dausseur
Licensed under the EUPL
EUROPEAN UNION PUBLIC LICENCE v. 1.2
EUPL © the European Union 2007, 2016
This European Union Public Licence (the 'EUPL') applies to the Work (as
defined below) which is provided under the terms of this Licence. Any use of
the Work, other than as authorised under this Licence is prohibited (to the
extent such use is covered by a right of the copyright holder of the Work).
The Work is provided under the terms of this Licence when the Licensor (as
defined below) has placed the following notice immediately following the
copyright notice for the Work:
Licensed under the EUPL
or has expressed by any other means his willingness to license under the EUPL.
1. Definitions
In this Licence, the following terms have the following meaning:
- 'The Licence': this Licence.
- 'The Original Work': the work or software distributed or communicated by the
Licensor under this Licence, available as Source Code and also as Executable
Code as the case may be.
- 'Derivative Works': the works or software that could be created by the
Licensee, based upon the Original Work or modifications thereof. This Licence
does not define the extent of modification or dependence on the Original
Work required in order to classify a work as a Derivative Work; this extent
is determined by copyright law applicable in the country mentioned in
Article 15.
- 'The Work': the Original Work or its Derivative Works.
- 'The Source Code': the human-readable form of the Work which is the most
convenient for people to study and modify.
- 'The Executable Code': any code which has generally been compiled and which
is meant to be interpreted by a computer as a program.
- 'The Licensor': the natural or legal person that distributes or communicates
the Work under the Licence.
- 'Contributor(s)': any natural or legal person who modifies the Work under
the Licence, or otherwise contributes to the creation of a Derivative Work.
- 'The Licensee' or 'You': any natural or legal person who makes any usage of
the Work under the terms of the Licence.
- 'Distribution' or 'Communication': any act of selling, giving, lending,
renting, distributing, communicating, transmitting, or otherwise making
available, online or offline, copies of the Work or providing access to its
essential functionalities at the disposal of any other natural or legal
person.
2. Scope of the rights granted by the Licence
The Licensor hereby grants You a worldwide, royalty-free, non-exclusive,
sublicensable licence to do the following, for the duration of copyright
vested in the Original Work:
- use the Work in any circumstance and for all usage,
- reproduce the Work,
- modify the Work, and make Derivative Works based upon the Work,
- communicate to the public, including the right to make available or display
the Work or copies thereof to the public and perform publicly, as the case
may be, the Work,
- distribute the Work or copies thereof,
- lend and rent the Work or copies thereof,
- sublicense rights in the Work or copies thereof.
Those rights can be exercised on any media, supports and formats, whether now
known or later invented, as far as the applicable law permits so.
In the countries where moral rights apply, the Licensor waives his right to
exercise his moral right to the extent allowed by law in order to make
effective the licence of the economic rights here above listed.
The Licensor grants to the Licensee royalty-free, non-exclusive usage rights
to any patents held by the Licensor, to the extent necessary to make use of
the rights granted on the Work under this Licence.
3. Communication of the Source Code
The Licensor may provide the Work either in its Source Code form, or as
Executable Code. If the Work is provided as Executable Code, the Licensor
provides in addition a machine-readable copy of the Source Code of the Work
along with each copy of the Work that the Licensor distributes or indicates,
in a notice following the copyright notice attached to the Work, a repository
where the Source Code is easily and freely accessible for as long as the
Licensor continues to distribute or communicate the Work.
4. Limitations on copyright
Nothing in this Licence is intended to deprive the Licensee of the benefits
from any exception or limitation to the exclusive rights of the rights owners
in the Work, of the exhaustion of those rights or of other applicable
limitations thereto.
5. Obligations of the Licensee
The grant of the rights mentioned above is subject to some restrictions and
obligations imposed on the Licensee. Those obligations are the following:
Attribution right: The Licensee shall keep intact all copyright, patent or
trademarks notices and all notices that refer to the Licence and to the
disclaimer of warranties. The Licensee must include a copy of such notices
and a copy of the Licence with every copy of the Work he/she distributes or
communicates. The Licensee must cause any Derivative Work to carry prominent
notices stating that the Work has been modified and the date of modification.
Copyleft clause: If the Licensee distributes or communicates copies of the
Original Works or Derivative Works, this Distribution or Communication will
be done under the terms of this Licence or of a later version of this Licence
unless the Original Work is expressly distributed only under this version of
the Licence — for example by communicating 'EUPL v. 1.2 only'. The Licensee
(becoming Licensor) cannot offer or impose any additional terms or conditions
on the Work or Derivative Work that alter or restrict the terms of the
Licence.
Compatibility clause: If the Licensee Distributes or Communicates Derivative
Works or copies thereof based upon both the Work and another work licensed
under a Compatible Licence, this Distribution or Communication can be done
under the terms of this Compatible Licence. For the sake of this clause,
'Compatible Licence' refers to the licences listed in the appendix attached
to this Licence. Should the Licensee's obligations under the Compatible
Licence conflict with his/her obligations under this Licence, the obligations
of the Compatible Licence shall prevail.
Provision of Source Code: When distributing or communicating copies of the
Work, the Licensee will provide a machine-readable copy of the Source Code or
indicate a repository where this Source will be easily and freely available
for as long as the Licensee continues to distribute or communicate the Work.
Legal Protection: This Licence does not grant permission to use the trade
names, trademarks, service marks, or names of the Licensor, except as
required for reasonable and customary use in describing the origin of the
Work and reproducing the content of the copyright notice.
6. Chain of Authorship
The original Licensor warrants that the copyright in the Original Work
granted hereunder is owned by him/her or licensed to him/her and that he/she
has the power and authority to grant the Licence.
Each Contributor warrants that the copyright in the modifications he/she
brings to the Work are owned by him/her or licensed to him/her and that
he/she has the power and authority to grant the Licence.
Each time You accept the Licence, the original Licensor and subsequent
Contributors grant You a licence to their contributions to the Work, under
the terms of this Licence.
7. Disclaimer of Warranty
The Work is a work in progress, which is continuously improved by numerous
Contributors. It is not a finished work and may therefore contain defects or
'bugs' inherent to this type of development.
For the above reason, the Work is provided under the Licence on an 'as is'
basis and without warranties of any kind concerning the Work, including
without limitation merchantability, fitness for a particular purpose, absence
of defects or errors, accuracy, non-infringement of intellectual property
rights other than copyright as stated in Article 6 of this Licence.
This disclaimer of warranty is an essential part of the Licence and a
condition for the grant of any rights to the Work.
8. Disclaimer of Liability
Except in the cases of wilful misconduct or damages directly caused to
natural persons, the Licensor will in no event be liable for any direct or
indirect, material or moral, damages of any kind, arising out of the Licence
or of the use of the Work, including without limitation, damages for loss of
goodwill, work stoppage, computer failure or malfunction, loss of data or any
commercial damage, even if the Licensor has been advised of the possibility
of such damage. However, the Licensor will be liable under statutory product
liability laws as far such laws apply to the Work.
9. Additional agreements
While distributing the Work, You may choose to conclude an additional
agreement, defining obligations or services consistent with this Licence.
However, if accepting obligations, You may act only on your own behalf and on
your sole responsibility, not on behalf of the original Licensor or any other
Contributor, and only if You agree to indemnify, defend, and hold each
Contributor harmless for any liability incurred by, or claims asserted
against such Contributor by the fact You have accepted any warranty or
additional liability.
10. Acceptance of the Licence
The provisions of this Licence can be accepted by clicking on an icon 'I
agree' placed under the bottom of a window displaying the text of this
Licence or by affirming consent in any other similar way, in accordance with
the rules of applicable law. Clicking on that icon indicates your clear and
irrevocable acceptance of this Licence and all of its terms and conditions.
Similarly, you irrevocably accept this Licence and all of its terms and
conditions by exercising any rights granted to You by Article 2 of this
Licence, such as the use of the Work, the creation by You of a Derivative
Work or the Distribution or Communication by You of the Work or copies
thereof.
11. Information to the public
In case of any Distribution or Communication of the Work by means of
electronic communication by You (for example, by offering to download the
Work from a remote location) the distribution channel or media (for example,
a website) must at least provide to the public the information requested by
the applicable law regarding the Licensor, the Licence and the way it may be
accessible, concluded, stored and reproduced by the Licensee.
12. Termination of the Licence
The Licence and the rights granted hereunder will terminate automatically
upon any breach by the Licensee of the terms of the Licence.
Such a termination will not terminate the licences of any person who has
received the Work from the Licensee under the Licence, provided such persons
remain in full compliance with the Licence.
13. Miscellaneous
Without prejudice of Article 9 above, the Licence represents the complete
agreement between the Parties as to the Work.
If any provision of the Licence is invalid or unenforceable under applicable
law, this will not affect the validity or enforceability of the Licence as a
whole. Such provision will be construed or reformed so as necessary to make
it valid and enforceable.
The European Commission may publish other linguistic versions or new versions
of this Licence or updated versions of the Appendix, so far this is required
and reasonable, without reducing the scope of the rights granted by the
Licence. New versions of the Licence will be published with a unique version
number.
All linguistic versions of this Licence, approved by the European Commission,
have identical value. Parties can take advantage of the linguistic version of
their choice.
14. Jurisdiction
Without prejudice to specific agreement between parties,
- any litigation resulting from the interpretation of this License, arising
between the European Union institutions, bodies, offices or agencies, as a
Licensor, and any Licensee, will be subject to the jurisdiction of the
Court of Justice of the European Union, as laid down in article 272 of the
Treaty on the Functioning of the European Union,
- any litigation arising between other parties and resulting from the
interpretation of this License, will be subject to the exclusive
jurisdiction of the competent court where the Licensor resides or conducts
its primary business.
15. Applicable Law
Without prejudice to specific agreement between parties,
- this Licence shall be governed by the law of the European Union Member
State where the Licensor has his seat, resides or has his registered
office,
- this licence shall be governed by Belgian law if the Licensor has no seat,
residence or registered office inside a European Union Member State.
Appendix
'Compatible Licences' according to Article 5 EUPL are:
- GNU General Public License (GPL) v. 2, v. 3
- GNU Affero General Public License (AGPL) v. 3
- Open Software License (OSL) v. 2.1, v. 3.0
- Eclipse Public License (EPL) v. 1.0
- CeCILL v. 2.0, v. 2.1
- Mozilla Public Licence (MPL) v. 2
- GNU Lesser General Public Licence (LGPL) v. 2.1, v. 3
- Creative Commons Attribution-ShareAlike v. 3.0 Unported (CC BY-SA 3.0) for
works other than software
- European Union Public Licence (EUPL) v. 1.1, v. 1.2
- Québec Free and Open-Source Licence — Reciprocity (LiLiQ-R) or Strong
Reciprocity (LiLiQ-R+).
The European Commission may update this Appendix to later versions of the
above licences without producing a new version of the EUPL, as long as they
provide the rights granted in Article 2 of this Licence and protect the
covered Source Code from exclusive appropriation.
All other changes or additions to this Appendix require the production of a
new EUPL version.

View File

@@ -2,6 +2,18 @@
[See here](doc/manual/testium_manual.pdf).
# License
Copyright (c) 2025-2026 François Dausseur.
testium is distributed under the **European Union Public Licence v. 1.2
(EUPL-1.2)** — see the [LICENSE](LICENSE) file for the full text.
SPDX identifier: `EUPL-1.2`
Contributions are accepted under the same licence (inbound = outbound). See
[CONTRIBUTING.md](CONTRIBUTING.md) for details.
# run testium
From the root path, on windows `cmd`:
@@ -52,7 +64,7 @@ from the testium path, execute
## Install sphinx
pip install sphinx
pip install sphinx linuxdoc
## Generate the doc

View File

@@ -1,4 +1,4 @@
- unittest_file:
- unittest:
name: Test 5
test_file: dummy.py

View File

@@ -3,7 +3,7 @@ sequence: &endurance_test
!include endurance.tum
sequence:
- unittest_file:
- unittest:
name: Test 3
test_file: dummy.py
test_method: test_01_pass
@@ -11,6 +11,6 @@ sequence:
iterator: 10
steps:
*endurance_test
- unittest_file:
- unittest:
name: Test 4
test_file: dummy.py

View File

@@ -15,7 +15,7 @@ main:
- $(reference_1)
- $(reference_2)
report_show_success: true
- unittest_file:
- unittest:
name: Test 1
test_file: dummy.py
doc: |
@@ -23,7 +23,7 @@ main:
Voilà...
- sleep:
{name: Sleep between one and two, timeout: 10, dialog: true}
- unittest_file:
- unittest:
{name: Test 2, test_file: dummy.py,execute_on_stop: true}
- loop:
name: Cycle Temperature

View File

@@ -14,7 +14,7 @@ main:
key: report-key-2
stop_on_failure: True
steps:
- unittest_file:
- unittest:
name: unittest item
doc: |
The purpose of this unittest test item is to demonstrate
@@ -41,7 +41,7 @@ main:
param:
- 123
- unittest_file:
- unittest:
name: Unittest item
test_file: dummy/dummy.py
test_method:
@@ -98,7 +98,7 @@ main:
name: Infine loop unittest step crashes
stop_on_failure: True
steps:
- unittest_file:
- unittest:
name: Unittest item
test_file: dummy/dummy.py
test_method:
@@ -243,7 +243,7 @@ main:
name: Infinite loop
skipped: True
steps:
- unittest_file:
- unittest:
name: Unittest item
test_file: dummy/dummy.py
test_method: test_01_pass

View File

@@ -9,7 +9,7 @@ main:
name: Test Sample number one
version: 0.1
steps:
- unittest_file:
- unittest:
name: Unittest item
test_file: dummy/unittest_str.py
doc: Unittest test
@@ -88,7 +88,7 @@ main:
name: cycle item
iterator : 3
steps:
- unittest_file:
- unittest:
name: Unittest item
test_file: dummy/dummy.py
test_method: test_01_pass
@@ -99,7 +99,7 @@ main:
name: cycle item
iterator : 3
steps:
- unittest_file:
- unittest:
name: Unittest item
test_file: dummy/dummy.py
test_method: test_01_pass

View File

@@ -3,7 +3,7 @@ Command Line Interface
.. code-block:: text
usage: testium.pyw [-h] [--version] [-b] [-m] [-c CONFIG_FILE [CONFIG_FILE ...]] [-r] [-l LOG_FILE]
usage: testium.pyw [-h] [--version] [-b] [-c CONFIG_FILE [CONFIG_FILE ...]] [-r] [-l LOG_FILE]
[-d DEFINE [DEFINE ...]] [-p REPORT_FILE] [-t {sqlite,json,junit,html,text}]
[-n REPORT_PATTERN [REPORT_PATTERN ...]] [-i INCLUDE_PATH [INCLUDE_PATH ...]] [-o] [-g]
[test_file]
@@ -16,9 +16,8 @@ Command Line Interface
--version Returns the version of testium
-b, --batch-execution
Executes the test in batch mode
-m, --terminal Starts terminal mode
-c CONFIG_FILE [CONFIG_FILE ...], --config-file CONFIG_FILE [CONFIG_FILE ...]
-o, --no-color Deactivates stdout colors in batch and terminal mode
-o, --no-color Deactivates stdout colors in batch mode
Configuration file
-r, --run-and-close Runs the test then closes the application
-l LOG_FILE, --log-file LOG_FILE
@@ -45,17 +44,10 @@ Returns what's in the previous section.
Executes the test in text mode. No need to have QT installed in that case.
``-m, --terminal``
------------------
Starts a testium interactive console. It allows to run commands and sub-tests manually
in a console.
``-o, --no-color``
------------------
Switch allowing to disable the colored output in terminal or batch modes.
Switch allowing to disable the colored output in batch mode.
``-c, --config-file``
---------------------

View File

@@ -14,6 +14,7 @@ import os
import sys
sys.path.insert(0, os.path.abspath('../../../../src/testium/'))
sys.path.insert(0, os.path.abspath('../../../../src/'))
# -- Project information -----------------------------------------------------

View File

@@ -21,7 +21,7 @@ Global variables helper functions
To manage values in the global variables dataset, the following testium library API
must be used:
.. automodule:: interpreter.utils.globdict
.. automodule:: py_func.tm
:members: gd, setgd, delgd
:undoc-members:
:no-index:

View File

@@ -23,23 +23,3 @@ graphical interface.
:caption: call a test in batch mode
testium -b test/my_test/main.tum
Terminal mode
-------------
The terminal mode starts *testium* in interactive mode. From this console, some tests and
sequences of tests can be called interactively.
.. code-block:: text
:caption: call a test in terminal mode
$ testium -m
Configuration file loaded: /my/execution/path/param.yaml
[...]
================================================================================
====== Test configuration
================================================================================
Test executed with testium : 2.4.0 (binary release)
(testium)~

View File

@@ -75,7 +75,7 @@ a tooltip on the test row.
name: Test example
steps:
- unittest_file:
- unittest:
name: unittest item
doc: |
The purpose of this unittest test item is to demonstrate
@@ -93,4 +93,4 @@ See illustration in :numref:`Figure %s<doc-illustration>`.
Unittest
^^^^^^^^^
For ``unittest_file`` type test items, the python docstring of the test method is used as documentation.
For ``unittest`` type test items, the python docstring of the test method is used as documentation.

View File

@@ -6,7 +6,7 @@ This software is developed in python and it implements the Qt6 graphical framewo
It has been developed since 2013 with production and development testing in mind.
It's function is to automate the execution of tests. It can be invoked either as command line terminal application or as a graphical interface application.
It's function is to automate the execution of tests. It can be invoked either as command line application or as a graphical interface application.
Tests reports generation and customization are also in this tool's scope.

View File

@@ -10,7 +10,7 @@ This element is of the following form:
name: Group Item
condition: <| "$(OS)" == "Linux" |>
steps:
- unittest_file:
- unittest:
test_file: test_prod_alpha_13.py
test_method:
...

View File

@@ -33,7 +33,7 @@ if not provided is given in the table as well.
| | | It depends on the test item to take it |
| | | into account or not. |
| | | For example it makes sense to use it |
| | | for ``unittest_file`` test type |
| | | for ``unittest`` test type |
| | | because it can contain many sub-tests, |
| | | but not for sleep test type. |
| | | In cycles, it means that the child |
@@ -87,6 +87,10 @@ if not provided is given in the table as well.
| | | see :ref:`Expected result<sec_expected_result>` |
| | | for details. |
+-----------------------+-------------------+-------------------------------------------------------+
| ``store_result`` | / | Store the test result in a global variable. |
| | | see :ref:`Store result<sec_store_result>` |
| | | for details. |
+-----------------------+-------------------+-------------------------------------------------------+
last test result
@@ -183,6 +187,61 @@ If the result and the expected_result is equal, the test will be *PASSED* if ``T
The special ``$(result)`` variable is replaced in the ``expected_result`` attribute content with the test result value.
.. _sec_store_result:
Store result
-----------------------------------------------
The ``store_result`` attribute stores the test result into a named global variable,
making it available to subsequent test items via ``$(variable_name)``.
If the test item returns a value (e.g. ``py_func``, ``json_rpc``), that value is stored.
If ``process_result`` is also specified, the stored value is the post-processed result.
If the test item produces no value (result is ``None``), the stored value is the
test status string: ``"PASS"`` or ``"FAIL"``, evaluated after ``expected_result``
but **before** ``no_fail``. This ensures the real outcome is captured even when
``no_fail: True`` would otherwise mask a failure.
.. code-block:: yaml
:caption: Store a function return value
- py_func:
name: Read sensor
func_name: read_temperature
store_result: temperature
- py_func:
name: Check temperature in range
func_name: check_range
param: [$(temperature), 20, 30]
.. code-block:: yaml
:caption: Store a post-processed value
- py_func:
name: Get firmware version string
func_name: get_version
process_result: "'$(result)'.split('.')[0]"
store_result: fw_major
.. code-block:: yaml
:caption: Store the pass/fail status of a test with no return value
- console:
name: Send command
console_name: device
steps:
- writeln: reboot
- read_until: {expected: "ready", timeout: 10}
store_result: reboot_status
- py_func:
name: Use reboot status
func_name: log_status
param: [$(reboot_status)]
Export attribute
-----------------------------------------------

View File

@@ -12,7 +12,7 @@ This element is of the following form:
name: Cycle Temperature
iterator: 10
steps:
- unittest_file:
- unittest:
test_file: test_prod_rio6_8093.py
- py_func:
name: function test item

View File

@@ -39,11 +39,14 @@ The ``lua_func`` test item is of the form:
Beside common test items attributes, lua_func item has specific attribute, some of which being mandatory.
* ``file``: the script file name that contains the function to be executed.
Only python script format is supported.
Only Lua script format is supported.
* ``func_name``: The function name to be executed.
* ``param``: This is a list of parameters that are passed to the function
in the order they are presented in the script. These parameters are not
mandatory and are highly dependent of the function prototype.
* ``context_id``: Optional. When set, all ``lua_func`` items sharing the same
``context_id`` value run inside the same persistent Lua subprocess for the
duration of the test. See :ref:`lua_func context<sec_lua_func_context>` for details.
.. code-block:: yaml
:caption: ``lua_func`` test item example of usage
@@ -56,16 +59,71 @@ Beside common test items attributes, lua_func item has specific attribute, some
- $(my_param)
The result of the function (after eventual post treatment) is stored in the global
variable named ``pfn_<func_name>``
variable named ``lfn_<item_name>``
(See :ref:`global variables<sec_global_variables>` for more detail
on how to access to global variables from test items and scripts).
In the example above, the global variable ``$(lfn_activity)``
would be created at the end of the item execution. It would contain the resulting
value of the funcToBeExecuted python function.
value of the methodName Lua function.
The ``lua_func`` will always result ``PASS``, except if the called function raises
and exception or if the ``expected_result`` attribute is used.
an exception or if the ``expected_result`` attribute is used.
.. _sec_lua_func_context:
Sharing state between ``lua_func`` calls
------------------------------------------
Each ``lua_func`` item without a ``context_id`` runs in a dedicated subprocess that
is started and stopped around the call. Module-level variables are not preserved
between two such items.
Inside a ``lua_func`` script, the ``tm`` module exposes ``tm.setgd`` and ``tm.gd``
to read and write the testium global dictionary of the test process. Values stored
this way are accessible from any subsequent test item without requiring a shared
subprocess.
.. code-block:: lua
:caption: sharing a value via the global dictionary
local tm = require("tm")
local module = {}
function module.produce(val)
tm.setgd("my_shared_value", val)
return val
end
function module.consume()
return tm.gd("my_shared_value")
end
return module
When ``context_id`` is set, all ``lua_func`` items that share the same identifier
reuse the same persistent subprocess. This allows Lua-side state (upvalues, module
cache) to be retained across calls beyond what ``tm.setgd`` persists.
.. code-block:: yaml
:caption: ``lua_func`` items sharing a persistent subprocess
- lua_func:
name: produce value
file: my_script.lua
func_name: produce
context_id: my_context
param:
- hello
- lua_func:
name: consume value
file: my_script.lua
func_name: consume
context_id: my_context
expected_result: hello
The shared subprocess is automatically stopped at the end of the test run.
**Lua Interpreter environment setup**

View File

@@ -0,0 +1,97 @@
.. _sec_parallel_item:
**parallel** test item
============================================================
This element is of the following form:
.. code-block:: yaml
:caption: ``parallel`` test item usage example
- parallel:
name: My parallel block
sync: all
branches:
- name: Branch A
steps:
- py_func:
name: Long operation
file: long_op.py
func_name: do_work
- name: Branch B
wait_for:
condition: <| "$(ready_flag)" == "True" |>
timeout: 30
steps:
- let:
name: Mark done
values:
- branch_b_done: true
The ``parallel`` element runs several sequences of items concurrently. Each
inner sequence is called a *branch* and runs in its own thread. The parent
test item waits for branches to finish according to the ``sync`` policy.
Attributes
--------------------
* ``branches``: required. A list of branches to execute concurrently. Each
branch has a ``name`` and a ``steps`` list (same structure as a ``group``
item). It can also declare a ``wait_for`` precondition (see below).
* ``sync``: optional, defaults to ``all``.
* ``all``: the parallel item completes when *every* branch has finished.
The result is ``PASS`` if no branch returned ``FAIL`` (skipped or
disabled branches are ignored, like in ``group``); otherwise ``FAIL``.
* ``any``: the parallel item completes as soon as the *first* branch
finishes. The remaining branches are stopped (their next test items
are not executed). The result is ``PASS`` if at least one branch
succeeded.
* ``no_fail``: optional. When ``true``, a ``FAIL`` result is forced to
``PASS`` for the parallel item itself (same semantics as for any test
item). Branches keep their own result.
Branch attributes
--------------------
Each entry of ``branches`` is a dict with the following attributes:
* ``name``: required. The branch name. Used in reports and as a prefix
in the live log output (each line printed by the branch is prefixed
with the branch name in square brackets, e.g. ``[Branch A]``, so
concurrent branches stay readable).
* ``steps``: required. The list of test items executed sequentially
inside the branch.
* ``wait_for``: optional. Forces the branch to wait until a condition is
met before running its steps. If the timeout elapses, the branch
returns ``FAIL`` (the steps are not run). Sub-attributes:
* ``condition``: a testium expression evaluated repeatedly (every
100 ms) until it returns ``True``.
* ``timeout``: maximum wait, in seconds. Defaults to 30.
Reporting
--------------------
Each branch produces its own row in the SQLite report (with type
``Parallel branch``), in addition to the parent ``Parallel`` row. The
``log`` column of each row contains only the output emitted from that
branch's thread, so logs are never mixed between concurrent branches.
In the live (terminal / GUI) output, lines emitted from a branch are
prefixed with the branch name in square brackets (e.g. ``[Branch A]``).
The prefix is not stored in the SQLite log column.
Notes
--------------------
* A ``sleep`` item inside a branch is interruptible: if another
``sync: any`` branch wins the race, slow ``sleep`` items are aborted
within ~50 ms.
* A ``py_func`` or ``console`` item inside a branch is **not**
interruptible: a ``sync: any`` stop will only take effect after the
current item returns. The branch will then skip its remaining steps.
* When a user disables a branch in the GUI tree, the branch returns
``SKIP`` instantly without affecting the others (it does *not* win a
``sync: any`` race).

View File

@@ -89,6 +89,9 @@ some of which being mandatory.
* ``param``: This is a list of parameters that are passed to the function
in the order they are presented in the script. These parameters are not
mandatory and are highly dependent of the function prototype.
* ``context_id``: Optional. When set, all ``py_func`` items sharing the same
``context_id`` value run inside the same persistent Python subprocess for the
duration of the test. See :ref:`py_func context<sec_py_func_context>` for details.
.. code-block:: yaml
:caption: ``py_func`` test item example of usage
@@ -111,6 +114,67 @@ value of the funcToBeExecuted python function.
The ``py_func`` will always result ``PASS``, except if the called function raises
and exception or if the ``expected_result`` attribute is used.
.. _sec_py_func_context:
Sharing state between ``py_func`` calls
------------------------------------------
Each ``py_func`` item without a ``context_id`` runs in a dedicated subprocess that
is started and stopped around the call. State cannot be shared between two such
items using module-level variables.
Inside a ``py_func`` script, ``tm.setgd`` and ``tm.gd`` read and write the testium
global dictionary. Values stored this way are accessible from any subsequent test
item, including other ``py_func`` items, without requiring a shared subprocess.
.. code-block:: python
:caption: sharing a value via the global dictionary
import py_func.tm as tm
def produce(val):
tm.setgd("my_shared_value", val)
return val
def consume():
return tm.gd("my_shared_value", None)
When ``context_id`` is set, all ``py_func`` items that share the same identifier
reuse the same persistent subprocess. This allows sharing any Python object across
calls — including objects that cannot be transmitted to other processes.
.. code-block:: python
:caption: sharing an object via ``context_id``
import py_func.tm as tm
def open_connection():
tm.setgd("conn", MyConnection())
return "ok"
def use_connection():
conn = tm.gd("conn")
return conn.status()
.. code-block:: yaml
:caption: ``py_func`` items sharing a persistent subprocess
- py_func:
name: open connection
file: my_script.py
func_name: open_connection
context_id: my_context
expected_result: ok
- py_func:
name: use connection
file: my_script.py
func_name: use_connection
context_id: my_context
expected_result: open
The shared subprocess is automatically stopped at the end of the test run.
**Python Interpreter environment setup**
Some global variables have an impact on the ``py_func`` test item behavior:

View File

@@ -1,16 +1,23 @@
**run** test item
============================================================
This test item executes a new instance of testium.
This test item executes a new instance of testium with the specified ``.tum`` file.
* In **batch mode** (``-b``): the sub-instance is started with ``-b``.
* In **GUI mode**: the sub-instance is started with ``-r`` (run and close).
The item result is **PASS** if the sub-instance launched and ran to completion,
regardless of whether the sub-tests passed or failed.
It is **FAIL** if the file could not be found, the sub-instance could not be
launched, or the time window was not reached (see ``start_time`` / ``end_time``).
.. code-block:: yaml
:caption: ``run`` test item usage example
- run:
name: Execute TUM
tum_fime: example_cycle.tum
tum: example_cycle.tum
python_bin: python3
testium_path: /home/francois/projets/testium-new-report/testium.pyw
log_file: $(home)/reports/test.log
report_file: $(home)/reports/test.rep
@@ -19,12 +26,12 @@ Attributes
run test item has the following specific attributes:
* ``tum_fime``: mandatory the path of the file to execute, it can be relative to current execution folder,
* ``param_file`` (optional) the path of the parameter file to use, otherwise default parameter file is used.
* ``python_bin`` (optional) the path of a specific python to run your scripts,
* ``testium_path`` (optional) the path of a specific testium to run your scripts,
* ``log_file`` (optional) the path of log file to register, if not provided a file is created with timestamp at the location of TUM file.
* ``report_file`` (optional), the path of report file to create
* ``start_time`` (optional), start time for the script execution, in HH:MM format.
* ``end_time`` (optional), end time for an execution within a time frame, in HH:MM format.
* ``wait_for_exec`` (optional). True or False, wait to be in the execution window defined by start_time and end_time to run the script.
* ``tum``: mandatory, the path of the file to execute. Can be relative to the current execution folder.
* ``param_file`` (optional): the path of the parameter file to use; otherwise the default parameter file is used.
* ``python_bin`` (optional): the path of a specific Python interpreter to use.
* ``testium_path`` (optional): the path of a specific testium executable to use.
* ``log_file`` (optional): the path of the log file. In GUI mode, if not provided, a file is created with a timestamp next to the ``.tum`` file. Not used in batch mode.
* ``report_file`` (optional): the path of the report file to create.
* ``start_time`` (optional): earliest time to execute the sub-instance, in ``HH:MM`` format.
* ``end_time`` (optional): latest time for execution within a time frame, in ``HH:MM`` format.
* ``wait_for_exec`` (optional): ``true`` to wait until the time window defined by ``start_time`` and ``end_time`` is reached before running. Requires both ``start_time`` and ``end_time``.

View File

@@ -1,15 +1,15 @@
**unittest_file** test item
**unittest** test item
============================================================
unittest_file test item allows the execution of unittest test script which
unittest test item allows the execution of unittest test script which
is part of python standard libraries.
The tum file prototype is as followed:
.. code-block:: yaml
:caption: ``unittest_file`` test item usage example
:caption: ``unittest`` test item usage example
- unittest_file:
- unittest:
name: unitTest test item
test_file: unitTestScript.py
test_method:
@@ -23,7 +23,7 @@ Beside common test items attributes, unittest test item has specific attribute,
* ``test_file``: it is the name (and eventually path) of the unittest file
to be processed.
* ``test_method``: it is an optional unittest_file test sub-item. If one or more
* ``test_method``: it is an optional unittest test sub-item. If one or more
elements are present, the unittest python script file is parsed and only
the corresponding methods are included in the test tree. Otherwise, all
the test methods are included in the test tree.
@@ -78,4 +78,4 @@ Here is an example how to use the console module from python ``unittest``.
@classmethod
def tearDownClass(cls):
cls.consA0.close()
cls.consA0.close()

View File

@@ -255,11 +255,12 @@ step list attributes.
test_items/let_test_item.rst
test_items/loop_test_item.rst
test_items/lua_func_test_item.rst
test_items/parallel_test_item.rst
test_items/plot_test_item.rst
test_items/report_test_item.rst
test_items/run_test_item.rst
test_items/sleep_test_item.rst
test_items/unittest_file_test_item.rst
test_items/unittest_test_item.rst

Binary file not shown.

View File

@@ -4,6 +4,7 @@ SUPPORTED_API = [
"setgd",
"delgd",
"add_plot_values",
"last_plot_value"
"last_plot_value",
"text_mode",
]

View File

@@ -228,7 +228,7 @@ class JsonRpcConnection:
self.recv_thread.join()
def is_alive(self):
self.recv_thread.is_alive()
return self.recv_thread.is_alive()
def wait_ready(self, timeout=None):
return self._event_ready.wait(timeout)

View File

@@ -1,8 +1,88 @@
import sys
import threading
from threading import (Thread, Event)
from lib.string_queue import StringQueue
from time import (sleep)
class StdoutProxy:
"""Thread-aware stdout proxy.
Each writing thread can be associated with:
- a per-thread buffer (StringQueue) where its writes are captured for the
per-item SQLite log column;
- a 'branch' label, used to prefix each line in the live (parent-visible)
output stream so concurrent branches are easy to read.
Threads with no association fall back to the default buffer (the "main"
thread's buffer) and write to live output without prefix.
"""
def __init__(self, live_stream, default_buffer):
self.live_stream = live_stream
self.default_buffer = default_buffer
self._buffers = {}
self._branches = {}
self._lock = threading.Lock()
def register(self, tid=None, buffer=None, branch=None):
if tid is None:
tid = threading.get_ident()
with self._lock:
if buffer is not None:
self._buffers[tid] = buffer
if branch is not None:
self._branches[tid] = branch
def unregister(self, tid=None):
if tid is None:
tid = threading.get_ident()
with self._lock:
self._buffers.pop(tid, None)
self._branches.pop(tid, None)
def get_buffer(self, tid=None):
if tid is None:
tid = threading.get_ident()
with self._lock:
return self._buffers.get(tid, self.default_buffer)
def write(self, s):
if not s:
return
tid = threading.get_ident()
with self._lock:
buf = self._buffers.get(tid, self.default_buffer)
branch = self._branches.get(tid)
# Per-thread capture: clean, no prefix
buf.write(s)
# Live stream: prefix each line with the branch label
if branch:
self.live_stream.write(self._prefix(s, f'[{branch}] '))
else:
self.live_stream.write(s)
@staticmethod
def _prefix(s, prefix):
ends_nl = s.endswith('\n')
body = s[:-1] if ends_nl else s
if body == '':
return s
prefixed = '\n'.join(prefix + line for line in body.split('\n'))
if ends_nl:
prefixed += '\n'
return prefixed
def writeln(self, s=''):
self.write(s + '\n')
def flush(self):
try:
self.live_stream.flush()
except AttributeError:
pass
class StdioRedirect:
def __init__(self):
@@ -28,48 +108,38 @@ class StdioRedirect:
def intercept(self):
if not self.spy_enabled:
self.thr_started = Event()
self.log_buf = StringQueue()
self.in_stream = StringQueue()
self.stop_output = Event()
self.thrd_out = Thread(target=self.interceptStdOut)
self.thrd_out.daemon = True
sys.stdout = self.in_stream
sys.stderr = self.in_stream
self.stream = self.in_stream
self.thrd_out.start()
self.thr_started.wait()
self.log_buf = StringQueue() # default buffer (main thread)
self.proxy = StdoutProxy(self.out_stream, self.log_buf)
sys.stdout = self.proxy
sys.stderr = self.proxy
self.stream = self.proxy
self.spy_enabled = True
def stop(self):
if self.spy_enabled:
sys.stdout = self.out_stream
sys.stderr = self.out_stream
self.stream = self.out_stream
self.stop_output.set()
self.thrd_out.join()
del self.log_buf
del self.in_stream
del self.stop_output
del self.thrd_out
del self.thr_started
del self.proxy
self.spy_enabled = False
def interceptStdOut(self):
self.thr_started.set()
while not self.stop_output.is_set():
data = self.in_stream.read()
self.log_buf.write(data)
self.out_stream.write(data)
if data == '':
sleep(0.1)
def read(self):
ret = ''
"""Read accumulated content from the calling thread's buffer."""
if not self.spy_enabled:
return ''
return self.proxy.get_buffer().read()
def register_thread(self, buffer=None, branch=None):
"""Register the calling thread's per-thread buffer and/or branch label."""
if self.spy_enabled:
ret = self.log_buf.read()
return ret
self.proxy.register(buffer=buffer, branch=branch)
def unregister_thread(self):
"""Drop the calling thread's registration."""
if self.spy_enabled:
self.proxy.unregister()
stdio_redir = StdioRedirect()

View File

@@ -1,8 +1,5 @@
# from io import (StringIO, SEEK_SET, SEEK_CUR, SEEK_END)
from multiprocessing import Queue
from queue import (Empty)
from threading import (Thread, Event, Condition)
from threading import Lock
class StringQueue(object):
""" Class used to store the buffered consoles data:

View File

@@ -1,5 +1,6 @@
import traceback
import textwrap
from contextlib import contextmanager
class ETUMError(Exception):
@@ -67,6 +68,28 @@ class ETUMParamError(ETUMError):
return lines
@contextmanager
def item_load_context(item_type: str, item_name: str, filename: str = ""):
"""Context manager that enriches ETUMSyntaxError with item context during loading.
Usage in test item __init__:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.param = self._prms.getParam("param", required=True)
"""
try:
yield
except ETUMSyntaxError as e:
raise ETUMSyntaxError(
f"In '{item_type}' item named '{item_name}':\n{e._message}",
filename or e._file,
) from e
except Exception as e:
raise ETUMSyntaxError(
f"In '{item_type}' item named '{item_name}':\nUnexpected error: {e}",
filename,
) from e
def print_exception(exc: ETUMError):
if not isinstance(exc, ETUMError):
print(traceback.format_exc(4))

View File

@@ -1,19 +1,26 @@
import json
import sys
from py_func.handle import FuncHandler
from lib.tum_except import ETUMRuntimeError
from lib.api import SUPPORTED_API
thismodule = sys.modules[__name__]
# Shared FuncHandler instance used to forward API calls. Remains None
# until `_init_api` is invoked.
_func_call_thread = None
# Local storage for non-JSON-serializable values
_local_dict = {}
def _is_json_serializable(value):
try:
json.dumps(value)
return True
except (TypeError, ValueError):
return False
###############################################################################
# Dynamically create module-level functions for each supported API name.
# Each generated function shares the implementation of `api_call` but
# has a distinct name used as the remote action identifier.
def _make_api(name):
def _wrapper(*params):
if _func_call_thread is not None:
@@ -31,21 +38,86 @@ def _make_api(name):
return _wrapper
for k in SUPPORTED_API:
setattr(thismodule, k, _make_api(k))
if k not in ('gd', 'setgd', 'delgd'):
setattr(thismodule, k, _make_api(k))
###############################################################################
# gd/setgd/delgd with local-dict fallback for non-serializable values
def gd(name, default=None):
"""Return a value from the testium global dictionary.
The value is accessible from any test item and from any ``py_func``
subprocess, regardless of the ``context_id`` used.
:param name: Name of the entry to retrieve.
:type name: str
:param default: Value returned when the key is absent. Defaults to ``None``.
:return: The stored value, or *default* if not found.
"""
if name is not None and name in _local_dict:
return _local_dict[name]
if _func_call_thread is not None:
res = _func_call_thread.call("gd", (name, default))
if "result" in res:
return res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.gd' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
raise ETUMRuntimeError("api not initialized")
def setgd(name, value):
"""Store a value in the testium global dictionary.
The stored value is accessible from any subsequent test item and from any
``py_func`` subprocess via :func:`gd`.
When ``context_id`` is used on the ``py_func`` item, any Python object
(including those that cannot be transmitted to other processes) can be
stored and shared between calls running in the same subprocess.
:param name: Name of the entry to set.
:type name: str
:param value: Value to store.
"""
if name is not None and not _is_json_serializable(value):
_local_dict[name] = value
return None
if _func_call_thread is not None:
res = _func_call_thread.call("setgd", (name, value))
if "result" in res:
return res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.setgd' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
raise ETUMRuntimeError("api not initialized")
def delgd(name):
"""Remove an entry from the testium global dictionary.
:param name: Name of the entry to remove.
:type name: str
"""
if name is not None and name in _local_dict:
del _local_dict[name]
return None
if _func_call_thread is not None:
res = _func_call_thread.call("delgd", (name,))
if "result" in res:
return res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.delgd' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
raise ETUMRuntimeError("api not initialized")
def _init_api(host, port, timeout):
"""Start and initialize the remote function handler.
Starts a ``FuncHandler`` bound to ``port``, runs it and blocks until
it signals readiness.
Args:
port: port number or identifier passed to ``FuncHandler``.
Returns:
The initialized ``FuncHandler`` instance assigned to
``_func_call_thread``.
"""
global _func_call_thread
_func_call_thread = FuncHandler(host, port, timeout=timeout)
return _func_call_thread
@@ -53,17 +125,10 @@ def _init_api(host, port, timeout):
###############################################################################
def _remote_print(*values):
"""Forward print-like output to the remote handler.
If a ``_func_call_thread`` is available, this function calls the
handler with action name ``"print"`` and the provided values. Errors
during forwarding are ignored because printing is best-effort.
"""
if _func_call_thread is not None:
try:
_func_call_thread.call("print", values)
except:
# Best-effort: ignore forwarding failures
pass

View File

@@ -8,9 +8,12 @@ requires-python = ">=3.11"
authors = [
{name = "François Dausseur", email = "francois@beafrancois.fr"},
]
license = "EUPL-1.2"
license-files = ["../LICENSE"]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Programming Language :: Python"
"Programming Language :: Python",
"License :: OSI Approved :: European Union Public Licence 1.2 (EUPL 1.2)",
]
dependencies = [
"setuptools",

View File

@@ -21,10 +21,8 @@ def main():
help="Returns the version of testium", action='store_true')
parser.add_argument("-b", "--batch-execution",
help="Executes the test in batch mode", action='store_true')
parser.add_argument("-m", "--terminal",
help="Starts terminal mode", action='store_true')
parser.add_argument("-o", "--no-color",
help="Deactivates stdout colors in batch and terminal mode", action='store_true')
help="Deactivates stdout colors in batch mode", action='store_true')
parser.add_argument("-c", "--config-file", help="Configuration file",
nargs='+',
default=[])
@@ -95,36 +93,13 @@ def main():
from interpreter.utils.version import get_testium_version
print(get_testium_version())
elif args.terminal:
import select
from interpreter.terminal import Terminal
if (lf != '') or (rf != '') or (tf != '') or (pn != []):
print('"-l", "-p", "-t", "-n" options are not supported in this mode.')
t = Terminal(os.getcwd(), cf, defines, args.no_color)
loop = 1
while loop:
try:
loop = 0
t.cmdloop()
except KeyboardInterrupt:
print("\n<ctrl-c>")
loop = 1
except Exception as exc:
if str(exc) == 'quit':
break
print(exc)
loop = 1
elif args.batch_execution:
if (lf != ''):
print('"-l" option is not supported in this mode.')
from interpreter.batch import Batch
b = Batch(tf, cf, defines, rf, args.report_type, pn, args.no_color)
b = Batch(tf, cf, defines, rf, args.report_type, pn, args.no_color, text_mode=True)
sys.exit(0 if b.success else 1)
else:
from main_win.testium_win import MainWin

View File

@@ -1,6 +1,7 @@
import os
import sys
import platform
import threading
from time import sleep
from signal import signal, SIGINT
from queue import Empty
@@ -8,7 +9,7 @@ from multiprocessing import Queue
from interpreter.process import TestProcess
from interpreter.utils.test_ctrl import TestSetController
from lib.tum_except import ETUMFileError
from lib.tum_except import ETUMFileError, ETUMRuntimeError
from lib.stdout_redirect import stdio_redir
@@ -22,6 +23,7 @@ class Batch:
report_type,
report_pattern,
no_color,
text_mode=False,
):
try:
try:
@@ -51,6 +53,7 @@ class Batch:
signal(SIGINT, self.sigint_handler)
self._success = False
msg_queue = Queue()
self.tst_ctrl = TestSetController()
tst_proc = TestProcess(
@@ -59,11 +62,21 @@ class Batch:
self.tst_ctrl,
config_files,
defines,
text_mode=text_mode,
)
tst_proc.start()
while not self.tst_ctrl.control("loaded"):
sleep(0.1)
# Wait for TestProcess to finish loading.
# Run the blocking control("loaded") in a daemon thread so we
# can watch for unexpected process death in the main thread.
_loaded_event = threading.Event()
def _wait_loaded():
self.tst_ctrl.control("loaded")
_loaded_event.set()
threading.Thread(target=_wait_loaded, daemon=True).start()
while not _loaded_event.wait(timeout=0.1):
if not tst_proc.is_alive():
raise ETUMRuntimeError("TestProcess terminated unexpectedly during load")
self.tst_ctrl.control(
"report",
@@ -78,14 +91,18 @@ class Batch:
while True:
try:
m = msg_queue.get(timeout=0.2)
if m.get("id", None) is None:
# No id -> finished
if "id" in m and m["id"] is None:
# id key present and None -> finished
self._success = m.get("success", False)
break
except Empty:
if not tst_proc.is_alive():
break
continue
# Close the process and wait for termination
self.tst_ctrl.control("close")
if tst_proc.is_alive():
self.tst_ctrl.control("close")
tst_proc.join()
except Exception as e:
@@ -94,5 +111,12 @@ class Batch:
finally:
stdio_redir.restore()
@property
def success(self):
return self._success
def sigint_handler(self, signal_received, frame):
self.tst_ctrl.control("stop")
try:
self.tst_ctrl.control("stop", timeout=5)
except Exception:
pass

View File

@@ -1,4 +1,5 @@
import os
import signal
from multiprocessing import Process, Queue, Pipe
from queue import Empty
from threading import Thread
@@ -8,6 +9,7 @@ import copy
from lib.string_queue import StringQueue
from lib.tum_except import print_exception, ETUMRuntimeError, ETUMSyntaxError
import libs.testium as tm
import interpreter.utils.globdict as globdict
from interpreter.utils.params import expanse
from interpreter.utils.test_ctrl import TestSetController
from interpreter.utils.test_init import (
@@ -40,6 +42,7 @@ class TestProcess(Process):
config_files,
defines,
gui_defaults={},
text_mode=False,
) -> None:
super().__init__()
self.__fname = file_name
@@ -48,6 +51,7 @@ class TestProcess(Process):
self.__cfgf = config_files
self.__defs = defines
self.__gui_defaults = gui_defaults # default values coming from GUI prefs
self.__text_mode = text_mode
self.__exec = False
self.__loaded = False
self.__closed = False
@@ -193,6 +197,7 @@ class TestProcess(Process):
def run(self):
signal.signal(signal.SIGINT, signal.SIG_IGN)
try:
try:
# Thread for stdout redirection
@@ -223,6 +228,10 @@ Is the python exec path correct ?"""
# Load the test file
test_dict, param_files = self._load_test(init_param_files, glob_variables)
if self.__text_mode:
tm.setgd("_text_mode", True)
tm.setgd("_interactive", False)
# Backup the global dict in case of restart of the test
gdict = backup_gd()
@@ -255,6 +264,7 @@ Is the python exec path correct ?"""
try:
test_run_init()
print(test_run_header())
globdict.set_update_queue(self.__squeue)
test_set.execute()
finally:
if test_set.success():
@@ -265,8 +275,16 @@ Is the python exec path correct ?"""
test_set.run_post_exec()
finally:
self.__exec = False
# Stop shared context engines before restore_gd wipes them
for engine in tm.gd("_py_func_contexts", {}).values():
engine.stop()
engine.join()
for engine in tm.gd("_lua_func_contexts", {}).values():
engine.stop()
engine.join()
# Sends signal to the GUI
self.send_finished()
self.send_finished(success=test_set.success())
globdict.set_update_queue(None)
restore_gd(gdict)
except Exception as e:
print_exception(e)
@@ -275,6 +293,13 @@ Is the python exec path correct ?"""
# Stop python eval execution process
eval_proc.stop()
eval_proc.join()
# Stop shared func context engines (keep_context_id)
for engine in tm.gd("_py_func_contexts", {}).values():
engine.stop()
engine.join()
for engine in tm.gd("_lua_func_contexts", {}).values():
engine.stop()
engine.join()
except Exception as e:
print_exception(e)
@@ -297,6 +322,9 @@ Is the python exec path correct ?"""
"enabled_state": test_set.getEnabledState,
"process_param": self.process_param,
"set_test_outputs": self.set_test_outputs,
"get_gd_vars": self.get_gd_vars,
"set_gd_var": self.set_gd_var,
"del_gd_var": self.del_gd_var,
"set_enabled_state": test_set.setEnabledState,
"check_uncheck_all": test_set.checkUncheckAll,
"get_folded": test_set.getFolded,
@@ -311,8 +339,10 @@ Is the python exec path correct ?"""
stdio_redir.restore()
stdio_redir.stop()
def send_finished(self):
def send_finished(self, success=None):
status = {"id": None, "name": "test_process", "status": "finished"}
if success is not None:
status["success"] = success
self.__squeue.put(status)
def execute(self):
@@ -330,6 +360,25 @@ Is the python exec path correct ?"""
def set_test_outputs(self, outputs: list):
tm.setgd("test_outputs", outputs)
def get_gd_vars(self):
import json
result = {}
for k, v in globdict.global_dict.items():
if k.startswith("_"):
continue
try:
json.dumps(v)
result[k] = v
except (TypeError, ValueError):
pass
return result
def set_gd_var(self, name: str, value):
tm.setgd(name, value)
def del_gd_var(self, name: str):
tm.delgd(name)
def process_control_commands(self, tctrl):
term = False
while (not term) and (not self.__closed):
@@ -382,7 +431,7 @@ Is the python exec path correct ?"""
try:
# read the pipe data
data = cconn.recv()
print(data, end="")
print(data, end="", flush=True)
except EOFError:
# exit the loop is the pipe is closed
break

View File

@@ -1,244 +0,0 @@
try:
import readline
except:
pass
from cmd import Cmd
import os
import sys
from yaml import load, Loader
import functools
import platform
import types
import inspect
# test modules
from interpreter.utils.test_init import (
env_init, prepare_global, set_standard_gd_keys,
update_global, test_run_init, test_run_header, load_test)
from interpreter.utils.globdict import (global_dict)
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.test_report.test_report import TestReport
class FakeQueue:
def put(self, arg):
pass
def func(self, args):
if not args.startswith("{"):
args = "{"+args+"}"
y = load(args, Loader)
obj = self.current_item(y, status_queue=FakeQueue())
obj.report = self.report
res = obj.execute()
if not (res.value is None):
print('result : {}'.format(res.value))
print(res.test_result)
class Terminal(Cmd):
SUPPORTED_TESTS = [
cst.TYPE_SLEEP,
cst.TYPE_LET,
cst.TYPE_PY_FUNCTION,
cst.TYPE_LUA_FUNCTION,
cst.TYPE_CONSOLE,
cst.TYPE_IMAGE_DLG,
cst.TYPE_MESSAGE_DLG,
cst.TYPE_QUESTION_DLG,
cst.TYPE_VALUE_DLG,
]
SUPPORTED_GROUPS = [
cst.TYPE_GROUP,
cst.TYPE_CYCLE
]
def __init__(self, working_dir, config_files, defines, no_color):
super().__init__()
self.working_dir = working_dir
self.config_files = config_files
self.current_item = None
report = TestReport(None)
self.report = report
env_init()
prepare_global()
# Define the builtin variables
set_standard_gd_keys("Unnamed", self.working_dir, '', config_files)
update_global([], defines)
# creation of the functions
for tst in self.SUPPORTED_TESTS:
meth_name = "do_" + tst.item_cmd
# copy of the function
f = types.FunctionType(func.__code__, func.__globals__, name=meth_name,
argdefs=func.__defaults__,
closure=func.__closure__)
f = functools.update_wrapper(f, func)
f.__kwdefaults__ = func.__kwdefaults__
f.__doc__ = tst.item_class.__doc__
setattr(self, meth_name, types.MethodType(f, self))
test_run_init()
self.prompt = "(testium)~ "
# display header
print(test_run_header())
# redirect output
if 'Linux' in platform.system() and not no_color:
from lib.stdout_redirect import stdio_redir
try:
from interpreter.utils.termlog import TermLog
stdio_redir.redirect(TermLog(sys.stdout))
except ModuleNotFoundError:
tm.print_info('Colored console not supported by the system.' +
' If you want it, please install colorama module')
def precmd(self, line: str) -> str:
c = line.split(" ", 1)[0].strip()
self.current_item = None
for tst in self.SUPPORTED_TESTS:
if c == tst.item_cmd:
self.current_item = tst.item_class
break
return line
def load_test_recursively(self, tree_parent, parent_seq, status_queue):
try:
parent_seq_name = parent_seq['name']
except KeyError:
parent_seq['name'] = "sequence"
except TypeError:
raise Exception("Syntax error in an item of type {} which is a child of {}".format(
tree_parent.type(), tree_parent.parent().name()))
try:
parent_seq_actions = parent_seq['steps']
except KeyError:
raise Exception(' No action list found for "%s" sequence'
% (parent_seq_name))
# if action is a dictionary , we assume it is a single action
# that has not been nested in a list, so do it
if isinstance(parent_seq_actions, (dict)):
parent_seq_actions = [parent_seq_actions]
if not isinstance(parent_seq_actions, (list, tuple)):
raise Exception('Actions list not valid.')
# first we merged to the same level 'sequence dict entries and list within the list
counter = 0
test_dir = tm.gd('test_directory')
while (counter < len(parent_seq_actions)):
action = parent_seq_actions[counter]
# if action is a list raise up to the the same level,
# ie insert action element into the parent_seq_actions
if isinstance(action, (list, tuple)):
parent_seq_actions[counter:counter+1] = action
continue
# if action is a NoneType skip and continue
# (when pointing to an unused alias for instance)
if action is None:
counter += 1
continue
# if action is a sequence we insert its entry into the action list
if 'sequence' in action:
parent_seq_actions[counter:counter+1] = action['sequence']
continue
else:
executed = False
for it in [*self.SUPPORTED_TESTS, *self.SUPPORTED_GROUPS]:
if it.item_cmd in action:
executed = True
item = (it.item_class)(action[it.item_cmd],
tree_parent,
status_queue)
# check for sequence type:
if it.item_cmd == cst.TYPE_UNITTEST_FILE.item_cmd:
item.setTestDir(test_dir)
item.load()
elif ((it.item_cmd == cst.TYPE_CYCLE.item_cmd) or
(it.item_cmd == cst.TYPE_GROUP.item_cmd)):
self.load_test_recursively(
item, action[it.item_cmd], status_queue)
if not executed:
raise Exception('action type is not known "{}"'.format(
list(action.keys())[0]))
counter += 1
def __setReportRecursively(self, parent):
for i in range(parent.childCount()):
parent.child(i).report = self.report
self.__setReportRecursively(parent.child(i))
def setReport(self, root_item):
root_item.report = self.report
self.__setReportRecursively(root_item)
def get_names(self):
memb = inspect.getmembers(self)
return [n[0] for n in memb if (inspect.ismethod(n[1]) and n[0].startswith("do_"))]
def do_load(self, args):
"""load function.
This function loads and executes a testium sub-script.
The loaded sequence can't be a main testium script ("testium -b" option is
defined for such a usage).
Accepted files are with extension "*.tum".
usage:
load path/to/my/sequence.tum
"""
file = args.strip()
suff = file[-4:]
if not suff in ['.tum']:
raise Exception('Wrong input file extension')
if not (os.path.exists(file) and os.path.isfile(file)):
raise Exception(
'"{}" does not exist or is not a file.'.format(file))
d, _ = load_test(file)
if not isinstance(d, list):
raise Exception(
"The file root object must be a list. A \"main\" tum can't be loaded from here (use batch mode instead).")
if (len(d) == 1) and isinstance(d[0], dict) and (not d[0].get('sequence', None) is None):
d = d[0]['sequence']
sq = FakeQueue()
root_item = (cst.TYPE_ROOT.item_class)(
dict_item={'steps': d}, status_queue=sq)
self.load_test_recursively(root_item, {'steps': d}, sq)
self.setReport(root_item)
res = root_item.execute()
if not (res.value is None):
print('"{}" execution overall result: {}'.format(file, res.value))
print(res.test_result)
def do_gd(self, args):
"""Variables lists and values.
usage:
gd
gd home
"""
if args != '':
res = tm.gd(args, None)
if res is None:
raise Exception(
'the variable: "{}" has not been found.'.format(args))
print(res)
return
for k in global_dict.keys():
print('{}: {}'.format(str(k), str(global_dict[k])))
def do_quit(self, args):
'''Quit the application.'''
raise Exception('quit')

View File

@@ -5,7 +5,7 @@ from itertools import chain
from PySide6.QtGui import QIcon, QPixmap
from PySide6.QtWidgets import QApplication, QDialog, QDialogButtonBox
from PySide6.QtCore import Qt, QSettings, QSize
from PySide6.QtCore import Qt, QSettings, QTimer, QSize
from PySide6.QtGui import QFont, QFontInfo
from PySide6.QtWidgets import QTreeWidgetItem
@@ -185,7 +185,9 @@ def main(args, conn=None):
SettingsApplication = "testium_choices_dlg_" + args[0]
SettingsLastChoices = "last_choice"
success = True
app = QApplication()
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = ChoicesDialog()
d.setFixedSize(800, 600)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -205,6 +207,9 @@ def main(args, conn=None):
d.connect_checked()
d.choicesView.setFocus()
auto_result = args[4] if len(args) > 4 else None
if auto_result is not None:
QTimer.singleShot(2000, lambda: d.accept() if auto_result.lower() == 'ok' else d.reject())
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'choices_dialog_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -0,0 +1,15 @@
"""Qt platform environment setup for dialog subprocesses.
Call setup() at the start of every dialog subprocess main() function
to ensure the correct Qt platform plugin is selected.
"""
import sys
import os
def setup():
"""Configure the Qt environment for dialog subprocess usage."""
if sys.platform.startswith('linux'):
# On Linux/Wayland, force X11 (via XWayland) to avoid crashes
# when Qt is initialized inside a multiprocessing subprocess.
os.environ['QT_QPA_PLATFORM'] = 'xcb'

View File

@@ -1,7 +1,6 @@
import sys
import os
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6 import (QtGui)
@@ -18,7 +17,9 @@ class TestDialogWindow(QDialog, dialog_image_win.Ui_Dialog):
def main(args, conn):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(700,600)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -37,6 +38,10 @@ def main(args, conn):
d.labelImage.setPixmap(QtGui.QPixmap.fromImage(image2))
auto_result = args[3] if len(args) > 3 else None
if auto_result is not None:
QTimer.singleShot(2000, lambda: d.accept() if auto_result.lower() == 'ok' else d.reject())
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'dialog_image_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -1,36 +1,39 @@
import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args):
app = QApplication(sys.argv)
reply = QMessageBox.information(None, args[0], args[1], QMessageBox.Ok)
if hasattr(sys, "frozen"):
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug.
class dummyStream:
''' dummyStream behaves like a stream but does nothing. '''
def __init__(self): pass
def write(self,data): pass
def read(self,data): pass
def flush(self): pass
def close(self): pass
# and now redirect all default streams to this dummyStream:
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()
if __name__ == '__main__':
main(sys.argv[1:])
import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt, QTimer
def main(args):
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
msg = QMessageBox()
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
msg.setText(args[1])
msg.setIcon(QMessageBox.Information)
msg.setStandardButtons(QMessageBox.Ok)
if len(args) > 2:
QTimer.singleShot(2000, lambda: msg.button(QMessageBox.Ok).click())
msg.exec()
if hasattr(sys, "frozen"):
class dummyStream:
def __init__(self): pass
def write(self, data): pass
def read(self, data): pass
def flush(self): pass
def close(self): pass
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()
if __name__ == '__main__':
main(sys.argv[1:])

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'dialog_note_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -2,7 +2,7 @@ import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from interpreter.test_items.dialog_note_files import dialog_note_win
from multiprocessing import freeze_support
@@ -14,13 +14,23 @@ class TestDialogWindow(QDialog, dialog_note_win.Ui_Dialog):
def main(args, conn=None):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
d.setWindowTitle(args[0])
d.labelDialog.setText(args[1])
d.textEdit.setFocus()
auto_result = args[2] if len(args) > 2 else None
if auto_result is not None:
auto_value = args[3] if len(args) > 3 else None
def _auto_close():
if auto_value is not None:
d.textEdit.setPlainText(auto_value)
d.accept() if auto_result.lower() == 'ok' else d.reject()
QTimer.singleShot(2000, _auto_close)
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -1,32 +1,43 @@
import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args, conn):
app = QApplication(sys.argv)
reply = QMessageBox.question(None, args[0], args[1], QMessageBox.Yes|QMessageBox.No)
conn.send(reply)
conn.close()
if hasattr(sys, "frozen"):
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug.
class dummyStream:
''' dummyStream behaves like a stream but does nothing. '''
def __init__(self): pass
def write(self,data): pass
def read(self,data): pass
def flush(self): pass
def close(self): pass
# and now redirect all default streams to this dummyStream:
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()
import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt, QTimer
def main(args, conn):
try:
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
msg = QMessageBox()
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
msg.setText(args[1])
msg.setIcon(QMessageBox.Question)
msg.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
auto_result = args[2] if len(args) > 2 else None
if auto_result is not None:
btn = QMessageBox.Yes if auto_result.lower() == 'yes' else QMessageBox.No
QTimer.singleShot(2000, lambda: msg.button(btn).click())
reply = msg.exec()
conn.send(reply)
except Exception as e:
print(f"dialog_question error: {e}", file=sys.stderr)
finally:
conn.close()
if hasattr(sys, "frozen"):
class dummyStream:
def __init__(self): pass
def write(self, data): pass
def read(self, data): pass
def flush(self): pass
def close(self): pass
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()

View File

@@ -39,7 +39,9 @@ class DialogSleepWindow(QDialog, dialog_sleep_win.Ui_SleepDialogWindow):
def main(args, conn=None):
success = True
app = QApplication(sys.argv)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = DialogSleepWindow()
d.setFixedSize(379,129)
d.setWindowFlags(Qt.WindowStaysOnTopHint)

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'dialog_sleep_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'dialog_value_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -2,7 +2,7 @@ import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from interpreter.test_items.dialog_value_files import dialog_value_win
from multiprocessing import freeze_support
@@ -15,7 +15,9 @@ class TestDialogWindow(QDialog, dialog_value_win.Ui_Dialog):
def main(args, conn=None):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -23,6 +25,14 @@ def main(args, conn=None):
d.labelDialog.setText(args[1])
d.lineEdit.setText(args[2])
d.lineEdit.setFocus()
auto_result = args[3] if len(args) > 3 else None
if auto_result is not None:
auto_value = args[4] if len(args) > 4 else None
def _auto_close():
if auto_value is not None:
d.lineEdit.setText(auto_value)
d.accept() if auto_result.lower() == 'ok' else d.reject()
QTimer.singleShot(2000, _auto_close)
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -7,7 +7,7 @@ import libs.testium as tm
from interpreter.utils.params import TestItemParams
from interpreter.utils.constants import TestItemType as cst_type
from interpreter.utils.eval import eval_to_boolean, evaluate, post_evaluate
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
LOG_TEST_STOP = '<----- step "{}" finished'
LOG_TEST_START = '-----> step "{}" started'
@@ -101,6 +101,7 @@ class TestItem:
self.status_queue = status_queue
self._execute_on_stop = False
self._post_eval = None
self._store_result = None
self._expected_result = None
self._no_fail = None
self._is_stopped = False
@@ -131,11 +132,11 @@ class TestItem:
if s:
try:
self.skipped = eval_to_boolean(s)
except:
except Exception as e:
raise ETUMSyntaxError(
f"'{self.cmd()}' test item named '{self.name()}':\nskipped expresion can only be a static expression as it is evaluated during loading of TUM : {s}",
self.seqFilename(),
)
) from e
# This allow disabling test item directly by using its name inside param.yaml file
elif self._name in tm.gd("skipped_test_item", []):
self.skipped = True
@@ -155,6 +156,9 @@ class TestItem:
if "process_result" in dict_item:
self._post_eval = dict_item["process_result"]
if "store_result" in dict_item:
self._store_result = dict_item["store_result"]
if "expected_result" in dict_item:
self._expected_result = dict_item["expected_result"]
@@ -164,11 +168,13 @@ class TestItem:
self.banner = LOG_TEST_START.format(self._name)
self.footer = LOG_TEST_STOP.format(self._name)
except:
except ETUMSyntaxError:
raise
except Exception as e:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
f"The '{self.cmd()}' test item named '{self.name()}' has an unexpected loading error: {e}",
self.seqFilename(),
)
) from e
self.result = TestResult(self, TestValue.FAILURE, "Failure by default")
@@ -275,6 +281,9 @@ class TestItem:
self.process_result()
# expected_result treatment
self.result_expected()
# Store result in a global variable if requested (before no_fail so
# the real outcome is captured when result.value is None)
self.store_result()
# Case of the no_fail true parameter
self.process_no_fail()
@@ -317,6 +326,17 @@ class TestItem:
print(e)
self.result.set(TestValue.FAILURE, "Result processing failed")
def store_result(self):
if self._store_result is None:
return
var_name = self._prms.expanse(self._store_result)
if self.result.value is None:
value = str(self.result.test_result)
else:
value = self.result.value
tm.setgd(var_name, value)
print(f"Stored result in '$({var_name})': {value}")
def process_report(self, report_eval):
tm.print_debug(f"Export reported values:")
rep_eval = self._prms.expanse(report_eval)

View File

@@ -1,7 +1,7 @@
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import TestValue
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
@@ -15,21 +15,16 @@ class TestItemCheckValue(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_CHECK
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._action_list = self._prms.getParamAll('steps', default=[], required=False)
if len(self._action_list) > 0:
tm.print_warn("'steps' argument of check test item is deprecated and is replaced by 'values'")
self._action_list += self._prms.getParamAll('values', default=[], required=False)
if len(self._action_list) <= 0:
raise ETUMSyntaxError(
f" The '{self.cmd()}' test item named '{self.name()}' must have a 'values' parameter",
f"Missing required 'values' parameter",
self.seqFilename()
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (a child of: '{self.parent().name()}') has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):

View File

@@ -1,50 +1,92 @@
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.dialog_choices_files import choices_dialog
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemChoicesDialog(TestItem):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_CHOICES_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_CHOICES_DLG
self.is_container = False
try:
self._question = self._prms.getParam("question", required=True)
self._choices = self._prms.getParam("choices", required=True)
self._default_icon = self._prms.getParam(
"icon", required=False, default=None
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (a child of: '{self.parent().name()}') has a missing or wrong parameter",
self.seqFilename()
)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
choices = self._prms.expanse(self._choices)
icon = self._prms.expanse(self._default_icon)
parent_conn, child_conn = Pipe()
p = Process(
target=choices_dialog.main, args=([self.name(), q, choices, icon], child_conn)
)
p.start()
val, succ = parent_conn.recv()
p.join()
self.result.value = val
if succ:
# The result of the test item is put into the global dict
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, str(val))
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
class TestItemChoicesDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_CHOICES_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_CHOICES_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam("question", required=True)
self._choices = self._prms.getParam("choices", required=True)
self._default_icon = self._prms.getParam("icon", required=False, default=None)
self._auto_result = self._prms.getParam("auto_result", required=False, default=None)
def _print_choices(self, choices, indent=0):
if not isinstance(choices, list):
return
for choice in choices:
name = choice.get("name", "")
desc = choice.get("description", "")
line = " " * indent + f"- {name}"
if desc:
line += f": {desc}"
print(line)
sub = choice.get("choices", None)
if sub:
self._print_choices(sub, indent + 1)
def _all_checked(self, choices):
result = []
if not isinstance(choices, list):
return result
for choice in choices:
item = {"name": choice.get("name", ""), "checked": True}
sub = choice.get("choices", None)
if sub is not None:
item["choices"] = self._all_checked(sub)
result.append(item)
return result
@test_run
def execute(self):
q = self._prms.expanse(self._question)
choices = self._prms.expanse(self._choices)
icon = self._prms.expanse(self._default_icon)
if _is_text_mode():
print(f"Choices: {q}")
self._print_choices(choices)
if _is_interactive():
ans = input("Accept all? (y/n) [default: y]: ").strip().lower()
if ans in ('n', 'no'):
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, "Cancelled")
else:
val = self._all_checked(choices)
self.result.value = val
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
elif ar == 'cancel':
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, "Cancelled")
else:
val = self._all_checked(choices)
self.result.value = val
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
return
from interpreter.test_items.dialog_choices_files import choices_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q, choices, icon] + ([ar] if ar is not None else [])
result = self._run_dialog_with_result(choices_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
val, succ = result
self.result.value = val
if succ:
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, str(val))

View File

@@ -0,0 +1,37 @@
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
class TestItemContainer(TestItem):
"""Base class for items that run a sequence of children sequentially."""
def __init__(self, item_type, dict_item, parent=None, status_queue=None, filename=""):
self._name = item_type.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = item_type
self.is_container = True
def _run_children_sequentially(self):
"""Execute all children in order, respecting stop_on_failure and stop requests.
Returns a TestResult aggregating all children outcomes."""
i = 0
to_be_stopped = False
while not self.isStopped() and i < self.childCount() and not to_be_stopped:
result = self.child(i).execute()
if result.test_result == TestValue.FAILURE and self._stop_on_failure:
to_be_stopped = True
i += 1
if self.isStopped() or to_be_stopped:
for j in range(self.childCount()):
if self.child(j).executedOnStop() and j >= i:
self.child(j).execute()
success = TestValue.SUCCESS
for j in range(i):
if self.child(j).result.test_result == TestValue.FAILURE:
success = TestValue.FAILURE
break
stopped = self.isStopped() or to_be_stopped
return TestResult(None, success, ""), stopped

View File

@@ -0,0 +1,58 @@
import multiprocessing
import libs.testium as tm
from interpreter.test_items.test_item import TestItem
def _is_text_mode():
return tm.text_mode()
def _is_interactive():
return bool(tm.gd("_interactive", True))
_spawn_ctx = multiprocessing.get_context('spawn')
class TestItemDialogBase(TestItem):
"""Base class for test items that launch a Qt dialog in a subprocess."""
def _cleanup_process(self, p):
if p.is_alive():
p.terminate()
p.join(timeout=0.2)
if p.is_alive():
p.kill()
p.join()
def _run_dialog(self, target, args):
"""Launch target(args) in a subprocess with no return value.
Returns the subprocess exit code.
"""
p = _spawn_ctx.Process(target=target, args=(args,))
p.start()
while p.is_alive() and not self._is_stopped:
p.join(timeout=0.5)
self._cleanup_process(p)
return p.exitcode
def _run_dialog_with_result(self, target, args):
"""Launch target(args, child_conn) in a subprocess and return what it sends.
Returns the received value, or None if stopped or if the subprocess crashed.
"""
parent_conn, child_conn = _spawn_ctx.Pipe()
p = _spawn_ctx.Process(target=target, args=(args, child_conn))
p.start()
child_conn.close()
result = None
while p.is_alive() and not self._is_stopped:
if parent_conn.poll(0.5):
try:
result = parent_conn.recv()
except EOFError:
pass
break
self._cleanup_process(p)
return result

View File

@@ -1,71 +1,56 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.dialog_image_files import dialog_image
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError
class TestItemImageDialog(TestItem):
"""dialog_image item usage.
dialog_image name: Nice image, question: could you press the red button, filename: img.jpg
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_IMAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_IMAGE_DLG
self.is_container = False
try:
self._question = self._prms.getParam("question", required=True)
self._filename = self._prms.getParam("filename", required=True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(
os.path.dirname(ourpath), "dialog_image_files", "dialog_image.py"
)
q = self._prms.expanse(self._question)
image_path = self._prms.expanse(self._filename)
print("Image Displayed:\n" + q + "\n" + image_path)
if not os.path.isfile(image_path):
image_path = os.path.normpath(
os.path.join(tm.gd("test_directory"), image_path)
)
parent_conn, child_conn = Pipe()
p = Process(
target=dialog_image.main, args=([self.name(), q, image_path], child_conn)
)
p.start()
succ = parent_conn.recv()
p.join()
if succ:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__ == "__main__":
p = Process(target=test_dialog.main, args=(["bob", "bab"],))
p.start()
p.join()
import os
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
class TestItemImageDialog(TestItemDialogBase):
"""dialog_image item usage.
dialog_image name: Nice image, question: could you press the red button, filename: img.jpg
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_IMAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_IMAGE_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam("question", required=True)
self._filename = self._prms.getParam("filename", required=True)
self._auto_result = self._prms.getParam("auto_result", required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
image_path = self._prms.expanse(self._filename)
print("Image Displayed:\n" + q + "\n" + image_path)
if not os.path.isfile(image_path):
image_path = os.path.normpath(
os.path.join(tm.gd("test_directory"), image_path)
)
if _is_text_mode():
if _is_interactive():
ans = input("Accept? (y/n) [default: y]: ").strip().lower()
self.result.set(TestValue.FAILURE if ans in ('n', 'no') else TestValue.SUCCESS)
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
elif ar == 'cancel':
self.result.set(TestValue.FAILURE)
else:
self.result.set(TestValue.SUCCESS)
return
from interpreter.test_items.dialog_image_files import dialog_image
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q, image_path] + ([ar] if ar is not None else [])
succ = self._run_dialog_with_result(dialog_image.main, args)
if succ is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
elif succ:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE)

View File

@@ -5,7 +5,7 @@ import time
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
@@ -19,18 +19,13 @@ class TestItemLet(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_LET
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._values_list = self._prms.getParamAll('values', default=[], required=False)
if len(self._values_list) <= 0:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' must have a 'values' parameter",
f"Missing required 'values' parameter",
self.seqFilename(),
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):

View File

@@ -4,7 +4,7 @@ import traceback
import pprint
import textwrap
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestValue
import libs.testium as tm
@@ -12,10 +12,13 @@ from interpreter.utils.lua_func_exec import LuaFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.constants import TestItemType as cst
_LUA_FUNC_CONTEXTS_KEY = "_lua_func_contexts"
class TestItemLuaFunc(TestItem):
"""lua_func item usage.
func file: func_file.lua, func_name: func, param: [$(variable1), [1, 2, 3], true]
Optional: context_id: <id> — share a persistent process with other lua_func items using the same id.
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
@@ -23,18 +26,25 @@ class TestItemLuaFunc(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_LUA_FUNCTION
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (child of '{self.parent.name()}') has a missing or wrong parameter",
self.seqFilename(),
)
# Lua functions call subprocess initialization
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
self._lua_func_proc = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
def _get_engine(self):
"""Return (engine, persistent). If context_id is set, use a shared persistent engine."""
if self._context_id is None:
return self._lua_func_proc, False
ctx_id = self._prms.expanse(self._context_id)
contexts = tm.gd(_LUA_FUNC_CONTEXTS_KEY, {})
if ctx_id not in contexts:
contexts[ctx_id] = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
tm.setgd(_LUA_FUNC_CONTEXTS_KEY, contexts)
return contexts[ctx_id], True
@test_run
def execute(self):
self.result.set(
@@ -48,22 +58,25 @@ class TestItemLuaFunc(TestItem):
print("Parameters list:")
print(textwrap.indent(pprint.pformat(pl), " |"))
self._lua_func_proc.start()
if not self._lua_func_proc.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external lua execution process.
engine, persistent = self._get_engine()
if not engine.is_alive():
engine.start()
if not engine.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external lua execution process.
Is the lua path correct ?
lua_bin = {tm.gd("lua_bin", "no lua path defined")}
Are "lua-sockets" and "lua-cjson" installed ?
Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables ?"""
)
)
try:
success, ret = self._lua_func_proc.func_call(self.file_name, self.func_name, pl)
success, ret = engine.func_call(self.file_name, self.func_name, pl)
finally:
# Stops lua function execution process
self._lua_func_proc.stop()
self._lua_func_proc.join()
if not persistent:
engine.stop()
engine.join()
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)
@@ -73,7 +86,6 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
print("Returned value:")
print(textwrap.indent(pprint.pformat(res), " |"))
# The result of the func test item is put in global dir and result
tm.setgd("lfn_" + self._name, res)
self.result.value = res
@@ -88,5 +100,5 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
traceback.print_exception(*sys.exc_info())
self.result.set(
TestValue.FAILURE,
'Unrecoverable "py_func" item error from {}'.format(self.func_name),
'Unrecoverable "lua_func" item error from {}'.format(self.func_name),
)

View File

@@ -1,54 +1,46 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from interpreter.test_items.dialog_msg_files import msg_dialog
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError
class TestItemMsgDialog(TestItem):
"""dialog_message item usage.
dialog_message name: Nice message, question: Open the door and press OK
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_MESSAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_MESSAGE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_msg_files',
'msg_dialog.py')
q = self._prms.expanse(self._question)
print("Message Displayed:\n" + q)
parent_conn, child_conn = Pipe()
p=Process(target=msg_dialog.main,
args=([self.name(), q],))
p.start()
p.join()
self.result.set(TestValue.SUCCESS)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=msg_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
import os
import sys
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
class TestItemMsgDialog(TestItemDialogBase):
"""dialog_message item usage.
dialog_message name: Nice message, question: Open the door and press OK
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_MESSAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_MESSAGE_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
print("Message Displayed:\n" + q)
if _is_text_mode():
if _is_interactive():
input("Press Enter to continue...")
self.result.set(TestValue.SUCCESS)
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is not None:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
from interpreter.test_items.dialog_msg_files import msg_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q] + ([ar] if ar is not None else [])
exitcode = self._run_dialog(msg_dialog.main, args)
if exitcode == 0:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE, f"Dialog subprocess exited with code {exitcode}")

View File

@@ -1,62 +1,75 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.dialog_note_files import test_dialog
from lib.tum_except import ETUMSyntaxError
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
class TestItemNoteDialog(TestItem):
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_NOTE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_NOTE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_note_files',
'test_dialog.py')
q = self._prms.expanse(self._question)
print("Question:\n" + q)
parent_conn, child_conn = Pipe()
p=Process(target=test_dialog.main, args=([self.name(), q],child_conn))
p.start()
val, succ = parent_conn.recv()
p.join()
tm.setgd(self.name(), val)
print("\n" + ("-" * 80) + "\n")
print("- Test note\n")
print("-" * 80 + "\n")
print(val)
print("-" * 80 + "\n")
self.result.reported = {'note': val}
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
class TestItemNoteDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_NOTE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_NOTE_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
self._auto_value = self._prms.getParam('auto_value', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
print("Question:\n" + q)
if _is_text_mode():
if _is_interactive():
print("Enter your note (type '.' on a new line to finish, empty line to cancel):")
lines = []
while True:
line = input()
if line == '.':
break
lines.append(line)
val = '\n'.join(lines)
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
if ar == 'cancel':
self.result.set(TestValue.FAILURE, 'Dialog cancelled')
return
val = av if av is not None else ''
tm.setgd(self.name(), val)
print("\n" + ("-" * 80) + "\n")
print("- Test note\n")
print("-" * 80 + "\n")
print(val)
print("-" * 80 + "\n")
self.result.reported = {'note': val}
if val:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
return
from interpreter.test_items.dialog_note_files import test_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
args = [self.name(), q] + ([ar, av] if ar is not None else [])
result = self._run_dialog_with_result(test_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
val, succ = result
tm.setgd(self.name(), val)
print("\n" + ("-" * 80) + "\n")
print("- Test note\n")
print("-" * 80 + "\n")
print(val)
print("-" * 80 + "\n")
self.result.reported = {'note': val}
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)

View File

@@ -0,0 +1,193 @@
import threading
from time import sleep, time
from interpreter.test_items.test_item_container import TestItemContainer
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import eval_to_boolean
from lib.tum_except import ETUMSyntaxError
from lib.string_queue import StringQueue
from lib.stdout_redirect import stdio_redir
class TestItemParallelBranch(TestItemContainer):
"""One branch of a parallel item. Runs its children sequentially,
optionally waiting for a condition before starting."""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
super().__init__(cst.TYPE_PARALLEL_BRANCH, dict_item, parent, status_queue, filename=filename)
self._wait_condition = None
self._wait_timeout = 30
if "wait_for" in dict_item:
wf = dict_item["wait_for"]
if not isinstance(wf, dict):
raise ETUMSyntaxError(
f"'wait_for' in branch '{self.name()}' must be a dict with 'condition' and optional 'timeout'",
self.seqFilename(),
)
self._wait_condition = wf.get("condition", None)
self._wait_timeout = float(wf.get("timeout", 30))
def _wait_start(self):
"""Block until wait_for condition is True, or timeout. Returns False on timeout."""
if self._wait_condition is None:
return True
deadline = time() + self._wait_timeout
while time() < deadline:
if self.isStopped():
return False
try:
c = self._prms.expanse(self._wait_condition)
if eval_to_boolean(c):
return True
except Exception:
pass
sleep(0.1)
return False
@test_run
def execute(self):
if not self._wait_start():
self.result.set(
TestValue.FAILURE,
f"wait_for timeout ({self._wait_timeout}s): condition '{self._wait_condition}' not met",
)
return
result, stopped = self._run_children_sequentially()
if stopped:
if result.test_result == TestValue.FAILURE:
self.result.set(TestValue.FAILURE, "Branch aborted on failure")
else:
self.result.set(TestValue.NORUN, "Branch aborted on user request")
else:
self.result.set(result.test_result, "")
class TestItemParallel(TestItemContainer):
"""Runs multiple branches concurrently.
YAML:
parallel:
name: ...
sync: all # all (default): wait for every branch
# any: stop as soon as one branch finishes
stop_on_failure: false
branches:
- name: Branch A
wait_for:
condition: "'$(ready)' == 'True'"
timeout: 30
steps:
- ...
- name: Branch B
steps:
- ...
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
branches = dict_item.get("branches", [])
if not branches:
raise ETUMSyntaxError(
f"'parallel' item requires at least one branch in 'branches'",
dict_item.get("seq_filename", ""),
)
# Inject a synthetic 'steps' key so load_test_recursively can load branches
# as TestItemParallelBranch children. The base class' _filter_dict_item
# drops 'steps'; we also drop 'branches' (overridden below) so the F1
# panel shows only the parallel's own attributes, not the duplicated
# tree of branches/steps already displayed in the test tree.
dict_item["steps"] = [{"parallel_branch": b} for b in branches]
super().__init__(cst.TYPE_PARALLEL, dict_item, parent, status_queue, filename=filename)
self._sync = str(dict_item.get("sync", "all")).lower()
if self._sync not in ("all", "any"):
raise ETUMSyntaxError(
f"'sync' must be 'all' or 'any', got '{self._sync}'",
self.seqFilename(),
)
def _filter_dict_item(self, dict_item):
c = super()._filter_dict_item(dict_item)
# Keep 'branches' so the F1 panel shows the branch list and their
# per-branch attributes (name, wait_for, condition, ...), but strip
# the 'steps' inside each branch — the steps are already visible as
# children in the test tree and would just duplicate the information.
if isinstance(c, dict) and isinstance(c.get("branches"), list):
stripped = []
for b in c["branches"]:
if isinstance(b, dict):
stripped.append({k: v for k, v in b.items() if k != "steps"})
else:
stripped.append(b)
c["branches"] = stripped
return c
def _stop_branch_recursively(self, item):
item.stop()
for i in range(item.childCount()):
self._stop_branch_recursively(item.child(i))
@test_run
def execute(self):
branch_results = [None] * self.childCount()
any_done = threading.Event()
def run_branch(idx):
branch = self.child(idx)
stdio_redir.register_thread(buffer=StringQueue(), branch=branch.name())
try:
# sync:any: if another branch already won the race, mark this
# branch as stopped so its execute() skips children but still
# goes through the normal addTest path (clean DB entry).
if self._sync == "any" and any_done.is_set():
branch.stop()
try:
result = branch.execute()
except Exception as e:
import traceback
print(f"[parallel] Branch '{branch.name()}' crashed: {e}")
traceback.print_exc()
branch.result.set(TestValue.FAILURE, f"Branch crashed: {e}")
result = branch.result
branch_results[idx] = result
# Only a branch that actually ran (SUCCESS or FAILURE) wins the
# sync:any race. A disabled or skipped branch returns NORUN
# almost instantly and must not stop legitimate branches.
if self._sync == "any" and result.test_result != TestValue.NORUN:
any_done.set()
for j in range(self.childCount()):
if j != idx:
self._stop_branch_recursively(self.child(j))
finally:
stdio_redir.unregister_thread()
threads = [
threading.Thread(target=run_branch, args=(i,), daemon=True)
for i in range(self.childCount())
]
for t in threads:
t.start()
for t in threads:
t.join()
if self._sync == "all":
# Pass if no branch failed; disabled/skipped branches (NORUN) are
# ignored, matching how Group/Cycle treat disabled children.
success = all(
r is not None and r.test_result != TestValue.FAILURE
for r in branch_results
)
else:
# Pass if at least one branch ran and succeeded.
success = any(
r is not None and r.test_result == TestValue.SUCCESS
for r in branch_results
)
self.result.set(
TestValue.SUCCESS if success else TestValue.FAILURE,
f"parallel sync={self._sync}",
)

View File

@@ -4,7 +4,7 @@ import time
import pprint
import textwrap
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestValue
import libs.testium as tm
@@ -12,10 +12,13 @@ from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.constants import TestItemType as cst
_PY_FUNC_CONTEXTS_KEY = "_py_func_contexts"
class TestItemPyFunc(TestItem):
"""py_func item usage.
func file: func_file.py, func_name: func, param: [$(variable1), [1, 2, 3], true]
Optional: context_id: <id> — share a persistent process with other py_func items using the same id.
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
@@ -23,17 +26,25 @@ class TestItemPyFunc(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_PY_FUNCTION
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (child of '{self.parent.name()}') has a missing or wrong parameter",
self.seqFilename(),
)
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
self._py_func_proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
def _get_engine(self):
"""Return (engine, persistent). If context_id is set, use a shared persistent engine."""
if self._context_id is None:
return self._py_func_proc, False
ctx_id = self._prms.expanse(self._context_id)
contexts = tm.gd(_PY_FUNC_CONTEXTS_KEY, {})
if ctx_id not in contexts:
contexts[ctx_id] = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
tm.setgd(_PY_FUNC_CONTEXTS_KEY, contexts)
return contexts[ctx_id], True
@test_run
def execute(self):
self.result.set(
@@ -47,20 +58,23 @@ class TestItemPyFunc(TestItem):
print("Parameters list:")
print(textwrap.indent(pprint.pformat(pl), " |"))
# start the process for executing external python
self._py_func_proc.start()
if not self._py_func_proc.wait_ready():
raise ETUMRuntimeError(
f"""Impossible to start the external python execution process.
engine, persistent = self._get_engine()
if not engine.is_alive():
engine.start()
if not engine.wait_ready():
raise ETUMRuntimeError(
f"""Impossible to start the external python execution process.
Is the python path correct ?
python_bin = {tm.gd("python_bin", "no python path defined")}"""
)
)
try:
success, ret = self._py_func_proc.func_call(self.file_name, self.func_name, pl)
success, ret = engine.func_call(self.file_name, self.func_name, pl)
finally:
# Stops python function execution process
self._py_func_proc.stop()
self._py_func_proc.join()
if not persistent:
engine.stop()
engine.join()
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)
@@ -70,7 +84,6 @@ python_bin = {tm.gd("python_bin", "no python path defined")}"""
print("Returned value:")
print(textwrap.indent(pprint.pformat(res), " |"))
# The result of the func test item is put in global dir and result
tm.setgd("pfn_" + self._name, res)
self.result.value = res

View File

@@ -1,62 +1,58 @@
import os
import sys
from multiprocessing import Process, Pipe
from PySide6.QtWidgets import QMessageBox
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.dialog_question_files import question_dialog
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemQuestionDialog(TestItem):
"""dialog_question item usage.
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_QUESTION_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_QUESTION_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_question_files',
'question_dialog.py')
q = self._prms.expanse(self._question)
print('Question asked:\n' + q + '\n')
parent_conn, child_conn = Pipe()
p=Process(target=question_dialog.main,
args=([self.name(), q],child_conn))
p.start()
succ = parent_conn.recv()
p.join()
if succ == QMessageBox.Yes:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
else:
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
class TestItemQuestionDialog(TestItemDialogBase):
"""dialog_question item usage.
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_QUESTION_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_QUESTION_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
print('Question asked:\n' + q + '\n')
if _is_text_mode():
if _is_interactive():
ans = input("Answer yes (y) or no (n) [default: y]: ").strip().lower()
if ans in ('n', 'no'):
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
else:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
elif ar in ('no', 'cancel'):
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
else:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
return
from interpreter.test_items.dialog_question_files import question_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q] + ([ar] if ar is not None else [])
succ = self._run_dialog_with_result(question_dialog.main, args)
if succ is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
from PySide6.QtWidgets import QMessageBox
if succ == QMessageBox.Yes:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
else:
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')

View File

@@ -10,7 +10,7 @@ from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
def nowInBetween(start, end):
@@ -30,8 +30,8 @@ class TestItemRun(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_RUN
self.is_container = False
try:
self.tum_fime = self._prms.getParam('tum_fime', required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.tum_file = self._prms.getParam('tum', required=True)
self.param_file = self._prms.getParam('param_file', default='')
self.python_bin = self._prms.getParam('python_bin', default='')
self.testium_path = self._prms.getParam('testium_path', default='')
@@ -40,47 +40,46 @@ class TestItemRun(TestItem):
self.start_time = self._prms.getParam('start_time')
self.end_time = self._prms.getParam('end_time')
self.wait_for_exec = self._prms.getParam('wait_for_exec')
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
res = -1
try:
file_path = self._prms.expanse(self.tum_fime)
file_path = self._prms.expanse(self.tum_file)
if not os.path.exists(file_path) and not os.path.isabs(file_path):
file_path = os.path.join(tm.gd('test_directory'), self.tum_fime)
file_path = os.path.join(tm.gd('test_directory'), file_path)
if not os.path.isfile(file_path):
raise ETUMRuntimeError(
'"{}" file could not be found'.format(file_path))
self.tum_fime = file_path
self.tum_file = file_path
pf = self._prms.expanse(self.param_file)
pp = self._prms.expanse(self.python_bin)
sp = self._prms.expanse(self.testium_path)
lp = self._prms.expanse(self.log_path)
rp = self._prms.expanse(self.report_path)
cmd = []
if sp == '':
sp = sys.argv[0]
if pp != '':
cmd.append(pp)
if sp == '':
sp = os.path.join(tm.get_main_dir(), "testium.pyw")
elif not os.path.isfile(sp) or not os.access(sp, os.X_OK):
cmd.append(sys.executable)
cmd.append(sp)
if lp == '':
lp = os.path.splitext(self.tum_fime)[0] + "_" + \
datetime.utcnow().isoformat(timespec='seconds') + '.log'
cmd.append("-r")
if tm.text_mode():
cmd.append("-b")
else:
cmd.append("-r")
if lp == '':
lp = os.path.splitext(self.tum_file)[0] + "_" + \
datetime.utcnow().isoformat(timespec='seconds') + '.log'
cmd.append("-l")
cmd.append('"' + lp + '"')
if pf != '':
cmd.append("-c")
cmd.append('"' + pf + '"')
cmd.append("-l")
cmd.append('"' + lp + '"')
if rp != '':
cmd.append("-p")
cmd.append('"' + rp + '"')
cmd.append(self.tum_fime)
cmd.append(self.tum_file)
for c in cmd:
print(c, end = ' ')
@@ -95,31 +94,23 @@ class TestItemRun(TestItem):
raise ETUMRuntimeError(
'"wait_for_exec" set but not start_time or end_time')
r = None
if self.wait_for_exec:
while not nowInBetween(self.start_time, self.end_time):
sleep(60)
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
elif self.start_time is not None and self.end_time is not None:
if nowInBetween(self.start_time, self.end_time):
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
elif self.start_time is not None:
if self.start_time < datetime.now().time():
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
else:
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
if isinstance(r, subprocess.CompletedProcess):
print((r.stdout).decode())
print(r.stderr.decode())
res = r.returncode
if res >= 0:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE,
'Test execution returned negative value.')
self.result.set(TestValue.FAILURE, 'Sub-test did not execute')
except:
traceback.print_exception(*sys.exc_info())
self.result.set(TestValue.FAILURE, 'Unrecoverable "run" item error')

View File

@@ -4,7 +4,7 @@ import traceback
from functools import wraps
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.item_actions import TestItemActions
@@ -108,17 +108,12 @@ class TestItemPlotActionPeriodic(TestItemPlotAction):
)
# Periodic function call
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.period = self._prms.getParam("period", required=True)
self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
self.post_eval = self._prms.getParam("eval", default="")
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' 'periodic' action settings syntax error",
self.seqFilename(),
)
@test_run
def execute(self):

View File

@@ -3,11 +3,11 @@ from time import sleep
from datetime import timedelta
from multiprocessing import Process, Pipe
import libs.testium as tm
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from interpreter.test_items.dialog_sleep_files import dialog_sleep
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
class TestItemSleep(TestItem):
"""sleep item usage.
@@ -19,14 +19,9 @@ class TestItemSleep(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_SLEEP
self.is_container = False
try:
self._timeout = self._prms.getParam('timeout', required = True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._timeout = self._prms.getParam('timeout', required=True)
self._has_dialog = self._prms.getParam('dialog', default=False)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
@@ -48,6 +43,20 @@ class TestItemSleep(TestItem):
#test core function
if has_dialog:
if tm.text_mode():
import time as _time
print(f"Sleep {timeout}s (press Ctrl+C to abort)...")
end_time = _time.time() + float(timeout)
while _time.time() < end_time and not self._is_stopped:
sleep(0.2)
if self._is_stopped:
print("Aborted")
self.result.set(TestValue.FAILURE, 'Sleep aborted')
else:
self.result.set(TestValue.SUCCESS, f'Sleep {timeout} sec')
return
from interpreter.test_items.dialog_sleep_files import dialog_sleep
parent_conn, child_conn = Pipe()
p=Process(target=dialog_sleep.main, args=([self.name(), timeout],child_conn))
p.start()
@@ -67,5 +76,8 @@ class TestItemSleep(TestItem):
else:
if not isinstance(timeout, (int, float)):
raise ETUMRuntimeError(f"Timeout value of sleep test item \"{self.name}\" is not valid: \"{timeout}\".")
sleep(timeout)
import time as _time
end_time = _time.time() + float(timeout)
while _time.time() < end_time and not self._is_stopped:
sleep(min(0.05, end_time - _time.time()))
self.result.set(TestValue.SUCCESS, 'Sleep %s sec' % (str(timeout)))

View File

@@ -1,77 +1,78 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.tested_references_files import tested_refs_dialog
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemTestedRefsDialog(TestItem):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_REFERENCE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_REFERENCE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required=True)
self._init_values = self._prms.getParamAll('reference', required=False, processed=True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath=__file__
test_file=os.path.join(os.path.dirname(ourpath),
'tested_references_files',
'tested_refs_dialog.py')
q=self._prms.expanse(self._question)
parent_conn, child_conn=Pipe()
init_values=','.join(self._init_values)
p=Process(target=tested_refs_dialog.main,
args=([self.name(), q, init_values],
child_conn))
p.start()
val, succ=parent_conn.recv()
p.join()
titems=[]
if len(val) > 0:
i = 0
for sitem in val.split(','):
titem={}
telems=sitem.split('/')
titem['reference']=telems[0]
titem['revision']=telems[1]
titem['serial']=telems[2]
print("Identification:\n" + str(titem))
titems.append(titem)
self.result.reported = {'reference_{}'.format(i): titem}
i = i + 1
self.result.value = titems
tm.setgd('tested_items', titems)
if len(val) > 0:
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__ == '__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
class TestItemTestedRefsDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_REFERENCE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_REFERENCE_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._init_values = self._prms.getParamAll('reference', required=False, processed=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
init_values = ','.join(self._init_values)
if _is_text_mode():
print(f"References: {q}")
rows = init_values.split(',') if init_values else ['']
result_rows = []
for i, row in enumerate(rows):
parts = (row.split('/') + ['', '', ''])[:3]
if _is_interactive():
ref = input(f"Row {i+1} - Reference [{parts[0]}]: ").strip() or parts[0]
rev = input(f"Row {i+1} - Revision [{parts[1]}]: ").strip() or parts[1]
serial = input(f"Row {i+1} - Serial [{parts[2]}]: ").strip() or parts[2]
else:
ref, rev, serial = parts[0], parts[1], parts[2]
result_rows.append(f"{ref}/{rev}/{serial}")
val = ','.join(result_rows)
if _is_interactive():
succ = True
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
succ = ar != 'cancel'
result = [val, succ]
else:
from interpreter.test_items.tested_references_files import tested_refs_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q, init_values] + ([ar] if ar is not None else [])
result = self._run_dialog_with_result(tested_refs_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
val, succ = result
titems = []
if len(val) > 0:
i = 0
for sitem in val.split(','):
titem = {}
telems = sitem.split('/')
titem['reference'] = telems[0]
titem['revision'] = telems[1]
titem['serial'] = telems[2]
print("Identification:\n" + str(titem))
titems.append(titem)
self.result.reported = {'reference_{}'.format(i): titem}
i += 1
self.result.value = titems
tm.setgd('tested_items', titems)
if len(val) > 0:
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')

View File

@@ -96,10 +96,10 @@ class TestItemUnittestElement(TestItem):
class TestItemUnittestFile(TestItem):
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_UNITTEST_FILE.item_name
self._name = cst.TYPE_UNITTEST.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self.is_container = True
self._type = cst.TYPE_UNITTEST_FILE
self._type = cst.TYPE_UNITTEST
self._fileName = self._prms.getParam('test_file', required = True, processed = True)
self._testDir = ''
self._test_methods = self._prms.getParamAll('test_method', processed=True)
@@ -161,7 +161,7 @@ class TestItemUnittestFile(TestItem):
if self.isStopped():
self.result.set(TestValue.NORUN, 'Group execution aborted on user request')
else:
self.result.set(result.test_result, 'unittest file ' + str(result.test_result))
self.result.set(result.test_result, 'unittest ' + str(result.test_result))
def load(self):
ret = {}

View File

@@ -1,67 +1,74 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.dialog_value_files import test_dialog
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemValueDialog(TestItem):
"""dialog_value item usage.
dialog_value name: Enter value, question: "Which value did you measure?"
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_VALUE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_VALUE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
self._default = self._prms.getParam('default', '')
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_value_files',
'test_dialog.py')
q = self._prms.expanse(self._question)
d = self._prms.expanse(self._default)
print("Question:\n" + q)
parent_conn, child_conn = Pipe()
p=Process(target=test_dialog.main, args=([self.name(), q, d],child_conn))
p.start()
val, succ = parent_conn.recv()
p.join()
tm.setgd(self.name(), val)
print("Answer: " + val)
if len(val) > 0:
self.result.reported = {'question': q, 'answer': val}
self.result.value = val
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
class TestItemValueDialog(TestItemDialogBase):
"""dialog_value item usage.
dialog_value name: Enter value, question: "Which value did you measure?"
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_VALUE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_VALUE_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._default = self._prms.getParam('default', '')
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
self._auto_value = self._prms.getParam('auto_value', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
d = self._prms.expanse(self._default)
print("Question:\n" + q)
if _is_text_mode():
if _is_interactive():
prompt = f"Enter value [{d}]: " if d else "Enter value: "
ans = input(prompt).strip()
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
if ar is None:
print("Answer: \nDialog not supported in batch mode")
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
if ar == 'cancel':
print("Answer: \nDialog cancelled")
self.result.set(TestValue.FAILURE, 'Dialog cancelled')
return
ans = av if av is not None else ''
val = ans if ans else d
tm.setgd(self.name(), val)
print("Answer: " + str(val))
if val:
self.result.reported = {'question': q, 'answer': val}
self.result.value = val
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, 'No value entered')
return
from interpreter.test_items.dialog_value_files import test_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
args = [self.name(), q, d] + ([ar, av] if ar is not None else [])
result = self._run_dialog_with_result(test_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
val, succ = result
tm.setgd(self.name(), val)
print("Answer: " + val)
if len(val) > 0:
self.result.reported = {'question': q, 'answer': val}
self.result.value = val
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')

View File

@@ -1,9 +1,8 @@
import sys
import os
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QDialog, QTableWidgetItem)
from PySide6.QtCore import (Qt, QSettings)
from PySide6.QtCore import Qt, QSettings, QTimer
try:
from interpreter.test_items.tested_references_files import tested_refs_win
@@ -20,7 +19,9 @@ def main(args, conn=None):
SettingsApplication = 'testium_ref_item'
SettingsLastReference = 'lastReference'
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestedRefsWindow()
d.setFixedSize(481,386)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -51,6 +52,9 @@ def main(args, conn=None):
i += 1
d.tableReferences.setFocus()
auto_result = args[3] if len(args) > 3 else None
if auto_result is not None:
QTimer.singleShot(2000, lambda: d.accept() if auto_result.lower() == 'ok' else d.reject())
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'tested_refs_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################
@@ -44,13 +44,13 @@ class Ui_Dialog(object):
font1 = QFont()
font1.setPointSize(10)
__qtablewidgetitem = QTableWidgetItem()
__qtablewidgetitem.setFont(font1);
__qtablewidgetitem.setFont(font1)
self.tableReferences.setHorizontalHeaderItem(0, __qtablewidgetitem)
__qtablewidgetitem1 = QTableWidgetItem()
__qtablewidgetitem1.setFont(font1);
__qtablewidgetitem1.setFont(font1)
self.tableReferences.setHorizontalHeaderItem(1, __qtablewidgetitem1)
__qtablewidgetitem2 = QTableWidgetItem()
__qtablewidgetitem2.setFont(font1);
__qtablewidgetitem2.setFont(font1)
self.tableReferences.setHorizontalHeaderItem(2, __qtablewidgetitem2)
self.tableReferences.setObjectName(u"tableReferences")
self.tableReferences.setGeometry(QRect(10, 130, 461, 211))
@@ -70,10 +70,10 @@ class Ui_Dialog(object):
Dialog.setWindowTitle(QCoreApplication.translate("Dialog", u"Dialog", None))
self.labelDialog.setText(QCoreApplication.translate("Dialog", u"TextLabel", None))
___qtablewidgetitem = self.tableReferences.horizontalHeaderItem(0)
___qtablewidgetitem.setText(QCoreApplication.translate("Dialog", u"Reference", None));
___qtablewidgetitem.setText(QCoreApplication.translate("Dialog", u"Reference", None))
___qtablewidgetitem1 = self.tableReferences.horizontalHeaderItem(1)
___qtablewidgetitem1.setText(QCoreApplication.translate("Dialog", u"Revision", None));
___qtablewidgetitem1.setText(QCoreApplication.translate("Dialog", u"Revision", None))
___qtablewidgetitem2 = self.tableReferences.horizontalHeaderItem(2)
___qtablewidgetitem2.setText(QCoreApplication.translate("Dialog", u"Serial number", None));
___qtablewidgetitem2.setText(QCoreApplication.translate("Dialog", u"Serial number", None))
# retranslateUi

View File

@@ -10,7 +10,7 @@ class ReportExportTxt(rpe.ReportExport):
no_value_types = [cst_type.TYPE_CONSOLE.item_name, cst_type.TYPE_SLEEP.item_name,
cst_type.TYPE_IMAGE_DLG.item_name, cst_type.TYPE_LET.item_name, cst_type.TYPE_CHECK,
cst_type.TYPE_CYCLE.item_name, cst_type.TYPE_GROUP.item_name,
cst_type.TYPE_UNITTEST_FILE.item_name, cst_type.TYPE_MESSAGE_DLG.item_name,
cst_type.TYPE_UNITTEST.item_name, cst_type.TYPE_MESSAGE_DLG.item_name,
cst_type.TYPE_QUESTION_DLG.item_name]
def __init__(self, name, report_db, report_file, pattern, key, no_header=False):

View File

@@ -1,4 +1,5 @@
import os
import threading
from functools import wraps
import sqlite3
from time import (time, sleep)
@@ -143,6 +144,7 @@ class TestReport:
self._level = 0
self._log_stored = False
self._con = None
self._lock = threading.Lock()
if dict_report is None:
self._active = False
@@ -231,7 +233,7 @@ class TestReport:
prepare_file_to_save(rep_path)
if not os.path.exists(os.path.dirname(rep_path)):
raise ETUMRuntimeError("Report path does not exist: " + rep_path)
self._con = sqlite3.connect(rep_path)
self._con = sqlite3.connect(rep_path, check_same_thread=False)
self.createHeader(header)
self.createTestTable()
self._con.commit()
@@ -334,7 +336,8 @@ class TestReport:
req = req + '?,'
req = req[:-1] + ')'
self._con.execute(req, param)
with self._lock:
self._con.execute(req, param)
def incLevel(self):
self._level = self._level + 1

View File

@@ -3,9 +3,7 @@ import datetime
from queue import Queue
from interpreter.utils.params import expanse
import libs.testium as tm
from lib.tum_except import (
ETUMSyntaxError,
)
from lib.tum_except import ETUMSyntaxError
import interpreter.utils.settings as prefs
from interpreter.test_report.test_report import TestReport
from interpreter.utils.py_func_exec import PyFuncExecEngine
@@ -19,6 +17,17 @@ from interpreter.test_items.item_actions import TestItemActions
from interpreter.test_items.test_result import TestValue
def _build_item_path(item) -> str:
"""Build a breadcrumb path like 'main > Group > sub-group' from an item to root."""
parts = []
current = item
while current is not None:
name = current.name()
parts.append(name if name else f"[{current.type()}]")
current = current.parent()
return " > ".join(reversed(parts))
class TestSet:
def __init__(
self,
@@ -479,16 +488,23 @@ class TestSet:
action_name = cst.FOLDED_CHAR + it.item_cmd
seq_filename = action[action_name]["seq_filename"]
item = (it.item_class)(
action[action_name],
tree_parent,
self.status_queue,
filename=seq_filename
)
try:
item = (it.item_class)(
action[action_name],
tree_parent,
self.status_queue,
filename=seq_filename
)
except ETUMSyntaxError as e:
path = _build_item_path(tree_parent)
raise ETUMSyntaxError(
f"In: {path}\n{e._message}",
e._file or seq_filename,
) from e
item.is_folded = is_folded
child = {}
# case where the test item loads itself its descendants
if it == cst_type.TYPE_UNITTEST_FILE:
if it == cst_type.TYPE_UNITTEST:
item.setTestDir(test_dir)
child = item.load()
elif issubclass(it.item_class, TestItemActions):

View File

@@ -8,7 +8,7 @@ class TestItemEnum():
self.item_class = item_class
class TestItemType(Enum):
TYPE_UNITTEST_FILE = TestItemEnum("unittest_file", "unittest file")
TYPE_UNITTEST = TestItemEnum("unittest", "unittest")
TYPE_UNITTEST_STEP = TestItemEnum("unittest_step", "unittest step")
TYPE_CONSOLE = TestItemEnum("console", "Console")
TYPE_CONSOLE_ACTION = TestItemEnum("console_action", "Console action")
@@ -33,6 +33,8 @@ class TestItemType(Enum):
TYPE_RUN = TestItemEnum("run", "Run tum")
TYPE_JSON_RPC = TestItemEnum("json_rpc", "JSON-RPC")
TYPE_JSON_RPC_ACTION = TestItemEnum("json_rpc_action", "JSON-RPC action")
TYPE_PARALLEL = TestItemEnum("parallel", "Parallel")
TYPE_PARALLEL_BRANCH = TestItemEnum("parallel_branch", "Parallel branch")
TYPE_ROOT = TestItemEnum("default", "default")
@staticmethod

View File

@@ -1,3 +1,4 @@
import json
from threading import Lock
@@ -5,6 +6,30 @@ global_dict = {}
global_dict_lock = Lock()
_update_queue = None
def set_update_queue(q):
global _update_queue
_update_queue = q
def _push_update(key, value):
if _update_queue is None or key.startswith("_"):
return
try:
json.dumps(value)
_update_queue.put({"type": "gd_update", "key": key, "value": value})
except (TypeError, ValueError):
pass
def _push_delete(key):
if _update_queue is None or key.startswith("_"):
return
_update_queue.put({"type": "gd_delete", "key": key})
# Global dictionnary helper functions
def gd(name, default=None):
''' Function which returns a variable from the global dictionary of testium
@@ -31,6 +56,7 @@ def setgd(name, value):
'''
with global_dict_lock:
global_dict.update({name: value})
_push_update(name, value)
def delgd(name):
''' Function which removes a variable from the global dictionary of testium
@@ -44,6 +70,7 @@ def delgd(name):
del global_dict[name]
except:
pass
_push_delete(name)
def cleargd():
with global_dict_lock:

View File

@@ -189,7 +189,13 @@ class LuaProcessBase:
if tm.debug_enabled() and tm.gd("debug_rpc", False):
params.append("--verbose")
self._process = subprocess.Popen(params, env=env, cwd=func_proc_path)
self._process = subprocess.Popen(
params, env=env, cwd=func_proc_path,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
restore_signals=False,
)
self._rpc = JsonRpcClient(
"localhost", self._port, req_handler=self._req_handler
@@ -221,6 +227,11 @@ class LuaProcessBase:
return self._rpc.wait_ready(timeout)
return False
def is_alive(self):
if self._rpc is not None:
return self._rpc.is_alive()
return False
def stop(self):
"""
Stops the RPC client.

View File

@@ -158,7 +158,13 @@ class PyProcessBase:
if tm.debug_enabled() and tm.gd("debug_rpc", False):
params.append("-v")
self._process = subprocess.Popen(params, env=env, cwd=func_proc_path)
self._process = subprocess.Popen(
params, env=env, cwd=func_proc_path,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
restore_signals=False,
)
self._rpc = JsonRpcClient(
"localhost", self._port, req_handler=self._req_handler

View File

@@ -1,32 +1,102 @@
import colorama
import os
import re
import sys
import colorama
from colorama import Fore, Style
COLOR_DEFAULT = Fore.WHITE
COLOR_RESET = Fore.RESET + Style.RESET_ALL + COLOR_DEFAULT
def _detect_dark_background() -> bool:
"""Detect whether the terminal has a dark background.
def colored_string(string: str, inputs: list) -> None:
"""Function which calculate the coloring of strings with many layers.
Overlap of layers and inner layers are managed.
Tries the following methods in order:
1. ``COLORFGBG`` environment variable (Konsole, rxvt, …)
2. OSC 11 terminal query — reads the actual background colour from the
terminal emulator (xterm, VTE, kitty, WezTerm, …)
3. ``darkdetect`` module — OS-level dark-mode preference (optional dep)
Returns ``True`` for a dark background (default assumption).
"""
cols = [COLOR_DEFAULT for i in range(len(string))]
for input in inputs:
for i in range(input[0][0], input[0][1]):
cols[i] = input[1]
# --- Method 1: COLORFGBG ---
colorfgbg = os.environ.get("COLORFGBG", "")
if colorfgbg:
try:
bg = int(colorfgbg.split(";")[-1])
# 0-6: dark palette entries, 7-15: light palette entries
return bg < 7
except (ValueError, IndexError):
pass
# --- Method 2: OSC 11 terminal query ---
if sys.stdin.isatty() and sys.stdout.isatty():
try:
import select
import termios
import tty
fd = sys.stdin.fileno()
old = termios.tcgetattr(fd)
try:
tty.setraw(fd)
# Query background colour
sys.stdout.write("\033]11;?\007")
sys.stdout.flush()
ready, _, _ = select.select([sys.stdin], [], [], 0.2)
if ready:
response = ""
while True:
r2, _, _ = select.select([sys.stdin], [], [], 0.05)
if not r2:
break
chunk = os.read(fd, 64).decode("latin-1", errors="replace")
response += chunk
# Terminal answers with ESC]11;rgb:RR../GG../BB..<BEL|ST>
if response.endswith("\007") or response.endswith("\033\\"):
break
m = re.search(
r"rgb:([0-9a-fA-F]+)/([0-9a-fA-F]+)/([0-9a-fA-F]+)",
response,
)
if m:
# Components are 8- or 16-bit hex; normalise to 0-255
def _norm(h: str) -> float:
return int(h[:2], 16)
r_v = _norm(m.group(1))
g_v = _norm(m.group(2))
b_v = _norm(m.group(3))
luminance = 0.299 * r_v + 0.587 * g_v + 0.114 * b_v
return luminance < 128
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old)
except Exception:
pass
# Default: assume dark terminal
return True
def _colored_string(string: str, inputs: list, color_default: str, color_reset: str) -> str:
"""Return *string* with ANSI colour codes applied according to *inputs*.
*inputs* is a list of ``[[start, end], color_code]`` pairs.
Overlapping layers are handled: the last listed colour wins.
"""
cols = [color_default for _ in range(len(string))]
for span, color in inputs:
for i in range(span[0], span[1]):
cols[i] = color
# construction of the string
s = ""
ilast = 0
last_col = COLOR_DEFAULT
last_col = color_default
for i in range(len(string)):
if last_col != cols[i]:
s = s + string[ilast:i] + COLOR_RESET + cols[i]
s = s + string[ilast:i] + color_reset + cols[i]
ilast = i
last_col = cols[i]
return s + string[ilast:] + COLOR_RESET
return s + string[ilast:] + color_reset
class TermLog:
@@ -37,46 +107,74 @@ class TermLog:
DEBUG = ["DEBUG"]
BOOL = ["False", "True", "false", "true", "FALSE", "TRUE"]
def __init__(self, out) -> None:
"""Class used to color the stdout in batch and terminal mode."""
def __init__(self, out, dark_bg: bool = None) -> None:
"""Class used to colour the stdout in batch and terminal mode.
:param out: Underlying output stream.
:param dark_bg: ``True`` for dark background, ``False`` for light.
``None`` (default) triggers auto-detection.
"""
colorama.init()
self.out = out
self.pats = []
self.pats = self.pats + [
[re.compile('(\\"[^\\"]+\\")'), Fore.LIGHTBLUE_EX + Style.BRIGHT],
[re.compile("(\\'[^\\']+\\')"), Fore.LIGHTBLUE_EX + Style.BRIGHT],
[re.compile("(<-----|----->) step"), Fore.BLUE],
[
re.compile(
r"([\d\.]+)",
),
Fore.MAGENTA,
],
[re.compile(r"(@@\d+@@)"), Fore.BLACK],
self.residue = ""
if dark_bg is None:
dark_bg = _detect_dark_background()
if dark_bg:
color_default = Fore.WHITE
color_string = Fore.LIGHTBLUE_EX + Style.BRIGHT
color_number = Fore.MAGENTA
color_bool = Fore.MAGENTA
color_step = Fore.BLUE
color_marker = Fore.BLACK
color_warn = Fore.YELLOW
color_info = Style.BRIGHT
color_debug = Fore.BLUE + Style.BRIGHT
color_pass = Fore.GREEN + Style.BRIGHT
color_fail = Fore.RED + Style.BRIGHT
else:
color_default = Fore.RESET
color_string = Fore.BLUE
color_number = Fore.MAGENTA
color_bool = Fore.MAGENTA
color_step = Fore.BLUE
color_marker = Fore.RESET
color_warn = Fore.YELLOW + Style.BRIGHT
color_info = Fore.CYAN
color_debug = Fore.BLUE
color_pass = Fore.GREEN
color_fail = Fore.RED + Style.BRIGHT
self._color_default = color_default
self._color_reset = Fore.RESET + Style.RESET_ALL + color_default
self.pats = [
[re.compile(r'("(?:[^"]+)")'), color_string],
[re.compile(r"('(?:[^']+)')"), color_string],
[re.compile(r"(<-----|----->) step"), color_step],
[re.compile(r"([\d\.]+)"), color_number],
[re.compile(r"(@@\d+@@)"), color_marker],
]
for word in self.BOOL:
self.pats.append([re.compile("({})".format(word)), Fore.MAGENTA])
self.pats.append([re.compile(r"({})".format(word)), color_bool])
for word in self.WARN:
self.pats.append([re.compile("({})".format(word)), Fore.YELLOW])
self.pats.append([re.compile(r"({})".format(word)), color_warn])
for word in self.INFO:
self.pats.append([re.compile("({})".format(word)), Style.BRIGHT])
self.pats.append([re.compile(r"({})".format(word)), color_info])
for word in self.DEBUG:
self.pats.append([re.compile("({})".format(word)), Fore.BLUE + Style.BRIGHT])
self.pats.append([re.compile(r"({})".format(word)), color_debug])
for word in self.PASS:
self.pats.append(
[re.compile("({})".format(word)), Fore.GREEN + Style.BRIGHT]
)
self.pats.append([re.compile(r"({})".format(word)), color_pass])
for word in self.FAIL:
self.pats.append([re.compile("({})".format(word)), Fore.RED + Style.BRIGHT])
self.residue = ""
self.pats.append([re.compile(r"({})".format(word)), color_fail])
def find_pats(self, line):
spans = []
for p in self.pats:
it = p[0].finditer(line)
for m in it:
for p, color in self.pats:
for m in p.finditer(line):
if m:
spans.append([m.span(), p[1]])
spans.append([m.span(), color])
return spans
def write(self, s: str) -> None:
@@ -87,15 +185,19 @@ class TermLog:
if s[-1:] != "\n":
pos = s.rfind("\n")
if pos >= 0:
self.residue = s[pos:]
s = s[:pos]
self.residue = s[pos + 1:]
s = s[:pos + 1]
else:
# only one line
self.out.write(colored_string(s, self.find_pats(s)))
# single incomplete line — output immediately
self.out.write(_colored_string(s, self.find_pats(s),
self._color_default, self._color_reset))
return
# multiline case
for l in s.splitlines():
self.out.write(colored_string(l, self.find_pats(l)) + "\n")
# one or more complete lines
for line in s.splitlines():
self.out.write(
_colored_string(line, self.find_pats(line),
self._color_default, self._color_reset) + "\n"
)
def flush(self):
if self.residue != "":

View File

@@ -44,6 +44,7 @@ from interpreter.test_items.test_item_choices_dialog import TestItemChoicesDialo
from interpreter.test_items.test_item_console import TestItemConsole
from interpreter.test_items.test_item_run import TestItemRun
from interpreter.test_items.test_item_report import TestItemReport
from interpreter.test_items.test_item_parallel import TestItemParallel, TestItemParallelBranch
def _constants_init():
@@ -67,8 +68,10 @@ def _constants_init():
cst.TYPE_ROOT.item_class = TestItem
cst.TYPE_RUN.item_class = TestItemRun
cst.TYPE_SLEEP.item_class = TestItemSleep
cst.TYPE_UNITTEST_FILE.item_class = TestItemUnittestFile
cst.TYPE_UNITTEST.item_class = TestItemUnittestFile
cst.TYPE_VALUE_DLG.item_class = TestItemValueDialog
cst.TYPE_PARALLEL.item_class = TestItemParallel
cst.TYPE_PARALLEL_BRANCH.item_class = TestItemParallelBranch
def locate_report_file(rep_file):

View File

@@ -209,6 +209,15 @@ def OS():
return platform.system()
def text_mode():
"""Whether testium is running in text mode (batch ``-b`` or terminal ``-m``).
:return: ``True`` if running in text mode, ``False`` otherwise.
:rtype: bool
"""
return bool(globdict.gd("_text_mode", False))
def sys_encoding():
if OS() == "Windows":
enc = "oem"

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'about_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -1,11 +1,16 @@
import ast
import json
import os
import sys
import subprocess
import re
import subprocess
import sys
from PySide6.QtWidgets import QDialog
from PySide6.QtWidgets import (
QDialog, QDialogButtonBox, QHeaderView, QMenu, QMessageBox,
QPushButton, QTextEdit, QVBoxLayout,
)
from PySide6.QtGui import QSyntaxHighlighter, QTextCharFormat, QColor, QFont, QDesktopServices
from PySide6.QtCore import Qt, QUrl
from PySide6.QtCore import Qt, QUrl, Slot
from main_win.f1_win.f1_win_core import Ui_F1Dialog
@@ -16,58 +21,253 @@ class YamlHighlighter(QSyntaxHighlighter):
self.highlightingRules = []
# --- KEY formatting (before colon) ---
key_format = QTextCharFormat()
key_format.setForeground(QColor("#268bd2")) # Solarized blue
key_format.setForeground(QColor("#268bd2"))
key_format.setFontWeight(QFont.Bold)
self.highlightingRules.append((r"^\s*[^:]+(?=:)", key_format))
# --- VALUE formatting (strings) ---
value_format = QTextCharFormat()
value_format.setForeground(QColor("#2aa198")) # teal
value_format.setForeground(QColor("#2aa198"))
self.highlightingRules.append((r":\s*[^#\n]+", value_format))
# --- Booleans (true/false) ---
bool_format = QTextCharFormat()
bool_format.setForeground(QColor("#b58900")) # yellow
bool_format.setForeground(QColor("#b58900"))
bool_format.setFontWeight(QFont.Bold)
self.highlightingRules.append((r"\b(true|false)\b", bool_format))
# --- Numbers ---
num_format = QTextCharFormat()
num_format.setForeground(QColor("#d33682")) # magenta
num_format.setForeground(QColor("#d33682"))
self.highlightingRules.append((r"\b[0-9]+\b", num_format))
# --- Comments (# ...) ---
comment_format = QTextCharFormat()
comment_format.setForeground(QColor("#586e75")) # gray
comment_format.setForeground(QColor("#586e75"))
self.highlightingRules.append((r"#.*", comment_format))
def highlightBlock(self, text):
for pattern, fmt in self.highlightingRules:
for match in re.finditer(pattern, text):
start, end = match.span()
self.setFormat(start, end-start, fmt)
self.setFormat(start, end - start, fmt)
class GdVarEditDialog(QDialog):
"""JSON editor dialog for dict/list values."""
def __init__(self, key, value, parent=None):
super().__init__(parent)
self.setWindowTitle(f"Edit: {key}")
self.result_value = None
layout = QVBoxLayout(self)
self._edit = QTextEdit()
self._edit.setPlainText(json.dumps(value, indent=2))
font = QFont("Monospace")
font.setStyleHint(QFont.StyleHint.TypeWriter)
font.setPointSize(9)
self._edit.setFont(font)
layout.addWidget(self._edit)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
buttons.accepted.connect(self._on_ok)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
self.resize(400, 300)
def _on_ok(self):
try:
self.result_value = json.loads(self._edit.toPlainText())
self.accept()
except json.JSONDecodeError as e:
QMessageBox.warning(self, "Invalid JSON", str(e))
class DialogF1(QDialog):
def __init__(self, parent = None):
def __init__(self, parent=None):
super().__init__(parent)
self.ui = Ui_F1Dialog()
self.ui.setupUi(self)
self.highlighter = YamlHighlighter(self.ui.TestContentEdit.document())
self.setWindowFlags(
Qt.Window | Qt.WindowStaysOnTopHint | Qt.Tool
)
self.setWindowFlags(Qt.Window | Qt.WindowStaysOnTopHint | Qt.Tool)
self.ui.ButtLocOpen.clicked.connect(self.on_butlocopen_click)
self.ui.ButtClose.clicked.connect(self.close)
self._service = None
self._key_rows = {}
self._updating = False
self._mono_font = QFont("Monospace")
self._mono_font.setStyleHint(QFont.StyleHint.TypeWriter)
self._mono_bold_font = QFont("Monospace")
self._mono_bold_font.setStyleHint(QFont.StyleHint.TypeWriter)
self._mono_bold_font.setBold(True)
self._setup_vars_tab()
def _setup_vars_tab(self):
table = self.ui.varsTable
table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Fixed)
table.setColumnWidth(2, 36)
table.verticalHeader().setVisible(False)
table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
table.customContextMenuRequested.connect(self._on_context_menu)
table.cellChanged.connect(self._on_cell_changed)
table.setEnabled(False)
self.ui.addVarButton.setEnabled(False)
self.ui.addVarButton.clicked.connect(self._on_add_var)
def load_initial_vars(self, vars_dict: dict):
for key, value in vars_dict.items():
self.gd_var_updated(key, value)
def set_service(self, service):
self._service = service
enabled = service is not None
self.ui.varsTable.setEnabled(enabled)
self.ui.addVarButton.setEnabled(enabled)
if not enabled:
self._updating = True
try:
self.ui.varsTable.setRowCount(0)
finally:
self._updating = False
self._key_rows.clear()
@Slot(str, object)
def gd_var_updated(self, key, value):
if key in self._key_rows:
self._refresh_row(self._key_rows[key], key, value)
else:
self._updating = True
try:
row = self.ui.varsTable.rowCount()
self.ui.varsTable.insertRow(row)
finally:
self._updating = False
self._key_rows[key] = row
self._refresh_row(row, key, value)
@Slot(str)
def gd_var_deleted(self, key):
if key not in self._key_rows:
return
row = self._key_rows.pop(key)
self._updating = True
try:
self.ui.varsTable.removeRow(row)
finally:
self._updating = False
self._key_rows = {k: (r - 1 if r > row else r) for k, r in self._key_rows.items()}
def _refresh_row(self, row, key, value):
from PySide6.QtWidgets import QTableWidgetItem
self._updating = True
try:
table = self.ui.varsTable
key_item = QTableWidgetItem(key)
key_item.setFlags(key_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
key_item.setFont(self._mono_bold_font)
table.setItem(row, 0, key_item)
display = self._display_value(value)
val_item = QTableWidgetItem(display)
val_item.setData(Qt.ItemDataRole.UserRole, value)
val_item.setToolTip(self._full_tooltip(value))
val_item.setFont(self._mono_font)
if self._is_complex(value):
val_item.setFlags(val_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
table.setItem(row, 1, val_item)
if self._is_complex(value):
btn = QPushButton("[…]")
captured_key = key
btn.clicked.connect(lambda: self._on_edit_complex(captured_key))
table.setCellWidget(row, 2, btn)
else:
table.setCellWidget(row, 2, None)
table.setItem(row, 2, QTableWidgetItem())
finally:
self._updating = False
def _is_complex(self, value):
return isinstance(value, (dict, list))
def _display_value(self, value):
if self._is_complex(value):
text = repr(value)
return (text[:60] + "") if len(text) > 60 else text
return repr(value)
def _full_tooltip(self, value):
try:
text = json.dumps(value, indent=2)
except (TypeError, ValueError):
text = repr(value)
escaped = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
return f"<pre>{escaped}</pre>"
def _on_cell_changed(self, row, col):
if self._updating or col != 1 or self._service is None:
return
from PySide6.QtWidgets import QTableWidgetItem
key_item = self.ui.varsTable.item(row, 0)
val_item = self.ui.varsTable.item(row, 1)
if key_item is None or val_item is None:
return
key = key_item.text()
text = val_item.text()
try:
value = ast.literal_eval(text)
except (ValueError, SyntaxError):
value = text
self._service.set_gd_var(key, value)
def _on_edit_complex(self, key):
if key not in self._key_rows:
return
val_item = self.ui.varsTable.item(self._key_rows[key], 1)
if val_item is None:
return
value = val_item.data(Qt.ItemDataRole.UserRole)
dlg = GdVarEditDialog(key, value, self)
if dlg.exec() == QDialog.DialogCode.Accepted and self._service is not None:
self._service.set_gd_var(key, dlg.result_value)
def _on_add_var(self):
key = self.ui.newKeyEdit.text().strip()
value_text = self.ui.newValueEdit.text().strip()
if not key or self._service is None:
return
try:
value = ast.literal_eval(value_text)
except (ValueError, SyntaxError):
value = value_text
self._service.set_gd_var(key, value)
self.ui.newKeyEdit.clear()
self.ui.newValueEdit.clear()
def _on_context_menu(self, pos):
row = self.ui.varsTable.rowAt(pos.y())
if row < 0:
return
key_item = self.ui.varsTable.item(row, 0)
if key_item is None or self._service is None:
return
key = key_item.text()
menu = QMenu(self)
delete_action = menu.addAction("Delete")
if menu.exec(self.ui.varsTable.mapToGlobal(pos)) == delete_action:
self._service.del_gd_var(key)
def on_butlocopen_click(self):
file = self.ui.sequenceFileNameLineEdit.text()
if os.path.exists(file):
if sys.platform.startswith("win"): # Windows
if sys.platform.startswith("win"):
subprocess.Popen(f'explorer "{file}"')
else: # Linux / autres
else:
subprocess.Popen(["xdg-open", file])
QDesktopServices.openUrl(QUrl.fromLocalFile(file))
QDesktopServices.openUrl(QUrl.fromLocalFile(file))

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'f1_win_core.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################
@@ -16,8 +16,9 @@ from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor,
QImage, QKeySequence, QLinearGradient, QPainter,
QPalette, QPixmap, QRadialGradient, QTransform)
from PySide6.QtWidgets import (QApplication, QDialog, QFormLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QSizePolicy,
QSpacerItem, QTextEdit, QToolButton, QVBoxLayout,
QHeaderView, QLabel, QLineEdit, QPushButton,
QSizePolicy, QSpacerItem, QTabWidget, QTableWidget,
QTableWidgetItem, QTextEdit, QToolButton, QVBoxLayout,
QWidget)
import f1_win_rc
@@ -25,7 +26,7 @@ class Ui_F1Dialog(object):
def setupUi(self, F1Dialog):
if not F1Dialog.objectName():
F1Dialog.setObjectName(u"F1Dialog")
F1Dialog.resize(400, 300)
F1Dialog.resize(550, 450)
icon = QIcon()
if QIcon.hasThemeIcon(QIcon.ThemeIcon.HelpAbout):
icon = QIcon.fromTheme(QIcon.ThemeIcon.HelpAbout)
@@ -36,19 +37,20 @@ class Ui_F1Dialog(object):
F1Dialog.setLayoutDirection(Qt.LayoutDirection.LeftToRight)
self.verticalLayout_2 = QVBoxLayout(F1Dialog)
self.verticalLayout_2.setObjectName(u"verticalLayout_2")
self.horizontalLayout_2 = QHBoxLayout()
self.horizontalLayout_2.setObjectName(u"horizontalLayout_2")
self.verticalLayout_2.addLayout(self.horizontalLayout_2)
self.tabWidget = QTabWidget(F1Dialog)
self.tabWidget.setObjectName(u"tabWidget")
self.tabTestItem = QWidget()
self.tabTestItem.setObjectName(u"tabTestItem")
self.verticalLayout_tab0 = QVBoxLayout(self.tabTestItem)
self.verticalLayout_tab0.setObjectName(u"verticalLayout_tab0")
self.formLayout = QFormLayout()
self.formLayout.setObjectName(u"formLayout")
self.typeLabel = QLabel(F1Dialog)
self.typeLabel = QLabel(self.tabTestItem)
self.typeLabel.setObjectName(u"typeLabel")
self.formLayout.setWidget(0, QFormLayout.ItemRole.LabelRole, self.typeLabel)
self.typeLineEdit = QLineEdit(F1Dialog)
self.typeLineEdit = QLineEdit(self.tabTestItem)
self.typeLineEdit.setObjectName(u"typeLineEdit")
sizePolicy = QSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
sizePolicy.setHorizontalStretch(0)
@@ -59,20 +61,20 @@ class Ui_F1Dialog(object):
self.formLayout.setWidget(0, QFormLayout.ItemRole.FieldRole, self.typeLineEdit)
self.sequenceFileNameLabel = QLabel(F1Dialog)
self.sequenceFileNameLabel = QLabel(self.tabTestItem)
self.sequenceFileNameLabel.setObjectName(u"sequenceFileNameLabel")
self.formLayout.setWidget(1, QFormLayout.ItemRole.LabelRole, self.sequenceFileNameLabel)
self.horizontalLayout_3 = QHBoxLayout()
self.horizontalLayout_3.setObjectName(u"horizontalLayout_3")
self.sequenceFileNameLineEdit = QLineEdit(F1Dialog)
self.sequenceFileNameLineEdit = QLineEdit(self.tabTestItem)
self.sequenceFileNameLineEdit.setObjectName(u"sequenceFileNameLineEdit")
self.sequenceFileNameLineEdit.setReadOnly(True)
self.horizontalLayout_3.addWidget(self.sequenceFileNameLineEdit)
self.ButtLocOpen = QToolButton(F1Dialog)
self.ButtLocOpen = QToolButton(self.tabTestItem)
self.ButtLocOpen.setObjectName(u"ButtLocOpen")
self.horizontalLayout_3.addWidget(self.ButtLocOpen)
@@ -81,18 +83,61 @@ class Ui_F1Dialog(object):
self.formLayout.setLayout(1, QFormLayout.ItemRole.FieldRole, self.horizontalLayout_3)
self.verticalLayout_2.addLayout(self.formLayout)
self.verticalLayout_tab0.addLayout(self.formLayout)
self.label = QLabel(F1Dialog)
self.label = QLabel(self.tabTestItem)
self.label.setObjectName(u"label")
self.verticalLayout_2.addWidget(self.label)
self.verticalLayout_tab0.addWidget(self.label)
self.TestContentEdit = QTextEdit(F1Dialog)
self.TestContentEdit = QTextEdit(self.tabTestItem)
self.TestContentEdit.setObjectName(u"TestContentEdit")
self.TestContentEdit.setReadOnly(True)
self.verticalLayout_2.addWidget(self.TestContentEdit)
self.verticalLayout_tab0.addWidget(self.TestContentEdit)
self.tabWidget.addTab(self.tabTestItem, "")
self.tabVariables = QWidget()
self.tabVariables.setObjectName(u"tabVariables")
self.verticalLayout_tab1 = QVBoxLayout(self.tabVariables)
self.verticalLayout_tab1.setObjectName(u"verticalLayout_tab1")
self.varsTable = QTableWidget(self.tabVariables)
if (self.varsTable.columnCount() < 3):
self.varsTable.setColumnCount(3)
__qtablewidgetitem = QTableWidgetItem()
self.varsTable.setHorizontalHeaderItem(0, __qtablewidgetitem)
__qtablewidgetitem1 = QTableWidgetItem()
self.varsTable.setHorizontalHeaderItem(1, __qtablewidgetitem1)
__qtablewidgetitem2 = QTableWidgetItem()
self.varsTable.setHorizontalHeaderItem(2, __qtablewidgetitem2)
self.varsTable.setObjectName(u"varsTable")
self.verticalLayout_tab1.addWidget(self.varsTable)
self.addVarLayout = QHBoxLayout()
self.addVarLayout.setObjectName(u"addVarLayout")
self.newKeyEdit = QLineEdit(self.tabVariables)
self.newKeyEdit.setObjectName(u"newKeyEdit")
self.addVarLayout.addWidget(self.newKeyEdit)
self.newValueEdit = QLineEdit(self.tabVariables)
self.newValueEdit.setObjectName(u"newValueEdit")
self.addVarLayout.addWidget(self.newValueEdit)
self.addVarButton = QPushButton(self.tabVariables)
self.addVarButton.setObjectName(u"addVarButton")
self.addVarButton.setMaximumSize(QSize(30, 16777215))
self.addVarLayout.addWidget(self.addVarButton)
self.verticalLayout_tab1.addLayout(self.addVarLayout)
self.tabWidget.addTab(self.tabVariables, "")
self.verticalLayout_2.addWidget(self.tabWidget)
self.horizontalLayout = QHBoxLayout()
self.horizontalLayout.setObjectName(u"horizontalLayout")
@@ -113,6 +158,9 @@ class Ui_F1Dialog(object):
self.retranslateUi(F1Dialog)
self.tabWidget.setCurrentIndex(0)
QMetaObject.connectSlotsByName(F1Dialog)
# setupUi
@@ -122,6 +170,15 @@ class Ui_F1Dialog(object):
self.sequenceFileNameLabel.setText(QCoreApplication.translate("F1Dialog", u"Test file name", None))
self.ButtLocOpen.setText(QCoreApplication.translate("F1Dialog", u"...", None))
self.label.setText(QCoreApplication.translate("F1Dialog", u"Test content:", None))
self.tabWidget.setTabText(self.tabWidget.indexOf(self.tabTestItem), QCoreApplication.translate("F1Dialog", u"Test item", None))
___qtablewidgetitem = self.varsTable.horizontalHeaderItem(0)
___qtablewidgetitem.setText(QCoreApplication.translate("F1Dialog", u"Key", None))
___qtablewidgetitem1 = self.varsTable.horizontalHeaderItem(1)
___qtablewidgetitem1.setText(QCoreApplication.translate("F1Dialog", u"Value", None))
self.newKeyEdit.setPlaceholderText(QCoreApplication.translate("F1Dialog", u"New key", None))
self.newValueEdit.setPlaceholderText(QCoreApplication.translate("F1Dialog", u"Value", None))
self.addVarButton.setText(QCoreApplication.translate("F1Dialog", u"+", None))
self.tabWidget.setTabText(self.tabWidget.indexOf(self.tabVariables), QCoreApplication.translate("F1Dialog", u"Variables", None))
self.ButtClose.setText(QCoreApplication.translate("F1Dialog", u"Close", None))
# retranslateUi

View File

@@ -6,8 +6,8 @@
<rect>
<x>0</x>
<y>0</y>
<width>400</width>
<height>300</height>
<width>550</width>
<height>450</height>
</rect>
</property>
<property name="windowTitle">
@@ -22,69 +22,139 @@
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<layout class="QHBoxLayout" name="horizontalLayout_2"/>
</item>
<item>
<layout class="QFormLayout" name="formLayout">
<item row="0" column="0">
<widget class="QLabel" name="typeLabel">
<property name="text">
<string>Test step type</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QLineEdit" name="typeLineEdit">
<property name="sizePolicy">
<sizepolicy hsizetype="Fixed" vsizetype="Fixed">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="sequenceFileNameLabel">
<property name="text">
<string>Test file name</string>
</property>
</widget>
</item>
<item row="1" column="1">
<layout class="QHBoxLayout" name="horizontalLayout_3">
<widget class="QTabWidget" name="tabWidget">
<property name="currentIndex">
<number>0</number>
</property>
<!-- Tab 0: Test item -->
<widget class="QWidget" name="tabTestItem">
<attribute name="title">
<string>Test item</string>
</attribute>
<layout class="QVBoxLayout" name="verticalLayout_tab0">
<item>
<widget class="QLineEdit" name="sequenceFileNameLineEdit">
<layout class="QFormLayout" name="formLayout">
<item row="0" column="0">
<widget class="QLabel" name="typeLabel">
<property name="text">
<string>Test step type</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QLineEdit" name="typeLineEdit">
<property name="sizePolicy">
<sizepolicy hsizetype="Fixed" vsizetype="Fixed">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="sequenceFileNameLabel">
<property name="text">
<string>Test file name</string>
</property>
</widget>
</item>
<item row="1" column="1">
<layout class="QHBoxLayout" name="horizontalLayout_3">
<item>
<widget class="QLineEdit" name="sequenceFileNameLineEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="ButtLocOpen">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
</layout>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="label">
<property name="text">
<string>Test content:</string>
</property>
</widget>
</item>
<item>
<widget class="QTextEdit" name="TestContentEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
</layout>
</widget>
<!-- Tab 1: Variables -->
<widget class="QWidget" name="tabVariables">
<attribute name="title">
<string>Variables</string>
</attribute>
<layout class="QVBoxLayout" name="verticalLayout_tab1">
<item>
<widget class="QToolButton" name="ButtLocOpen">
<property name="text">
<string>...</string>
</property>
<widget class="QTableWidget" name="varsTable">
<column>
<property name="text">
<string>Key</string>
</property>
</column>
<column>
<property name="text">
<string>Value</string>
</property>
</column>
<column>
<property name="text">
<string/>
</property>
</column>
</widget>
</item>
<item>
<layout class="QHBoxLayout" name="addVarLayout">
<item>
<widget class="QLineEdit" name="newKeyEdit">
<property name="placeholderText">
<string>New key</string>
</property>
</widget>
</item>
<item>
<widget class="QLineEdit" name="newValueEdit">
<property name="placeholderText">
<string>Value</string>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="addVarButton">
<property name="text">
<string>+</string>
</property>
<property name="maximumSize">
<size>
<width>30</width>
<height>16777215</height>
</size>
</property>
</widget>
</item>
</layout>
</item>
</layout>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="label">
<property name="text">
<string>Test content:</string>
</property>
</widget>
</item>
<item>
<widget class="QTextEdit" name="TestContentEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</widget>
</item>
<item>

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'preference_core_win.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################

View File

@@ -1,6 +1,6 @@
# Resource object code (Python 3)
# Created by: object code
# Created by: The Resource Compiler for Qt version 6.10.1
# Created by: The Resource Compiler for Qt version 6.11.0
# WARNING! All changes made in this file will be lost!
from PySide6 import QtCore
@@ -1826,7 +1826,7 @@ qt_resource_struct = b"\
\x00\x00\x00\x00\x00\x02\x00\x00\x00\x01\x00\x00\x00\x01\
\x00\x00\x00\x00\x00\x00\x00\x00\
\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\
\x00\x00\x01\x9bi\x96\x0e\x1c\
\x00\x00\x01\x9b\x8f'M\xbb\
"
def qInitResources():

Binary file not shown.

After

Width:  |  Height:  |  Size: 654 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 245 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.0 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 313 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.5 KiB

View File

@@ -1,6 +1,6 @@
# Resource object code (Python 3)
# Created by: object code
# Created by: The Resource Compiler for Qt version 6.10.1
# Created by: The Resource Compiler for Qt version 6.11.0
# WARNING! All changes made in this file will be lost!
from PySide6 import QtCore
@@ -1832,7 +1832,7 @@ qt_resource_struct = b"\
\x00\x00\x00\x00\x00\x02\x00\x00\x00\x01\x00\x00\x00\x02\
\x00\x00\x00\x00\x00\x00\x00\x00\
\x00\x00\x00\x10\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\
\x00\x00\x01\x9bi\x96\x0e\x1c\
\x00\x00\x01\x9b\x8f'M\xbb\
"
def qInitResources():

View File

@@ -1,6 +1,7 @@
<RCC>
<qresource prefix="/black">
<file alias="testium_logo.png">testium_logo.png</file>
<file alias="run.png">black/run.png</file>
<file alias="red.png">red.png</file>
<file alias="gray.png">gray.png</file>
<file alias="green.png">green.png</file>
@@ -44,9 +45,12 @@
<file alias="lua.png">black/lua.png</file>
<file alias="verif.png">black/verif.png</file>
<file alias="view-refresh.png">black/view-refresh.png</file>
<file alias="parallel.png">black/parallel.png</file>
<file alias="parallel_branch.png">black/parallel_branch.png</file>
</qresource>
<qresource prefix="/white">
<file alias="testium_logo.png">testium_logo.png</file>
<file alias="run.png">white/run.png</file>
<file alias="red.png">red.png</file>
<file alias="gray.png">gray.png</file>
<file alias="green.png">green.png</file>
@@ -90,9 +94,12 @@
<file alias="lua.png">white/lua.png</file>
<file alias="verif.png">white/verif.png</file>
<file alias="view-refresh.png">white/view-refresh.png</file>
<file alias="parallel.png">white/parallel.png</file>
<file alias="parallel_branch.png">white/parallel_branch.png</file>
</qresource>
<qresource prefix="/color">
<file alias="testium_logo.png">testium_logo.png</file>
<file alias="run.png">color/run.png</file>
<file alias="red.png">red.png</file>
<file alias="gray.png">gray.png</file>
<file alias="green.png">green.png</file>
@@ -136,5 +143,7 @@
<file alias="lua.png">color/lua.png</file>
<file alias="verif.png">color/verif.png</file>
<file alias="view-refresh.png">color/view-refresh.png</file>
<file alias="parallel.png">color/parallel.png</file>
<file alias="parallel_branch.png">color/parallel_branch.png</file>
</qresource>
</RCC>

File diff suppressed because it is too large Load Diff

Binary file not shown.

After

Width:  |  Height:  |  Size: 480 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 245 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.6 KiB

View File

@@ -69,3 +69,12 @@ class TestControllerService:
def set_test_outputs(self, outputs: list) -> None:
self._ctrl.control("set_test_outputs", outputs=outputs)
def get_gd_vars(self) -> dict:
return self._ctrl.control("get_gd_vars")
def set_gd_var(self, name: str, value) -> None:
self._ctrl.control("set_gd_var", name=name, value=value)
def del_gd_var(self, name: str) -> None:
self._ctrl.control("del_gd_var", name=name)

View File

@@ -3,7 +3,8 @@ import sys
import traceback
from queue import Empty
from PySide6.QtWidgets import QApplication, QFileDialog
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication, QFileDialog, QProgressDialog
from interpreter.process import TestProcess
from interpreter.utils.test_ctrl import TestSetController
@@ -29,24 +30,55 @@ class TestFileManager:
):
w.test_service.stop()
w.test_service.close()
w.test_proc.join()
w.test_proc.join(timeout=5)
if w.test_proc.is_alive():
w.test_proc.terminate()
w.test_proc.join(timeout=2)
if w.test_proc.is_alive():
w.test_proc.kill()
w.test_proc.join()
del w.test_proc
w.test_proc = None
del w.test_service
w.test_service = None
w.d_f1_win.set_service(None)
del w.ts_controller
w.ts_controller = None
def reload(self, file_name: str):
w = self._win
w.disconnect_signals()
# Snapshot user-selected checkboxes and fold state so they survive a
# reload of the same file (same logic as session-restore through prefs).
previous_check_list = w.treeTests.getCheckList()
previous_fold_list = w.treeTests.getFoldList()
previous_count = w.treeTests.getItemCount()
self.clear_process()
self.load(file_name)
if self.load(file_name) and w.test_service is not None:
if w.treeTests.getItemCount() == previous_count:
w.treeTests.restoreCheckList(previous_check_list, w.test_service)
w.treeTests.restoreFoldList(previous_fold_list)
w.reconnect_signals()
def _make_progress(self, w):
progress = QProgressDialog("Starting test process…", None, 0, 0, w)
progress.setWindowTitle("Loading")
progress.setWindowFlags(Qt.Dialog | Qt.CustomizeWindowHint | Qt.WindowTitleHint)
progress.setWindowModality(Qt.WindowModal)
progress.setMinimumDuration(0)
progress.setMinimumWidth(320)
progress._force_close = False
progress.closeEvent = lambda e: e.accept() if progress._force_close else e.ignore()
return progress
def _close_progress(self, progress):
progress._force_close = True
progress.close()
def load(self, file_name: str) -> bool:
"""Load a test file. Returns True on success, False otherwise."""
w = self._win
progress = None
try:
if not file_name:
raise ETUMFileError("No file to load")
@@ -59,9 +91,14 @@ class TestFileManager:
if not os.path.isfile(file_name):
raise ETUMFileError("Could not find %s file" % file_name)
progress = self._make_progress(w)
progress.show()
QApplication.processEvents()
w.testFile = None
w.ts_controller = TestSetController()
w.test_service = TestControllerService(w.ts_controller)
w.d_f1_win.set_service(w.test_service)
w.test_proc = TestProcess(
file_name,
w.status_queue,
@@ -71,29 +108,38 @@ class TestFileManager:
self._defaults_for_process(),
)
w.test_proc.start()
progress.setLabelText("Loading test file…")
while w.test_proc.is_alive():
try:
if w.test_service.loaded(timeout=1.0):
if w.test_service.loaded(timeout=0.05):
break
except Empty:
w.test_service.clear()
QApplication.processEvents()
if not w.test_proc.is_alive():
del w.test_proc
w.test_proc = None
del w.test_service
w.test_service = None
w.d_f1_win.set_service(None)
del w.ts_controller
w.ts_controller = None
raise ETUMRuntimeError(
"Test could not be loaded (test process crashed for any reason)"
)
progress.setLabelText("Building test tree…")
QApplication.processEvents()
test_data = w.test_service.tree()
w.treeTests.clear()
QApplication.processEvents()
w.treeTests.loadTestRecursively(w.treeTests.invisibleRootItem(), test_data)
self._close_progress(progress)
progress = None
w.treeTests.setFoldDefault()
w.treeTests.updateTreeSkipState(w.test_service)
w.d_f1_win.load_initial_vars(w.test_service.get_gd_vars())
w.checkSelect.setChecked(True)
w.testFile = file_name
@@ -109,6 +155,8 @@ class TestFileManager:
w.show_checkboxes()
return True
except:
if progress is not None:
self._close_progress(progress)
w.statusBar().showMessage("No test file could be loaded", 10000)
w.treeTests.clear()
print(traceback.format_exc())

Some files were not shown because too many files have changed in this diff Show More