Andrey
876b58517c
Сохрани адресацию отдельных регистров GAS 0x3 и выдели слотам тип 0x4 с маской Word0. Добавь единое C-ядро GAS_WATCH2, тонкие Python/JNI-порты и проверки общих эталонных байтов. Совместимость: тип 0x4 больше не обслуживает Modbus Coil. Потребителям PCAN_MSG_MODBUS_COIL и ProtoCanMsgToModbusCoil нужно перейти на GAS_SLOT и отдельный API слотов; регистровый GAS остаётся на 0x3.
170 lines
26 KiB
Python
170 lines
26 KiB
Python
# Проверяем неизменяемые модели и настоящий C ABI GAS_WATCH2 на общих эталонных байтах.
|
|
# Python не подменяет C-кодек: проверки wire-пакетов выполняются только с указанной SETPROTOCOL_LIBRARY.
|
|
|
|
import ctypes # Загружаем настоящую общую библиотеку для проверки границы Python/C.
|
|
import json # Читаем общий JSON-fixture, используемый также C- и JVM-проверками.
|
|
import os # Путь к проверяемой сборке получает тестовый процесс через SETPROTOCOL_LIBRARY.
|
|
import unittest # Запускаем воспроизводимые проверки модели, подписок и ошибочных данных.
|
|
from dataclasses import FrozenInstanceError # Проверяем запрет изменения уже созданных моделей.
|
|
from pathlib import Path # Находим эталон относительно исходников независимо от текущего каталога.
|
|
|
|
from protocan.gas_watch import GasWatchConfig, GasWatchSample, NativeGasWatch # Проверяем публичный тонкий порт без доступа к другому Python-кодеку.
|
|
from protocan.protocol import MessageType # Проверяем выделение новых типов сообщений без изменения старых значений.
|
|
|
|
FIXTURE_PATH = Path(__file__).resolve().parents[2] / "c/set-protocol/tests/fixtures/gas-watch-v2.json" # Общие байты находятся рядом с C-тестами как единый контракт портов.
|
|
|
|
|
|
class GasWatchModelTests(unittest.TestCase): # Эти проверки не требуют DLL, поскольку относятся к представлению моделей Python.
|
|
def test_models_copy_sequences_and_are_frozen(self): # Изменение исходных списков не должно менять ожидаемый ACK и принятую выборку.
|
|
addresses = [0, 1, 0xFF01] # Повторяем три разрозненных адреса из пользовательского примера.
|
|
config = GasWatchConfig(1, 100, addresses) # Переданный список должен быть независимо сохранён как кортеж.
|
|
addresses[0] = 99 # Моделируем последующее редактирование таблицы сигналов GUI.
|
|
self.assertEqual((0, 1, 0xFF01), config.addresses) # Проверяем сохранение исходного порядка подписки.
|
|
with self.assertRaises(FrozenInstanceError): # Идентификатор уже сформированной модели нельзя менять на месте.
|
|
config.config_id = 2 # Присваивание проверяет защиту frozen dataclass, а не изменение состояния устройства.
|
|
values = [0, None, 65535] # Корректный ноль и недостоверное значение должны оставаться различимыми.
|
|
sample = GasWatchSample(1, 2, 3, values) # Публичная модель также должна копировать изменяемый контейнер.
|
|
values[0] = 5 # Моделируем повторное использование входного списка вызывающим приложением.
|
|
self.assertEqual((0, None, 65535), sample.values) # Ранее созданная выборка не должна изменяться вслед за списком.
|
|
with self.assertRaises(FrozenInstanceError): # Метка времени проверенной выборки остаётся неизменяемой.
|
|
sample.timestamp_ms = 4 # Присваивание должно быть отклонено механизмом dataclass.
|
|
|
|
def test_integer_ranges_prevent_ctypes_narrowing(self): # Python обязан отвергать значения, которые ctypes незаметно усекло бы до C-типа.
|
|
for config_id in (0, -1, 0x100000000, True, 1.5): # Проверяем нулевой, отрицательный, переполненный и нецелые идентификаторы.
|
|
with self.subTest(config_id=config_id), self.assertRaises(ValueError): # Каждый ошибочный id должен отказать до вызова C.
|
|
GasWatchConfig(config_id, 100, (0,)) # Остальные поля допустимы и не маскируют проверяемую ошибку.
|
|
for period in (-1, 65536, True, 1.5): # Период представляет uint16_t в миллисекундах без преобразования bool и float.
|
|
with self.subTest(period=period), self.assertRaises(ValueError): # Проверяем все источники возможного сужения периода.
|
|
GasWatchConfig(1, period, (0,)) # Валидный идентификатор и адрес изолируют неверный период.
|
|
for address in (-1, 65536, True, 1.5): # GAS-регистр задаётся uint16_t, а не произвольным Python-числом.
|
|
with self.subTest(address=address), self.assertRaises(ValueError): # Любой недопустимый элемент массива должен отказать.
|
|
GasWatchConfig(1, 100, (address,)) # Один адрес позволяет проверить границу независимо от числа каналов.
|
|
with self.assertRaises(ValueError): # Подписка не должна превышать общий предел 64 каналов.
|
|
GasWatchConfig(1, 100, tuple(range(65))) # 65 уникальных допустимых адресов изолируют ограничение размера.
|
|
with self.assertRaises(ValueError): # Значения публичной модели также не должны выходить за 16 бит.
|
|
GasWatchSample(1, 2, 3, (65536,)) # Отдельная ошибочная величина проверяет защиту модели результата.
|
|
|
|
def test_old_library_has_explicit_rebuild_error(self): # Устаревшая библиотека должна дать понятную ошибку без Python fallback.
|
|
with self.assertRaisesRegex(RuntimeError, "set_gas_watch.c"): # Сообщение должно назвать недостающую реализацию для пересборки.
|
|
NativeGasWatch(object()) # Объект без новых ABI-символов воспроизводит отсутствие функции в старой DLL.
|
|
|
|
def test_message_types_preserve_legacy_values(self): # Новые wire-типы не должны переопределять существующий GAS_WATCH.
|
|
self.assertEqual((0x12, 0x13), (MessageType.GAS_WATCH_SET, MessageType.GAS_WATCH_DATA)) # Сохраняем совместимость старого профиля v1.
|
|
self.assertEqual((0x14, 0x15), (MessageType.GAS_WATCH2_SET, MessageType.GAS_WATCH2_DATA)) # Выделенные типы однозначно выбирают новый формат.
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("SETPROTOCOL_LIBRARY"), "Укажите SETPROTOCOL_LIBRARY для проверки настоящего C-ядра") # Без явного пути не подменяем C тестовым Python-кодеком.
|
|
class NativeGasWatchTests(unittest.TestCase): # Проверяем реальные builder, ACK-валидатор и DATA-декодер общей библиотеки.
|
|
@classmethod # Одну DLL и общий fixture безопасно переиспользуют отдельные экземпляры адаптера каждого теста.
|
|
def setUpClass(cls): # Загружаем внешний контракт один раз до тестов C ABI.
|
|
cls.library = ctypes.CDLL(os.environ["SETPROTOCOL_LIBRARY"]) # Указанный путь должен вести к сборке с set_gas_watch.c.
|
|
cls.fixture = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) # Эталонные данные читаются в обязательной кодировке UTF-8.
|
|
cls.request_bytes = bytes.fromhex(cls.fixture["request_hex"]) # Фиксированные байты отделяют ожидаемый wire-вектор от реализации C.
|
|
cls.sample_bytes = bytes.fromhex(cls.fixture["sample_hex"]) # DATA проверяется теми же байтами, что и другие языковые порты.
|
|
|
|
def setUp(self): # Изолируем pending и active между независимыми сценариями.
|
|
self.core = NativeGasWatch(self.library) # Каждый тест начинает с отсутствующей подтверждённой конфигурацией.
|
|
self.config = GasWatchConfig(self.fixture["config_id"], self.fixture["period_ms"], self.fixture["addresses"]) # Сохраняем порядок общих трёх адресов.
|
|
|
|
def activate(self): # Повторяем нормальный цикл отправки конфигурации и её точного подтверждения.
|
|
payload = self.core.request(self.config) # Запрос формирует настоящий C-builder.
|
|
self.core.validate_ack(payload, self.config) # Полное эхо устройства проверяет настоящий C ACK-валидатор.
|
|
|
|
def test_shared_fixture_request_ack_and_sample(self): # Основной сценарий фиксирует совместимость байтов и семантику признаков достоверности.
|
|
self.assertEqual(self.request_bytes, self.core.request(self.config)) # Сопоставляем результат C-builder с общим независимым эталоном.
|
|
with self.assertRaises(RuntimeError): # DATA до ACK не имеет подтверждённого соответствия каналам.
|
|
self.core.sample(self.sample_bytes) # Даже корректный пакет должен быть отвергнут до активации конфигурации.
|
|
self.core.validate_ack(self.request_bytes, self.config) # Активируем подписку только по полному точному эху.
|
|
sample = self.core.sample(self.sample_bytes) # C проверяет config_id, количество, длину, padding и значения invalid-каналов.
|
|
expected_values = tuple(value if valid else None for value, valid in zip(self.fixture["values"], self.fixture["valid"])) # Преобразуем семантику fixture, не декодируя wire-пакет.
|
|
self.assertEqual(GasWatchSample(self.config.config_id, self.fixture["sequence"], self.fixture["timestamp_ms"], expected_values), sample) # Сверяем всю неизменяемую модель результата.
|
|
zero_sample = bytes.fromhex("0200030078563412040302014433221105000000003412") # Фиксированный вектор содержит достоверный ноль, invalid-ноль и слово 0x1234.
|
|
self.assertEqual((0, None, 0x1234), self.core.sample(zero_sample).values) # Недостоверное значение не должно превращаться в отображаемый ноль.
|
|
|
|
def test_native_request_rejects_duplicates_and_inconsistent_stop(self): # Логические ограничения протокола должны проверяться C, а не вторым валидатором Python.
|
|
self.activate() # Действующая подписка нужна для проверки сохранения состояния при отказе builder.
|
|
invalid = (GasWatchConfig(2, 100, (0, 0)), GasWatchConfig(2, 0, (0,)), GasWatchConfig(2, 100, ())) # Проверяем дубликаты и оба несовместимых сочетания периода с количеством.
|
|
for config in invalid: # Каждая ошибочная конфигурация должна быть отвергнута независимо от предыдущей попытки.
|
|
with self.subTest(config=config), self.assertRaises(ValueError): # Отказ C-builder преобразуется в явную ошибку Python.
|
|
self.core.request(config) # Никакой неправильный запрос не должен менять активную подписку.
|
|
self.assertEqual(self.config.config_id, self.core.sample(self.sample_bytes).config_id) # После отказа прежние подтверждённые DATA продолжают приниматься.
|
|
stop = GasWatchConfig(self.config.config_id + 1, 0, ()) # Пустой список с нулевым периодом представляет правильную команду отключения.
|
|
self.assertEqual(bytes.fromhex("020000007956341200000000"), self.core.request(stop)) # C-builder должен поддерживать отдельный идентификатор остановки.
|
|
self.core.validate_ack(bytes.fromhex("020000007956341200000000"), stop) # Полный ACK пустой конфигурации также должен проходить C-проверку.
|
|
|
|
def test_ack_checks_full_configuration_and_preserves_active(self): # Равного количества каналов недостаточно для принятия подтверждения.
|
|
self.activate() # Создаём активную подписку, которая должна пережить ошибочный ACK следующей команды.
|
|
changed = GasWatchConfig(self.config.config_id + 1, 200, (0xFF01, 1, 0)) # Новый порядок того же числа адресов получает новый идентификатор.
|
|
changed_request = self.core.request(changed) # Pending меняется, а действующий профиль остаётся активным до ACK.
|
|
self.assertEqual(self.config.config_id, self.core.sample(self.sample_bytes).config_id) # Старые DATA разрешены, пока устройство не подтвердило переключение.
|
|
for supplied in (self.config, changed): # Отдельно проверяем поздний ACK и пакет, не соответствующий новой конфигурации.
|
|
with self.subTest(supplied=supplied), self.assertRaises(ValueError): # Ни один вариант не должен активировать неподтверждённую подписку.
|
|
self.core.validate_ack(self.request_bytes, supplied) # Вариант old отвергает pending-проверка, вариант new — C-сравнение эха.
|
|
altered = bytearray(changed_request) # Моделируем ACK с тем же id, периодом и count, но другим порядком адресов.
|
|
altered[-2:] = b"\x02\x00" # Меняем один адрес в конце вектора, не затрагивая геометрию ответа.
|
|
with self.assertRaises(ValueError): # Полное сравнение адресов должно обнаружить подмену при прежнем количестве каналов.
|
|
self.core.validate_ack(bytes(altered), changed) # C проверяет весь список и не разрешает частичную подписку.
|
|
self.core.validate_ack(changed_request, changed) # Корректное полное эхо атомарно переводит pending в active.
|
|
with self.assertRaises(ValueError): # Пакет прежнего id после успешного переключения становится устаревшим.
|
|
self.core.sample(self.sample_bytes) # Количество осталось тем же, поэтому проверка должна опираться на идентификатор.
|
|
new_sample = bytes.fromhex("0200030079563412fffffffff0ffffff037b00feff0000") # Фиксированный DATA-вектор относится к новому config_id 0x12345679.
|
|
self.assertEqual(changed.config_id, self.core.sample(new_sample).config_id) # Новый подтверждённый id разрешает данные именно новой конфигурации.
|
|
|
|
def test_malformed_ack_and_sample_do_not_poison_subscription(self): # Ошибки отдельных пакетов не должны разрушать рабочую подписку.
|
|
self.core.request(self.config) # Pending необходим для проверки точного формата первого подтверждения.
|
|
for payload in (self.request_bytes[:-1], self.request_bytes + b"\x00"): # Отсечённые и дополненные ACK имеют неправильную длину.
|
|
with self.subTest(ack=payload), self.assertRaises(ValueError): # C должен отвергнуть и неполное, и нестрогое эхо.
|
|
self.core.validate_ack(payload, self.config) # Pending сохраняется, позволяя принять следующий корректный ACK.
|
|
self.core.validate_ack(self.request_bytes, self.config) # Корректное подтверждение после ошибок всё ещё активирует исходную подписку.
|
|
malformed = ( # Список фиксированных искажений покрывает геометрию, достоверность и привязку к конфигурации.
|
|
self.sample_bytes[:-1], # Отбрасываем последнее слово частично: общая длина больше не соответствует count.
|
|
self.sample_bytes + b"\x00", # Лишний байт также нарушает точную геометрию пакета.
|
|
bytes.fromhex("0100030078563412fffffffff0ffffff037b00feff0000"), # Версия 1 не должна интерпретироваться как новый формат версии 2.
|
|
bytes.fromhex("0200030079563412fffffffff0ffffff037b00feff0000"), # Другой id с прежним count моделирует поздние DATA другой подписки.
|
|
bytes.fromhex("0200020078563412fffffffff0ffffff037b00feff"), # Корректная геометрия двух значений не совпадает с тремя ожидаемыми адресами.
|
|
bytes.fromhex("0200030078563412fffffffff0ffffff837b00feff0000"), # Старший padding-бит bitmap для трёх каналов обязан быть нулём.
|
|
bytes.fromhex("0200030078563412fffffffff0ffffff037b00feff0100"), # Значение invalid-канала обязано быть нулём, иначе данные неоднозначны.
|
|
) # Закрываем перечень независимых отрицательных векторов C-декодера.
|
|
for payload in malformed: # Проверяем каждый отказ с тем же неизменным состоянием подписки.
|
|
with self.subTest(data=payload), self.assertRaises(ValueError): # Ни один некорректный пакет не должен превращаться в модель.
|
|
self.core.sample(payload) # Все проверки структуры пакета выполняются общим C-декодером.
|
|
self.assertEqual(self.config.config_id, self.core.sample(self.sample_bytes).config_id) # Следующий корректный пакет доказывает сохранение active после ошибки.
|
|
|
|
def test_failed_decode_keeps_all_c_outputs_unchanged(self): # Проверяем обещание ABI не публиковать частичные результаты при ошибке пакета.
|
|
data = (ctypes.c_uint8 * len(self.sample_bytes)).from_buffer_copy(self.sample_bytes) # Передаём эталонные байты, изменяя только ожидаемый id C-вызова.
|
|
sequence = ctypes.c_uint32(11) # Отличимый sentinel обнаруживает запись счётчика до окончания проверки.
|
|
timestamp = ctypes.c_uint32(22) # Независимый sentinel защищает проверку метки времени.
|
|
values = (ctypes.c_uint16 * 3)(31, 32, 33) # Отличимые слова обнаруживают частичное заполнение выходного массива.
|
|
valid = (ctypes.c_uint8 * 3)(41, 42, 43) # Признаки также не должны перезаписываться при устаревшем id.
|
|
result = self.core.lib.set_gas_watch_decode(data, len(data), self.config.config_id + 1, 3, ctypes.byref(sequence), ctypes.byref(timestamp), values, valid, 3) # Проверяем настоящий C ABI без обёртки создания модели.
|
|
self.assertEqual(-1, result) # Несовпадающий expected_config_id обязан отвергнуть даже корректно сформированный пакет.
|
|
self.assertEqual((11, 22, (31, 32, 33), (41, 42, 43)), (sequence.value, timestamp.value, tuple(values), tuple(valid))) # Каждый выход должен сохранить исходный sentinel.
|
|
|
|
def test_pending_replacement_id_reuse_and_reset(self): # Проверяем гонки нескольких запросов и очистку состояния при переподключении.
|
|
self.activate() # Ранее подтверждённая конфигурация задаёт исходный active.
|
|
changed = GasWatchConfig(self.config.config_id + 1, 100, (1, 0, 0xFF01)) # Первый новый порядок начинает ожидать подтверждение.
|
|
self.core.request(changed) # Старый active остаётся доступным, пока новый pending не подтверждён.
|
|
newest = GasWatchConfig(self.config.config_id + 2, 100, (0xFF01, 0, 1)) # Следующая команда заменяет ожидаемый список ещё до первого ACK.
|
|
newest_request = self.core.request(newest) # Актуальным pending становится только последний запрос.
|
|
with self.assertRaises(ValueError): # ACK уже заменённого pending не должен возвращать старое ожидание.
|
|
self.core.validate_ack(self.request_bytes, changed) # Проверка pending срабатывает до анализа неподходящего payload.
|
|
for reused in (GasWatchConfig(self.config.config_id, 200, self.config.addresses), GasWatchConfig(newest.config_id, 200, newest.addresses)): # Active и pending id нельзя переиспользовать с другим содержимым.
|
|
with self.subTest(reused=reused), self.assertRaises(ValueError): # Такая подмена лишила бы DATA возможности определить порядок значений.
|
|
self.core.request(reused) # Ошибка не должна уничтожать последний ожидаемый запрос.
|
|
self.core.validate_ack(newest_request, newest) # Последний корректный pending всё ещё подтверждается после отвергнутых попыток.
|
|
self.core.reset() # Переподключение удаляет как ожидаемую, так и подтверждённую подписку.
|
|
with self.assertRaises(RuntimeError): # Старая выборка после reset требует нового цикла подтверждения.
|
|
self.core.sample(self.sample_bytes) # Данные прошлого соединения не должны отображаться автоматически.
|
|
with self.assertRaises(ValueError): # Ни один ACK прошлого соединения больше не ожидается.
|
|
self.core.validate_ack(newest_request, newest) # Reset запрещает позднее восстановление прежнего active.
|
|
|
|
def test_maximum_subscription_uses_real_c_builder(self): # Проверяем безопасную передачу полного массива из 64 каналов через ctypes.
|
|
config = GasWatchConfig(0xFFFFFFFF, 65535, tuple(range(64))) # Используем максимальные id и период при допустимом числе уникальных адресов.
|
|
payload = self.core.request(config) # Размер payload определяет настоящий C-builder в общем транспортном буфере.
|
|
self.assertEqual(140, len(payload)) # Версия 2 задаёт 12 байт заголовка и 64 двухбайтных адреса, всего 140 байт.
|
|
self.core.validate_ack(payload, config) # Полное эхо максимального массива проверяется без усечения и частичного принятия.
|
|
|
|
|
|
if __name__ == "__main__": # Разрешаем запуск этого набора проверок напрямую наряду с unittest discovery.
|
|
unittest.main() # Стандартный runner сообщает результаты, ошибки DLL и пропущенные C-проверки.
|