get_path_to_file_with_original_name

securedrop_client/export.py

@@ -83,6 +83,7 @@ class Export(QObject):
     begin_print = pyqtSignal(list)
     print_call_failure = pyqtSignal(object)
     print_call_success = pyqtSignal()
+    export_completed = pyqtSignal(list)
 
     def __init__(self) -> None:
         super().__init__()
@@ -282,10 +283,7 @@ def send_file_to_usb_device(self, filepaths: List[str], passphrase: str) -> None
                 logger.error(e)
                 self.export_usb_call_failure.emit(e)
 
-        # Export is finished, now remove files created when export began
-        for filepath in filepaths:
-            if os.path.exists(filepath):
-                os.remove(filepath)
+        self.export_completed.emit(filepaths)
 
     @pyqtSlot(list)
     def print(self, filepaths: List[str]) -> None:
@@ -306,7 +304,4 @@ def print(self, filepaths: List[str]) -> None:
                 logger.error(e)
                 self.print_call_failure.emit(e)
 
-        # Print is finished, now remove files created when print began
-        for filepath in filepaths:
-            if os.path.exists(filepath):
-                os.remove(filepath)
+        self.export_completed.emit(filepaths)

securedrop_client/logic.py

@@ -181,6 +181,7 @@ def __init__(self, hostname: str, gui, session_maker: sessionmaker,
         self.gpg = GpgHelper(home, self.session_maker, proxy)
 
         self.export = Export()
+        self.export.export_completed.connect(self.cleanup_hardlinked_file)
 
         self.sync_flag = os.path.join(home, 'sync_flag')
 
@@ -588,16 +589,40 @@ def downloaded_file_exists(self, file_uuid: str) -> bool:
             return False
         return True
 
-    def on_file_open(self, file_uuid: str) -> None:
+    def cleanup_hardlinked_file(cls, filepaths):
+        '''
+        Once inter-vm communication is complete and we no longer need to keep file copies with the
+        original filenames, delete them.
         '''
-        Open the file specified by file_uuid.
+        for filepath in filepaths:
+            if os.path.exists(filepath):
+                os.remove(filepath)
+
+    def get_path_to_file_with_original_name(
+        cls, file_dir: str, filename: str, filename_orig: str
+    ) -> str:
+        '''
+        Create a hardlink with the original filename and return its path.
 
         Once a file is downloaded, it exists in the data directory with the same filename as the
         server, except with the .gz.gpg stripped off. In order for the Display VM to know which
         application to open the file in, we create a hard link to this file with the original file
         name, including its extension.
+        '''
+
+        path_to_file_with_original_name = os.path.join(file_dir, filename_orig)
+
+        if not os.path.exists(path_to_file_with_original_name):
+            fn_no_ext, dummy = os.path.splitext(os.path.splitext(filename)[0])
+            filepath = os.path.join(file_dir, fn_no_ext)
+            os.link(filepath, path_to_file_with_original_name)
+
+        return path_to_file_with_original_name
 
-        If the file is missing, update the db so that is_downloaded is set to False.
+    def on_file_open(self, file_uuid: str) -> None:
+        '''
+        Open the file specified by file_uuid. If the file is missing, update the db so that
+        is_downloaded is set to False.
         '''
         file = self.get_file(file_uuid)
         logger.info('Opening file "{}".'.format(file.original_filename))
@@ -606,16 +631,12 @@ def on_file_open(self, file_uuid: str) -> None:
             self.sync_api()
             return
 
-        path_to_file_with_original_name = os.path.join(self.data_dir, file.original_filename)
-
-        if not os.path.exists(path_to_file_with_original_name):
-            fn_no_ext, dummy = os.path.splitext(os.path.splitext(file.filename)[0])
-            filepath = os.path.join(self.data_dir, fn_no_ext)
-            os.link(filepath, path_to_file_with_original_name)
-
         if not self.qubes:
             return
 
+        path_to_file_with_original_name = self.get_path_to_file_with_original_name(
+            self.data_dir, file.filename, file.original_filename)
+
         command = "qvm-open-in-vm"
         args = ['$dispvm:sd-svs-disp', path_to_file_with_original_name]
         process = QProcess(self)
@@ -635,14 +656,8 @@ def run_export_preflight_checks(self):
     def export_file_to_usb_drive(self, file_uuid: str, passphrase: str) -> None:
         '''
         Send the file specified by file_uuid to the Export VM with the user-provided passphrase for
-        unlocking the attached transfer device.
-
-        Once a file is downloaded, it exists in the data directory with the same filename as the
-        server, except with the .gz.gpg stripped off. In order for the user to know which
-        application to open the file in, we export the file with a different name: the original
-        filename which includes the file extesion.
-
-        If the file is missing, update the db so that is_downloaded is set to False.
+        unlocking the attached transfer device.  If the file is missing, update the db so that
+        is_downloaded is set to False.
         '''
         file = self.get_file(file_uuid)
         logger.info('Exporting file {}'.format(file.original_filename))
@@ -651,28 +666,18 @@ def export_file_to_usb_drive(self, file_uuid: str, passphrase: str) -> None:
             self.sync_api()
             return
 
-        path_to_file_with_original_name = os.path.join(self.data_dir, file.original_filename)
-
-        if not os.path.exists(path_to_file_with_original_name):
-            fn_no_ext, dummy = os.path.splitext(os.path.splitext(file.filename)[0])
-            filepath = os.path.join(self.data_dir, fn_no_ext)
-            os.link(filepath, path_to_file_with_original_name)
-
         if not self.qubes:
             return
 
-        fn_no_ext, dummy = os.path.splitext(os.path.splitext(file.filename)[0])
-        filepath = os.path.join(self.data_dir, fn_no_ext)
+        path_to_file_with_original_name = self.get_path_to_file_with_original_name(
+            self.data_dir, file.filename, file.original_filename)
+
         self.export.begin_usb_export.emit([path_to_file_with_original_name], passphrase)
 
     def print_file(self, file_uuid: str) -> None:
         '''
-        Send the file specified by file_uuid to the Export VM.
-
-        Once a file is downloaded, it exists in the data directory with the same filename as the
-        server, except with the .gz.gpg stripped off.
-
-        If the file is missing, update the db so that is_downloaded is set to False.
+        Send the file specified by file_uuid to the Export VM. If the file is missing, update the db
+        so that is_downloaded is set to False.
         '''
         file = self.get_file(file_uuid)
         logger.info('Printing file {}'.format(file.original_filename))
@@ -681,18 +686,12 @@ def print_file(self, file_uuid: str) -> None:
             self.sync_api()
             return
 
-        path_to_file_with_original_name = os.path.join(self.data_dir, file.original_filename)
-
-        if not os.path.exists(path_to_file_with_original_name):
-            fn_no_ext, dummy = os.path.splitext(os.path.splitext(file.filename)[0])
-            filepath = os.path.join(self.data_dir, fn_no_ext)
-            os.link(filepath, path_to_file_with_original_name)
-
         if not self.qubes:
             return
 
-        fn_no_ext, dummy = os.path.splitext(os.path.splitext(file.filename)[0])
-        filepath = os.path.join(self.data_dir, fn_no_ext)
+        path_to_file_with_original_name = self.get_path_to_file_with_original_name(
+            self.data_dir, file.filename, file.original_filename)
+
         self.export.begin_print.emit([path_to_file_with_original_name])
 
     def on_submission_download(

tests/gui/test_widgets.py

@@ -1492,7 +1492,7 @@ def test_FileWidget__on_export_clicked_missing_file(mocker, session, source):
 
 def test_FileWidget__on_print_clicked(mocker, session, source):
     """
-    Ensure preflight checks start when the EXPORT button is clicked and that password is requested
+    Ensure print_file is called when the PRINT button is clicked
     """
     file = factory.File(source=source['source'], is_downloaded=True)
     session.add(file)
@@ -1764,7 +1764,7 @@ def test_ExportDialog__update_after_CALLED_PROCESS_ERROR(mocker):
 
 def test_PrintDialog__on_retry_button_clicked(mocker):
     """
-    Ensure happy path runs preflight checks.
+    Ensure happy path prints the file.
     """
     controller = mocker.MagicMock()
     dialog = PrintDialog(controller, 'mock_uuid')

tests/test_export.py

@@ -17,17 +17,16 @@ def test_print(mocker):
     export = Export()
     export.print_call_success = mocker.MagicMock()
     export.print_call_success.emit = mocker.MagicMock()
+    export.export_completed = mocker.MagicMock()
+    export.export_completed.emit = mocker.MagicMock()
     _run_print = mocker.patch.object(export, '_run_print')
     mocker.patch('os.path.exists', return_value=True)
-    os_remove = mocker.patch('os.remove')
 
     export.print(['path1', 'path2'])
 
     _run_print.assert_called_once_with('mock_temp_dir', ['path1', 'path2'])
     export.print_call_success.emit.assert_called_once_with()
-    assert os_remove.call_count == 2
-    assert os_remove.call_args_list[0][0][0] == 'path1'
-    assert os_remove.call_args_list[1][0][0] == 'path2'
+    export.export_completed.emit.assert_called_once_with(['path1', 'path2'])
 
 
 def test_print_error(mocker):
@@ -41,18 +40,17 @@ def test_print_error(mocker):
     export = Export()
     export.print_call_failure = mocker.MagicMock()
     export.print_call_failure.emit = mocker.MagicMock()
+    export.export_completed = mocker.MagicMock()
+    export.export_completed.emit = mocker.MagicMock()
     error = ExportError('[mock_filepath]')
     _run_print = mocker.patch.object(export, '_run_print', side_effect=error)
     mocker.patch('os.path.exists', return_value=True)
-    os_remove = mocker.patch('os.remove')
 
     export.print(['path1', 'path2'])
 
     _run_print.assert_called_once_with('mock_temp_dir', ['path1', 'path2'])
     export.print_call_failure.emit.assert_called_once_with(error)
-    assert os_remove.call_count == 2
-    assert os_remove.call_args_list[0][0][0] == 'path1'
-    assert os_remove.call_args_list[1][0][0] == 'path2'
+    export.export_completed.emit.assert_called_once_with(['path1', 'path2'])
 
 
 def test__run_print(mocker):
@@ -99,17 +97,16 @@ def test_send_file_to_usb_device(mocker):
     export = Export()
     export.export_usb_call_success = mocker.MagicMock()
     export.export_usb_call_success.emit = mocker.MagicMock()
+    export.export_completed = mocker.MagicMock()
+    export.export_completed.emit = mocker.MagicMock()
     _run_disk_export = mocker.patch.object(export, '_run_disk_export')
     mocker.patch('os.path.exists', return_value=True)
-    os_remove = mocker.patch('os.remove')
 
     export.send_file_to_usb_device(['path1', 'path2'], 'mock passphrase')
 
     _run_disk_export.assert_called_once_with('mock_temp_dir', ['path1', 'path2'], 'mock passphrase')
     export.export_usb_call_success.emit.assert_called_once_with()
-    assert os_remove.call_count == 2
-    assert os_remove.call_args_list[0][0][0] == 'path1'
-    assert os_remove.call_args_list[1][0][0] == 'path2'
+    export.export_completed.emit.assert_called_once_with(['path1', 'path2'])
 
 
 def test_send_file_to_usb_device_error(mocker):
@@ -123,18 +120,17 @@ def test_send_file_to_usb_device_error(mocker):
     export = Export()
     export.export_usb_call_failure = mocker.MagicMock()
     export.export_usb_call_failure.emit = mocker.MagicMock()
+    export.export_completed = mocker.MagicMock()
+    export.export_completed.emit = mocker.MagicMock()
     error = ExportError('[mock_filepath]')
     _run_disk_export = mocker.patch.object(export, '_run_disk_export', side_effect=error)
     mocker.patch('os.path.exists', return_value=True)
-    os_remove = mocker.patch('os.remove')
 
     export.send_file_to_usb_device(['path1', 'path2'], 'mock passphrase')
 
     _run_disk_export.assert_called_once_with('mock_temp_dir', ['path1', 'path2'], 'mock passphrase')
     export.export_usb_call_failure.emit.assert_called_once_with(error)
-    assert os_remove.call_count == 2
-    assert os_remove.call_args_list[0][0][0] == 'path1'
-    assert os_remove.call_args_list[1][0][0] == 'path2'
+    export.export_completed.emit.assert_called_once_with(['path1', 'path2'])
 
 
 def test_run_preflight_checks(mocker):