autocal layout first design

This commit is contained in:
ykcheungab
2026-09-20 21:15:39 +08:00
parent 07d7c3bad6
commit e276b5de0f
92 changed files with 18071 additions and 64 deletions
+137
View File
@@ -0,0 +1,137 @@
"""Fixtures for the AutoCal tests.
Everything here runs offscreen with no controller and no hardware. Two
environment facts are set before Qt is imported:
* ``QT_QPA_PLATFORM=offscreen`` so a window can be built without a display;
* ``QT_QPA_FONTDIR`` because the offscreen platform finds no fonts on Windows
otherwise, and every label would measure as zero-width.
The suite requires PySide6 — the app's only Qt binding. Where it is missing
(an older interpreter, or a machine with no way to install it) the tests are
reported as skipped rather than failing, so a broken environment cannot look
like a broken application.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
import pytest
os.environ.setdefault("ORISYS_APP_ID", "autocal")
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
if sys.platform == "win32" and "QT_QPA_FONTDIR" not in os.environ:
_fonts = Path(os.environ.get("WINDIR", r"C:\Windows")) / "Fonts"
if _fonts.is_dir():
os.environ["QT_QPA_FONTDIR"] = str(_fonts)
def _have_pyside6() -> tuple[bool, str]:
try:
import PySide6
except ImportError as exc:
return False, str(exc)
return True, getattr(PySide6, "__version__", "unknown")
HAVE_PYSIDE6, PYSIDE6_VERSION = _have_pyside6()
if not HAVE_PYSIDE6: # pragma: no cover - depends on the machine
collect_ignore_glob = ["test_*.py"]
@pytest.fixture(scope="session")
def qt_app():
"""One ``QApplication`` per session (Qt forbids more than one)."""
if not HAVE_PYSIDE6: # pragma: no cover
pytest.skip("PySide6 is not installed")
from PySide6.QtWidgets import QApplication
app = QApplication.instance() or QApplication([])
yield app
SCRATCH_ROOT = Path(__file__).resolve().parents[1] / "outputs" / "_test_scratch"
def _scratch_dir() -> Path:
"""A fresh scratch directory under the app's own (gitignored) outputs."""
import time
path = SCRATCH_ROOT / f"{os.getpid()}-{int(time.time() * 1000) % 1_000_000}"
path.mkdir(parents=True, exist_ok=True)
return path
@pytest.fixture
def make_window(qt_app, monkeypatch):
"""Build a window whose preference writes stay inside a scratch directory.
Deliberately not ``tmp_path``: this suite has to run inside a sandboxed
shell, and ``tmp_path`` depends on a writable process temp root that such an
environment may not provide. Using the app's own ``outputs/`` (already
gitignored) removes that dependency entirely.
"""
import shutil
from orisys_app_shared import paths as paths_mod
from orisys_app_shared import prefs as prefs_pkg
from orisys_autocal import preferences as prefs_mod
from orisys_autocal.motion import DemoTransport
from orisys_autocal.ui.window import AutoCalMainWindow
scratch = _scratch_dir()
prefs_file = scratch / "preference.json"
monkeypatch.setattr(paths_mod, "preference_path", lambda: prefs_file)
monkeypatch.setattr(prefs_pkg.store, "preference_path", lambda: prefs_file)
monkeypatch.setattr(prefs_mod, "preference_path", lambda: prefs_file)
created = []
def _make(**overrides):
prefs = prefs_mod.default_preferences()
prefs.update(overrides)
# The translator is process-global, so a window built after another test
# switched language would otherwise keep that language and then be
# labelled with the wrong locale.
from orisys_app_shared.i18n import set_locale
set_locale(str(prefs.get("language") or "zh_CN"))
window = AutoCalMainWindow(prefs=prefs)
# Swap in the simulated platform: no test in this suite touches a DLL.
window.poller.service._transport = DemoTransport() # noqa: SLF001
created.append(window)
return window
yield _make
for window in created:
try:
window.poller.shutdown()
window.close()
except Exception: # noqa: BLE001 - teardown must not mask a failure
pass
shutil.rmtree(scratch, ignore_errors=True)
def pump(app, seconds: float = 0.2) -> None:
"""Run the event loop for a short, bounded time."""
import time
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
app.processEvents()
time.sleep(0.005)
def visible_descendants(widget):
"""Yield every visible descendant of ``widget``."""
for child in widget.findChildren(object):
if not hasattr(child, "isVisible") or not hasattr(child, "geometry"):
continue
if child.isVisible():
yield child
+185
View File
@@ -0,0 +1,185 @@
"""Binding policy for AutoCal.
AutoCal is **PySide6 only**, and this file is the guard for that:
* no module may import PyQt5, PySide2 or PyQt6 — the app's environment has
PySide6 and nothing else, so such an import is a hard start-up failure;
* no module may use an unscoped Qt enumerator (``Qt.AlignLeft``), which PySide6
removed in favour of ``Qt.AlignmentFlag.AlignLeft``;
* ``shared/`` — which serves both this app and the PyQt5-based studio — must
import no Qt binding at all;
* the controller layer must stay importable from a headless process.
The reason the binding matters at all is licensing, not taste: PyQt5's free
option is GPLv3-or-commercial, while PySide6 is LGPLv3, so an internal
closed-source tool needs no licence purchase. See ``README.md``.
"""
from __future__ import annotations
import ast
from pathlib import Path
APP_SRC = Path(__file__).resolve().parents[1] / "src"
SHARED_SRC = Path(__file__).resolve().parents[3] / "shared" / "src"
#: Bindings the app must never import.
FORBIDDEN_BINDINGS = {"PyQt5", "PyQt6", "PySide2"}
#: The one binding allowed, and where it may be named.
ALLOWED_BINDING = "PySide6"
#: Scoped enum holders we care about, with the value names actually used.
_SCOPED_ENUMS = {
"Qt": (
"AlignmentFlag",
"ItemDataRole",
"Orientation",
"ConnectionType",
"TextInteractionFlag",
"TransformationMode",
"ToolButtonStyle",
"FocusPolicy",
"ScrollBarPolicy",
),
"QTableWidget": ("SelectionMode", "EditTrigger"),
"QSizePolicy": ("Policy",),
"QHeaderView": ("ResizeMode",),
"QMessageBox": ("StandardButton", "Icon"),
"QDialog": ("DialogCode",),
"QAbstractItemView": ("SelectionMode", "EditTrigger"),
}
def _python_files(root: Path) -> list[Path]:
return sorted(path for path in root.rglob("*.py") if "__pycache__" not in path.parts)
def _imported_packages(path: Path) -> set[str]:
"""Every top-level package name imported by ``path``."""
tree = ast.parse(path.read_text(encoding="utf-8"))
found: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
found.add(alias.name.split(".")[0])
elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
found.add(node.module.split(".")[0])
return found
def _unscoped_enum_references(path: Path) -> list[str]:
"""Attribute reads like ``Qt.AlignLeft`` — the spelling PySide6 rejects.
Parsed rather than pattern-matched for two reasons: a regex also hits prose
in a comment or docstring (which is exactly where the scoped spelling gets
*explained*), and ``QMessageBox.information(...)`` looks like an enumerator
but is a static method. The parent map separates the two, and
``Qt.AlignmentFlag.AlignLeft`` is an inner attribute on an allow-listed enum
class rather than a plain name, so it is never a candidate.
"""
tree = ast.parse(path.read_text(encoding="utf-8"))
called = {
id(node.func)
for node in ast.walk(tree)
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
}
found: set[str] = set()
for node in ast.walk(tree):
if not isinstance(node, ast.Attribute) or not isinstance(node.value, ast.Name):
continue
if id(node) in called:
continue # a method or function call, not an enumerator
holder = node.value.id
enum_names = _SCOPED_ENUMS.get(holder)
if not enum_names or node.attr in enum_names:
continue
found.add(f"{holder}.{node.attr}")
return sorted(found)
def test_no_module_imports_a_forbidden_binding():
offenders = []
for path in _python_files(APP_SRC):
bad = _imported_packages(path) & FORBIDDEN_BINDINGS
if bad:
offenders.append(f"{path.name}: {sorted(bad)}")
assert not offenders, (
f"AutoCal is {ALLOWED_BINDING}-only; these modules import another "
f"binding: {offenders}"
)
def test_every_ui_module_imports_the_allowed_binding():
"""A UI file that names no binding would be using one it does not declare."""
offenders = []
for path in _python_files(APP_SRC / "orisys_autocal" / "ui"):
if path.name == "__init__.py":
continue # a re-export barrel, not a widget module
if ALLOWED_BINDING not in _imported_packages(path):
offenders.append(path.name)
assert not offenders, f"these UI modules do not import {ALLOWED_BINDING}: {offenders}"
def test_no_module_uses_an_unscoped_enum():
offenders = []
for path in _python_files(APP_SRC):
references = _unscoped_enum_references(path)
if references:
offenders.append(f"{path.name}: {references}")
assert not offenders, (
"unscoped Qt enumerators were removed in PySide6; use the scoped "
f"spelling: {offenders}"
)
def test_the_controller_interface_is_qt_free():
"""The controller and its guards stay importable from a headless process.
``poller`` is deliberately absent from this check: it *is* the Qt worker
bridge (``QThread``, signals), and it lives beside the service precisely so
that the service, the transports, the models and the error map do not need
a GUI to be exercised.
"""
qt_free = ("models.py", "errors.py", "units.py", "service.py", "transport.py")
offenders = []
for path in _python_files(APP_SRC / "orisys_autocal" / "motion"):
if path.name not in qt_free:
continue
qt_imports = _imported_packages(path) & {"PySide2", "PySide6", "PyQt5", "PyQt6"}
if qt_imports:
offenders.append(f"{path.name}: {sorted(qt_imports)}")
assert not offenders, offenders
def test_shared_imports_no_qt_binding():
"""Shared code serves both products, so it must not pick a side."""
offenders = []
for path in _python_files(SHARED_SRC):
qt_imports = _imported_packages(path) & {"PySide2", "PySide6"}
if qt_imports:
offenders.append(f"{path.name}: {sorted(qt_imports)}")
assert not offenders, (
"shared code must not import PySide2/PySide6 (studio is PyQt5): "
f"{offenders}"
)
def test_zmotion_client_imports_no_qt_binding():
zmotion = SHARED_SRC / "orisys_app_shared" / "zmotion"
offenders = []
for path in _python_files(zmotion):
qt_imports = _imported_packages(path) & {"PySide2", "PySide6", "PyQt5", "PyQt6"}
if qt_imports:
offenders.append(f"{path.name}: {sorted(qt_imports)}")
assert not offenders, offenders
def test_the_app_declares_its_binding_dependency():
"""The packaging must state PySide6, not leave it to chance."""
pyproject = (Path(__file__).resolve().parents[1] / "pyproject.toml").read_text(
encoding="utf-8"
)
assert ALLOWED_BINDING in pyproject
for forbidden in FORBIDDEN_BINDINGS:
assert f'"{forbidden}' not in pyproject, forbidden
+54
View File
@@ -0,0 +1,54 @@
"""Controller return-code translation.
The vendor SDK header defines the auxiliary error block at 30000; everything
else is the firmware's own number. The point of these tests is that AutoCal
never invents a meaning for a code it cannot read out of that header.
"""
from __future__ import annotations
import pytest
from orisys_autocal.motion import errors
DOCUMENTED = sorted(errors.MESSAGES)
def test_zero_is_success_in_both_languages():
assert errors.describe(0) == "成功"
assert errors.describe(0, language="en_US") == "OK"
def test_the_documented_block_is_the_vendor_offset():
assert errors.ERR_AUX_OFFSET == 30000
assert DOCUMENTED == [30000, 30001, 30002, 30003, 30004, 30005]
@pytest.mark.parametrize("code", DOCUMENTED)
def test_every_documented_code_has_both_languages(code):
zh = errors.describe(code)
en = errors.describe(code, language="en_US")
assert zh and en and zh != en
@pytest.mark.parametrize("code", DOCUMENTED)
def test_no_documented_message_is_empty(code):
assert errors.describe(code).strip()
assert errors.describe(code, language="en_US").strip()
def test_an_unknown_code_keeps_its_number():
"""A guessed meaning on a motion controller is worse than 'unknown'."""
message = errors.describe(12345)
assert "12345" in message
assert message == "错误码 12345"
def test_an_unknown_code_points_at_zdevelop():
message = errors.describe_with_hint(-1)
assert "-1" in message
assert errors.UNKNOWN_HINT in message
def test_a_documented_code_gets_no_hint_appended():
assert errors.describe_with_hint(30000) == errors.describe(30000)
+469
View File
@@ -0,0 +1,469 @@
"""The single-page shell, from an operator's point of view.
These assertions are about the layout contract rather than any one panel:
* the motion status strip is outside the tabs, so it survives every tab change;
* all five tabs exist, are labelled, and their pages fit;
* the right-hand readout rail shows every drive and issues no per-axis command;
* the window refuses to shrink below the size at which the page would clip.
The overflow check lives in ``test_panel_smoke.py`` because it needs the colour
tables; what is here is the structure.
"""
from __future__ import annotations
import pytest
from conftest import pump
from orisys_autocal.ui.workspace import TABS
def _show(window, size=(1600, 900)):
from PySide6.QtWidgets import QApplication
window.resize(*size)
window.show()
QApplication.processEvents()
return window
def test_the_tab_set_is_the_documented_one(make_window):
window = _show(make_window())
assert window.workspace.page_keys() == [key for key, _ in TABS]
def test_every_tab_has_a_localised_label(make_window):
from orisys_app_shared.i18n import t
window = _show(make_window())
for index, (_key, locale_key) in enumerate(TABS):
label = window.workspace.tabText(index)
assert label, f"tab {index} has no label"
assert label == t(locale_key)
assert label != locale_key, f"tab {index} shows its raw key"
def test_the_status_strip_survives_every_tab_change(make_window):
"""The whole point of putting the strip above the tabs."""
window = _show(make_window())
for key in window.workspace.page_keys():
window.workspace.set_current_key(key)
assert window.status_strip.isVisible(), f"the strip hid on tab {key}"
assert window.status_strip.height() <= 48
def test_the_strip_reports_connection_and_bus_state(make_window):
from orisys_app_shared.i18n import t
from orisys_autocal.motion import BusStatus, Snapshot
window = _show(make_window())
strip = window.status_strip
strip.set_snapshot(Snapshot(connected=False))
assert strip.chip_connection.text() == t("connection.state_closed")
assert strip.chip_init.text() == t("bus.state_pending")
strip.set_snapshot(
Snapshot(connected=True, bus=BusStatus(bus_type="ecat", init_status=1, total_axes=5))
)
assert strip.chip_connection.text() == t("connection.state_open")
assert strip.chip_bus.text() == t("bus.type_ecat")
assert strip.chip_init.text() == t("bus.state_ok")
def test_a_refusal_stays_visible_on_the_strip(make_window):
"""A command refused while the operator was on another tab must not vanish."""
window = _show(make_window())
window.workspace.set_current_key("sensor")
window.status_strip.set_action_result("轴0 回零模式未设置", False)
assert "回零模式" in window.status_strip.action_label.text()
def test_the_readout_rail_covers_every_axis(make_window):
window = _show(make_window())
assert len(window.axis_readout._rows) == window.axis_panel.axis_count == 5
def test_the_readout_rail_shows_position_and_state(make_window):
from orisys_autocal.motion import AxisState, Snapshot
window = _show(make_window())
snapshot = Snapshot(
connected=True,
axes=(
AxisState(axis=0, enabled=True, units=10.0, dpos=1000.0, idle=True),
AxisState(axis=1, enabled=False, units=10.0, dpos=0.0, idle=True),
),
)
window.axis_readout.update_snapshot(snapshot)
# 1000 user units at units=10 is the documented 2.000 mm, and the readout
# shows both so an operator never has to convert in their head.
text = window.axis_readout._rows[0]["position"].text()
assert "1000.000" in text and "2.000 mm" in text, text
assert window.axis_readout._rows[1]["state"].text()
def test_the_readout_rail_issues_no_per_axis_motion(make_window):
"""Per-axis commands live in one place only: the Axis-debug tab."""
window = _show(make_window())
rail = window.axis_readout
assert not hasattr(rail, "jogRequested")
assert not hasattr(rail, "moveAbsRequested")
assert hasattr(rail, "stopAllRequested")
assert hasattr(rail, "enableAllRequested")
def test_enable_all_toggles_with_the_current_state(make_window):
from orisys_autocal.motion import AxisState, Snapshot
window = _show(make_window())
requested: list[bool] = []
window.axis_readout.enableAllRequested.connect(requested.append)
window.axis_readout.update_snapshot(
Snapshot(connected=True, axes=tuple(AxisState(axis=a, enabled=True) for a in range(5)))
)
window.axis_readout.enable_all_button.click()
assert requested == [False], "enable-all did not become disable-all"
window.axis_readout.update_snapshot(Snapshot(connected=True, axes=()))
window.axis_readout.enable_all_button.click()
assert requested == [False, True]
def test_the_window_refuses_to_shrink_below_the_page(make_window):
window = _show(make_window())
window.resize(800, 600)
assert window.width() >= window.minimumWidth()
assert window.height() >= window.minimumHeight()
assert window.minimumWidth() >= 1300
def test_the_bus_data_group_starts_collapsed(make_window):
"""It is commissioning-only; expanded, it cost the page ~270 px."""
window = _show(make_window())
assert not window.bus_data_group.isChecked()
assert not window.bus_data_panel.isVisible()
def test_expanding_bus_data_does_not_resize_the_window(make_window):
window = _show(make_window())
before = (window.width(), window.height())
window.bus_data_group.setChecked(True)
pump(_app(), 0.1)
assert (window.width(), window.height()) == before
def _app():
from PySide6.QtWidgets import QApplication
return QApplication.instance()
def test_the_sequence_tab_shows_the_documented_profile_fields(make_window):
from orisys_autocal.ui.sequence_panel import PROFILE_FIELDS
window = _show(make_window())
panel = window.sequence_panel
assert panel.profile_keys() == [key for key, _locale, _kind in PROFILE_FIELDS]
def test_the_sequence_panel_exposes_its_three_zones(make_window):
window = _show(make_window())
panel = window.sequence_panel
assert panel.validation_checks() == [
"on_surface",
"within_travel",
"contact_confirmable",
]
assert panel.point_row_count() > 0, "the point list is empty"
assert panel.summary_label.text()
def test_every_pending_control_names_the_phase_that_owns_it(make_window):
"""A disabled control must say why, or it reads as a broken button."""
window = _show(make_window())
panel = window.sequence_panel
assert panel.pending_controls() == [
"generate",
"simulate",
"dry_run",
"save",
"load",
]
for name, phase in panel.pending_phases().items():
button = panel._pending_buttons[name]
assert not button.isEnabled(), f"{name} is usable but not implemented"
assert phase in button.toolTip(), f"{name} does not name its phase"
assert button.text(), f"{name} has no label"
def test_the_sequence_tab_hands_out_no_runnable_steps(make_window):
"""A runner must not be able to start from points nothing has validated."""
window = _show(make_window())
assert window.sequence_panel.planned_steps() == []
def test_point_coordinates_are_generated_not_typed(make_window):
"""A typed coordinate is one nothing has checked against the sensor."""
from orisys_autocal.ui.sequence_panel import _EDITABLE_POINT_COLUMNS
assert _EDITABLE_POINT_COLUMNS == ()
window = _show(make_window())
panel = window.sequence_panel
# The point list is a real table, so it scrolls instead of clipping rows.
assert panel.table.rowCount() == panel.point_row_count()
assert panel.point_row(0)["x"] == "0.000"
# Every cell is read-only: the table's edit triggers are off, so a generated
# coordinate cannot be typed over.
from PySide6.QtWidgets import QTableWidget
assert panel.table.editTriggers() == QTableWidget.EditTrigger.NoEditTriggers
def test_the_profile_round_trips_through_preferences(make_window):
window = _show(make_window())
panel = window.sequence_panel
panel.field("pitch_mm").setText("3.5")
panel.field("repeats").setText("4")
window._persist_prefs()
stored = window.prefs["sequence_profile"]
assert stored["pitch_mm"] == 3.5
assert stored["repeats"] == 4
def test_an_unusable_profile_number_falls_back_rather_than_raising(make_window):
from orisys_autocal.preferences import default_preferences
window = _show(make_window())
panel = window.sequence_panel
panel.field("pitch_mm").setText("not a number")
assert panel.profile()["pitch_mm"] == 0.0
# A blank repeats field must not become zero repeats.
panel.field("repeats").setText("")
assert panel.profile()["repeats"] >= 1
assert "pitch_mm" in default_preferences()["sequence_profile"]
def test_the_stored_profile_is_loaded_into_the_form(make_window):
window = _show(make_window(sequence_profile={"pitch_mm": 7.5, "order": "row"}))
panel = window.sequence_panel
assert panel.profile()["pitch_mm"] == 7.5
assert panel.profile()["order"] == "row"
def test_the_axis_table_shows_seven_columns(make_window):
from orisys_autocal.ui.axis_panel import COLUMNS
window = _show(make_window())
assert len(COLUMNS) == 7
assert window.axis_panel.table.columnCount() == 8 # seven data + actions
def test_the_axis_detail_line_carries_what_the_table_dropped(make_window):
"""Axis type, accel, feedback and the status word must stay reachable."""
from orisys_autocal.motion import AxisState, Snapshot
window = _show(make_window())
window.axis_panel.set_selected_axis(3)
window.axis_panel.update_snapshot(
Snapshot(
connected=True,
axes=(AxisState(axis=3, enabled=True, atype=65, accel=1234.0, mpos=5.0),),
)
)
detail = window.axis_panel._detail_label.text()
assert "65" in detail and "1234" in detail
assert "3" in detail
def test_the_axis_detail_follows_the_selected_axis(make_window):
from orisys_autocal.motion import AxisState, Snapshot
window = _show(make_window())
window.axis_panel.update_snapshot(
Snapshot(
connected=True,
axes=(AxisState(axis=0, atype=65), AxisState(axis=1, atype=67)),
)
)
window.axis_panel.set_selected_axis(0)
assert "65" in window.axis_panel._detail_label.text()
window.axis_panel.set_selected_axis(1)
assert "67" in window.axis_panel._detail_label.text()
def test_the_window_defaults_to_the_design_resolution(make_window):
from orisys_autocal.ui.window import DEFAULT_WINDOW_SIZE, MINIMUM_WINDOW_SIZE
assert DEFAULT_WINDOW_SIZE == (1920, 1080)
assert MINIMUM_WINDOW_SIZE[0] >= 1440
def test_the_default_theme_is_light(make_window):
from orisys_autocal.preferences import DEFAULT_THEME
assert DEFAULT_THEME == "light"
window = _show(make_window())
assert window._theme == "light"
def test_pausing_the_strip_pauses_the_poller(make_window):
window = _show(make_window())
window.status_strip.pause_button.setChecked(True)
window._on_pause_polling()
assert window.poller.paused
window.status_strip.pause_button.setChecked(False)
window._on_pause_polling()
assert not window.poller.paused
def test_the_diagnostics_footer_does_not_move_the_layout(make_window):
window = _show(make_window())
splitters = window.main_splitter.sizes()
window._set_diagnostics(True)
window._set_diagnostics(False)
assert window.main_splitter.sizes() == splitters
def test_splitter_sizes_round_trip_through_preferences(make_window):
window = _show(make_window())
window.main_splitter.setSizes([340, 1500])
window._persist_prefs()
stored = window.prefs["main_splitter_sizes"]
assert len(stored) == 2
assert stored[1] > stored[0]
def test_the_rail_is_a_single_scrollable_column(make_window):
"""Two thin rails bracketing the workspace cost it 240 px at every size."""
from orisys_autocal.ui.window import MAIN_SPLITTER_SIZES
assert len(MAIN_SPLITTER_SIZES) == 2, "the layout is back to three columns"
window = _show(make_window())
assert window.main_splitter.count() == 2
assert window.left_rail.widget() is window.rail_content
# The readout lives in the rail now, not in a rail of its own.
assert window.rail_content.findChild(type(window.axis_readout)) is window.axis_readout
def test_the_rail_scrolls_rather_than_clipping_its_controls(make_window):
"""At 900 px high the three rail groups do not fit; the rail must scroll."""
window = _show(make_window(), (1440, 900))
viewport = window.left_rail.viewport().height()
content = window.rail_content.sizeHint().height()
scroll = window.left_rail.verticalScrollBar()
assert content <= viewport or scroll.maximum() > 0, (
f"rail content is {content}px in a {viewport}px viewport and cannot scroll"
)
# --------------------------------------------------------------- run controls
def test_the_run_bar_is_visible_from_every_tab(make_window):
"""Abort must never require finding the tab that started the run."""
window = _show(make_window())
for key in window.workspace.page_keys():
window.workspace.set_current_key(key)
assert window.run_bar.isVisible(), f"the run bar hid on tab {key}"
def test_run_is_wired_and_refuses_by_naming_its_phase(make_window):
window = _show(make_window())
window.run_bar.run_button.click()
assert "P7" in window.status_strip.action_label.text()
def test_abort_stops_the_axes_and_is_enabled_once_a_run_exists(make_window):
window = _show(make_window())
assert not window.run_bar.abort_button.isEnabled()
requested = []
window.run_bar.abortRequested.connect(lambda: requested.append(True))
window.run_bar.set_run(True)
assert window.run_bar.abort_button.isEnabled()
window.run_bar.abort_button.click()
assert requested == [True]
assert window.run_bar.phase == "failed"
def test_the_run_bar_shows_progress_and_timing(make_window):
window = _show(make_window())
window.run_bar.set_progress(7, 40)
assert "7" in window.run_bar.step_label.text()
assert "40" in window.run_bar.step_label.text()
assert window.run_bar.progress.maximum() == 40
assert window.run_bar.progress.value() == 7
window.run_bar.set_timing(125.0, 300.0)
assert "2:05" in window.run_bar.time_label.text()
# An unknown estimate must say so rather than showing a wrong number.
window.run_bar.set_timing(125.0, None)
assert "2:05" in window.run_bar.time_label.text()
# ---------------------------------------------------------------- poll health
def test_the_health_chip_reports_the_age_of_the_last_reading(make_window):
from orisys_autocal.motion import Snapshot
window = _show(make_window())
strip = window.status_strip
strip.set_snapshot(Snapshot(connected=True))
assert strip.chip_health.text()
assert strip.chip_health._tone == "ok"
def test_the_health_chip_ages_when_the_poll_stops(make_window):
"""A stalled poll must not keep looking healthy."""
from orisys_autocal.motion import Snapshot
window = _show(make_window())
strip = window.status_strip
strip.set_poll_interval_ms(100)
strip.set_snapshot(Snapshot(connected=True))
strip._last_snapshot_at -= 1.5 # fifteen intervals ago
strip.refresh_health()
assert strip.chip_health._tone == "error", strip.chip_health.text()
def test_the_health_chip_says_paused_rather_than_ageing(make_window):
from orisys_autocal.motion import Snapshot
window = _show(make_window())
strip = window.status_strip
strip.set_snapshot(Snapshot(connected=True))
strip.set_paused(True)
strip.refresh_health()
assert strip.chip_health._tone == "muted"
assert strip.chip_health.text() != ""
def test_the_status_bar_uses_its_own_theme_token(make_window):
"""The brand navy reads well as a header and heavy as a footer."""
from orisys_app_shared.theme import load_colors
for theme in ("light", "dark"):
colours = load_colors(theme)
assert "status_bar_bg" in colours, theme
assert "status_bar_text" in colours, theme
window = _show(make_window())
assert "QStatusBar" in window.statusBar().styleSheet()
def test_language_switch_relabels_the_new_shell(make_window):
from orisys_app_shared.i18n import t
window = _show(make_window(language="zh_CN"))
zh_tabs = [window.workspace.tabText(i) for i in range(window.workspace.count())]
window._set_language("en_US")
assert window.status_strip.title.text() == t("status.motion")
assert window.run_bar.title.text() == t("run.title")
en_tabs = [window.workspace.tabText(i) for i in range(window.workspace.count())]
assert en_tabs != zh_tabs
assert all(en_tabs)
@@ -0,0 +1,88 @@
"""Locale files: same keys everywhere, and a real fallback.
``Translator`` returns the key itself when a string is missing, which is silent
in production — a label would simply read ``axes.nope``. The studio suite guards
this by comparing key sets; the same guard applies here.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from orisys_autocal.preferences import load_preferences
LOCALES = Path(__file__).resolve().parents[1] / "config" / "locales"
LANGUAGES = ("zh_CN", "en_US")
def _load(language: str) -> dict:
return json.loads((LOCALES / f"{language}.json").read_text(encoding="utf-8"))
def _flatten(node: dict, prefix: str = "") -> set[str]:
keys: set[str] = set()
for key, value in node.items():
path = f"{prefix}{key}"
if isinstance(value, dict):
keys |= _flatten(value, f"{path}.")
else:
keys.add(path)
return keys
def test_every_locale_file_exists():
for language in LANGUAGES:
assert (LOCALES / f"{language}.json").is_file(), language
def test_locales_have_identical_key_sets():
reference = _flatten(_load("zh_CN"))
for language in LANGUAGES[1:]:
other = _flatten(_load(language))
assert other == reference, {
f"{language}_only": sorted(other - reference),
f"zh_CN_only": sorted(reference - other),
}
def test_no_translation_is_empty_where_a_label_is_expected():
"""A blank string renders as an empty widget, which looks like a bug."""
for language in LANGUAGES:
for key, value in _walk(_load(language)):
assert str(value).strip(), f"{language}: {key} is blank"
def test_the_product_name_is_localised():
"""The window title is Chinese in zh_CN and English in en_US."""
assert _load("zh_CN")["app"]["title"] != _load("en_US")["app"]["title"]
@pytest.mark.parametrize("language", LANGUAGES)
def test_every_locale_loads_through_the_translator(language):
from orisys_app_shared.i18n import Translator
translator = Translator(LOCALES, locale=language)
for key in sorted(_flatten(_load(language))):
assert translator.t(key) != key, f"{language}: {key} did not resolve"
def test_the_default_preferences_file_matches_the_schema_shape():
"""A shipped default that names a key the app does not have is dead weight."""
shipped = json.loads(
(LOCALES.parent / "preferences.default.json").read_text(encoding="utf-8")
)
schema = load_preferences()
for key in shipped:
assert key in schema, f"preferences.default.json has an unknown key: {key}"
def _walk(node, prefix: str = ""):
for key, value in node.items():
path = f"{prefix}{key}"
if isinstance(value, dict):
yield from _walk(value, f"{path}.")
else:
yield path, value
+328
View File
@@ -0,0 +1,328 @@
"""Guard and state-machine behaviour of the motion service.
Driven against the simulated platform, so every assertion here is about AutoCal's
own logic: what it refuses, what it reads, and when. No test in this file touches
a controller or a DLL.
"""
from __future__ import annotations
import time
import pytest
from orisys_autocal.motion import (
INIT_FAILED,
INIT_OK,
INIT_RUNNING,
AxisParams,
ConnectionSpec,
DemoTransport,
MotionService,
)
@pytest.fixture
def service():
return MotionService(DemoTransport())
def _initialise(service, timeout: float = 5.0) -> None:
"""Download a BAS program and wait for the simulated bus to report ready."""
service.download_bas(__file__, 1)
deadline = time.monotonic() + timeout
while time.monotonic() < deadline and service.bus.init_status != INIT_OK:
service.poll([0])
time.sleep(0.05)
# ----------------------------------------------------------------- connection
def test_a_fresh_service_is_disconnected_and_touches_nothing(service):
assert not service.connected
assert not service.bas_loaded
assert service.poll([0]).connected is False
def test_connect_reports_success_and_an_unknown_bus_without_a_bas(service):
result = service.connect(ConnectionSpec(mode="demo"))
assert result.ok
assert service.connected
# No BAS program has run, so the bus type must not be guessed at.
snapshot = service.poll([0])
assert snapshot.bus.bus_type == "unknown"
assert not snapshot.bus.initialized
def test_connect_failure_is_reported_without_raising(service):
result = service.connect(ConnectionSpec(mode="eth", ip="127.0.0.1"))
assert not result.ok
assert not service.connected
def test_disconnect_clears_bus_state(service):
service.connect(ConnectionSpec(mode="demo"))
service.disconnect()
assert not service.connected
assert service.bus.bus_type == "unknown"
# ---------------------------------------------------------------- bus init
def test_download_requires_a_real_file(service):
service.connect(ConnectionSpec(mode="demo"))
assert not service.download_bas("", 1).ok
assert not service.download_bas(r"D:\nope\missing.bas", 1).ok
def test_download_needs_a_connection(service):
assert not service.download_bas(__file__, 1).ok
def test_download_then_poll_reaches_initialised(service):
service.connect(ConnectionSpec(mode="demo"))
assert service.download_bas(__file__, 1).ok
_initialise(service)
assert service.bus.init_status == INIT_OK
assert service.bus.bus_type == "ecat"
assert service.bus.total_axes == 5
assert service.bus.node_count == 5
def test_bus_is_running_immediately_after_download(service):
service.connect(ConnectionSpec(mode="demo"))
service.download_bas(__file__, 1)
assert service.bus.init_status == INIT_RUNNING
def test_reinitialise_is_refused_while_already_running(service):
service.connect(ConnectionSpec(mode="demo"))
service.download_bas(__file__, 1)
assert not service.reinit_bus().ok
def test_reinitialise_is_refused_without_a_bas_program(service):
service.connect(ConnectionSpec(mode="demo"))
result = service.reinit_bus()
assert not result.ok
assert "BAS" in result.message or "Bas" in result.message
def test_init_timeout_is_reported_rather_than_raising():
"""A controller that never publishes its globals must time out, not hang."""
class SilentTransport(DemoTransport):
def user_var(self, name):
from orisys_app_shared.zmotion import ZAuxError
raise ZAuxError(30002, f"{name} is not defined")
service = MotionService(SilentTransport())
service._init_timeout = 0.05 # noqa: SLF001 - shortening the vendor window
service.connect(ConnectionSpec(mode="demo"))
service.download_bas(__file__, 1)
time.sleep(0.1)
snapshot = service.poll([0])
assert snapshot.bus.init_status == INIT_FAILED
assert snapshot.bus.message # carries the ZDevelop hint, not an empty string
# ------------------------------------------------------------------- guards
def test_motion_is_refused_before_the_bus_is_initialised(service):
service.connect(ConnectionSpec(mode="demo"))
for attempt in (
service.jog(0, True),
service.move_abs(0, 100.0),
service.move_rel(0, 10.0),
service.home(0),
):
assert not attempt.ok
assert attempt.message
def test_motion_is_refused_when_not_connected(service):
assert not service.jog(0, True).ok
assert not service.move_abs(0, 1.0).ok
assert not service.stop(0).ok
def test_motion_on_an_unknown_axis_is_refused(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
result = service.jog(99, True)
assert not result.ok
assert "99" in result.message or "0-4" in result.message
@pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")])
def test_a_non_finite_target_is_refused(service, bad):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
result = service.move_abs(0, bad)
assert not result.ok
assert result.message
def test_home_needs_a_configured_mode(service):
"""0 is not a valid homing mode; inheriting one would be worse than saying so."""
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
result = service.home(0)
assert not result.ok
assert "0" in result.message
def test_home_runs_once_a_mode_is_set(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
service.set_axis_params(0, AxisParams(home_mode=17))
assert service.home(0).ok
def test_home_is_refused_on_rtex(service):
"""The vendor example refuses RTEX homing; so do we."""
class RtexTransport(DemoTransport):
def user_var(self, name):
return 1.0 if name == "BUS_TYPE" else super().user_var(name)
rtex = MotionService(RtexTransport())
rtex.connect(ConnectionSpec(mode="demo"))
_initialise(rtex)
assert rtex.bus.bus_type == "rtex"
result = rtex.home(0)
assert not result.ok
assert result.message
# ------------------------------------------------------------------- motion
def test_jog_and_stop_move_the_state_machine(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
assert service.jog(0, True).ok
state = service.poll([0]).axis(0)
assert state is not None and not state.idle
assert service.stop(0).ok
state = service.poll([0]).axis(0)
assert state is not None and state.idle
def test_absolute_and_relative_moves_change_the_command_position(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
assert service.move_abs(1, 250.0).ok
assert service.poll([1]).axis(1).dpos == pytest.approx(250.0)
assert service.move_rel(1, -50.0).ok
assert service.poll([1]).axis(1).dpos == pytest.approx(200.0)
def test_set_axis_params_writes_and_remembers(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
params = AxisParams(units=5.0, speed=200.0, accel=2000.0)
assert service.set_axis_params(2, params).ok
assert service.params(2) == params
assert service.to_mm(2, 1000.0) == pytest.approx(1.0)
def test_enable_toggles_the_axis(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
assert service.set_axis_enable(3, False).ok
assert service.poll([3]).axis(3).enabled is False
assert service.set_axis_enable(3, True).ok
assert service.poll([3]).axis(3).enabled is True
def test_stop_all_covers_every_axis(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
results = service.stop_all()
assert len(results) == service.axis_count
assert all(result.ok for result in results)
def test_clear_alarm_reaches_the_drive(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
assert service.clear_alarm(0).ok
# ----------------------------------------------------------------- bus data
def test_sdo_round_trips_on_ethercat(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
assert service.sdo_write(0, 0x6040, 0, 3, 6).ok
result, value = service.sdo_read(0, 0x6040, 0, 3)
assert result.ok
assert value == 6
def test_sdo_is_refused_on_rtex(service):
class RtexTransport(DemoTransport):
def user_var(self, name):
return 1.0 if name == "BUS_TYPE" else super().user_var(name)
rtex = MotionService(RtexTransport())
rtex.connect(ConnectionSpec(mode="demo"))
_initialise(rtex)
result, _value = rtex.sdo_read(0, 0x6040, 0, 3)
assert not result.ok
def test_rtex_is_refused_on_ethercat(service):
service.connect(ConnectionSpec(mode="demo"))
_initialise(service)
result, _value = service.rtex_read(0, 1)
assert not result.ok
def test_bus_data_is_refused_before_the_bus_type_is_known(service):
service.connect(ConnectionSpec(mode="demo"))
result, _value = service.sdo_read(0, 0x6040, 0, 3)
assert not result.ok
# ------------------------------------------------------------------- polling
def test_polling_reports_every_requested_axis(service):
service.connect(ConnectionSpec(mode="demo"))
snapshot = service.poll([0, 1, 2, 3, 4])
assert [state.axis for state in snapshot.axes] == [0, 1, 2, 3, 4]
assert all(state.read_ok for state in snapshot.axes)
def test_a_read_failure_is_marked_not_silently_zeroed():
"""A failed read must not look like a position of zero."""
class FailingReads(DemoTransport):
def get_dpos(self, axis):
from orisys_app_shared.zmotion import ZAuxError
raise ZAuxError(30000, "no response")
service = MotionService(FailingReads())
service.connect(ConnectionSpec(mode="demo"))
snapshot = service.poll([0])
state = snapshot.axis(0)
assert state is not None
assert state.read_ok is False
assert state.error
def test_language_switches_the_refusal_text(service):
assert not service.jog(0, True).ok
zh = service.jog(0, True).message
service.set_language("en_US")
en = service.jog(0, True).message
assert zh != en
+331
View File
@@ -0,0 +1,331 @@
"""Offscreen UI smoke tests: construction, layout, gating, and preferences.
The window is built for real, with the simulated platform standing in for the
controller. Nothing here needs a display or hardware.
"""
from __future__ import annotations
import pytest
from conftest import pump, visible_descendants
RESOLUTIONS = ((1440, 900), (1600, 900), (1920, 1080))
def _show(window, size=(1440, 880)):
from PySide6.QtWidgets import QApplication
window.resize(*size)
window.show()
QApplication.processEvents()
return window
def _enabled(widget) -> bool:
"""Whether ``widget`` is effectively enabled, shown or not.
A plain ``isEnabled()`` is False for every child of a window that has not
been shown, which makes an assertion about gating depend on whether the
test happened to realise the window. ``isEnabledTo`` answers the question
the tests actually mean: would a user be able to operate this control?
"""
return widget.isEnabledTo(widget.window())
# ------------------------------------------------------------------ structure
@pytest.mark.parametrize(("theme", "language"), [
("dark", "zh_CN"),
("light", "zh_CN"),
("dark", "en_US"),
("light", "en_US"),
])
def test_the_window_builds_in_every_theme_and_language(make_window, theme, language):
window = _show(make_window(theme=theme, language=language))
assert window.windowTitle()
assert window.axis_panel.table.rowCount() == 5
def test_the_product_name_is_localised_in_the_window_title(make_window):
window = _show(make_window(language="zh_CN"))
zh = window.windowTitle()
window._set_language("en_US")
assert window.windowTitle() != zh
def test_every_axis_row_has_a_labelled_control(make_window):
window = _show(make_window())
panel = window.axis_panel
for row in range(panel.axis_count):
widgets = panel._rows[row]
for key in ("enable_button", "fwd_button", "rev_button", "stop_button"):
assert widgets[key].text().strip(), f"row {row} {key} has no label"
def test_the_table_shows_one_column_per_documented_quantity(make_window):
from orisys_app_shared.i18n import t
from orisys_autocal.ui.axis_panel import COLUMNS
window = _show(make_window())
headers = [
window.axis_panel.table.horizontalHeaderItem(index).text()
for index in range(window.axis_panel.table.columnCount())
]
assert headers == [t(key) for key in COLUMNS] + [t("axes.controls")]
@pytest.mark.parametrize("size", RESOLUTIONS)
def test_every_axis_row_is_reachable(make_window, size):
"""No drive may be hidden by the splitter.
The panel asks for three rows without scrolling; at the default window size
all five must be visible, because a clipped fifth row on a five-axis platform
reads as a missing drive rather than as a scrollbar. If the panel is ever
squeezed below that, the table has to scroll rather than silently cut a row.
"""
window = _show(make_window(), size)
table = window.axis_panel.table
rows_height = sum(
table.rowHeight(row) for row in range(window.axis_panel.axis_count)
)
header = table.horizontalHeader().height()
available = table.viewport().height()
assert available >= rows_height or table.verticalScrollBar().maximum() > 0, (
f"{size}: rows need {rows_height}px, viewport has {available}px "
f"(header {header}px) and it cannot scroll"
)
@pytest.mark.parametrize("size", RESOLUTIONS)
def test_no_visible_child_escapes_its_parent(make_window, size):
window = _show(make_window(), size)
oversize = []
for child in visible_descendants(window):
if _inside_scroll_area(child):
# A scroll area's content is *meant* to exceed its viewport: the rail
# scrolls instead of squeezing its controls into an unreadable column.
continue
parent = child.parentWidget()
if parent is None or not parent.isVisible():
continue
geometry = child.geometry()
if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1:
oversize.append(
f"{type(child).__name__} {geometry.width()}x{geometry.height()}"
f" > parent {parent.width()}x{parent.height()}"
)
assert not oversize, oversize
def _inside_scroll_area(widget) -> bool:
"""Whether ``widget`` is content of a QAbstractScrollArea, not a laid-out child."""
from PySide6.QtWidgets import QAbstractScrollArea
parent = widget.parentWidget()
while parent is not None:
if isinstance(parent, QAbstractScrollArea):
return True
parent = parent.parentWidget()
return False
# ----------------------------------------------------------------------- gating
def test_motion_controls_start_disabled_without_a_bus(make_window):
window = _show(make_window())
assert not _enabled(window.axis_panel._rows[0]["fwd_button"])
assert not _enabled(window.axis_panel._rows[0]["rev_button"])
def test_bus_data_panels_follow_the_reported_bus_type(make_window):
window = _show(make_window())
window.bus_data_panel.set_connected(True)
window.bus_data_panel.set_bus_type("ecat")
assert _enabled(window.bus_data_panel.ecat_box)
assert not _enabled(window.bus_data_panel.rtex_box)
window.bus_data_panel.set_bus_type("rtex")
assert _enabled(window.bus_data_panel.rtex_box)
assert not _enabled(window.bus_data_panel.ecat_box)
window.bus_data_panel.set_bus_type("unknown")
assert not _enabled(window.bus_data_panel.ecat_box)
assert not _enabled(window.bus_data_panel.rtex_box)
def test_bus_data_panels_are_disabled_while_disconnected(make_window):
window = _show(make_window())
window.bus_data_panel.set_bus_type("ecat")
window.bus_data_panel.set_connected(False)
assert not _enabled(window.bus_data_panel.ecat_box)
def test_axis_rows_dim_when_the_bus_reports_fewer_drives(make_window):
window = _show(make_window())
window.axis_panel.set_connected(True)
window.axis_panel.set_axis_count(3)
assert _enabled(window.axis_panel._rows[2]["enable_button"])
assert not _enabled(window.axis_panel._rows[3]["enable_button"])
def test_unknown_connection_modes_report_instead_of_pretending(make_window):
from orisys_app_shared.i18n import t
window = _show(make_window())
panel = window.connection_panel
index = panel.mode_combo.findData("com")
panel.mode_combo.setCurrentIndex(index)
panel._emit_connect()
assert panel.status_label.text() != t("connection.state_open")
# ------------------------------------------------------------------- actions
def test_the_selected_axis_follows_a_table_click(make_window):
window = _show(make_window())
window.axis_panel._on_cell_clicked(3, 0)
assert window.axis_panel.selected_axis == 3
assert window.axis_panel.axis_index_label.text() == "3"
def test_blank_numeric_fields_fall_back_to_defaults(make_window):
from orisys_autocal.motion import DEFAULT_UNITS
window = _show(make_window())
panel = window.axis_panel
panel.units_edit.setText("")
assert panel.current_params().units == DEFAULT_UNITS
panel.units_edit.setText("not a number")
assert panel.current_params().units == DEFAULT_UNITS
def test_a_blank_target_does_not_emit_a_move(make_window):
window = _show(make_window())
emitted = []
window.axis_panel.moveAbsRequested.connect(lambda axis, value: emitted.append(value))
window.axis_panel.target_edit.setText("")
window.axis_panel._on_move_abs()
assert emitted == []
def test_applying_parameters_persists_them_in_the_panel(make_window):
from orisys_autocal.motion import AxisParams
window = _show(make_window())
panel = window.axis_panel
panel.set_selected_axis(1)
panel.units_edit.setText("5")
panel.speed_edit.setText("250")
panel.accel_edit.setText("2500")
panel._on_apply()
stored = panel.stored_params()[1]
assert isinstance(stored, AxisParams)
assert stored.units == 5.0
# --------------------------------------------------------------- preferences
def test_the_window_geometry_round_trips_through_preferences(make_window):
# Above the enforced minimum (1400x860): a smaller request would be clamped,
# which tests Qt's constraint rather than the preference round-trip.
window = _show(make_window(), (1520, 920))
window._persist_prefs()
geometry = window.prefs["window_geometry"]
assert len(geometry) == 4
assert geometry[2] == pytest.approx(1520, abs=5)
assert geometry[3] == pytest.approx(920, abs=5)
def test_a_malformed_geometry_is_ignored_rather_than_crashing(make_window):
window = _show(make_window(window_geometry=["nonsense"]))
assert window.width() > 0
def test_axis_parameters_survive_a_persist_reload(make_window):
from orisys_autocal import preferences as prefs_mod
window = _show(make_window())
window.axis_panel.set_selected_axis(2)
window.axis_panel.units_edit.setText("7")
window.axis_panel._on_apply()
window._persist_prefs()
reloaded = prefs_mod.load_preferences()
assert reloaded["axis_params"]["2"]["units"] == 7.0
def test_the_bus_path_is_remembered(make_window):
window = _show(make_window())
window.bus_panel.set_bas_path(r"D:\somewhere\ECAT.bas")
window._persist_prefs()
assert window.prefs["bas_path"] == r"D:\somewhere\ECAT.bas"
def test_sdo_fields_are_remembered(make_window):
window = _show(make_window())
panel = window.bus_data_panel
panel.node_edit.setText("2")
panel.index_edit.setText("24608")
panel.subindex_edit.setText("1")
window._persist_prefs()
assert window.prefs["sdo_last"]["node"] == 2
assert window.prefs["sdo_last"]["index"] == 24608
# --------------------------------------------------------------- demo session
def test_a_demo_session_fills_the_table_and_gates_motion(make_window, qt_app):
"""End to end through the real signal path: connect, initialise, jog."""
from orisys_autocal.motion import ConnectionSpec
window = _show(make_window())
window.connect_controller(ConnectionSpec(mode="demo"), label="connect")
assert _wait_for(lambda: window._last_snapshot.connected, qt_app)
# Download the shipped ECAT.bas, then wait for the simulated scan.
window.bus_panel.set_bas_path(str(_shipped_bas("ECAT.bas")))
window._on_download_bas()
assert _wait_for(lambda: window._last_snapshot.bus.init_status == 1, qt_app, timeout=8.0)
assert window._last_snapshot.bus.bus_type == "ecat"
assert _enabled(window.bus_data_panel.ecat_box)
assert _enabled(window.axis_panel._rows[0]["fwd_button"])
# ATYPE is no longer a table column — the seven-column table shows position
# and status, and the drive's full frame is on the selected-axis detail line.
window.axis_panel.set_selected_axis(0)
assert "65" in window.axis_panel._detail_label.text()
pump(qt_app, 0.2)
def test_closing_the_window_stops_the_poller(make_window, qt_app):
window = _show(make_window())
window.close()
pump(qt_app, 0.2)
assert not window.poller._thread.isRunning()
def _shipped_bas(name: str):
from orisys_autocal.ui.bus_panel import default_bas_dir
return default_bas_dir() / name
def _wait_for(predicate, app, timeout: float = 5.0) -> bool:
import time
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
app.processEvents()
if predicate():
return True
time.sleep(0.02)
return False
+233
View File
@@ -0,0 +1,233 @@
"""The poller thread: serialisation, failure isolation, and clean teardown.
The two things that must never happen are a controller call on the GUI thread
and two controller calls at once. Both are asserted here against a transport
that records how it was used.
"""
from __future__ import annotations
import threading
import time
import pytest
from orisys_autocal.motion import ConnectionSpec, DemoTransport, MotionPoller, MotionService
class RecordingTransport(DemoTransport):
"""A *connected* demo platform that records thread and overlap of every call.
Connected on purpose: the service polls only ``BUS_TYPE`` and ``Bus_InitStatus``
without a BAS program, so an unconnected transport would keep every read
short and nothing could be observed. A ``_concurrent`` guard asserts no two
calls ever overlap rather than merely asserting the maximum sum of started
calls, which would be trivially satisfied.
"""
def __init__(
self,
*,
delay: float = 0.0,
unexpected_on: set[str] | None = None,
) -> None:
super().__init__()
self.delay = delay
self.unexpected_on = unexpected_on or set()
self.thread_ids: set[int] = set()
self.max_concurrent = 0
self.overlaps: list[str] = []
self.calls: list[str] = []
self._lock = threading.Lock()
self._concurrent = 0
# A fake "BAS program has run": the bus reports EtherCAT and ready.
self._init_started = None
self._preloaded = True
def open(self, spec) -> None:
super().open(ConnectionSpec(mode="demo"))
self._bas_loaded = True
def user_var(self, name: str) -> float:
if name == "BUS_TYPE":
return 0.0
if name == "Bus_InitStatus":
return 1.0
if name == "Bus_TotalAxisnum":
return 5.0
return super().user_var(name)
def _enter(self, name: str) -> None:
with self._lock:
self._concurrent += 1
if self._concurrent > 1:
self.overlaps.append(name)
self.max_concurrent = max(self.max_concurrent, self._concurrent)
self.calls.append(name)
self.thread_ids.add(threading.get_ident())
if self.delay:
time.sleep(self.delay)
if name in self.unexpected_on:
raise RuntimeError(f"simulated unexpected fault in {name}")
def _exit(self) -> None:
with self._lock:
self._concurrent -= 1
def get_dpos(self, axis: int) -> float:
self._enter("get_dpos")
try:
return super().get_dpos(axis)
finally:
self._exit()
def get_mpos(self, axis: int) -> float:
self._enter("get_mpos")
try:
return super().get_mpos(axis)
finally:
self._exit()
def get_axis_status(self, axis: int) -> int:
self._enter("get_axis_status")
try:
return super().get_axis_status(axis)
finally:
self._exit()
def get_if_idle(self, axis: int) -> int:
self._enter("get_if_idle")
try:
return super().get_if_idle(axis)
finally:
self._exit()
def _connected_poller(transport: RecordingTransport, *, axes=(0,)) -> MotionPoller:
service = MotionService(transport)
assert service.connect(ConnectionSpec(mode="demo")).ok
return MotionPoller(service, axes=list(axes))
@pytest.fixture
def qt_app_session(qt_app):
return qt_app
def _drain(app, seconds: float = 1.0) -> None:
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
app.processEvents()
time.sleep(0.005)
def _wait_for(predicate, app, timeout: float = 5.0) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
app.processEvents()
if predicate():
return True
time.sleep(0.01)
return False
def test_completed_commands_are_reported_and_none_are_lost(qt_app):
transport = RecordingTransport()
poller = _connected_poller(transport)
try:
repo = {"finished": [], "failed": []}
poller.commandFinished.connect(lambda token, result: repo["finished"].append(token))
poller.commandFailed.connect(lambda token, message: repo["failed"].append(token))
tokens = [poller.submit(lambda service: None) for _ in range(5)]
assert _wait_for(lambda: len(repo["finished"]) == 5, qt_app)
assert sorted(repo["finished"]) == sorted(tokens)
assert repo["failed"] == []
finally:
poller.shutdown()
def test_every_controller_call_happens_on_one_worker_thread(qt_app):
transport = RecordingTransport()
poller = _connected_poller(transport, axes=(0, 1))
try:
assert _wait_for(lambda: "get_dpos" in transport.calls, qt_app)
poller.submit(lambda service: None)
_drain(qt_app, 0.3)
assert len(transport.thread_ids) == 1, transport.thread_ids
assert threading.get_ident() not in transport.thread_ids, (
"a controller call ran on the test's (GUI) thread"
)
finally:
poller.shutdown()
def test_two_controller_calls_never_overlap(qt_app):
"""A slow poll plus queued commands must still serialise."""
transport = RecordingTransport(delay=0.02)
poller = _connected_poller(transport)
try:
for _ in range(4):
poller.submit(lambda service: None)
_drain(qt_app, 0.8)
assert transport.max_concurrent == 1, (
f"{transport.max_concurrent} controller calls overlapped at once"
)
assert not transport.overlaps, transport.overlaps
finally:
poller.shutdown()
def test_an_unexpected_fault_is_reported_and_does_not_kill_the_poller(qt_app):
"""A non-``ZAuxError`` from inside a command must not stop the timer."""
transport = RecordingTransport(unexpected_on={"get_dpos"})
poller = _connected_poller(transport)
try:
failures = []
poller.commandFailed.connect(lambda token, message: failures.append((token, message)))
# The submitted action asks for the position directly, so the fault
# surfaces from the command rather than being absorbed by a poll.
poller.submit(lambda service: service.transport.get_dpos(0))
assert _wait_for(lambda: bool(failures), qt_app)
assert failures[0][1]
# The poller is still alive: later snapshots keep arriving.
before = len(transport.calls)
_drain(qt_app, 0.4)
assert len(transport.calls) > before
finally:
poller.shutdown()
def test_shutdown_stops_the_thread_and_is_idempotent(qt_app):
poller = _connected_poller(RecordingTransport())
_drain(qt_app, 0.2)
poller.shutdown(timeout_ms=3000)
poller.shutdown(timeout_ms=3000) # must not raise or hang
def test_a_refused_command_still_arrives_with_its_reason(qt_app):
"""An unconnected service refuses in the guard, without touching the DLL."""
poller = MotionPoller(MotionService(RecordingTransport()), axes=[0])
try:
results = []
poller.commandFinished.connect(lambda token, result: results.append(result))
poller.submit(lambda service: service.jog(0, True))
assert _wait_for(lambda: bool(results), qt_app)
assert results[0].ok is False
assert results[0].message
finally:
poller.shutdown()
def test_connect_and_disconnect_round_trip_through_the_poller(qt_app):
poller = MotionPoller(MotionService(RecordingTransport()), axes=[0])
try:
states = []
poller.connectionChanged.connect(states.append)
poller.connect_controller(ConnectionSpec(mode="demo"))
assert _wait_for(lambda: poller.service.connected, qt_app)
poller.disconnect_controller()
assert _wait_for(lambda: not poller.service.connected, qt_app)
finally:
poller.shutdown()
+73
View File
@@ -0,0 +1,73 @@
"""Unit-to-millimetre conversion.
The reference program carries the bench measurement in a comment::
# 轴0(X)/轴1(Y) 实测: UNITS=10 时 1000用户单位 = 2mm → 每毫米 5000 脉冲
# 当量5 → 1000单位=1mm,不是100单位=1mm
These are the assertions that keep that comment honest.
"""
from __future__ import annotations
import math
import pytest
from orisys_autocal.motion.units import (
PULSES_PER_MM,
mm_to_user_units,
safe_float,
user_units_to_mm,
)
@pytest.mark.parametrize(
("units", "user_units", "expected_mm"),
[
(10.0, 1000.0, 2.0), # the measured case from the vendor comment
(5.0, 1000.0, 1.0), # ...and the second one
(10.0, 0.0, 0.0),
(10.0, 5000.0, 10.0),
(10.0, -1000.0, -2.0), # negative travel must stay negative
],
)
def test_conversion_matches_the_bench_measurement(units, user_units, expected_mm):
assert user_units_to_mm(user_units, units=units) == pytest.approx(expected_mm)
def test_default_units_is_the_platform_value():
assert user_units_to_mm(1000.0) == pytest.approx(2.0)
assert PULSES_PER_MM == 5000.0
def test_conversion_round_trips():
for units in (1.0, 5.0, 10.0, 37.5):
millimetres = user_units_to_mm(1234.5, units=units)
assert mm_to_user_units(millimetres, units=units) == pytest.approx(1234.5)
def test_a_zero_or_negative_scale_is_not_a_number():
"""A bad constant must be visibly unusable, not a plausible number."""
assert math.isnan(user_units_to_mm(1000.0, pulses_per_mm=0.0))
assert math.isnan(user_units_to_mm(1000.0, pulses_per_mm=-1.0))
assert math.isnan(mm_to_user_units(1.0, units=0.0))
@pytest.mark.parametrize(
"text",
["", " ", "abc", "1,5", None, "nan", "inf", "-inf"],
)
def test_unusable_entries_fall_back(text):
assert safe_float(text, 42.0) == 42.0
@pytest.mark.parametrize("text", ["0", "0.0", " 0 "])
def test_zero_falls_back_because_it_would_divide(text):
"""A pulse equivalent of zero must never reach the controller."""
assert safe_float(text, 7.0) == 7.0
@pytest.mark.parametrize(("text", "expected"), [("5", 5.0), (" 2.5 ", 2.5), ("-3", -3.0)])
def test_usable_entries_are_parsed(text, expected):
assert safe_float(text, 42.0) == expected