Add hex+b64 universal decoder

fetchai/ledger/api/tx.py

@@ -1,8 +1,8 @@
-import base64
 import json
 from typing import Union, List, Dict, Optional
 
 from fetchai.ledger.crypto import Address, Identity
+from fetchai.ledger.decode import decode_hex_or_b64
 from .common import ApiEndpoint
 
 AddressLike = Union[Address, Identity, bytes, str]
@@ -122,7 +122,7 @@ def _status(self, tx_digest):
         response = self._session.get(url).json()
 
         return TxStatus(
-            digest=base64.b64decode(response['tx'].encode()),
+            digest=decode_hex_or_b64(response['tx']),
             status=str(response['status']),
             exit_code=int(response['exit_code']),
             charge=int(response['charge']),

fetchai/ledger/decode.py

@@ -0,0 +1,24 @@
+import base64
+
+
+def decode_hex_or_b64(encoded) -> bytes:
+    hex_prefix = '0x'
+    b64_padding = '='
+
+    if isinstance(encoded, str):
+        encoded_str = encoded
+    elif isinstance(encoded, bytes):
+        encoded_str = encoded.decode()
+    else:
+        raise TypeError('Expected argument to be bytes or str')
+
+    try:
+        if encoded_str.startswith(hex_prefix) and not encoded_str.endswith(b64_padding) and len(encoded_str) > 2:
+            return bytes.fromhex(encoded_str[2:])
+        else:
+            return bytes.fromhex(encoded_str)
+    except ValueError:
+        # hex decoding did not work - attempt base64
+        padding = b64_padding * (len(encoded_str) % 4)
+
+        return base64.b64decode(encoded_str + padding)

tests/test_decode.py

@@ -0,0 +1,29 @@
+import unittest
+
+from fetchai.ledger.decode import decode_hex_or_b64
+
+
+class UniversalDecodeTests(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual(decode_hex_or_b64(''), b'')
+        self.assertEqual(decode_hex_or_b64(b''), b'')
+
+    def test_unprefixed_hex(self):
+        self.assertEqual(decode_hex_or_b64('deadbeef'), b'\xde\xad\xbe\xef')
+        self.assertEqual(decode_hex_or_b64(b'deadbeef'), b'\xde\xad\xbe\xef')
+
+    def test_prefixed_hex(self):
+        self.assertEqual(decode_hex_or_b64('0xdeadbeef'), b'\xde\xad\xbe\xef')
+        self.assertEqual(decode_hex_or_b64(b'0xdeadbeef'), b'\xde\xad\xbe\xef')
+
+    def test_padded_base64(self):
+        self.assertEqual(decode_hex_or_b64('YWJjZDEyMzQ='), b'abcd1234')
+        self.assertEqual(decode_hex_or_b64(b'YWJjZDEyMzQ='), b'abcd1234')
+
+    def test_unpadded_base64(self):
+        self.assertEqual(decode_hex_or_b64('YWJjZDEyMzQ'), b'abcd1234')
+        self.assertEqual(decode_hex_or_b64(b'YWJjZDEyMzQ'), b'abcd1234')
+
+    def test_0x(self):
+        self.assertEqual(decode_hex_or_b64('0x'), b'\xd3')
+        self.assertEqual(decode_hex_or_b64(b'0x'), b'\xd3')