diff --git a/backend/fluksio/flow/logs.py b/backend/fluksio/flow/logs.py index 9cb60bf..e8f1957 100644 --- a/backend/fluksio/flow/logs.py +++ b/backend/fluksio/flow/logs.py @@ -36,17 +36,19 @@ class _Tee(io.TextIOBase): sink(text) try: return self._real.write(text) - except OSError: + except (OSError, ValueError): # A dead stdout — `fluksio serve` runs the engine as a child of the # dashboard, which holds the far end of that pipe — must not fail # the node whose output was being teed. The capture above has it, - # and there is nowhere left to report the loss to anyway. + # and there is nowhere left to report the loss to anyway. A closed + # one (ValueError) is the same at interpreter exit, when whatever + # was wrapped — pytest's capture, say — has been closed first. return len(text) def flush(self) -> None: try: self._real.flush() - except OSError: + except (OSError, ValueError): pass def isatty(self) -> bool: diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py index 4abde58..258ceb6 100644 --- a/backend/tests/flow/test_connector.py +++ b/backend/tests/flow/test_connector.py @@ -1,6 +1,7 @@ """The connector contract: polling, deduplication, health and discovery.""" import asyncio +from collections.abc import Callable from types import SimpleNamespace from typing import Any @@ -48,12 +49,28 @@ def a_sensor(readings: list[Any], **params: Any) -> Sensor: return node -def run_briefly(node: ConnectorNode, seconds: float = 0.12, app: Any = None) -> None: - """Start the poll loop, let it tick a few times, stop it.""" +def run_briefly( + node: ConnectorNode, + seconds: float = 0.12, + app: Any = None, + until: Callable[[], bool] | None = None, +) -> None: + """Start the poll loop, let it tick a few times, stop it. + + With ``until``, stop once that holds instead, giving up after 5 s: a fixed + pause is too short when the machine is busy. The caller's asserts then say + what never happened. + """ async def cycle() -> None: await node.start(app) - await asyncio.sleep(seconds) + if until is None: + await asyncio.sleep(seconds) + else: + for _ in range(500): + if until(): + break + await asyncio.sleep(0.01) await node.stop() asyncio.run(cycle()) @@ -97,7 +114,8 @@ def test_a_failing_poll_reports_down_and_keeps_going(): node._on_health = lambda _node, status, detail: health.append((status, detail)) Pipeline(nodes=[node]) - run_briefly(node) + # The first poll fails, so an "ok" after it is the recovery. + run_briefly(node, until=lambda: len(health) > 1 and health[-1][0] == "ok") assert ("down", "RuntimeError: device unplugged") in health # It recovered rather than giving up. diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 61cb816..355d218 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -636,6 +636,8 @@ def test_two_files_of_one_name_are_refused(tmp_path) -> None: where it sits; this is the spelling that still gets there — two files named on the command line, each rooted at its own directory. """ + import sys + import pytest from fluksio.sdk import SyncError @@ -647,9 +649,14 @@ def test_two_files_of_one_name_are_refused(tmp_path) -> None: first = tmp_path / "s1" / "study.py" second = tmp_path / "s2" / "study.py" - _import(*_module_of(first), first) - with pytest.raises(SyncError, match="both import as 'study'"): - _import(*_module_of(second), second) + roots = {str(tmp_path / "s1"), str(tmp_path / "s2")} + try: + _import(*_module_of(first), first) + with pytest.raises(SyncError, match="both import as 'study'"): + _import(*_module_of(second), second) + finally: + sys.modules.pop("study", None) + sys.path[:] = [entry for entry in sys.path if entry not in roots] def test_how_long_ago_reads_like_a_duration() -> None: