diff --git a/docs/api-changes-dev-src.md b/docs/api-changes-dev-src.md index 0236902..7ea8e7a 100644 --- a/docs/api-changes-dev-src.md +++ b/docs/api-changes-dev-src.md @@ -30,7 +30,9 @@ the HW definitions. Breaking changes come first. Every change names its commits - **Watchdog:** — `6ac4eb0`, `f298e87`, `fce8d1d` - its circuit is the name of the unit (`1`) instead of `1_01`, the saved aliases of watchdogs are not assigned, - `nv_save` returns an error and is not in `full()`, - - `value` is 0/1 instead of 0–3, a restart is reported only by `was_wd_reset`. + - `value` is 0/1 instead of 0–3, a restart is reported only by `was_wd_reset`, + - `timeout` must be a whole number of ms, at most 65535 (a larger one is cut to it as before); an invalid + `timeout`, `value` or `reset` is refused before any change; a changed timeout is an event (#280). - **AI and AO:** `modes` has only `unit` and `range`, without the code of the register and the `transformation`. — `e792e17` - **OwPower:** `value` is always 0/1, before `true`/`false` after a write. — `2042b3d` @@ -116,7 +118,13 @@ the HW definitions. Breaking changes come first. Every change names its commits - **Sequencer** of each output (RO, DO, LED), the device `sequencer` with the circuit `_` (`/rest/sequencer/do_1_01`): a pulse (`value`, `pulse_duration`), a sequence of the changes of the output (`steps`, `final`) or `cancel`; its state has `running` and the progress, an event at the start, steps and end. - The RPC method `relay_set_for_time`. — `e16a3bf`, `7789635`, `a85fb14` + A step failed by an error of the communication (e.g. a noise on the bus) is written again after 1 s up to + 30 s, the sequence stays running, the output does not stay in the state of a pulse; a step refused by the unit + ends the sequence after its final state (if it has one) is written. The RPC method `relay_set_for_time`. + — `e16a3bf`, `7789635`, `a85fb14`, #280 +- An invalid alias in a change (a circuit, reserved, invalid or already used) is refused before the first write, + the output was switched and then the request failed. The aliases of a bulk are checked in the order of the + changes, an alias released by an earlier change can be taken. — #280 - The reply to POST, bulk and the WebSocket `set` has the state after the writes and the event comes at once. — `a3f4679` - **New events:** — `60e2287`, `08d9f56`, `62b37ef`, `891bd4a` @@ -190,7 +198,7 @@ An invalid value is an error: the unit or the feature is not created and the err a next write renews it; then the periodic scan reads its block, a value refused by the unit is seen then. A unit may set a written register with a delay, e.g. xS52 the mode of its AI in 0.3–0.8 s: the read returned the old mode and the switch seemed to fail, it was seen by the next scan of the slow block after several seconds. - A unit without the periodic scan (`scan_enabled: false`) shows the value read after the change. + A unit without the periodic scan (`scan_enabled: false`) shows the value read after the change. — #278, #279 - **Modbus addresses:** a serial port can be used by one `MODBUSRTU` bus only, the `slave-id` must be unique on an RTU bus and on a TCP server (`hostname`, `port`) over all `MODBUSTCP` buses. The second bus or device is not created, before both read and wrote the same unit. — `dc7b1ee` diff --git a/docs/apis/sequencer.md b/docs/apis/sequencer.md index 82d7a32..c7968c9 100644 --- a/docs/apis/sequencer.md +++ b/docs/apis/sequencer.md @@ -9,7 +9,14 @@ At most one pulse or sequence runs for an output, a new one replaces it. While i itself (except its alias) is refused: REST, JSON and bulk with the status `409` (`SequenceRunning`), the WebSocket with the same error, RPC with the code `-32002`; cancel the sequence by its sequencer first. When Evok stops, it writes the final state of the running pulses and sequences which have one (`final`), an output is not left -in the state of a pulse; a sequence without `final` leaves the output in its state. After a crash of Evok +in the state of a pulse; a sequence without `final` leaves the output in its state. A step which fails by an error of +the communication, e.g. a noise on the bus or an unavailable unit, is written again after 1 s, the delay doubles +up to 30 s, and the sequence stays running until it succeeds or the sequence is cancelled; the failure is logged, +again when the error changes and every 5 minutes. Another error, e.g. a value refused by the unit (an exception +response other than Server Device Busy, Acknowledge or of a gateway), ends the sequence and it is logged; the final +state of the sequence, if it has one (`final`, the end of a pulse), is written before the sequence ends, written +again as a step on an error of the communication. When Evok stops while the last step is written and it fails, the final state +is written. After a crash of Evok the output stays in its state. ## State diff --git a/src/evok/blocks/sequencer.py b/src/evok/blocks/sequencer.py index 85aeba3..7efc0ad 100644 --- a/src/evok/blocks/sequencer.py +++ b/src/evok/blocks/sequencer.py @@ -14,12 +14,49 @@ import asyncio import logging +import time from dataclasses import dataclass +from tmodbus.exceptions import ( + AcknowledgeError, + GatewayPathUnavailableError, + GatewayTargetDeviceFailedToRespondError, + InvalidRequestError, + ModbusResponseError, + ServerDeviceBusyError, + TModbusError, +) + from ..devices import devents, to_bool, to_float +from ..errors import UnitCommunicationError, describe_error, root_cause logger = logging.getLogger(__name__) +STEP_RETRY_DELAY = 1.0 # s, a failed step is written again after it, the output does not stay in its state +STEP_RETRY_MAX_DELAY = 30.0 # s, the delay doubles up to it, a missing unit does not load its bus +STEP_FAILURE_LOG = 300.0 # s, a step still failing is logged again after it, also a changed error at once +# the errors of the communication, they pass; another error of a step ends its sequence and writes its final +# state, e.g. a value refused by the device, it would be written again forever and the output refused the changes +STEP_RETRIED = (UnitCommunicationError, TModbusError, TimeoutError, OSError) +# the request refused by the unit (an exception response, e.g. Illegal Data Value or Server Device Failure) +# or invalid, it would be refused again +STEP_REFUSED = (ModbusResponseError, InvalidRequestError) +# the exception responses which pass, e.g. a busy unit or a gateway without its target +STEP_TRANSIENT = ( + ServerDeviceBusyError, + AcknowledgeError, + GatewayPathUnavailableError, + GatewayTargetDeviceFailedToRespondError, +) + + +def _retried(error: Exception) -> bool: + """An error of the communication, not a request refused by the unit""" + cause = root_cause(error) + return isinstance(error, STEP_RETRIED) and ( + not isinstance(cause, STEP_REFUSED) or isinstance(cause, STEP_TRANSIENT) + ) + @dataclass class Step: @@ -66,6 +103,7 @@ class Running: task: asyncio.Task sequence: Sequence step: int = 0 # the number of the done steps + finished: bool = False # the last step was written, the sequence ended by it class Sequencer: @@ -141,12 +179,12 @@ def _end(self, device, task: asyncio.Task) -> bool: async def _run(self, device, sequence: Sequence): task = asyncio.current_task() - step_number = 0 + step_number, step = 0, None try: for step_number, step in enumerate(sequence.steps, 1): await asyncio.sleep(step.delay) last = step_number == len(sequence.steps) - if not await self._apply(device, step.params, task, step_number, last): + if not await self._apply_step(device, sequence, step, task, step_number, last): return if not last: if not self._current(device, task): @@ -156,10 +194,63 @@ async def _run(self, device, sequence: Sequence): except asyncio.CancelledError: raise except Exception: - logger.exception(f"{_name(device)}: step {step_number} of the {sequence.name} failed") + logger.exception(f"{_name(device)}: step {step_number} of the {sequence.name} failed, it is ended") + # not the refused step again, e.g. the end of a pulse is its final state + refused = step.params if step is not None else None + if sequence.final is not None and sequence.final != refused and self._current(device, task): + # the output does not stay in the state of the sequence: the final state is written as its last + # step, the sequence runs until it is written (a change of the output meanwhile is refused, + # finish_all() waits for it), an error of the communication is written again + try: + await self._apply_step( + device, sequence, Step(sequence.final), task, step_number - 1, True, what="final state" + ) + except asyncio.CancelledError: + raise + except Exception: + logger.exception(f"{_name(device)}: final state of the {sequence.name} failed") finally: self._end(device, task) + async def _apply_step( + self, device, sequence: Sequence, step: Step, task, step_number: int, last: bool, what: str | None = None + ) -> bool: + """Write the step until it succeeds, the sequence stays running meanwhile, e.g. a noise on the bus at + the end of a pulse; False if the sequence was cancelled or replaced meanwhile + """ + what = what or f"step {step_number}" + failures, delay = 0, STEP_RETRY_DELAY + logged, logged_at = None, 0.0 + while True: + try: + applied = await self._apply(device, step.params, task, step_number, last) + except Exception as E: + if not _retried(E): + raise + if not self._current(device, task): + return False + failures += 1 + # the kind of the error, its text may differ by each attempt + now, kind = time.monotonic(), type(root_cause(E)) + if logged is None: + logger.exception( + f"{_name(device)}: {what} of the {sequence.name} failed, it is written again " + f"after {STEP_RETRY_DELAY}..{STEP_RETRY_MAX_DELAY} s" + ) + logged, logged_at = kind, now + elif kind != logged or now - logged_at >= STEP_FAILURE_LOG: + # a changed error, or a reminder that the output is still in the state of the sequence + logger.error( + f"{_name(device)}: {what} of the {sequence.name} failed {failures}x: {describe_error(E)}" + ) + logged, logged_at = kind, now + await asyncio.sleep(delay) + delay = min(delay * 2, STEP_RETRY_MAX_DELAY) + continue + if failures and applied: + logger.info(f"{_name(device)}: {what} of the {sequence.name} written after {failures} failures") + return applied + async def _apply(self, device, params: dict, task: asyncio.Task | None = None, step_number=0, last=False) -> bool: """Set the params of a step, False if the sequence was cancelled or replaced meanwhile""" applied = False @@ -170,15 +261,18 @@ async def operation(): entry = self._running.get(device) if entry is None or entry.task is not task: return - entry.step = step_number - if last: - # the state read after the write is not running - self._end(device, task) self._writing.add(task) try: await device.set(**params) finally: self._writing.discard(task) + if task is not None: + # done after the successful write, a failed step is written again and it is not reported as done + entry.step = step_number + if last: + # the state read after the write is not running; finish_all() does not write the final state + entry.finished = True + self._end(device, task) applied = True change = getattr(getattr(device, "client", None), "change", None) @@ -199,15 +293,18 @@ async def finish_all(self, timeout: float = 5.0): for device, entry in entries: self._stop(entry) self._notify(device) - finals = [(device, entry.sequence) for device, entry in entries if entry.sequence.final is not None] + candidates = [(device, entry) for device, entry in entries if entry.sequence.final is not None] writing = list(self._writing) - if not finals and not writing: + if not candidates and not writing: return - logger.info(f"Finishing {len(finals)} running sequences") + logger.info(f"Finishing {len(candidates)} running sequences") async def finish(): if writing: await asyncio.wait(writing) + # a sequence whose last step was written ended by it; a failed write of the last step is written + # by the final state + finals = [(device, entry.sequence) for device, entry in candidates if not entry.finished] await asyncio.gather(*(self._finish(device, sequence) for device, sequence in finals)) try: diff --git a/src/evok/devices.py b/src/evok/devices.py index bcaebdc..2b6318f 100644 --- a/src/evok/devices.py +++ b/src/evok/devices.py @@ -88,9 +88,10 @@ def set_force_save(self) -> None: if self.save_callback: self.save_callback() - def validate(self, alias: str, device: Device) -> None: + def validate(self, alias: str, device: Device, free=()) -> None: + """Raise ValueError for an invalid alias; free are the aliases released by the change before it""" # check duplicity - if alias in self.alias_dict: + if alias in self.alias_dict and alias not in free: raise ValueError(f"Duplicate alias {alias}") # an alias loaded from the file belongs to its device, even when the device is not registered yet rec = self.initial_dict.get(alias) @@ -228,6 +229,20 @@ def register_device(self, devtype_name, device): devents.config(device) logger.debug(f"Registered new device '{devtype_name}' with circuit {device.circuit} \t ({device})") + def check_alias(self, alias: str | None, device: Device, free=()) -> None: + """Raise ValueError if set_alias() would refuse the alias, before any change of the device; + free are the aliases released by the changes before it, e.g. an alias moved to another device + """ + if alias in (None, "") or alias == device.alias: + return + if not isinstance(alias, str): + # e.g. 0 or false of a request without the schema, it is not a reset of the alias + raise ValueError(f"Invalid alias {alias!r}") + # by_name() finds the circuit first, such alias would never be used + if alias in self[device.devtype]: + raise ValueError(f"Alias {alias} is a circuit of {device.devtype}") + self.aliases.validate(alias, device, free) + def set_alias(self, alias: str | None, device: Device, file_update: bool = False) -> None: """Set or reset ('' or None) the alias, an invalid alias raises ValueError logged by the caller""" if alias == device.alias: @@ -239,9 +254,7 @@ def set_alias(self, alias: str | None, device: Device, file_update: bool = False device.alias = "" logger.debug(f"Reset alias of {device.devtype}[{device.circuit}]") else: - # by_name() finds the circuit first, such alias would never be used - if alias in self[device.devtype]: - raise ValueError(f"Alias {alias} is a circuit of {device.devtype}") + self.check_alias(alias, device) self.aliases.add(alias, device, file_update) device.alias = alias logger.debug(f"Set alias {alias} of {device.devtype}[{device.circuit}]") diff --git a/src/evok/modbus/base.py b/src/evok/modbus/base.py index 912ea91..b38deae 100644 --- a/src/evok/modbus/base.py +++ b/src/evok/modbus/base.py @@ -80,6 +80,8 @@ async def set_devices( unit = device.client if isinstance(device, IODevice) else device units.setdefault(unit, []).append((index, device, kw)) + _check_aliases(units) + for unit, items in units.items(): done = [] @@ -108,3 +110,33 @@ async def operation(): for index, device in done: states[index] = device.full() return [states[index] for index in range(len(assignments))] + + +def _check_aliases(units: dict): + """Check the aliases of the assignments in the order of their changes, before the first write: an invalid + alias was refused after the write of the value, the output was switched and the request failed. An alias + released by an earlier assignment can be taken, an alias taken by an earlier one cannot. + """ + planned: dict = {} # the alias of a device after the earlier assignments + free: set[str] = set() + taken: set[str] = set() + for items in units.values(): + for _index, device, kw in items: + # None is not given, set_alias() keeps the alias; "" resets it + if kw.get("alias") is None: + continue + alias = kw["alias"] + current = planned.get(device, device.alias) + if alias == current: + continue + if alias != "": + # first, an alias which is not a string is invalid, e.g. a list is not hashable for taken + Devices.check_alias(alias, device, free) + if alias in taken: + raise ValueError(f"Duplicate alias {alias}") + taken.add(alias) + free.discard(alias) + if current: + free.add(current) + taken.discard(current) + planned[device] = alias diff --git a/src/evok/modbus/client.py b/src/evok/modbus/client.py index d1d1e5d..5296730 100644 --- a/src/evok/modbus/client.py +++ b/src/evok/modbus/client.py @@ -439,7 +439,7 @@ def read_raw(self, client: Client) -> int: async def write_raw(self, client: Client, raw: int): """Read-modify-write, the register is read from the unit, not from the cache""" curr = (await client.cache.get_register_async(self.index))[0] - await client.write_registers(self.index, [curr | self.mask if raw else curr & ~self.mask]) + await client.write_registers(self.index, [curr | self.mask if raw else curr & ~self.mask], pin=self.pin) class Accessor32(Accessor): diff --git a/src/evok/modbus/special.py b/src/evok/modbus/special.py index 8b1ce9b..ca494c2 100644 --- a/src/evok/modbus/special.py +++ b/src/evok/modbus/special.py @@ -1,7 +1,7 @@ import asyncio import logging -from ..devices import NV_SAVE, OWPOWER, WATCHDOG, to_bool +from ..devices import NV_SAVE, OWPOWER, WATCHDOG, to_bool, to_float from .base import IODevice from .client import AccessorU16, Client @@ -124,12 +124,12 @@ def full(self): async def check_new_data(self): """Bit 0 of the register enables the watchdog, bit 1 is a restart by the watchdog""" - old_state = (self.value, self.was_wd_boot_value) + old_state = (self.value, self.was_wd_boot_value, self.timeout) register = self.accessor.read(self.client) self.value = register & 0b01 self.was_wd_boot_value = 1 if register & 0b10 else 0 self.timeout = self.accessor_timeout.read(self.client) - return old_state != (self.value, self.was_wd_boot_value) + return old_state != (self.value, self.was_wd_boot_value, self.timeout) async def set(self, value=None, timeout=None, reset=None, nv_save=None, alias=None): """Sets new on/off status. Disable pending timeouts @@ -141,18 +141,26 @@ async def set(self, value=None, timeout=None, reset=None, nv_save=None, alias=No raise ValueError( f"WD {self.circuit}: nv_save is deprecated and not supported, use the device nv_save {self.major_group}" ) + # all params are checked before any change, e.g. the watchdog enabled with an invalid timeout + if value is not None: + value = 1 if to_bool(value) else 0 + if timeout is not None: + number = to_float(timeout) + if number != int(number) or number < 0: + raise ValueError(f"WD {self.circuit}: timeout must be a whole number of ms >= 0, not {timeout}") + timeout = min(int(number), 65535) + # ignored by a unit without the reset coil, as before + reset = self.reset_coil >= 0 and reset is not None and to_bool(reset) self.set_alias(alias) if value is not None: # the whole register is written, it clears the flag of a restart by the watchdog too - await self.accessor.write(self.client, 1 if to_bool(value) else 0) + await self.accessor.write(self.client, value) if timeout is not None: - timeout = min(int(timeout), 65535) await self.accessor_timeout.write(self.client, timeout) - if self.reset_coil >= 0 and reset is not None: - if to_bool(reset): - # the unit restarts at once, it does not respond and a retry would restart it again - await self.client.write_coil_without_response(self.reset_coil, 1) - logger.info(f"Performed reset of board {self.circuit}") + if reset: + # the unit restarts at once, it does not respond and a retry would restart it again + await self.client.write_coil_without_response(self.reset_coil, 1) + logger.info(f"Performed reset of board {self.circuit}") diff --git a/tests/test_blocks.py b/tests/test_blocks.py index 88c3196..b2a121c 100644 --- a/tests/test_blocks.py +++ b/tests/test_blocks.py @@ -1,10 +1,13 @@ import asyncio +import sys import pytest +from tmodbus.exceptions import IllegalDataValueError, ServerDeviceBusyError, ServerDeviceFailureError from evok import devents from evok.blocks import Sequence, Sequencer, Step, pulse, pulse_params, sequencer from evok.devices import DO, LED, RO +from evok.errors import UnitCommunicationError class FakeDevice: @@ -68,13 +71,229 @@ async def test_finish_all_writes_the_final_states(): assert not seq.is_running(first) and not seq.is_running(second) -async def test_failed_step_ends_the_sequence(caplog): +async def test_failed_step_is_written_again(caplog, monkeypatch): + """ A failed step ended the sequence, the end of a pulse was lost and the output stayed in its state """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) device = FakeDevice(DO, '2_01') seq = Sequencer() seq.start(device, Sequence([Step({'value': 'fail'}), Step({'value': 1}, 0.01)], name='test')) + await asyncio.sleep(0.05) + assert device.changes == [] and seq.is_running(device) + assert caplog.text.count('DO 2_01: step 1 of the test failed') == 1 # logged once + seq.cancel(device) await asyncio.sleep(0.03) - assert device.changes == [] and not seq.is_running(device) - assert 'DO 2_01: step 1 of the test failed' in caplog.text + assert not seq.is_running(device) + + +async def test_refused_step_ends_the_sequence(caplog, monkeypatch): + """ A value refused by the device is not an error of the communication, it was written again forever and + the output refused every change + """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + + async def refuse(**kw): + raise ValueError('not in this mode') + device.set = refuse + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 1})], name='test')) + await asyncio.sleep(0.03) + assert not seq.is_running(device) + assert caplog.text.count('DO 2_01: step 1 of the test failed') == 1 + + +def unit_error(cause): + """ The error of a change of a Modbus unit, as Client.change() raises it """ + try: + raise UnitCommunicationError('Unit 2') from cause + except UnitCommunicationError as E: + return E + + +async def test_step_refused_by_the_unit_ends_the_sequence_with_its_final(monkeypatch): + """ An exception response (Illegal Data Value) was retried as an error of the communication forever; + the sequence ends and its final state is written, the output does not stay in its state + """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + set_ = device.set + + async def refuse(**kw): + if kw.get('value') == 2: + raise unit_error(IllegalDataValueError(6)) + await set_(**kw) + device.set = refuse + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 1}), Step({'value': 2}, 0.01)], final={'value': 0}, name='test')) + await asyncio.sleep(0.05) + assert not seq.is_running(device) + assert device.changes == [{'value': 1}, {'value': 0}] + + +async def test_failed_step_is_written_again_less_often(monkeypatch): + """ A missing unit: the delay doubles, its bus is not loaded every second """ + module = sys.modules['evok.blocks.sequencer'] + monkeypatch.setattr(module, 'STEP_RETRY_DELAY', 0.01) + monkeypatch.setattr(module, 'STEP_RETRY_MAX_DELAY', 0.04) + device = FakeDevice(DO, '2_01') + attempts = [] + + async def lost(**kw): + attempts.append(kw) + raise unit_error(TimeoutError('no response')) + device.set = lost + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 1})], name='test')) + await asyncio.sleep(0.2) + seq.cancel(device) + assert 4 <= len(attempts) <= 8 # 0.01, 0.02, 0.04, 0.04... instead of 20 + + +async def test_changed_error_of_a_step_is_logged(monkeypatch, caplog): + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + errors = [TimeoutError('no response')] * 3 + [ConnectionError('port closed')] * 3 + + async def failing(**kw): + raise unit_error(errors.pop(0)) if errors else None + device.set = failing + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 1})], name='test')) + await asyncio.sleep(0.3) + logged = [r.getMessage() for r in caplog.records if 'step 1 of the test failed' in r.getMessage()] + assert len(logged) == 2 and 'port closed' in logged[1] + + +async def test_final_after_a_refused_step_is_written_while_the_sequence_runs(monkeypatch): + """ The sequence ended before its final state was written: a new pulse in the gap was overwritten by it and + finish_all() did not wait for it; a noise on the bus lost the final state + """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + release = asyncio.Event() + noise = [unit_error(TimeoutError('noise'))] + set_ = device.set + + async def slow(**kw): + if kw.get('value') == 2: + raise unit_error(IllegalDataValueError(6)) + if kw.get('value') == 0: + await release.wait() + if noise: + raise noise.pop() + await set_(**kw) + device.set = slow + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 2})], final={'value': 0}, name='test')) + await asyncio.sleep(0.02) + assert seq.is_running(device) # the final state is being written + release.set() + await asyncio.sleep(0.05) + assert device.changes == [{'value': 0}] and not seq.is_running(device) + + +async def test_refused_end_of_a_pulse_is_not_written_again(caplog): + """ The only step of a pulse is its final state, it was sent again after it was refused """ + device = FakeDevice(DO, '2_01') + attempts = [] + + async def refuse(**kw): + attempts.append(kw) + raise unit_error(IllegalDataValueError(5)) + device.set = refuse + seq = Sequencer() + seq.start(device, pulse(1, 0.01)) + await asyncio.sleep(0.05) + assert attempts == [{'value': 0}] and not seq.is_running(device) + assert len([r for r in caplog.records if r.exc_info]) == 1 + + +async def test_finish_all_waits_for_the_final_after_a_refused_step(): + device = FakeDevice(DO, '2_01') + release = asyncio.Event() + set_ = device.set + + async def slow(**kw): + if kw.get('value') == 2: + raise unit_error(IllegalDataValueError(6)) + await release.wait() + await set_(**kw) + device.set = slow + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 2})], final={'value': 0}, name='test')) + await asyncio.sleep(0.01) + finish = asyncio.create_task(seq.finish_all()) + await asyncio.sleep(0.01) + assert not finish.done() + release.set() + await finish + assert device.changes == [{'value': 0}] # once + + +@pytest.mark.parametrize('cause, retried', [(ServerDeviceFailureError(6), False), (ServerDeviceBusyError(6), True)]) +async def test_exception_responses_of_the_unit(monkeypatch, cause, retried): + """ Server Device Failure is refused again, it was retried forever; a busy unit passes """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + attempts = [] + + async def refuse(**kw): + attempts.append(kw) + raise unit_error(cause) + device.set = refuse + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 1})], name='test')) + await asyncio.sleep(0.05) + assert seq.is_running(device) is retried and (len(attempts) > 1) is retried + seq.cancel(device) + + +async def test_error_with_a_changing_text_is_logged_once(monkeypatch, caplog): + """ The text of a timeout may differ by each attempt, it is the same error """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + attempt = [] + + async def lost(**kw): + attempt.append(1) + raise unit_error(TimeoutError(f'no response to the request {len(attempt)}')) + device.set = lost + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 1})], name='test')) + await asyncio.sleep(0.2) + seq.cancel(device) + assert len(attempt) > 2 + assert len([r for r in caplog.records if 'step 1 of the test failed' in r.getMessage()]) == 1 + + +async def test_step_is_done_after_its_write(monkeypatch): + """ A step written again was reported as done, the end of a pulse seemed written while the relay was on """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 'fail'}), Step({'value': 1})], name='test')) + await asyncio.sleep(0.03) + assert seq.running(device).step == 0 + seq.cancel(device) + + +async def test_last_step_ends_the_sequence_after_its_write(monkeypatch): + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) + device = FakeDevice(DO, '2_01') + failures = [ConnectionError('lost')] * 2 + set_ = device.set + + async def flaky(**kw): + if failures: + raise failures.pop() + await set_(**kw) + device.set = flaky + seq = Sequencer() + seq.start(device, Sequence([Step({'value': 0}, 0.01)], final={'value': 0}, name='pulse')) + await asyncio.sleep(0.02) + assert seq.is_running(device) # the end failed, it is written again + await asyncio.sleep(0.05) + assert device.changes == [{'value': 0}] and not seq.is_running(device) def test_global_sequencer(): diff --git a/tests/test_changes.py b/tests/test_changes.py index 0e74582..c673b9f 100644 --- a/tests/test_changes.py +++ b/tests/test_changes.py @@ -215,6 +215,78 @@ async def test_unit_without_the_periodic_scan_shows_the_read(monkeypatch): assert client.cache.groups[2].pins == {} +@pytest.mark.parametrize('alias, error', [('1_02', 'is a circuit of do'), ('all', 'reserved'), ('a b', 'Invalid alias'), + ('taken', 'Duplicate alias')]) +async def test_invalid_alias_is_refused_before_the_write(units, alias, error): + """ The output was switched and then the request failed on its alias """ + client, = await units('1') + Devices.set_alias('taken', Devices.by_name(DO, '1_02')) + do = Devices.by_name(DO, '1_01') + with pytest.raises(ValueError, match=error): + await set_devices([(do, {'value': 1, 'alias': alias})]) + assert client.mb_client.writes == [] + assert do.alias == '' + + +async def test_same_alias_twice_in_one_request(units): + """ Both passed the check against the aliases before the request, the second failed after the first write """ + client1, client2 = await units('1', '2') + with pytest.raises(ValueError, match='Duplicate alias x'): + await set_devices([(Devices.by_name(DO, '1_01'), {'value': 1, 'alias': 'x'}), + (Devices.by_name(DO, '2_01'), {'value': 1, 'alias': 'x'})]) + assert client1.mb_client.writes == [] and client2.mb_client.writes == [] + + +async def test_alias_moved_in_one_request(units): + """ An alias released by an earlier assignment was refused as a duplicate """ + await units('1') + do1, do2 = Devices.by_name(DO, '1_01'), Devices.by_name(DO, '1_02') + Devices.set_alias('pump', do1) + await set_devices([(do1, {'alias': 'old_pump'}), (do2, {'alias': 'pump'})]) + assert (do1.alias, do2.alias) == ('old_pump', 'pump') + + +async def test_alias_released_by_a_later_unit_is_refused(units): + """ The units are changed in the order of their first assignment: unit 1 takes the alias before unit 2 + releases it, it is refused before any write + """ + client1, client2 = await units('1', '2') + do1, do2, do12 = Devices.by_name(DO, '1_01'), Devices.by_name(DO, '2_01'), Devices.by_name(DO, '1_02') + Devices.set_alias('pump', do2) + with pytest.raises(ValueError, match='Duplicate alias pump'): + await set_devices([(do1, {'value': 1}), (do2, {'alias': ''}), (do12, {'alias': 'pump'})]) + assert client1.mb_client.writes == client2.mb_client.writes == [] and do2.alias == 'pump' + + +@pytest.mark.parametrize('alias', [0, False, 1, ['x'], {'x': 1}]) +async def test_alias_which_is_not_a_string_is_refused_before_the_write(units, alias): + """ 0 passed the check as a reset of the alias, set_alias(0) failed by TypeError after the write """ + client, = await units('1') + do = Devices.by_name(DO, '1_01') + with pytest.raises(ValueError, match='Invalid alias'): + await set_devices([(do, {'value': 1, 'alias': alias})]) + assert client.mb_client.writes == [] and do.alias == '' + + +async def test_alias_null_keeps_the_alias(units): + """ null was checked as a reset of the alias, set_alias() keeps it: the alias passed as released to another + device, the duplicate failed after the write + """ + client, = await units('1') + do1, do2 = Devices.by_name(DO, '1_01'), Devices.by_name(DO, '1_02') + Devices.set_alias('a', do1) + with pytest.raises(ValueError, match='Duplicate alias a'): + await set_devices([(do1, {'value': 1, 'alias': None}), (do2, {'value': 1, 'alias': 'a'})]) + assert client.mb_client.writes == [] and (do1.alias, do2.alias) == ('a', '') + + +async def test_bit_without_a_pin_is_not_pinned(): + client = make_client(SLOW_BLOCKS, holding={20: 0}) + await client.cache.do_scan(initial=True) + await client.change(lambda: AccessorBit(20, 0b01, pin=False).write_raw(client, 1), devices=[]) + assert client.cache.groups[2].pins == {} + + async def test_restart_drops_the_pins(monkeypatch): """ A write and a reset of the unit in one change: the unit has its saved settings, the read shows them """ client = make_client(SLOW_BLOCKS, holding={20: 1}) @@ -627,6 +699,30 @@ async def test_finish_all_waits_for_the_write_of_a_step(units): assert client.mb_client.writes == [('coil', 0, 1), ('coil', 0, 0)] +async def test_finish_all_writes_the_final_state_after_a_failed_last_step(units): + """ The end of a pulse being written when Evok stops fails: its final state is written, the relay stayed on """ + client, = await units('1') + do = Devices.by_name(DO, '1_01') + await start_pulse(do, 1, 0.01) + release, started = asyncio.Event(), asyncio.Event() + write = client.mb_client.write_single_coil + failures = [ConnectionError('noise')] + + async def flaky(address, value): + started.set() + await release.wait() + if failures: + raise failures.pop() + await write(address, value) + client.mb_client.write_single_coil = flaky + await asyncio.wait_for(started.wait(), 1) + finish = asyncio.create_task(sequencer.finish_all()) + await asyncio.sleep(0.01) + release.set() + await finish + assert client.mb_client.writes == [('coil', 0, 1), ('coil', 0, 0)] + + async def test_nv_save_is_changed_under_the_lock(units): client, = await units('1') nv_save = Devices.by_name(NV_SAVE, '1') diff --git a/tests/test_devices.py b/tests/test_devices.py index 718cdd3..3341172 100644 --- a/tests/test_devices.py +++ b/tests/test_devices.py @@ -1,6 +1,7 @@ import asyncio import json import math +import sys import pytest from conftest import make_client, scan @@ -14,7 +15,7 @@ from evok.modbus.builder import IOParser from evok.modbus.client import FLOAT32_BE, FLOAT32_LE, to_registers from evok.modbus.digital import ULED, BinaryOutput, DigitalInput, DigitalOutput, Relay -from evok.modbus.special import NvSave, OwPower +from evok.modbus.special import NvSave, OwPower, Watchdog @pytest.fixture @@ -407,16 +408,23 @@ async def hang(address, value): assert 'timed out' in caplog.text -async def test_do_pulse_end_error_is_logged(unit, caplog): +async def test_do_pulse_end_error_is_logged_and_written_again(unit, caplog, monkeypatch): + """ The output stayed on after a failed end of the pulse """ + monkeypatch.setattr(sys.modules['evok.blocks.sequencer'], 'STEP_RETRY_DELAY', 0.01) client = await unit() do = dev(DO, '1_01') await pulse(do, value=1, pulse_duration=0.01) + write = client.mb_client.write_single_coil async def fail(address, value): raise ConnectionError('lost') client.mb_client.write_single_coil = fail await asyncio.sleep(0.05) assert 'DO 1_01: step 1 of the pulse failed' in caplog.text + assert running(do) is True + client.mb_client.write_single_coil = write + await asyncio.sleep(0.05) + assert client.mb_client.coils[0] == 0 assert running(do) is False @@ -607,6 +615,51 @@ async def test_watchdog_restart_flag_is_a_change(unit): assert not await wd.check_new_data() +async def test_watchdog_timeout_is_a_change(unit): + """ A changed timeout was not an event, the clients kept the old one """ + client = await unit({12: 0b01, 1008: 500}) + wd = dev(WATCHDOG, '1') + client.cache.set_register(1008, [800]) + assert await wd.check_new_data() + assert wd.timeout == 800 + assert not await wd.check_new_data() + + +@pytest.mark.parametrize('timeout', ['1.5', 2.9, -1, 'x']) +async def test_watchdog_invalid_timeout_changes_nothing(unit, timeout): + """ The watchdog was enabled and the alias set before int() of the timeout failed """ + client = await unit() + wd = dev(WATCHDOG, '1') + with pytest.raises(ValueError): + await wd.set(value=1, timeout=timeout, alias='wd') + assert client.mb_client.writes == [] and wd.alias == '' + + +@pytest.mark.parametrize('kw', [{'value': 'maybe'}, {'value': 1, 'timeout': 1000, 'reset': 'x'}]) +async def test_watchdog_invalid_value_or_reset_changes_nothing(unit, kw): + """ The alias was set, the watchdog enabled and the timeout written before to_bool() failed """ + client = await unit() + wd = dev(WATCHDOG, '1') + with pytest.raises(ValueError, match='Invalid boolean'): + await wd.set(alias='wd', **kw) + assert client.mb_client.writes == [] and wd.alias == '' + + +async def test_watchdog_without_a_reset_coil_ignores_reset(): + """ reset of a unit without the reset coil was ignored, the check of all params refused an invalid one """ + client = make_client([{'start_reg': 0, 'count': 2, 'scan_divider': 1}]) + await client.cache.do_scan(initial=True) + wd = Watchdog('x', client, 0, 1, -1) + await wd.set(reset='maybe') + assert client.mb_client.writes == [] + + +async def test_watchdog_whole_timeout(unit): + client = await unit() + await dev(WATCHDOG, '1').set(timeout='3000') + assert client.mb_client.holding[1008] == 3000 + + @pytest.mark.parametrize('nv_save', [1, 0, 'true']) async def test_watchdog_nv_save_is_rejected(unit, nv_save): """ nv_save=0 saved the settings, it is deprecated and not supported """