"""Datastore tests: WAGO register-map areas, mirrors, bit addressing.""" from __future__ import annotations import pytest from wago_sim.pdu import EXC_ILLEGAL_DATA_ADDRESS, ExceptionResponse def _is_exc(resp) -> bool: return isinstance(resp, ExceptionResponse) class TestWordRead: def test_input_area1_readable(self, datastore): datastore.set_input_word(0, 0xBEEF) assert datastore.read_words(0x0000, 1) == [0xBEEF] def test_input_area1_upper_edge(self, datastore): datastore.set_input_word(255, 0x1234) assert datastore.read_words(0x00FF, 1) == [0x1234] def test_hole_0x0100_illegal(self, datastore): assert _is_exc(datastore.read_words(0x0100, 1)) def test_output_readback_area1(self, datastore): result = datastore.write_words(0x0000, [0xABCD]) assert result is None assert datastore.read_words(0x0200, 1) == [0xABCD] def test_hole_0x0300_illegal(self, datastore): assert _is_exc(datastore.read_words(0x0300, 1)) def test_config_area_readable(self, datastore): # 0x2002 must return the byte-order constant 0x1234 assert datastore.read_words(0x2002, 1) == [0x1234] def test_input_area2(self, datastore): datastore.set_input_word(256, 0x00FF) # %IW256 -> 0x6000 assert datastore.read_words(0x6000, 1) == [0x00FF] def test_output_readback_area2(self, datastore): assert datastore.write_words(0x6000, [0x5A5A]) is None assert datastore.read_words(0x7000, 1) == [0x5A5A] def test_hole_0x62FC_illegal(self, datastore): assert _is_exc(datastore.read_words(0x62FC, 1)) def test_area_end_0x72FB(self, datastore): assert datastore.write_words(0x6000 + 763, [1]) is None assert datastore.read_words(0x7000 + 763, 1) == [1] def test_beyond_0x72FC_illegal(self, datastore): assert _is_exc(datastore.read_words(0x72FC, 1)) def test_spanning_areas_rejected(self, datastore): # a read crossing from input area 1 into the 0x0100 hole must fail assert _is_exc(datastore.read_words(0x00FE, 4)) class TestWordWrite: def test_write_area1(self, datastore): assert datastore.write_words(0x0000, [0xFFFF]) is None assert datastore.output_words[0] == 0xFFFF def test_write_mirror_area1(self, datastore): assert datastore.write_words(0x0200, [0x7777]) is None assert datastore.output_words[0] == 0x7777 def test_write_config_area_watchdog_time(self, datastore): assert datastore.write_words(0x1000, [10]) is None def test_write_hole_0x0100_illegal(self, datastore): assert _is_exc(datastore.write_words(0x0100, [1])) def test_write_input_area_illegal(self, datastore): # 0x0300 hole on write path assert _is_exc(datastore.write_words(0x0300, [1])) # reading input image via write FC6 at 0x0200 goes to outputs, not inputs assert _is_exc(datastore.write_words(0x62FC, [1])) def test_word_write_limit_100(self, datastore): values = [1] * 101 resp = datastore.write_words(0x0000, values) assert _is_exc(resp) def test_word_read_limit_125(self, datastore): assert _is_exc(datastore.read_words(0x0000, 126)) def test_mask_write(self, datastore): assert datastore.write_words(0x0000, [0x12]) is None result = datastore.mask_write_word(0x0000, and_mask=0xF2, or_mask=0x25) assert result == ((0x12 & 0xF2) | (0x25 & ~0xF2 & 0xFFFF)) assert datastore.output_words[0] == result def test_mask_write_illegal_address(self, datastore): assert _is_exc(datastore.mask_write_word(0x0100, 0xFF, 0x00)) class TestBits: def test_read_discrete_input_area1(self, datastore): datastore.set_digital_input_channel(4, True) assert datastore.read_bits(3, 1) == [True] def test_read_coil_area1_mirror(self, datastore): assert datastore.write_bits(0x0200, [True, False, True]) is None assert datastore.read_bits(0x0200, 3) == [True, False, True] def test_write_coil_area1(self, datastore): # bit writes at 0x0000-0x01FF target OUTPUTS (Table 122), but reads # there return INPUTS (Table 121) — readback uses the 0x0200 mirror assert datastore.write_bits(0x0000, [True] * 8) is None assert datastore.digital_output_channel(1) is True assert datastore.read_bits(0x0200, 8) == [True] * 8 def test_digital_channel_addressing(self, datastore): # channel 47 -> bit address 46 (manual §11.2.4) datastore.set_digital_input_channel(47, True) assert datastore.read_bits(46, 1) == [True] def test_bit_area2_input(self, datastore): # DI 513/514 live in input area 2 (0x8000+); digital storage starts # after the analog words — default offset 0, so channels 513/514 map # to storage word 32/33 via the global bit index 512/513 datastore.set_digital_input_channel(513, True) datastore.set_digital_input_channel(514, True) assert datastore.read_bits(0x8000 + 1, 1) == [True] def test_bit_illegal_ranges(self, datastore): assert _is_exc(datastore.read_bits(0x0400, 1)) assert _is_exc(datastore.read_bits(0x35F8, 1)) assert _is_exc(datastore.write_bits(0x0400, [True])) assert _is_exc(datastore.read_bits(0x85F8, 1)) assert _is_exc(datastore.read_bits(0x95F8, 1)) def test_bit_write_limit_800(self, datastore): assert _is_exc(datastore.write_bits(0x0000, [True] * 801)) def test_bit_read_limit_2000(self, datastore): assert _is_exc(datastore.read_bits(0x0000, 2001)) class TestWatchdogReaction: def test_clear_outputs(self, datastore): assert datastore.write_words(0x0000, [0xFFFF]) is None assert datastore.write_bits(0x0000, [True] * 16) is None datastore.clear_outputs() assert datastore.output_words == [0] * len(datastore.output_words) class TestListeners: def test_listener_notified_on_write(self, datastore): calls = [] datastore.add_listener(lambda: calls.append(1)) datastore.write_words(0x0000, [1]) datastore.write_bits(0x0000, [True]) assert len(calls) == 2