Migrate payment to taskpayment (#4297)

* Migrate payment to taskpayment

* Update migration numbering

golem/database/database.py

@@ -55,7 +55,7 @@ def execute_sql(self, sql, params=None, require_commit=True):
 
 
 class Database:
-    SCHEMA_VERSION = 30
+    SCHEMA_VERSION = 31
 
     def __init__(self,  # noqa pylint: disable=too-many-arguments
                  db: peewee.Database,

golem/database/schemas/031_migrate_payment_to_task_payment.py

@@ -0,0 +1,74 @@
+# pylint: disable=no-member,unused-argument
+import json
+import logging
+
+SCHEMA_VERSION = 31
+
+logger = logging.getLogger('golem.database')
+
+
+STATUS_MAPPING = {
+    1: 'awaiting',
+    2: 'sent',
+    3: 'confirmed',
+    4: 'overdue',
+}
+
+
+def migrate_payment(database, db_row):
+    details = json.loads(db_row['details'])
+    status = STATUS_MAPPING[db_row['status']]
+    cursor = database.execute_sql(
+        "INSERT INTO walletoperation"
+        " (tx_hash, direction, operation_type, status, sender_address,"
+        "  recipient_address, amount, currency, gas_cost,"
+        "  created_date, modified_date)"
+        " VALUES (?, 'outgoing', 'task_payment', ?, '', ?, ?, 'GNT', ?,"
+        "        ?, datetime('now'))",
+        (
+            f"0x{details['tx']}",
+            status,
+            f'0x{db_row["payee"]}',
+            db_row['value'],
+            details['fee'],
+            db_row['created_date'],
+        ),
+    )
+    wallet_operation_id = cursor.lastrowid
+    cursor.execute(
+        "INSERT INTO taskpayment"
+        " (wallet_operation_id, node, task, subtask,"
+        "  expected_amount, created_date, modified_date)"
+        " VALUES (?, ?, '', ?, ?, ?, datetime('now'))",
+        (
+            wallet_operation_id,
+            details['node_info']['key'],
+            db_row['subtask'],
+            db_row['value'],
+            db_row['created_date'],
+        ),
+    )
+
+
+def migrate(migrator, database, fake=False, **kwargs):
+    cursor = database.execute_sql(
+        'SELECT details, status, payee, value, subtask, created_date'
+        ' FROM payment'
+    )
+    for db_row in cursor.fetchall():
+        dict_row = {
+            'details': db_row[0],
+            'status': db_row[1],
+            'payee': db_row[2],
+            'value': db_row[3],
+            'subtask': db_row[4],
+            'created_date': db_row[5],
+        }
+        try:
+            migrate_payment(database, dict_row)
+        except Exception:  # pylint: disable=broad-except
+            logger.error("Migration problem. db_row=%s", db_row, exc_info=True)
+
+
+def rollback(migrator, database, fake=False, **kwargs):
+    pass

tests/golem/database/test_migration.py

@@ -3,6 +3,7 @@
 from contextlib import contextmanager
 from unittest import TestCase
 from unittest.mock import patch
+import uuid
 
 import os
 
@@ -312,22 +313,32 @@ def test_same_version(self):
     def test_18(self, _create_tables_mock):
         with self.database_context() as database:
             database._migrate_schema(6, 17)
+            sender_node = 'adbeef' + 'deadbeef' * 15
+            subtask_id = str(uuid.uuid4())
             database.db.execute_sql(
                 "INSERT INTO income ("
                 "sender_node, subtask, value, created_date, modified_date,"
                 " overdue"
                 ")"
-                " VALUES ('0xdead', '0xdead', 10, datetime('now'),"
-                "         datetime('now'), 0)"
+                f" VALUES (?, ?, 10, datetime('now'),"
+                "           datetime('now'), 0)",
+                (
+                    sender_node,
+                    subtask_id,
+                ),
             )
             database._migrate_schema(17, 18)
             cursor = database.db.execute_sql(
                 "SELECT payer_address FROM income"
-                " WHERE sender_node = '0xdead' AND subtask = '0xdead'"
-                " LIMIT 1"
+                f" WHERE sender_node = ? AND subtask = ?"
+                " LIMIT 1",
+                (
+                    sender_node,
+                    subtask_id,
+                ),
             )
             value = cursor.fetchone()[0]
-            self.assertEqual(value, '0eeA941c1244ADC31F53525D0eC1397ff6951C9C')
+            self.assertEqual(value, 'c106A6f2534E74b9D5890d13C5991A3fB146Ae52')
 
     @patch('golem.database.Database._create_tables')
     def test_20_income_value_received(self, _create_tables_mock):
@@ -351,6 +362,36 @@ def test_20_income_value_received(self, _create_tables_mock):
             value = cursor.fetchone()[0]
             self.assertEqual(value, '10')
 
+    @patch('golem.database.Database._create_tables')
+    def test_31_payments_migration(self, *_args):
+        with self.database_context() as database:
+            database._migrate_schema(6, 30)
+
+            details = '{"node_info": {"node_name": "Laughing Octopus", "key": "392e54805752937326aa87da97a69c9271f7b4423382fb2563a349d54c44d9a904f38b4f2e3a022572c8257220426d8e5e34198be2cc8971bc149f1a368161e3", "prv_port": 40201, "pub_port": 40201, "p2p_prv_port": 40200, "p2p_pub_port": 40200, "prv_addr": "10.30.8.12", "pub_addr": "194.181.80.91", "prv_addresses": ["10.30.8.12", "172.17.0.1"], "hyperdrive_prv_port": 3282, "hyperdrive_pub_port": 3282, "port_statuses": {"3282": "timeout", "40200": "timeout", "40201": "timeout"}, "nat_type": "Symmetric NAT"}, "fee": 116264444444444, "block_hash": "184575de00b91fdac0ccd1c763d5b56b967898e3a541f400480b01a6dbf1fef9", "block_number": 1937551, "check": null, "tx": "4b9f628f16c82d0fe3f3ab144feef7940a0093107d521b45a8a0bfb5739400be"}'  # noqa pylint: disable=line-too-long
+            database.db.execute_sql(
+                "INSERT INTO payment ("
+                "    subtask, created_date, modified_date, status,"
+                "    payee, value, details)"
+                " VALUES ('0xdead', datetime('now'), datetime('now'), 1,"
+                "         '0x0eeA941c1244ADC31F53525D0eC1397ff6951C9C', 10,"
+                f"        '{details}')"
+            )
+            database._migrate_schema(30, 31)
+
+            # UNIONS don't work here. Do it manually
+            cursor = database.db.execute_sql("SELECT count(*) FROM payment")
+            payment_count = cursor.fetchone()[0]
+            cursor = database.db.execute_sql(
+                "SELECT count(*) FROM walletoperation",
+            )
+            wo_count = cursor.fetchone()[0]
+            cursor = database.db.execute_sql("SELECT count(*) FROM taskpayment")
+            tp_count = cursor.fetchone()[0]
+            # Migrated payments shouldn't be removed
+            self.assertEqual(payment_count, 1)
+            self.assertEqual(wo_count, 1)
+            self.assertEqual(tp_count, 1)
+
 
 def generate(start, stop):
     return ['{:03}_script'.format(i) for i in range(start, stop + 1)]