SpoolBuddy: check a tag write instead of always reporting success (#3076)

- the PN5180 can't see the tag's ACK, so the written pages are read
  back after a fresh select; other data fails the write and names the
  page, no answer leaves it "not verified" instead of failing it
- a tag whose Capability Container isn't NDEF (a blank Ultralight) or
  is smaller than the payload is refused before writing
- a write whose result is still in the daemon's buffer is not run
  again when the next heartbeat brings the same command back

spoolbuddy/daemon/api_client.py

@@ -11,6 +11,10 @@ logger = logging.getLogger(__name__)
 MAX_BUFFER_SIZE = 100
 
 
+# Where a tag write's result goes; main checks the buffer for it (#3076).
+WRITE_RESULT_PATH = "/nfc/write-result"
+
+
 def _says_unregistered(response: httpx.Response) -> bool:
     """The backend's own answer for a device it does not know -- not a bare
     404 from a reverse proxy while Bambuddy is restarting."""
@@ -73,6 +77,10 @@ class APIClient:
             logger.warning("GET %s failed: %s", path, e)
             return None
 
+    def has_buffered(self, path: str) -> bool:
+        """Whether a POST to *path* is still waiting in the buffer to be sent."""
+        return any(item["path"] == path for item in self._buffer)
+
     async def _flush_buffer(self):
         while self._buffer:
             item = self._buffer[0]
@@ -214,7 +222,7 @@ class APIClient:
         self, device_id: str, spool_id: int, tag_uid: str, success: bool, message: str | None = None
     ) -> dict | None:
         return await self._post(
-            "/nfc/write-result",
+            WRITE_RESULT_PATH,
             {
                 "device_id": device_id,
                 "spool_id": spool_id,

spoolbuddy/daemon/main.py

@@ -11,7 +11,7 @@ import time
 from pathlib import Path
 
 from . import __version__, system_stats
-from .api_client import APIClient
+from .api_client import WRITE_RESULT_PATH, APIClient
 from .config import Config
 from .display_control import DisplayControl
 from .nfc_reader import NFCReader, NFCState
@@ -154,6 +154,11 @@ async def nfc_poll_loop(config: Config, api: APIClient, shared: dict):
                         message=msg,
                     )
                     shared.pop("pending_write", None)
+                    if api.has_buffered(WRITE_RESULT_PATH):
+                        # The backend keeps the write command until it has the
+                        # result, so heartbeats keep bringing it back meanwhile
+                        # (#3076). See the write_tag handling in heartbeat_loop.
+                        shared["write_result_unsent"] = pending["spool_id"]
                 else:
                     # Fail fast when a non-NTAG is presented during write mode.
                     # Without this, UI can appear stuck on "waiting for SpoolBuddy".
@@ -415,7 +420,21 @@ async def heartbeat_loop(config: Config, api: APIClient, start_time: float, shar
                 continue
             elif cmd == "write_tag":
                 write_payload = result.get("pending_write_payload")
-                if write_payload:
+                unsent = shared.get("write_result_unsent")
+                if write_payload and unsent is not None and write_payload["spool_id"] == unsent:
+                    # The write was done; its result hadn't reached the backend
+                    # when this heartbeat was answered, so the command is the
+                    # same write again. Running it would write the tag on the
+                    # reader a second time, a different one if it was swapped
+                    # (#3076). Once the result is out, the backend clears the
+                    # command and a new write request arrives as usual.
+                    if not api.has_buffered(WRITE_RESULT_PATH):
+                        shared.pop("write_result_unsent", None)
+                    logger.info(
+                        "Write tag command for spool %d ignored: the result of that write is still on its way",
+                        unsent,
+                    )
+                elif write_payload:
                     shared["pending_write"] = {
                         "spool_id": write_payload["spool_id"],
                         "ndef_data": bytes.fromhex(write_payload["ndef_data_hex"]),

spoolbuddy/daemon/nfc_reader.py

@@ -97,6 +97,13 @@ class NFCReader:
         2.004, #3500) is written instead when the tag's Capability Container
         says it has room for it; v2 does not fit an NTAG213.
 
+        The Capability Container is read first: a tag whose CC is not NDEF
+        (a blank MIFARE Ultralight also answers with SAK 0x00) or whose data
+        area is smaller than the payload is refused rather than written past
+        its end (#3076). Afterwards the pages are read back, since the PN5180
+        cannot see the tag's ACK: data that reads back different fails the
+        write. A read-back that gets no answer leaves it "not verified".
+
         Requires tag in TAG_PRESENT state with SAK=0x00.
         Returns (success, message).
         """
@@ -117,32 +124,78 @@ class NFCReader:
             if uid_bytes.hex().upper() != self._current_uid:
                 return False, "Tag UID changed during write"
 
+            # Page 3 is the Capability Container; READ returns pages 3-6.
+            try:
+                cc_pages = self._nfc.ntag_read_pages(start_page=3, num_pages=1)
+            except Exception as e:
+                logger.warning("Could not read the tag's Capability Container: %s", e)
+                cc_pages = None
+
             version = "v1"
             if data_v2 is not None:
-                # Page 3 is the Capability Container; READ returns pages 3-6.
-                try:
-                    cc_pages = self._nfc.ntag_read_pages(start_page=3, num_pages=1)
-                except Exception as e:
-                    logger.warning("Could not read the tag's Capability Container: %s", e)
-                    cc_pages = None
                 data, version = choose_ndef(cc_pages, data, data_v2)
-                # A fresh select before writing, as the write path always had.
-                if self._nfc.reactivate_card() is None:
-                    return False, "Failed to reactivate card for write"
+            refusal = capacity_refusal(cc_pages, len(data))
+            if refusal:
+                logger.warning("NTAG write refused for tag %s: %s", self._current_uid, refusal)
+                return False, refusal
+
+            # A fresh select before writing, as the write path always had.
+            if self._nfc.reactivate_card() is None:
+                return False, "Failed to reactivate card for write"
 
             # Write starting at page 4
-            success = self._nfc.ntag_write_pages(start_page=4, data=data)
-            if success:
-                logger.info(
-                    "NTAG write successful: %d bytes (OpenTag3D %s) to tag %s", len(data), version, self._current_uid
-                )
-                return True, f"Write successful (OpenTag3D {version})"
-            else:
-                return False, "Write or verification failed"
+            if not self._nfc.ntag_write_pages(start_page=4, data=data):
+                return False, "Write failed"
+
+            verified, mismatch = self._verify_written(data)
+            if verified is False:
+                return False, mismatch
+            note = "" if verified else ", not verified"
+            logger.info(
+                "NTAG write successful%s: %d bytes (OpenTag3D %s) to tag %s",
+                note,
+                len(data),
+                version,
+                self._current_uid,
+            )
+            return True, f"Write successful (OpenTag3D {version}{note})"
         except Exception as e:
             logger.error("NTAG write error: %s", e)
             return False, f"Write error: {e}"
 
+    def _verify_written(self, data: bytes) -> tuple[bool | None, str]:
+        """Read the written pages back and compare (#3076).
+
+        (True, "") when they match, (False, where) when they differ, (None, "")
+        when the tag could not be read back. The PN5180 used to
+        get no answer to a READ right after a batch write, so the tag is
+        selected afresh first; a read that still fails proves nothing either
+        way, and the write keeps counting as done, as it did before.
+        """
+        padded = data + bytes(-len(data) % 4)
+        try:
+            if self._nfc.reactivate_card() is None:
+                logger.warning("NTAG read-back skipped: the tag did not answer the reselect")
+                return None, ""
+            read = self._nfc.ntag_read_pages(start_page=4, num_pages=len(padded) // 4)
+        except Exception as e:
+            logger.warning("NTAG read-back failed: %s", e)
+            return None, ""
+        if read is None or len(read) < len(padded):
+            logger.warning("NTAG read-back got no complete answer; the write is not verified")
+            return None, ""
+        for offset in range(0, len(padded), 4):
+            if read[offset : offset + 4] != padded[offset : offset + 4]:
+                page = 4 + offset // 4
+                logger.warning(
+                    "NTAG read-back mismatch at page %d: wrote %s, read %s",
+                    page,
+                    padded[offset : offset + 4].hex(),
+                    read[offset : offset + 4].hex(),
+                )
+                return False, f"Verification failed at page {page}: the tag holds other data"
+        return True, ""
+
     def poll(self) -> tuple[str, dict | None]:
         """Poll for tag. Returns (event_type, event_data).
 
@@ -284,6 +337,24 @@ def _extract_tray_uuid(blocks: dict[int, bytes]) -> str | None:
     return uuid_str
 
 
+def capacity_refusal(cc_pages: bytes | None, size: int) -> str | None:
+    """Why *size* bytes can't go onto this tag, or None (#3076).
+
+    *cc_pages* starts with page 3, the Capability Container. NTAG21x tags come
+    with an NDEF CC (E1, version, data area size / 8, access); a blank MIFARE
+    Ultralight, which also reports SAK 0x00, has none and holds 48 bytes. An
+    unreadable CC proves nothing, so the write goes ahead as it always did.
+    """
+    if not cc_pages or len(cc_pages) < 4:
+        return None
+    if cc_pages[0] != 0xE1:
+        return "Not an NDEF-formatted NTAG tag. Use an NTAG213, 215 or 216."
+    capacity = cc_pages[2] * 8
+    if size > capacity:
+        return f"Tag too small: {size} bytes to write, {capacity} bytes of space"
+    return None
+
+
 def choose_ndef(cc_pages: bytes | None, data_v1: bytes, data_v2: bytes) -> tuple[bytes, str]:
     """Pick the OpenTag3D encoding the tag has room for (#3500).
 

spoolbuddy/daemon/pn5180.py

@@ -573,8 +573,8 @@ class PN5180:
 
         NTAG WRITE command: 0xA2 + page_number + 4 bytes data.
         TX CRC on (tag requires it). Always returns True — the 4-bit ACK
-        cannot be captured by the PN5180, so verification is deferred to
-        ntag_write_pages() which reads back all written data.
+        cannot be captured by the PN5180, so NFCReader.write_ntag() reads
+        the written pages back after a fresh select (#3076).
         """
         if len(data) != 4:
             return False
@@ -604,10 +604,10 @@ class PN5180:
     def ntag_write_pages(self, start_page: int, data: bytes) -> bool:
         """Write data to consecutive NTAG pages starting at start_page.
 
-        Pads last chunk to 4 bytes. Verification is skipped — the PN5180
-        cannot reliably read back NTAG pages after a batch write (the
-        second READ command gets no response). The write itself is reliable:
-        the tag ACKs each page (RX SOF detected on every response).
+        Pads last chunk to 4 bytes. Nothing is verified here: the PN5180
+        cannot see the tag's ACK, and a READ straight after a batch write gets
+        no response. NFCReader.write_ntag() selects the tag afresh and reads
+        the pages back (#3076).
         """
         # Pad to 4-byte boundary
         padded = bytearray(data)

spoolbuddy/tests/test_api_client.py

@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
 import pytest
-from daemon.api_client import MAX_BUFFER_SIZE, APIClient
+from daemon.api_client import MAX_BUFFER_SIZE, WRITE_RESULT_PATH, APIClient
 
 
 @pytest.fixture
@@ -57,6 +57,17 @@ class TestPost:
         assert len(api._buffer) == 1
         assert api._buffer[0] == {"path": "/test", "data": {"data": 1}}
 
+    @pytest.mark.asyncio
+    async def test_a_buffered_write_result_is_reported(self, api):
+        # #3076: main must know a tag write's result is still unsent.
+        api._client.post = AsyncMock(side_effect=httpx.ConnectError("down"))
+        assert not api.has_buffered(WRITE_RESULT_PATH)
+
+        await api.write_tag_result("dev-1", 7, "04A1", True, "ok")
+
+        assert api.has_buffered(WRITE_RESULT_PATH)
+        assert not api.has_buffered("/heartbeat")
+
     @pytest.mark.asyncio
     async def test_post_failure_logs_connection_lost_once(self, api):
         api._connected = True

spoolbuddy/tests/test_main.py

@@ -26,6 +26,7 @@ def _make_api():
     api = AsyncMock()
     api.heartbeat = AsyncMock(return_value=None)
     api.update_tare = AsyncMock(return_value={"ok": True})
+    api.has_buffered = MagicMock(return_value=False)
     return api
 
 
@@ -450,6 +451,7 @@ class TestNfcPollLoopWrite:
         nfc.current_uid = "04A1B2C3"
         nfc.write_ntag = MagicMock(return_value=(True, "Write successful (OpenTag3D v2)"))
         api = AsyncMock()
+        api.has_buffered = MagicMock(return_value=False)
         config = _make_config(nfc_poll_interval=0.0)
         shared = {
             "nfc": nfc,
@@ -473,3 +475,105 @@ class TestNfcPollLoopWrite:
                 pass
 
         nfc.write_ntag.assert_called_once_with(b"v1", b"v2")
+
+
+class TestNoSecondWrite:
+    """#3076: the backend keeps the write command until it has the result, so a
+    result stuck in the buffer brought the same write back with the next
+    heartbeat, and it ran again on whatever tag was on the reader."""
+
+    async def _write_once(self, buffered: bool) -> dict:
+        from daemon.nfc_reader import NFCState
+
+        nfc = MagicMock()
+        nfc.ok = True
+        nfc.poll = lambda: ("none", None)
+        nfc.state = NFCState.TAG_PRESENT
+        nfc.current_sak = 0x00
+        nfc.current_uid = "04A1B2C3"
+        nfc.write_ntag = MagicMock(return_value=(True, "Write successful (OpenTag3D v1)"))
+        api = AsyncMock()
+        api.has_buffered = MagicMock(return_value=buffered)
+        shared = {
+            "nfc": nfc,
+            "display": MagicMock(),
+            "pending_write": {"spool_id": 7, "ndef_data": b"v1", "ndef_v2": None},
+        }
+
+        async def _inline_to_thread(fn, *args, **kwargs):
+            return fn(*args, **kwargs)
+
+        with patch("daemon.main.asyncio.to_thread", _inline_to_thread):
+            task = asyncio.create_task(nfc_poll_loop(_make_config(nfc_poll_interval=0.0), api, shared))
+            for _ in range(20):
+                await asyncio.sleep(0)
+                if "pending_write" not in shared:
+                    break
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        return shared
+
+    @pytest.mark.asyncio
+    async def test_an_unsent_result_is_remembered(self):
+        shared = await self._write_once(buffered=True)
+        assert shared["write_result_unsent"] == 7
+
+    @pytest.mark.asyncio
+    async def test_a_delivered_result_leaves_nothing_behind(self):
+        shared = await self._write_once(buffered=False)
+        assert "write_result_unsent" not in shared
+
+    async def _heartbeat(self, shared: dict, spool_id: int, still_buffered: bool) -> None:
+        api = _make_api()
+        api.has_buffered = MagicMock(return_value=still_buffered)
+        sent = False
+
+        async def mock_heartbeat(*args, **kwargs):
+            nonlocal sent
+            if sent:
+                return None
+            sent = True
+            return {
+                "pending_command": "write_tag",
+                "pending_write_payload": {"spool_id": spool_id, "ndef_data_hex": "DEADBEEF"},
+            }
+
+        api.heartbeat = mock_heartbeat
+        task = asyncio.create_task(heartbeat_loop(_make_config(), api, time.monotonic(), shared))
+        await asyncio.sleep(0.1)
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    @pytest.mark.asyncio
+    async def test_the_same_write_is_not_armed_while_its_result_is_unsent(self):
+        shared = {"nfc": None, "scale": None, "display": MagicMock(), "write_result_unsent": 7}
+
+        await self._heartbeat(shared, spool_id=7, still_buffered=True)
+
+        assert "pending_write" not in shared
+        assert shared["write_result_unsent"] == 7
+
+    @pytest.mark.asyncio
+    async def test_a_result_sent_during_the_heartbeat_still_skips_its_stale_command(self):
+        # The heartbeat was answered before the buffer went out, so its
+        # command is the old one; the next heartbeat no longer carries it.
+        shared = {"nfc": None, "scale": None, "display": MagicMock(), "write_result_unsent": 7}
+
+        await self._heartbeat(shared, spool_id=7, still_buffered=False)
+
+        assert "pending_write" not in shared
+        assert "write_result_unsent" not in shared
+
+    @pytest.mark.asyncio
+    async def test_a_write_for_another_spool_is_armed(self):
+        shared = {"nfc": None, "scale": None, "display": MagicMock(), "write_result_unsent": 7}
+
+        await self._heartbeat(shared, spool_id=8, still_buffered=True)
+
+        assert shared["pending_write"]["spool_id"] == 8

spoolbuddy/tests/test_nfc_reader.py

@@ -2,7 +2,7 @@
 
 from unittest.mock import MagicMock
 
-from daemon.nfc_reader import NFCReader, NFCState, _extract_tray_uuid, _raw_blocks, choose_ndef
+from daemon.nfc_reader import NFCReader, NFCState, _extract_tray_uuid, _raw_blocks, capacity_refusal, choose_ndef
 
 # Measured on a real spool: block 9 equals the tray_uuid the AMS reports (#984).
 BLOCK9 = bytes.fromhex("9E0B0717BEE94D7887EB1D8DFD1A14F3")
@@ -67,18 +67,44 @@ class TestChooseNdef:
 class TestWriteNtag:
     UID = bytes.fromhex("04A1B2C3D4E5F6")
 
-    def _reader(self, cc: bytes | None = None, read_error: Exception | None = None):
+    NTAG215_CC = bytes([0xE1, 0x10, 0x3E, 0x00])
+
+    def _reader(
+        self,
+        cc: bytes | None = NTAG215_CC,
+        read_error: Exception | None = None,
+        read_back=None,
+    ):
+        """A reader over a fake PN5180 whose tag keeps what is written to it.
+
+        *read_back* replaces what reading the written pages returns: a
+        callable taking the written bytes, or an exception to raise.
+        """
         reader = NFCReader.__new__(NFCReader)
         reader._state = NFCState.TAG_PRESENT
         reader._current_uid = self.UID.hex().upper()
         reader._current_sak = 0x00
         nfc = MagicMock()
         nfc.reactivate_card.return_value = (self.UID, 0x00)
-        if read_error is not None:
-            nfc.ntag_read_pages.side_effect = read_error
-        else:
-            nfc.ntag_read_pages.return_value = cc
-        nfc.ntag_write_pages.return_value = True
+        tag = bytearray()
+
+        def write_pages(start_page, data):
+            tag[:] = data + bytes(-len(data) % 4)
+            return True
+
+        def read_pages(start_page, num_pages):
+            if start_page == 3:
+                if read_error is not None:
+                    raise read_error
+                return cc
+            if isinstance(read_back, Exception):
+                raise read_back
+            if read_back is not None:
+                return read_back(bytes(tag))
+            return bytes(tag[: num_pages * 4])
+
+        nfc.ntag_read_pages.side_effect = read_pages
+        nfc.ntag_write_pages.side_effect = write_pages
         reader._nfc = nfc
         return reader, nfc
 
@@ -92,7 +118,7 @@ class TestWriteNtag:
 
         assert ok
         assert "v2" in msg
-        nfc.ntag_read_pages.assert_called_once_with(start_page=3, num_pages=1)
+        assert nfc.ntag_read_pages.call_args_list[0].kwargs == {"start_page": 3, "num_pages": 1}
         assert self._written(nfc) == TestChooseNdef.V2
         assert nfc.ntag_write_pages.call_args.kwargs["start_page"] == 4
 
@@ -113,11 +139,110 @@ class TestWriteNtag:
         assert ok
         assert self._written(nfc) == TestChooseNdef.V1
 
-    def test_without_v2_the_tag_is_not_read_first(self):
-        reader, nfc = self._reader()
+    def test_without_v2_the_capacity_is_still_checked(self):
+        # #3076: v1 alone is checked against the tag's size too.
+        reader, nfc = self._reader(cc=bytes([0xE1, 0x10, 0x06, 0x00]))  # 48 bytes
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1)
+
+        assert not ok
+        assert msg == "Tag too small: 129 bytes to write, 48 bytes of space"
+        nfc.ntag_write_pages.assert_not_called()
+
+    def test_a_tag_without_an_ndef_cc_is_refused(self):
+        # A blank MIFARE Ultralight reports SAK 0x00 like an NTAG, but has no
+        # NDEF Capability Container (#3076).
+        reader, nfc = self._reader(cc=bytes(4))
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert not ok
+        assert "Not an NDEF-formatted NTAG" in msg
+        nfc.ntag_write_pages.assert_not_called()
+
+    def test_v1_that_fits_an_ntag213_is_written(self):
+        reader, nfc = self._reader(cc=bytes([0xE1, 0x10, 0x12, 0x00]))  # 144 bytes
 
         ok, _ = reader.write_ntag(TestChooseNdef.V1)
 
         assert ok
-        nfc.ntag_read_pages.assert_not_called()
         assert self._written(nfc) == TestChooseNdef.V1
+
+
+class TestWriteNtagReadBack:
+    """#3076: the PN5180 can't see the tag's ACK, so the pages are read back."""
+
+    def test_matching_data_is_a_verified_write(self):
+        reader, nfc = TestWriteNtag()._reader()
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert (ok, msg) == (True, "Write successful (OpenTag3D v2)")
+        # Read from page 4 over the whole payload, after a fresh select.
+        assert nfc.ntag_read_pages.call_args_list[-1].kwargs == {"start_page": 4, "num_pages": 48}
+        assert nfc.reactivate_card.call_count == 3
+
+    def test_other_data_fails_the_write_and_names_the_page(self):
+        def corrupt_page_9(written: bytes) -> bytes:
+            data = bytearray(written)
+            data[20] ^= 0xFF  # page 4 + 20 // 4
+            return bytes(data)
+
+        reader, _ = TestWriteNtag()._reader(read_back=corrupt_page_9)
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert not ok
+        assert msg == "Verification failed at page 9: the tag holds other data"
+
+    def test_no_answer_to_the_read_back_leaves_it_unverified(self):
+        reader, _ = TestWriteNtag()._reader(read_back=lambda written: None)
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert (ok, msg) == (True, "Write successful (OpenTag3D v2, not verified)")
+
+    def test_a_short_answer_leaves_it_unverified(self):
+        reader, _ = TestWriteNtag()._reader(read_back=lambda written: written[:16])
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert (ok, msg) == (True, "Write successful (OpenTag3D v2, not verified)")
+
+    def test_a_read_error_leaves_it_unverified(self):
+        reader, _ = TestWriteNtag()._reader(read_back=RuntimeError("BUSY timeout"))
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert (ok, msg) == (True, "Write successful (OpenTag3D v2, not verified)")
+
+    def test_no_answer_to_the_reselect_leaves_it_unverified(self):
+        reader, nfc = TestWriteNtag()._reader()
+        nfc.reactivate_card.side_effect = [(TestWriteNtag.UID, 0x00), (TestWriteNtag.UID, 0x00), None]
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert (ok, msg) == (True, "Write successful (OpenTag3D v2, not verified)")
+
+    def test_a_failed_write_is_reported_as_such(self):
+        reader, nfc = TestWriteNtag()._reader()
+        nfc.ntag_write_pages.side_effect = None
+        nfc.ntag_write_pages.return_value = False
+
+        ok, msg = reader.write_ntag(TestChooseNdef.V1, TestChooseNdef.V2)
+
+        assert (ok, msg) == (False, "Write failed")
+
+
+class TestCapacityRefusal:
+    def test_an_ntag213_holds_144_bytes(self):
+        cc = bytes([0xE1, 0x10, 0x12, 0x00])
+        assert capacity_refusal(cc, 144) is None
+        assert capacity_refusal(cc, 145) == "Tag too small: 145 bytes to write, 144 bytes of space"
+
+    def test_a_cc_that_is_not_ndef_is_refused(self):
+        assert capacity_refusal(bytes([0x00, 0x00, 0x3E, 0x00]), 10) is not None
+
+    def test_an_unread_cc_proves_nothing(self):
+        assert capacity_refusal(None, 500) is None
+        assert capacity_refusal(b"\xe1", 500) is None