diff --git a/tests/test_bpsk_radio.py b/tests/test_bpsk_radio.py new file mode 100644 index 0000000..96c777b --- /dev/null +++ b/tests/test_bpsk_radio.py @@ -0,0 +1,212 @@ +"""Проверки protocol/bpsk_radio.py. + +Модуль был собран переносом функций из Lab018, Lab019 и Lab023. Главная +задача этих проверок — доказать, что перенос ничего не изменил: результаты +сравниваются с оригиналами, загруженными из лабораторных в изоляции, без +выполнения их модульного кода. +""" + +from __future__ import annotations + +import ast +import pathlib +import struct + +import numpy as np +import pytest + +from protocol import bpsk_radio as radio +from protocol.packet import MESSAGE_TYPE_IMAGE_FRAGMENT, build_packet, parse_packet + + +PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent +SHARED_CONSTANTS = { + "RADIO_SYNC_WORD", + "PREAMBLE_BIT_COUNT", + "SYMBOL_RATE", + "CFO_REFINEMENT_HALF_WIDTH_HZ", + "CFO_REFINEMENT_STEP_HZ", +} + + +def _load_isolated(relative_path: str, function_names: set[str]) -> dict: + """Выполнить только заданные функции лабораторной, без кода модуля. + + Лабораторные при импорте создают каталоги и пишут файлы, поэтому + импортировать их в тестах нельзя. + """ + + path = PROJECT_ROOT / relative_path + tree = ast.parse(path.read_text(encoding="utf-8")) + kept = [ + node + for node in tree.body + if (isinstance(node, ast.FunctionDef) and node.name in function_names) + or (isinstance(node, ast.Assign) and ast.unparse(node.targets[0]) in SHARED_CONSTANTS) + ] + namespace: dict = {"np": np, "struct": struct} + exec(compile(ast.Module(body=kept, type_ignores=[]), str(path), "exec"), namespace) + return namespace + + +@pytest.fixture(scope="module") +def original_lab023() -> dict: + return _load_isolated( + "experiments/lab023_guarded_cfo_correction.py", + { + "bytes_to_bits", + "bits_to_bytes", + "bpsk_modulate", + "bpsk_demodulate", + "estimate_carrier_parameters", + "correct_phase_and_frequency", + }, + ) + + +@pytest.fixture(scope="module") +def original_lab019() -> dict: + return _load_isolated( + "experiments/lab019_bpsk_receiver.py", + { + "build_frame_marker", + "find_radio_frame", + "bytes_to_bits", + "bits_to_bytes", + "bpsk_modulate", + "bpsk_demodulate", + "root_raised_cosine_taps", + }, + ) + + +@pytest.fixture(scope="module") +def original_lab018() -> dict: + return _load_isolated("experiments/lab018_bpsk_radio_frame.py", {"root_raised_cosine_taps"}) + + +def _identical(left, right) -> bool: + if isinstance(left, np.ndarray) or isinstance(right, np.ndarray): + return np.array_equal(np.asarray(left), np.asarray(right)) + if isinstance(left, dict): + return sorted(left) == sorted(right) and all(_identical(left[k], right[k]) for k in left) + if isinstance(left, (tuple, list)): + return len(left) == len(right) and all(_identical(a, b) for a, b in zip(left, right)) + return left == right + + +# ---------------------------------------------------------------- эквивалентность + + +@pytest.mark.parametrize("length", [1, 7, 64, 255]) +def test_bit_conversion_matches_lab023(original_lab023: dict, length: int) -> None: + data = bytes(np.random.default_rng(length).integers(0, 256, size=length, dtype=np.uint8)) + ours = radio.bytes_to_bits(data) + assert _identical(ours, original_lab023["bytes_to_bits"](data)) + assert _identical(radio.bits_to_bytes(ours), original_lab023["bits_to_bytes"](ours)) + + +@pytest.mark.parametrize("length", [8, 64, 256]) +def test_modulation_matches_lab023(original_lab023: dict, length: int) -> None: + bits = np.random.default_rng(length).integers(0, 2, size=length).astype(np.uint8) + ours = radio.bpsk_modulate(bits) + assert _identical(ours, original_lab023["bpsk_modulate"](bits)) + assert _identical(radio.bpsk_demodulate(ours), original_lab023["bpsk_demodulate"](ours)) + + +@pytest.mark.parametrize("rolloff,sps,span", [(0.35, 32, 10), (0.35, 128, 10), (0.5, 64, 8)]) +def test_shaping_filter_matches_lab018(original_lab018: dict, rolloff: float, sps: int, span: int) -> None: + assert _identical( + radio.root_raised_cosine_taps(rolloff, sps, span), + original_lab018["root_raised_cosine_taps"](rolloff, sps, span), + ) + + +def test_frame_marker_matches_lab019(original_lab019: dict) -> None: + """Маркер возвращает пару: биты и символы. Совпасть должны обе.""" + + assert _identical(radio.build_frame_marker(), original_lab019["build_frame_marker"]()) + + +@pytest.mark.parametrize("seed", [1, 2, 3]) +def test_carrier_estimation_matches_lab023(original_lab023: dict, seed: int) -> None: + """Оценка ухода частоты обязана совпасть при произвольных фазе и уходе.""" + + rng = np.random.default_rng(seed) + _, marker = radio.build_frame_marker() + phase = float(rng.uniform(-3.0, 3.0)) + increment = float(rng.uniform(-0.005, 0.005)) + impaired = marker * np.exp(1j * (phase + increment * np.arange(len(marker)))) + impaired = impaired + 0.02 * rng.standard_normal(len(marker)) + + assert _identical( + radio.estimate_carrier_parameters(impaired, marker), + original_lab023["estimate_carrier_parameters"](impaired, marker), + ) + assert _identical( + radio.correct_phase_and_frequency(impaired, phase, increment), + original_lab023["correct_phase_and_frequency"](impaired, phase, increment), + ) + + +# ---------------------------------------------------------------- новые функции кадра + + +@pytest.mark.parametrize("payload_size", [1, 17, 512, radio.MAXIMUM_PROTOCOL_PACKET_BYTES]) +def test_frame_roundtrip(payload_size: int) -> None: + packet = bytes(np.random.default_rng(payload_size).integers(0, 256, size=payload_size, dtype=np.uint8)) + bits, symbols, marker = radio.build_radio_frame(packet) + + assert len(bits) == radio.radio_frame_bit_count(payload_size) + assert len(symbols) == len(bits) + assert len(marker) == radio.MARKER_BIT_COUNT + assert radio.parse_radio_frame(bits) == packet + + +def test_frame_rejects_sizes_outside_range() -> None: + with pytest.raises(ValueError): + radio.build_radio_frame(b"") + with pytest.raises(ValueError): + radio.build_radio_frame(b"\x00" * (radio.MAXIMUM_PROTOCOL_PACKET_BYTES + 1)) + with pytest.raises(TypeError): + radio.build_radio_frame("не байты") + + +def test_frame_with_broken_sync_word_is_rejected() -> None: + bits, _, _ = radio.build_radio_frame(b"payload") + broken = bits.copy() + broken[radio.PREAMBLE_BIT_COUNT] ^= 1 + assert radio.parse_radio_frame(broken) is None + + +def test_truncated_frame_is_rejected() -> None: + bits, _, _ = radio.build_radio_frame(b"payload") + assert radio.parse_radio_frame(bits[:10]) is None + assert radio.parse_radio_frame(bits[:-8]) is None + + +def test_frame_carries_a_protocol_packet() -> None: + """Кадр должен переносить настоящий пакет так, чтобы CRC32 сошлась.""" + + packet = build_packet(b"fragment payload", MESSAGE_TYPE_IMAGE_FRAGMENT, 5) + bits, _, _ = radio.build_radio_frame(packet) + assert parse_packet(radio.parse_radio_frame(bits)).payload == b"fragment payload" + + +def test_frame_search_finds_the_start_in_a_shaped_signal() -> None: + """Сквозная проверка: формирование, фильтрация и поиск кадра.""" + + samples_per_symbol = 32 + taps = radio.root_raised_cosine_taps(0.35, samples_per_symbol, 10) + packet = build_packet(b"HELLO SDR", MESSAGE_TYPE_IMAGE_FRAGMENT, 1) + bits, symbols, marker = radio.build_radio_frame(packet) + + upsampled = np.zeros(len(symbols) * samples_per_symbol, dtype=complex) + upsampled[::samples_per_symbol] = symbols + matched = np.convolve(np.convolve(upsampled, taps, mode="full"), taps, mode="full") + + found = radio.find_radio_frame(matched, marker, samples_per_symbol) + start = found["start_symbol_index"] + recovered = found["symbol_samples"][start : start + len(bits)] + + assert radio.parse_radio_frame(radio.bpsk_demodulate(recovered)) == packet diff --git a/tests/test_packet.py b/tests/test_packet.py new file mode 100644 index 0000000..edaa9a9 --- /dev/null +++ b/tests/test_packet.py @@ -0,0 +1,97 @@ +"""Быстрые проверки пакетного протокола: protocol/packet.py.""" + +from __future__ import annotations + +import struct + +import pytest + +from protocol.packet import ( + CRC_SIZE, + HEADER_SIZE, + MAX_PAYLOAD_SIZE, + MESSAGE_TYPE_ACK, + MESSAGE_TYPE_IMAGE_FRAGMENT, + MESSAGE_TYPE_TELEMETRY, + MESSAGE_TYPE_TEXT, + PROTOCOL_VERSION, + SYNC_WORD, + CRCError, + PacketError, + build_packet, + parse_packet, +) + + +MESSAGE_TYPES = ( + MESSAGE_TYPE_TEXT, + MESSAGE_TYPE_TELEMETRY, + MESSAGE_TYPE_IMAGE_FRAGMENT, + MESSAGE_TYPE_ACK, +) + + +@pytest.mark.parametrize("message_type", MESSAGE_TYPES) +@pytest.mark.parametrize("payload", [b"", b"H", b"HELLO SDR", bytes(range(256)), b"\x00" * 1024]) +def test_roundtrip_preserves_payload(payload: bytes, message_type: int) -> None: + parsed = parse_packet(build_packet(payload, message_type, 7)) + assert parsed.payload == payload + assert parsed.message_type == message_type + assert parsed.sequence_number == 7 + assert parsed.version == PROTOCOL_VERSION + + +def test_packet_starts_with_sync_word() -> None: + packet = build_packet(b"data", MESSAGE_TYPE_TEXT, 1) + assert struct.unpack(">H", packet[:2])[0] == SYNC_WORD + + +def test_length_is_header_plus_payload_plus_crc() -> None: + payload = b"x" * 300 + packet = build_packet(payload, MESSAGE_TYPE_TEXT, 1) + assert len(packet) == HEADER_SIZE + len(payload) + CRC_SIZE + + +def test_sequence_number_survives_full_range() -> None: + for sequence_number in (0, 1, 65534, 65535): + assert parse_packet(build_packet(b"a", MESSAGE_TYPE_TEXT, sequence_number)).sequence_number == sequence_number + + +@pytest.mark.parametrize("bit_index", [0, 7, 8, 23, 64, 100]) +def test_single_bit_corruption_is_detected(bit_index: int) -> None: + """CRC32 обязан ловить одиночное искажение бита в любом месте пакета.""" + + packet = bytearray(build_packet(b"HELLO SDR TELEMETRY", MESSAGE_TYPE_TELEMETRY, 3)) + byte_index, offset = divmod(bit_index, 8) + assert byte_index < len(packet) + packet[byte_index] ^= 1 << offset + + with pytest.raises((CRCError, PacketError)): + parse_packet(bytes(packet)) + + +def test_truncated_packet_is_rejected() -> None: + packet = build_packet(b"HELLO", MESSAGE_TYPE_TEXT, 1) + for cut in (0, 1, HEADER_SIZE - 1, len(packet) - 1): + with pytest.raises(PacketError): + parse_packet(packet[:cut]) + + +def test_oversized_payload_is_rejected() -> None: + with pytest.raises((PacketError, ValueError)): + build_packet(b"\x00" * (MAX_PAYLOAD_SIZE + 1), MESSAGE_TYPE_TEXT, 1) + + +def test_wrong_sync_word_is_rejected() -> None: + packet = bytearray(build_packet(b"HELLO", MESSAGE_TYPE_TEXT, 1)) + packet[0] ^= 0xFF + with pytest.raises(PacketError): + parse_packet(bytes(packet)) + + +def test_payload_is_not_shared_with_caller() -> None: + """Разбор не должен возвращать ссылку на изменяемый буфер.""" + + packet = bytearray(build_packet(b"HELLO", MESSAGE_TYPE_TEXT, 1)) + parsed = parse_packet(bytes(packet)) + assert isinstance(parsed.payload, bytes)