diff --git a/tests/decommutator/test_ccsds_reader_pipeline.py b/tests/decommutator/test_ccsds_reader_pipeline.py new file mode 100644 index 0000000..65657aa --- /dev/null +++ b/tests/decommutator/test_ccsds_reader_pipeline.py @@ -0,0 +1,146 @@ +""" +Tests of the private helper functions in ccsds_reader_pipeline. + +The end-to-end test in ``test_decommutator.py`` checks that a whole L0 +file is converted to the expected L05 CDF files, but when it fails it +does not show which step went wrong. The tests here check individual +functions so that a failure points to the function responsible. + +These tests were generated by Claude. These tests should be treated +as regression tests to catch unexpected changes in behavior, rather +than checks to make sure that functions are behaving correctly. +""" + +from __future__ import annotations + +import struct + +import pytest + +from pyfaradaycup.decommutator.ccsds_reader_pipeline import _parse_ccsds_head + +_FIELD_WIDTHS = { + "CCSDS_Version": 3, + "CCSDS_PacketType": 1, + "CCSDS_SecHdrFlag": 1, + "CCSDS_ApID": 11, + "CCSDS_GroupFlags": 2, + "CCSDS_SeqCnt": 14, + "CCSDS_PacketLen": 16, + "CCSDS_MET": 32, +} + + +def _make_ccsds_head(fields: dict[str, int]) -> bytes: + """ + Encode CCSDS header fields into a 10-byte header. + + This is the inverse of `_parse_ccsds_head`. It is written with + `struct.pack` and bit shifts rather than by reusing any code from + the package, so that a mistake in the decoder is not repeated in + the encoder and hidden from the tests. + + Parameters + ---------- + fields : dict of str to int + The header fields, with the same keys that `_parse_ccsds_head` + returns. + + Returns + ------- + bytes + The 6-byte CCSDS primary header followed by the 4-byte MET. + """ + word0 = ( + fields["CCSDS_Version"] << 13 + | fields["CCSDS_PacketType"] << 12 + | fields["CCSDS_SecHdrFlag"] << 11 + | fields["CCSDS_ApID"] + ) + word1 = fields["CCSDS_GroupFlags"] << 14 | fields["CCSDS_SeqCnt"] + return struct.pack( + ">HHHI", word0, word1, fields["CCSDS_PacketLen"], fields["CCSDS_MET"] + ) + + +_typical_fields = { + "CCSDS_Version": 0, + "CCSDS_PacketType": 0, + "CCSDS_SecHdrFlag": 1, + "CCSDS_ApID": 0x352, + "CCSDS_GroupFlags": 3, + "CCSDS_SeqCnt": 0x1234, + "CCSDS_PacketLen": 0x098D, + "CCSDS_MET": 523462910, +} + + +@pytest.mark.parametrize( + "fields", + [ + _typical_fields, + {**_typical_fields, "CCSDS_ApID": 0x081}, + {**_typical_fields, "CCSDS_ApID": 0x35E}, + {**_typical_fields, "CCSDS_ApID": 0x7FF}, + dict.fromkeys(_FIELD_WIDTHS, 0), + {name: 2**width - 1 for name, width in _FIELD_WIDTHS.items()}, + ], +) +def test_parse_ccsds_head_round_trip(fields: dict[str, int]) -> None: + """ + Test that decoding an encoded header recovers the original fields. + + Every packet that the decommutator reads is identified and sized + from its header, so a wrong field here means packets are skipped, + assigned to the wrong APID, or given wrong times. The cases include + the SPC and spacecraft APIDs that the decommutator looks for, the + largest 11-bit APID (to check that the top 3 APID bits, which share + a byte with the version and flags, are decoded), and headers with + every field at its minimum or maximum value. + """ + assert _parse_ccsds_head(_make_ccsds_head(fields)) == fields + + +@pytest.mark.parametrize("name", _FIELD_WIDTHS) +def test_parse_ccsds_head_field_isolation(name: str) -> None: + """ + Test that setting every bit of one field does not leak into the others. + + Several fields share a byte, so a mask or shift that is off by one + bit would move a bit from one field into its neighbor. Round-trip + tests with typical values can miss this, because the bits next to a + field boundary are often zero. Setting one field to its maximum and + the rest to zero makes any such leak show up as a nonzero value in + a field that should be zero, or a smaller value in the field + being tested. + """ + fields = dict.fromkeys(_FIELD_WIDTHS, 0) + fields[name] = 2 ** _FIELD_WIDTHS[name] - 1 + assert _parse_ccsds_head(_make_ccsds_head(fields)) == fields + + +def test_parse_ccsds_head_ignores_extra_bytes() -> None: + """ + Test that only the first 10 bytes are decoded. + + Callers do not always pass exactly 10 bytes: ``read_file_sc`` passes + the contents of the whole file to read the first header. The bytes + after the header must not change the result. + """ + header = _make_ccsds_head(_typical_fields) + assert _parse_ccsds_head(header + b"\xff" * 20) == _typical_fields + + +@pytest.mark.parametrize("length", range(10)) +def test_parse_ccsds_head_too_short(length: int) -> None: + """ + Test that a header shorter than 10 bytes raises a ValueError. + + A packet can be cut off at the end of an L0 file. ``_read_bytestr`` + and ``read_file_sc`` catch `ValueError` to skip such a packet, so a + short header must raise `ValueError` rather than another exception + or a header with missing fields. + """ + header = _make_ccsds_head(_typical_fields)[:length] + with pytest.raises(ValueError, match="not as long as expected"): + _parse_ccsds_head(header)