Merge pull request #86 from nfcpy/fix_issue_84

Raise TagCommandError when NDEF write fails.

HISTORY.rst

@@ -1,6 +1,13 @@
 Changelog for nfcpy
 ===================
 
+0.13.4 (2017-11-10)
+-------------------
+
+* Raise nfc.tag.TagCommandError when NDEF data could not be written to
+  the tag. Previously this was captured within the tag memory cache
+  for Type1Tag and Type2Tag and raised as IndexError.
+
 0.13.3 (2017-11-02)
 -------------------
 

src/nfc/__init__.py

@@ -32,7 +32,7 @@
 
 # METADATA ####################################################################
 
-__version__ = "0.13.3"
+__version__ = "0.13.4"
 
 __title__ = "nfcpy"
 __description__ = "Python module for Near Field Communication."

src/nfc/tag/__init__.py

@@ -182,11 +182,14 @@ def records(self):
             from the NDEF message data or set the message data from a
             list of records. ::
 
+                from ndef import TextRecord
                 if tag.ndef is not None:
                     for record in tag.ndef.records:
                         print(record)
-                    import ndef
-                    tag.ndef.records = [ndef.TextRecord('Hello World')]
+                    try:
+                        tag.ndef.records = [TextRecord('Hello World')]
+                    except nfc.tag.TagCommandError as err:
+                        print("NDEF write failed: " + str(err))
 
             Decoding is performed with a relaxed error handling
             strategy that ignores minor errors in the NDEF data. The

src/nfc/tag/tt1.py

@@ -60,7 +60,7 @@ def read_tlv(memory, offset, skip_bytes):
     # length (big endian).
     try:
         tlv_t, offset = (memory[offset], offset+1)
-    except IndexError:
+    except Type1TagCommandError:
         return (None, None, None)
 
     if tlv_t in (0x00, 0xFE):
@@ -144,9 +144,9 @@ def _read_ndef_data(self):
             # Otherwise, set state variables and return the ndef
             # message data as a bytearray (may be zero length).
             log.debug("read ndef data")
-            tag_memory = Type1TagMemoryReader(self.tag)
-
             try:
+                tag_memory = Type1TagMemoryReader(self.tag)
+
                 if tag_memory._header_rom[0] >> 4 != 1:
                     log.debug("proprietary type 1 tag memory structure")
                     return None
@@ -164,7 +164,7 @@ def _read_ndef_data(self):
 
                 tag_memory_size = (tag_memory[10] + 1) * 8
                 log.debug("tag memory size is %d byte" % tag_memory_size)
-            except IndexError:
+            except Type1TagCommandError:
                 log.debug("header rom and static memory were unreadable")
                 return None
 
@@ -511,48 +511,35 @@ def __delitem__(self, key):
 
     def _read_from_tag(self, stop):
         if len(self) < 120:
-            try:
-                read_all_data_response = self._tag.read_all()
-            except Type1TagCommandError:
-                return
-            else:
-                self._header_rom = read_all_data_response[0:2]
-                self._data_from_tag[0:] = read_all_data_response[2:]
-                self._data_in_cache[0:] = self._data_from_tag[0:]
+            read_all_data_response = self._tag.read_all()
+            self._header_rom = read_all_data_response[0:2]
+            self._data_from_tag[0:] = read_all_data_response[2:]
+            self._data_in_cache[0:] = self._data_from_tag[0:]
+
         if stop > 120 and len(self) < 128:
-            try:
-                read_block_response = self._tag.read_block(15)
-            except Type1TagCommandError:
-                return
-            else:
-                self._data_from_tag[120:128] = read_block_response
-                self._data_in_cache[120:128] = read_block_response
+            read_block_response = self._tag.read_block(15)
+            self._data_from_tag[120:128] = read_block_response
+            self._data_in_cache[120:128] = read_block_response
+
         while len(self) < stop:
-            try:
-                data = self._tag.read_segment(len(self) >> 7)
-            except Type1TagCommandError:
-                return
-            else:
-                self._data_from_tag.extend(data)
-                self._data_in_cache.extend(data)
+            data = self._tag.read_segment(len(self) >> 7)
+            self._data_from_tag.extend(data)
+            self._data_in_cache.extend(data)
 
     def _write_to_tag(self, stop):
-        try:
-            hr0 = self._header_rom[0]
-            if hr0 >> 4 == 1 and hr0 & 0x0F != 1:
-                for i in xrange(0, stop, 8):
-                    data = self._data_in_cache[i:i+8]
-                    if data != self._data_from_tag[i:i+8]:
-                        self._tag.write_block(i//8, data)
-                        self._data_from_tag[i:i+8] = data
-            else:
-                for i in xrange(0, stop):
-                    data = self._data_in_cache[i]
-                    if data != self._data_from_tag[i]:
-                        self._tag.write_byte(i, data)
-                        self._data_from_tag[i] = data
-        except Type1TagCommandError as error:
-            log.error(str(error))
+        hr0 = self._header_rom[0]
+        if hr0 >> 4 == 1 and hr0 & 0x0F != 1:
+            for i in xrange(0, stop, 8):
+                data = self._data_in_cache[i:i+8]
+                if data != self._data_from_tag[i:i+8]:
+                    self._tag.write_block(i//8, data)
+                    self._data_from_tag[i:i+8] = data
+        else:
+            for i in xrange(0, stop):
+                data = self._data_in_cache[i]
+                if data != self._data_from_tag[i]:
+                    self._tag.write_byte(i, data)
+                    self._data_from_tag[i] = data
 
     def synchronize(self):
         """Write pages that contain modified data back to tag memory."""

src/nfc/tag/tt2.py

@@ -153,7 +153,7 @@ def _read_capability_data(self, tag_memory):
                 self._readable = bool(tag_memory[15] >> 4 == 0)
                 self._writeable = bool(tag_memory[15] & 0xF == 0)
                 return True
-            except IndexError:
+            except Type2TagCommandError:
                 log.debug("first four memory pages were unreadable")
                 return False
 
@@ -178,7 +178,7 @@ def _read_ndef_data(self):
                 try:
                     tlv = read_tlv(tag_memory, offset, skip_bytes)
                     tlv_t, tlv_l, tlv_v = tlv
-                except IndexError:
+                except Type2TagCommandError:
                     return None
                 else:
                     logmsg = "tlv type {0} length {1} at offset {2}"
@@ -640,7 +640,7 @@ def __getitem__(self, key):
     def __setitem__(self, key, value):
         self.__getitem__(key)
         if isinstance(key, slice):
-            if len(value) != len(xrange(*key.indices(0x100000))):
+            if len(value) != len(range(*key.indices(0x100000))):
                 msg = "{cls} requires item assignment of identical length"
                 raise ValueError(msg.format(cls=self.__class__.__name__))
         self._data_in_cache[key] = value
@@ -651,25 +651,23 @@ def __delitem__(self, key):
         raise TypeError(msg.format(cls=self.__class__.__name__))
 
     def _read_from_tag(self, stop):
-        start = len(self)
-        try:
-            for i in xrange((start >> 4) << 4, stop, 16):
-                self._tag.sector_select(i >> 10)
-                self._data_from_tag[i:i+16] = self._tag.read(i >> 2)
-                self._data_in_cache[i:i+16] = self._data_from_tag[i:i+16]
-        except Type2TagCommandError:
-            pass
+        index = (len(self) >> 4) << 4
+        while index < stop:
+            self._tag.sector_select(index >> 10)
+            data = self._tag.read(index >> 2)
+            self._data_from_tag[index:] = data
+            self._data_in_cache[index:] = data
+            index += 16
 
     def _write_to_tag(self, stop):
-        try:
-            for i in xrange(0, stop, 4):
-                data = self._data_in_cache[i:i+4]
-                if data != self._data_from_tag[i:i+4]:
-                    self._tag.sector_select(i >> 10)
-                    self._tag.write(i >> 2, data)
-                    self._data_from_tag[i:i+4] = data
-        except Type2TagCommandError:
-            pass
+        index = 0
+        while index < stop:
+            data = self._data_in_cache[index:index+4]
+            if data != self._data_from_tag[index:index+4]:
+                self._tag.sector_select(index >> 10)
+                self._tag.write(index >> 2, data)
+                self._data_from_tag[index:index+4] = data
+            index += 4
 
     def synchronize(self):
         """Write pages that contain modified data back to tag memory."""

tests/test_tag_tt1.py

@@ -269,7 +269,7 @@ def test_read_from_dynamic_memory(self, tag, ndef_octets):
         assert tag.ndef.length == 461
         assert tag.ndef.octets == ndef_octets
 
-    def test_read_null_tlv_until_key_error(self, tag, mmap):
+    def test_read_null_tlv_until_read_error(self, tag, mmap):
         tag.clf.exchange.side_effect = [
             tag.target.rid_res[:2] + mmap[:12] + bytearray(108),  # RALL
             HEX("0F") + bytearray(8),  # READ8(15)
@@ -795,14 +795,15 @@ def test_slice_assign_with_different_length(self, tag):
         assert str(excinfo.value) == \
             "Type1TagMemoryReader requires item assignment of identical length"
 
-    @pytest.mark.parametrize("offset", [0, 120, 128])
+    @pytest.mark.parametrize("offset", [0, 1, 120, 121, 128])
     def test_read_mute_tag_at_offset(self, tag, offset):
         tag.clf.exchange.side_effect \
             = nfc.tag.tt1.Type1TagCommandError(nfc.tag.TIMEOUT_ERROR)
-        tag_memory = nfc.tag.tt1.Type1TagMemoryReader(tag)
-        tag_memory._data_from_tag = self.mmap[:offset]
-        with pytest.raises(IndexError):
+        with pytest.raises(nfc.tag.TagCommandError) as excinfo:
+            tag_memory = nfc.tag.tt1.Type1TagMemoryReader(tag)
+            tag_memory._data_from_tag = self.mmap[:offset]
             tag_memory[offset]
+        assert str(excinfo.value) == "unrecoverable timeout error"
 
     def test_write_raises_command_error(self, tag):
         tag.clf.exchange.side_effect = [
@@ -813,7 +814,9 @@ def test_write_raises_command_error(self, tag):
         ]
         tag_memory = nfc.tag.tt1.Type1TagMemoryReader(tag)
         tag_memory[128] = 0x5A
-        tag_memory.synchronize()
+        with pytest.raises(nfc.tag.TagCommandError) as excinfo:
+            tag_memory.synchronize()
+        assert str(excinfo.value) == "invalid response data"
         tag.clf.exchange.assert_has_calls([
             mock.call(HEX('00 00 00 01020304'), 0.1),
             mock.call(HEX('02 0f 00000000 00000000 01020304'), 0.1),

tests/test_tag_tt2.py

@@ -819,8 +819,9 @@ def test_read_error(self, tag):
         ]
         tag.clf.exchange.side_effect = responses
         tag_memory = nfc.tag.tt2.Type2TagMemoryReader(tag)
-        with pytest.raises(IndexError):
+        with pytest.raises(nfc.tag.TagCommandError) as excinfo:
             tag_memory[16] = 0xfe
+        assert str(excinfo.value) == "unrecoverable timeout error"
         assert tag.clf.exchange.mock_calls == [mock.call(*_) for _ in commands]
 
     def test_write_error(self, tag):
@@ -841,5 +842,7 @@ def test_write_error(self, tag):
         tag.clf.exchange.side_effect = responses
         tag_memory = nfc.tag.tt2.Type2TagMemoryReader(tag)
         tag_memory[16] = 0xfe
-        tag_memory.synchronize()
+        with pytest.raises(nfc.tag.TagCommandError) as excinfo:
+            tag_memory.synchronize()
+        assert str(excinfo.value) == "unrecoverable timeout error"
         assert tag.clf.exchange.mock_calls == [mock.call(*_) for _ in commands]