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

CHANGELOG.md

@@ -21,6 +21,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **"Any model" jobs go to the printer with the fewest print hours (#3480, sponsor vote Q4 2026)** — When several printers of a model can take a job, the queue used to pick the first one in its list, so on a farm the same printer got most of the work. It now picks the one with the fewest print hours (the Maintenance counter, including baseline hours entered there), so jobs and wear spread across the farm. Preferred colours still come first, and the queue powers on the least used printer too when it has to switch one on with an Auto Power On plug.
 
 ### Fixed
+- **SpoolBuddy checks a tag write instead of always reporting success (#3076, reported by @jinjiangz, gaps found by @pleasantone)** — The PN5180 can't see a tag's answer to a write, so **Write Tag** reported success for every page it sent. SpoolBuddy now reads the written data back and fails the write if the tag holds something else, naming the page. If the tag doesn't answer the read-back, the write still counts, reported as not verified. A tag too small for the data is refused before anything is written, and so is a tag that isn't NDEF-formatted, such as a blank MIFARE Ultralight, which looks like an NTAG to the reader. Before, a write past its end was reported as successful. When the write's result couldn't reach Bambuddy straight away, the next heartbeat brought the same write back and SpoolBuddy ran it a second time, on whatever tag was then on the reader. It now waits until the result is through.
 - **HMS notifications are no longer sent again after a restart (#3502, reported by @pleasantone)** — The fix for #3486 remembers which printer errors were already notified, in a table that was never created: startup did not load its model, so every start logged "no such table: hms_notified_faults" and the errors were sent again. The table is now created on the next start.
 - **A backup from Windows restored on Linux or macOS finds its files (#3501, reported by @Tech0verlord)** — A Windows install stores the location of each archive, library file and print log entry with backslashes, which Linux and macOS read as part of the file name. After restoring a Windows backup on Docker or another Linux install, every archive showed no thumbnail and refused its 3D preview and reprint with "Archive data unavailable". The restore now turns those backslashes into slashes. It runs only during a restore, does nothing on Windows, and leaves a path alone when a file with that exact name exists. An install already restored from a Windows backup is fixed by restoring the backup again after updating.
 - **Tags written by SpoolBuddy can be read by phones and other readers (#3500, reported by @pleasantone; likely #3076)** — SpoolBuddy's **Write Tag** wrote a copy of the tag's Capability Container in front of the OpenTag3D data. On an NTAG tag that copy landed in the data area, where a phone, NFC Tools or another OpenTag3D reader took it for unknown data, skipped over the spool record and reported an empty tag, although the write had succeeded. Bambuddy itself was unaffected, since it links spools by the tag's ID. The data now starts with the spool record, and the tag's own Capability Container is left as it is. Tags written before need writing again for other readers to see them.

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