data duplicator test

integration-tests/__tests__/data-duplicator.spec.js

@@ -1,33 +1,54 @@
 const engines = require('../engines');
+const stream = require('stream');
 const { testWrapper } = require('../tools');
 const dataDuplicator = require('dbgate-api/src/shell/dataDuplicator');
-const fakeObjectReader = require('dbgate-api/src/shell/fakeObjectReader');
+const { runCommandOnDriver } = require('dbgate-tools');
-
-const t1Sql = 'CREATE TABLE t1 (id int not null primary key, val varchar(50) null)';
-const t2Sql =
-  'CREATE TABLE t2 (id int not null primary key, val varchar(50) null, valfk int, foreign key (valfk) references t2(id))';
 
 describe('Data duplicator', () => {
   test.each(engines.map(engine => [engine.label, engine]))(
     'Insert simple data - %s',
     testWrapper(async (conn, driver, engine) => {
-      await driver.query(conn, t1Sql);
+      runCommandOnDriver(conn, driver, dmp =>
-      await driver.query(conn, t2Sql);
+        dmp.createTable({
+          pureName: 't1',
+          columns: [
+            { columnName: 'id', dataType: 'int', autoIncrement: true, notNull: true },
+            { columnName: 'val', dataType: 'varchar(50)' },
+          ],
+          primaryKey: {
+            columns: [{ columnName: 'id' }],
+          },
+        })
+      );
+      runCommandOnDriver(conn, driver, dmp =>
+        dmp.createTable({
+          pureName: 't2',
+          columns: [
+            { columnName: 'id', dataType: 'int', autoIncrement: true, notNull: true },
+            { columnName: 'val', dataType: 'varchar(50)' },
+            { columnName: 'valfk', dataType: 'int', notNull: true },
+          ],
+          primaryKey: {
+            columns: [{ columnName: 'id' }],
+          },
+          foreignKeys: [{ refTableName: 't1', columns: [{ columnName: 'valfk', refColumnName: 'id' }] }],
+        })
+      );
 
-      const t1 = await fakeObjectReader({
+      const gett1 = () =>
-        dynamicData: [
+        stream.Readable.from([
+          { __isStreamHeader: true, __isDynamicStructure: true },
           { id: 1, val: 'v1' },
           { id: 2, val: 'v2' },
           { id: 3, val: 'v3' },
-        ],
+        ]);
-      });
+      const gett2 = () =>
-      const t2 = await fakeObjectReader({
+        stream.Readable.from([
-        dynamicData: [
+          { __isStreamHeader: true, __isDynamicStructure: true },
           { id: 1, val: 'v1', valfk: 1 },
           { id: 2, val: 'v2', valfk: 2 },
           { id: 3, val: 'v3', valfk: 3 },
-        ],
+        ]);
-      });
 
       await dataDuplicator({
         systemConnection: conn,
@@ -36,12 +57,12 @@ describe('Data duplicator', () => {
           {
             name: 't1',
             operation: 'copy',
-            openStream: () => t1,
+            openStream: gett1,
           },
           {
             name: 't2',
             operation: 'copy',
-            openStream: () => t2,
+            openStream: gett2,
           },
         ],
       });
@@ -53,12 +74,12 @@ describe('Data duplicator', () => {
           {
             name: 't1',
             operation: 'copy',
-            openStream: () => t1,
+            openStream: gett1,
           },
           {
             name: 't2',
             operation: 'copy',
-            openStream: () => t2,
+            openStream: gett2,
           },
         ],
       });

packages/api/src/shell/dataDuplicator.js

@@ -35,7 +35,9 @@ async function dataDuplicator({
       name: item.name,
       operation: item.operation,
       matchColumns: item.matchColumns,
-      openStream: () => jsonLinesReader({ fileName: path.join(resolveArchiveFolder(archive), `${item.name}.jsonl`) }),
+      openStream:
+        item.openStream ||
+        (() => jsonLinesReader({ fileName: path.join(resolveArchiveFolder(archive), `${item.name}.jsonl`) })),
     })),
     stream,
     copyStream,

packages/datalib/src/DataDuplicator.ts

@@ -244,6 +244,7 @@ export class DataDuplicator {
     } catch (err) {
       logger.error({ err }, 'Failed duplicator job, rollbacking');
       await runCommandOnDriver(this.pool, this.driver, dmp => dmp.rollbackTransaction());
+      throw err;
     }
     if (this.options.rollbackAfterFinish) {
       logger.info('Rollbacking transaction, nothing was changed');