[DOP-17107] dump connection auth data

docs/changelog/next_release/63.bugfix.rst

@@ -0,0 +1 @@
+When starting the transfer, the connections' credentials were not dumped

syncmaster/backend/api/v1/transfers.py

@@ -411,8 +411,19 @@ async def start_run(
 
     transfer = await unit_of_work.transfer.read_by_id(transfer_id=create_run_data.transfer_id)
 
+    credentials_source = await unit_of_work.credentials.read(
+        transfer.source_connection_id,
+    )
+    credentials_target = await unit_of_work.credentials.read(
+        transfer.target_connection_id,
+    )
+
     async with unit_of_work:
-        run = await unit_of_work.run.create(transfer_id=create_run_data.transfer_id)
+        run = await unit_of_work.run.create(
+            transfer_id=create_run_data.transfer_id,
+            source_creds=credentials_source,
+            target_creds=credentials_target,
+        )
     try:
         celery.send_task("run_transfer_task", kwargs={"run_id": run.id}, queue=transfer.queue.name)
     except KombuError as e:

syncmaster/db/repositories/run.py

@@ -13,6 +13,7 @@
 from syncmaster.exceptions import SyncmasterError
 from syncmaster.exceptions.run import CannotStopRunError, RunNotFoundError
 from syncmaster.exceptions.transfer import TransferNotFoundError
+from syncmaster.schemas.v1.connections.connection import ReadConnectionSchema
 from syncmaster.schemas.v1.transfers import ReadFullTransferSchema
 
 
@@ -35,10 +36,15 @@ async def read_by_id(self, run_id: int) -> Run:
             raise RunNotFoundError
         return run
 
-    async def create(self, transfer_id: int) -> Run:
+    async def create(
+        self,
+        transfer_id: int,
+        source_creds: dict,
+        target_creds: dict,
+    ) -> Run:
         run = Run()
         run.transfer_id = transfer_id
-        run.transfer_dump = await self.read_full_serialized_transfer(transfer_id)
+        run.transfer_dump = await self.read_full_serialized_transfer(transfer_id, source_creds, target_creds)
         try:
             self._session.add(run)
             await self._session.flush()
@@ -60,7 +66,12 @@ async def stop(self, run_id: int) -> Run:
         await self._session.flush()
         return run
 
-    async def read_full_serialized_transfer(self, transfer_id: int) -> dict[str, Any]:
+    async def read_full_serialized_transfer(
+        self,
+        transfer_id: int,
+        source_creds: dict,
+        target_creds: dict,
+    ) -> dict[str, Any]:
         transfer = await self._session.scalars(
             select(Transfer)
             .where(Transfer.id == transfer_id)
@@ -69,7 +80,38 @@ async def read_full_serialized_transfer(self, transfer_id: int) -> dict[str, Any
                 selectinload(Transfer.target_connection),
             ),
         )
-        return ReadFullTransferSchema.from_orm(transfer.one()).dict()
+        transfer = transfer.one()
+
+        return ReadFullTransferSchema(
+            id=transfer.id,
+            name=transfer.name,
+            group_id=transfer.group_id,
+            queue_id=transfer.queue_id,
+            source_connection_id=transfer.source_connection_id,
+            target_connection_id=transfer.target_connection_id,
+            is_scheduled=transfer.is_scheduled,
+            description=transfer.description,
+            schedule=transfer.schedule,
+            source_params=transfer.source_params,
+            target_params=transfer.target_params,
+            strategy_params=transfer.strategy_params,
+            source_connection=ReadConnectionSchema(
+                id=transfer.source_connection.id,
+                group_id=transfer.source_connection.group_id,
+                name=transfer.source_connection.name,
+                description=transfer.source_connection.description,
+                data=transfer.source_connection.data,
+                auth_data=source_creds,
+            ),
+            target_connection=ReadConnectionSchema(
+                id=transfer.target_connection.id,
+                group_id=transfer.target_connection.group_id,
+                name=transfer.target_connection.name,
+                description=transfer.target_connection.description,
+                data=transfer.target_connection.data,
+                auth_data=target_creds,
+            ),
+        ).dict()
 
     def _raise_error(self, e: DBAPIError) -> NoReturn:
         constraint = e.__cause__.__cause__.constraint_name

tests/test_integration/test_run_transfer/test_hdfs.py

@@ -151,6 +151,8 @@ async def test_run_transfer_hdfs_to_postgres(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=postgres,
@@ -211,6 +213,8 @@ async def test_run_transfer_postgres_to_hdfs(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = FileDFReader(
         connection=hdfs_file_df_connection,

tests/test_integration/test_run_transfer/test_hve.py

@@ -101,6 +101,8 @@ async def test_run_transfer_postgres_to_hive(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=hive,
@@ -141,6 +143,8 @@ async def test_run_transfer_postgres_to_hive_mixed_naming(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
     reader = DBReader(
         connection=hive,
         table="default.target_table",
@@ -184,6 +188,8 @@ async def test_run_transfer_hive_to_postgres(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
     reader = DBReader(
         connection=postgres,
         table="public.target_table",
@@ -223,6 +229,8 @@ async def test_run_transfer_hive_to_postgres_mixes_naming(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=postgres,

tests/test_integration/test_run_transfer/test_oracle.py

@@ -104,6 +104,8 @@ async def test_run_transfer_postgres_to_oracle(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
     reader = DBReader(
         connection=oracle,
         table=f"{oracle.user}.target_table",
@@ -143,6 +145,8 @@ async def test_run_transfer_postgres_to_oracle_mixed_naming(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=oracle,
@@ -186,6 +190,8 @@ async def test_run_transfer_oracle_to_postgres(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=postgres,
@@ -227,6 +233,8 @@ async def test_run_transfer_oracle_to_postgres_mixed_naming(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=postgres,

tests/test_integration/test_run_transfer/test_s3.py

@@ -151,6 +151,8 @@ async def test_run_transfer_s3_to_postgres(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = DBReader(
         connection=postgres,
@@ -210,6 +212,8 @@ async def test_run_transfer_postgres_to_s3(
         token=group_owner.token,
     )
     assert run_data["status"] == Status.FINISHED.value
+    assert run_data["transfer_dump"]["source_connection"]["auth_data"]
+    assert run_data["transfer_dump"]["target_connection"]["auth_data"]
 
     reader = FileDFReader(
         connection=s3_file_df_connection,