nem: mosaics canonicalization

src/apps/nem/mosaic.py

@@ -1,6 +1,7 @@
 
 from .helpers import *
 from .writers import *
+from trezor.messages.NEMMosaic import NEMMosaic
 
 
 def nem_transaction_create_mosaic_creation(network: int, timestamp: int, signer_public_key: bytes, fee:int,
@@ -100,3 +101,35 @@ def nem_transaction_write_mosaic(w: bytearray, namespace: str, mosaic: str, quan
     write_bytes_with_length(w, bytearray(namespace))
     write_bytes_with_length(w, bytearray(mosaic))
     write_uint64(w, quantity)
+
+
+def nem_canonicalize_mosaics(mosaics: list):
+    if len(mosaics) <= 1:
+        return mosaics
+    mosaics = nem_merge_mosaics(mosaics)
+    return nem_sort_mosaics(mosaics)
+
+
+def are_mosaics_equal(a: NEMMosaic, b: NEMMosaic) -> bool:
+    if a.namespace == b.namespace and a.mosaic == b.mosaic:
+        return True
+    return False
+
+
+def nem_merge_mosaics(mosaics: list) -> list:
+    if not len(mosaics):
+        return list()
+    ret = list()
+    for i in mosaics:
+        found = False
+        for k, y in enumerate(ret):
+            if are_mosaics_equal(i, y):
+                ret[k].quantity += i.quantity
+                found = True
+        if not found:
+            ret.append(i)
+    return ret
+
+
+def nem_sort_mosaics(mosaics: list) -> list:
+    return sorted(mosaics, key=lambda m: (m.namespace, m.mosaic))

src/apps/nem/signing.py

@@ -15,8 +15,8 @@ async def nem_sign_tx(ctx, msg: NEMSignTx):
     node = await seed.derive_node(ctx, msg.transaction.address_n, NEM_CURVE)
 
     if msg.transfer:
+        msg.transfer.mosaics = nem_canonicalize_mosaics(msg.transfer.mosaics)
         tx = await _transfer(ctx, node, msg)
-        # todo msg.transfer.mosaics = canonicalize_mosaics(msg.transfer.mosaics)
     elif msg.provision_namespace:
         tx = await _provision_namespace(ctx, node, msg)
     elif msg.mosaic_creation:

tests/test_apps.nem.transaction.mosaic.canonicalization.py

@@ -0,0 +1,147 @@
+from common import *
+
+from apps.nem.mosaic import *
+
+
+class TestNemTransactionMosaicCanonicalization(unittest.TestCase):
+
+    def test_nem_transaction_mosaic_canonicalization(self):
+        a = NEMMosaic()
+        a.namespace = 'abc'
+        a.quantity = 3
+        a.mosaic = 'mosaic'
+        b = NEMMosaic()
+        b.namespace = 'abc'
+        b.quantity = 4
+        b.mosaic = 'a'
+        c = NEMMosaic()
+        c.namespace = 'zzz'
+        c.quantity = 3
+        c.mosaic = 'mosaic'
+        d = NEMMosaic()
+        d.namespace = 'abc'
+        d.quantity = 8
+        d.mosaic = 'mosaic'
+        e = NEMMosaic()
+        e.namespace = 'aaa'
+        e.quantity = 1
+        e.mosaic = 'mosaic'
+        f = NEMMosaic()
+        f.namespace = 'aaa'
+        f.quantity = 1
+        f.mosaic = 'mosaicz'
+        g = NEMMosaic()
+        g.namespace = 'zzz'
+        g.quantity = 30
+        g.mosaic = 'mosaic'
+
+        res = nem_canonicalize_mosaics([a, b, c, d, e, f, g])
+        self.assertEqual(res, [e, f, b, a, c])
+        self.assertEqual(res[2].quantity, b.quantity)
+        self.assertEqual(res[3].quantity, 3 + 8)  # a + d
+        self.assertEqual(res[4].quantity, 3 + 30)  # c + g
+
+    def test_nem_transaction_mosaic_merge(self):
+        a = NEMMosaic()
+        a.namespace = 'abc'
+        a.quantity = 1
+        a.mosaic = 'mosaic'
+        b = NEMMosaic()
+        b.namespace = 'abc'
+        b.quantity = 1
+        b.mosaic = 'mosaic'
+
+        merged = nem_merge_mosaics([a, b])
+        self.assertEqual(merged[0].quantity, 2)
+        self.assertEqual(len(merged), 1)
+
+        a.quantity = 1
+        b.quantity = 10
+        merged = nem_merge_mosaics([a, b])
+        self.assertEqual(merged[0].quantity, 11)
+
+        a.namespace = 'abcdef'
+        merged = nem_merge_mosaics([a, b])
+        self.assertEqual(len(merged), 2)
+
+        c = NEMMosaic()
+        c.namespace = 'abc'
+        c.mosaic = 'xxx'
+        c.quantity = 2
+        merged = nem_merge_mosaics([a, b, c])
+        self.assertEqual(len(merged), 3)
+
+        a.namespace = 'abcdef'
+        a.quantity = 1
+        a.mosaic = 'mosaic'
+        b.namespace = 'abc'
+        b.quantity = 2
+        b.mosaic = 'mosaic'
+        c.namespace = 'abc'
+        c.mosaic = 'mosaic'
+        c.quantity = 3
+        merged = nem_merge_mosaics([a, b, c])
+        self.assertEqual(merged[0].quantity, 1)
+        self.assertEqual(merged[1].quantity, 5)
+        self.assertEqual(len(merged), 2)
+
+        a.namespace = 'abc'
+        a.quantity = 1
+        a.mosaic = 'mosaic'
+        b.namespace = 'abc'
+        b.quantity = 2
+        b.mosaic = 'mosaic'
+        c.namespace = 'abc'
+        c.mosaic = 'mosaic'
+        c.quantity = 3
+        merged = nem_merge_mosaics([a, b, c])
+        self.assertEqual(merged[0].quantity, 6)
+        self.assertEqual(len(merged), 1)
+
+    def test_nem_transaction_mosaic_sort(self):
+        a = NEMMosaic()
+        a.namespace = 'abcz'
+        a.quantity = 1
+        a.mosaic = 'mosaic'
+        b = NEMMosaic()
+        b.namespace = 'abca'
+        b.quantity = 1
+        b.mosaic = 'mosaic'
+        res = nem_sort_mosaics([a, b])
+        self.assertEqual(res, [b, a])
+
+        a.namespace = ''
+        b.namespace = 'a.b.c'
+        res = nem_sort_mosaics([a, b])
+        self.assertEqual(res, [a, b])
+
+        a.namespace = 'z.z.z'
+        b.namespace = 'a.b.c'
+        res = nem_sort_mosaics([a, b])
+        self.assertEqual(res, [b, a])
+
+        a.namespace = 'a'
+        b.namespace = 'a'
+        a.mosaic = 'mosaic'
+        b.mosaic = 'mosaic'
+        res = nem_sort_mosaics([a, b])
+        self.assertEqual(res, [a, b])
+
+        a.mosaic = 'www'
+        b.mosaic = 'aaa'
+        res = nem_sort_mosaics([a, b])
+        self.assertEqual(res, [b, a])
+
+        c = NEMMosaic()
+        c.namespace = 'a'
+        c.mosaic = 'zzz'
+        res = nem_sort_mosaics([a, b, c])
+        self.assertEqual(res, [b, a, c])
+
+        c.mosaic = 'bbb'
+        res = nem_sort_mosaics([a, b, c])
+        self.assertEqual(res, [b, c, a])
+
+
+if __name__ == '__main__':
+    unittest.main()