diff --git a/client/securedrop_client/export.py b/client/securedrop_client/export.py index 8ab0a61de5..9a79a463f4 100644 --- a/client/securedrop_client/export.py +++ b/client/securedrop_client/export.py @@ -62,14 +62,19 @@ def run_printer_preflight_checks(self) -> None: """ logger.info("Beginning printer preflight check") self.tmpdir = mkdtemp() - archive_path = self._create_archive( - archive_dir=self.tmpdir, - archive_fn=self._PRINTER_PREFLIGHT_FN, - metadata=self._PRINTER_PREFLIGHT_METADATA, - ) - self._run_qrexec_export( - archive_path, self._on_print_preflight_success, self._on_print_prefight_error - ) + + try: + archive_path = self._create_archive( + archive_dir=self.tmpdir, + archive_fn=self._PRINTER_PREFLIGHT_FN, + metadata=self._PRINTER_PREFLIGHT_METADATA, + ) + self._run_qrexec_export( + archive_path, self._on_print_preflight_complete, self._on_print_prefight_error + ) + except ExportError as e: + logger.error("Error creating archive: {e}") + self._on_print_prefight_error() def run_export_preflight_checks(self) -> None: """ @@ -77,17 +82,21 @@ def run_export_preflight_checks(self) -> None: """ logger.debug("Beginning export preflight check") - self.tmpdir = mkdtemp() + try: + self.tmpdir = mkdtemp() - archive_path = self._create_archive( - archive_dir=self.tmpdir, - archive_fn=self._USB_TEST_FN, - metadata=self._USB_TEST_METADATA, - ) - # Emits status via on_process_completed() - self._run_qrexec_export( - archive_path, self._on_export_process_finished, self._on_export_process_error - ) + archive_path = self._create_archive( + archive_dir=self.tmpdir, + archive_fn=self._USB_TEST_FN, + metadata=self._USB_TEST_METADATA, + ) + # Emits status via on_process_completed() + self._run_qrexec_export( + archive_path, self._on_export_process_complete, self._on_export_process_error + ) + except ExportError: + logger.error("Export preflight check failed during archive creation") + self._on_export_process_error() def export(self, filepaths: List[str], passphrase: Optional[str]) -> None: """ @@ -112,7 +121,7 @@ def export(self, filepaths: List[str], passphrase: Optional[str]) -> None: # Emits status through callbacks self._run_qrexec_export( - archive_path, self._on_export_process_finished, self._on_export_process_error + archive_path, self._on_export_process_complete, self._on_export_process_error ) except IOError as e: @@ -120,13 +129,23 @@ def export(self, filepaths: List[str], passphrase: Optional[str]) -> None: logger.debug(f"Export failed: {e}") self.export_state_changed.emit(ExportStatus.ERROR_EXPORT) + # ExportStatus.ERROR_MISSING_FILES + except ExportError as err: + if err.status: + logger.error("Export failed while creating archive") + self.export_state_changed.emit(ExportError(err.status)) + else: + logger.error("Export failed while creating archive (no status supplied)") + self.export_state_changed.emit(ExportError(ExportStatus.ERROR_EXPORT)) + def _run_qrexec_export( self, archive_path: str, success_callback: Callable, error_callback: Callable ) -> None: """ Send the archive to the Export VM, where the archive will be processed. Uses qrexec-client-vm (via QProcess). Results are emitted via the - `on_process_finished` callback; errors are reported via `on_process_error`. + `finished` signal; errors are reported via `onError`. User-defined callback + functions must be connected to those signals. Args: archive_path (str): The path to the archive to be processed. @@ -168,33 +187,35 @@ def _cleanup_tmpdir(self): if self.tmpdir and os.path.exists(self.tmpdir): shutil.rmtree(self.tmpdir) - def _on_export_process_finished(self): + def _on_export_process_complete(self): """ Callback, handle and emit QProcess result. As with all such callbacks, the method signature cannot change. """ self._cleanup_tmpdir() # securedrop-export writes status to stderr - err = self.process.readAllStandardError() + if self.process: + err = self.process.readAllStandardError() - logger.debug(f"stderr: {err}") + logger.debug(f"stderr: {err}") - try: - result = err.data().decode("utf-8").strip() - if result: - logger.debug(f"Result is {result}") - # This is a bit messy, but make sure we are just taking the last line - # (no-op if no newline, since we already stripped whitespace above) - status_string = result.split("\n")[-1] - self.export_state_changed.emit(ExportStatus(status_string)) + try: + result = err.data().decode("utf-8").strip() + if result: + logger.debug(f"Result is {result}") + # This is a bit messy, but make sure we are just taking the last line + # (no-op if no newline, since we already stripped whitespace above) + status_string = result.split("\n")[-1] + self.export_state_changed.emit(ExportStatus(status_string)) - else: - logger.error("Export subprocess did not return a value we could parse") - self.export_state_changed.emit(ExportStatus.UNEXPECTED_RETURN_STATUS) + else: + logger.error("Export preflight returned empty result") + self.export_state_changed.emit(ExportStatus.UNEXPECTED_RETURN_STATUS) - except ValueError as e: - logger.debug(f"Export subprocess returned unexpected value: {e}") - self.export_state_changed.emit(ExportStatus.UNEXPECTED_RETURN_STATUS) + except ValueError as e: + logger.debug(f"Export preflight returned unexpected value: {e}") + logger.error("Export preflight returned unexpected value") + self.export_state_changed.emit(ExportStatus.UNEXPECTED_RETURN_STATUS) def _on_export_process_error(self): """ @@ -202,34 +223,41 @@ def _on_export_process_error(self): signature cannot change. """ self._cleanup_tmpdir() - err = self.process.readAllStandardError().data().decode("utf-8").strip() + if self.process: + err = self.process.readAllStandardError().data().decode("utf-8").strip() - logger.error(f"Export process error: {err}") + logger.error(f"Export process error: {err}") self.export_state_changed.emit(ExportStatus.CALLED_PROCESS_ERROR) - def _on_print_preflight_success(self): + def _on_print_preflight_complete(self): """ - Print preflight success callback. + Print preflight completion callback. """ self._cleanup_tmpdir() - output = self.process.readAllStandardError().data().decode("utf-8").strip() - try: - status = ExportStatus(output) - self.print_preflight_check_succeeded.emit(status) - logger.debug("Print preflight success") - - except ValueError as error: - logger.debug(f"Print preflight check failed: {error}") - logger.error("Print preflight check failed") - self.print_preflight_check_failed.emit(ExportError(ExportStatus.ERROR_PRINT)) - + if self.process: + output = self.process.readAllStandardError().data().decode("utf-8").strip() + try: + status = ExportStatus(output) + if status == ExportStatus.PRINT_PREFLIGHT_SUCCESS: + self.print_preflight_check_succeeded.emit(status) + logger.debug("Print preflight success") + else: + logger.debug(f"Print preflight failure ({status.value})") + self.print_preflight_check_failed.emit(status) + + except ValueError as error: + logger.debug(f"Print preflight check failed: {error}") + logger.error("Print preflight check failed") + self.print_preflight_check_failed.emit(ExportError(ExportStatus.ERROR_PRINT)) + def _on_print_prefight_error(self): """ - Print Preflight error callback. + Print Preflight error callback. Occurs when the QProcess itself fails. """ self._cleanup_tmpdir() - err = self.process.readAllStandardError().data().decode("utf-8").strip() - logger.debug(f"Print preflight error: {err}") + if self.process: + err = self.process.readAllStandardError().data().decode("utf-8").strip() + logger.debug(f"Print preflight error: {err}") self.print_preflight_check_failed.emit(ExportError(ExportStatus.ERROR_PRINT)) # Todo: not sure if we need to connect here, since the print dialog is managed by sd-devices. @@ -283,6 +311,15 @@ def print(self, filepaths: List[str]) -> None: logger.debug(f"Export failed: {e}") self.print_failed.emit(ExportError(ExportStatus.ERROR_PRINT)) + # ExportStatus.ERROR_MISSING_FILES + except ExportError as err: + if err.status: + logger.error("Print failed while creating archive") + self.print_failed.emit(ExportError(err.status)) + else: + logger.error("Print failed while creating archive (no status supplied)") + self.print_failed.emit(ExportError(ExportStatus.ERROR_PRINT)) + self.export_completed.emit(filepaths) def _create_archive( diff --git a/client/securedrop_client/gui/conversation/export/export_wizard_constants.py b/client/securedrop_client/gui/conversation/export/export_wizard_constants.py index eccb767a00..ce8c0f3af7 100644 --- a/client/securedrop_client/gui/conversation/export/export_wizard_constants.py +++ b/client/securedrop_client/gui/conversation/export/export_wizard_constants.py @@ -7,12 +7,13 @@ Export wizard page ordering, human-readable status messages """ + # Sequential list of pages (the enum value matters as a ranked ordering.) # The reason the 'error' page is second is because the other pages have # validation logic that means they can't be bypassed by QWizard::next. # When we need to show an error, it's easier to go 'back' to the error # page and set it to be a FinalPage than it is to try to skip the conditional -# pages. PyQt6 introduces behaviour that may deprecate this requirement. +# pages. PyQt6 introduces behaviour that may deprecate this requirement. class Pages(IntEnum): PREFLIGHT = 0 ERROR = 1 @@ -20,6 +21,7 @@ class Pages(IntEnum): UNLOCK_USB = 3 EXPORT_DONE = 4 + # Human-readable status info STATUS_MESSAGES = { ExportStatus.NO_DEVICE_DETECTED: _("No device detected"), diff --git a/client/securedrop_client/gui/conversation/export/export_wizard_page.py b/client/securedrop_client/gui/conversation/export/export_wizard_page.py index a3b2cf93a2..0609bc086f 100644 --- a/client/securedrop_client/gui/conversation/export/export_wizard_page.py +++ b/client/securedrop_client/gui/conversation/export/export_wizard_page.py @@ -102,7 +102,7 @@ def _build_layout(self) -> QVBoxLayout: self.header.setObjectName("ModalDialog_header") header_container_layout.addWidget(self.header_icon) header_container_layout.addWidget(self.header_spinner_label) - header_container_layout.addWidget(self.header, alignment=Qt.AlignLeft) # Prev: AlignCenter + header_container_layout.addWidget(self.header, alignment=Qt.AlignLeft) # Prev: AlignCenter header_container_layout.addStretch() self.header_line = QWidget() self.header_line.setObjectName("ModalDialog_header_line") @@ -122,7 +122,7 @@ def _build_layout(self) -> QVBoxLayout: self.body_layout.addWidget(self.body) self.body_layout.setSizeConstraint(QLayout.SetMinimumSize) - # TODO: it's either like this, or in the parent layout elements + # TODO: it's either like this, or in the parent layout elements self.body_layout.setSizeConstraint(QLayout.SetMinimumSize) # Widget for displaying error messages (hidden by default) @@ -258,6 +258,7 @@ def on_status_received(self, status: ExportStatus): self.header.setText(header) self.status = status + class ErrorPage(ExportWizardPage): def __init__(self, export, summary): header = _("Export Failed") @@ -267,11 +268,12 @@ def __init__(self, export, summary): def isComplete(self) -> bool: return False - + @pyqtSlot(object) def on_status_received(self, status: ExportStatus): pass + class InsertUSBPage(ExportWizardPage): def __init__(self, export, summary): self.summary = summary @@ -290,12 +292,12 @@ def on_status_received(self, status: ExportStatus) -> None: should_show_hint = status in ( ExportStatus.MULTI_DEVICE_DETECTED, ExportStatus.INVALID_DEVICE_DETECTED, - ) or (self.status == status == ExportStatus.NO_DEVICE_DETECTED) + ) or (self.status == status == ExportStatus.NO_DEVICE_DETECTED) self.update_content(status, should_show_hint) self.status = status self.completeChanged.emit() if status in (ExportStatus.DEVICE_LOCKED, ExportStatus.DEVICE_WRITABLE): - self.wizard().next() + self.wizard().next() def validatePage(self) -> bool: """ @@ -321,7 +323,6 @@ def validatePage(self) -> bool: logger.warning("InsertUSBPage encountered unexpected status") return super().validatePage() - def nextId(self): """ Override builtin to allow bypassing the password page if device unlocked @@ -435,7 +436,11 @@ def on_status_received(self, status: ExportStatus) -> None: def validatePage(self): # Also to add: DEVICE_BUSY for unmounting. # This shouldn't stop us from going "back" to an error page - return self.status in (ExportStatus.DEVICE_WRITABLE, ExportStatus.SUCCESS_EXPORT, ExportStatus.ERROR_EXPORT_CLEANUP) + return self.status in ( + ExportStatus.DEVICE_WRITABLE, + ExportStatus.SUCCESS_EXPORT, + ExportStatus.ERROR_EXPORT_CLEANUP, + ) def nextId(self): if self.status == ExportStatus.SUCCESS_EXPORT: diff --git a/client/tests/gui/conversation/export/test_device.py b/client/tests/gui/conversation/export/test_device.py index b8683b5408..5313630b3a 100644 --- a/client/tests/gui/conversation/export/test_device.py +++ b/client/tests/gui/conversation/export/test_device.py @@ -5,26 +5,27 @@ import pytest -from PyQt5.QtCore import QProcess +from PyQt5.QtTest import QSignalSpy from securedrop_client.export_status import ExportError, ExportStatus from securedrop_client.gui.conversation.export import Export from tests import factory _PATH_TO_PRETEND_ARCHIVE = "/tmp/archive-pretend" -_QREXEC_EXPORT_COMMAND = [ - "qrexec-client-vm", - "--", - "sd-devices", - "qubes.OpenInVM", - "/usr/lib/qubes/qopen-in-vm", - "--view-only", - "--", - f"{_PATH_TO_PRETEND_ARCHIVE}", -] +_QREXEC_EXPORT_COMMAND = ( + "/usr/bin/qrexec-client-vm", + [ + "--", + "sd-devices", + "qubes.OpenInVM", + "/usr/lib/qubes/qopen-in-vm", + "--view-only", + "--", + f"{_PATH_TO_PRETEND_ARCHIVE}", + ], +) _MOCK_FILEDIR = "/tmp/mock_tmpdir/" -@mock.patch("PyQt5.QtCore.QProcess") class TestDevice: @classmethod def setup_class(cls): @@ -46,89 +47,127 @@ def teardown_method(cls): cls.mock_file = None cls.device._create_archive = None - def test_Device_run_printer_preflight_checks(self, mock_subprocess): - mock_qp = mock_subprocess() - mock_qp.start = mock.MagicMock() - mock_qp.readAllStandardError.return_value = ( - ExportStatus.PRINT_PREFLIGHT_SUCCESS.value.encode("utf-8") - ) + def test_Device_run_printer_preflight_checks(self): device = Export() device._create_archive = mock.MagicMock() device._create_archive.return_value = _PATH_TO_PRETEND_ARCHIVE with mock.patch( - "securedrop_client.export.TemporaryDirectory", + "securedrop_client.export.mkdtemp", return_value=self.mock_tmpdir, - ), mock.patch("securedrop_client.export.QProcess", return_value=mock_qp): + ), mock.patch("securedrop_client.export.QProcess") as mock_qprocess: + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() + mock_qproc.readAllStandardError.return_value = ( + ExportStatus.PRINT_PREFLIGHT_SUCCESS.value.encode("utf-8") + ) device.run_printer_preflight_checks() - mock_qp.start.assert_called_once() - assert ( - _QREXEC_EXPORT_COMMAND in mock_subprocess.call_args[0] - ), f"Actual: {mock_subprocess.call_args[0]}" + mock_qproc.start.assert_called_once() + assert ( + mock_qproc.start.call_args[0] == _QREXEC_EXPORT_COMMAND + ), f"Actual: {mock_qproc.start.call_args[0]}" - def test_Device_run_print_preflight_checks_with_error(self, mock_sp): - mock_qp = mock_sp() - mock_qp.start = mock.MagicMock() - mock_qp.readAllStandardError.return_value = b"Not a real status\n" + def test_Device_run_print_preflight_checks_with_error(self): + spy = QSignalSpy(self.device.export_state_changed) + with mock.patch( + "securedrop_client.export.mkdtemp", + return_value=self.mock_tmpdir, + ), mock.patch("securedrop_client.export.QProcess") as mock_qprocess: + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() + mock_qproc.start.side_effect = ( + lambda proc, args: mock_qproc.finished.emit() + ) # This ain't doin it + mock_qproc.readAllStandardError.return_value = b"Not a real status\n" - with mock.patch("securedrop_client.export.logger.error") as err: self.device.run_printer_preflight_checks() - assert "Print preflight failed" in err.call_args[0] + mock_qproc.start.assert_called_once() - def test_Device_print(self, mock_subprocess): - with mock.patch( - "securedrop_client.export.TemporaryDirectory", + # TODO + # assert len(spy) == 1 and spy[0] == ExportStatus.UNEXPECTED_RETURN_STATUS + + def test_Device_print(self): + with mock.patch("securedrop_client.export.QProcess") as mock_qprocess, mock.patch( + "securedrop_client.export.mkdtemp", return_value=self.mock_tmpdir, ): + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() self.device.print([self.mock_file_location]) - self.device._create_archive.assert_called_once_with( - archive_dir=_MOCK_FILEDIR, - archive_fn=self.device._PRINT_FN, - metadata=self.device._PRINT_METADATA, - filepaths=[self.mock_file_location], - ) - mock_subprocess.assert_called_once() - assert _QREXEC_EXPORT_COMMAND in mock_subprocess.call_args[0] + mock_qprocess.start.assert_called_once() + assert mock_qprocess.start.call_args[0] == _QREXEC_EXPORT_COMMAND + + self.device._create_archive.assert_called_once_with( + archive_dir=self.mock_tmpdir, + archive_fn=self.device._PRINT_FN, + metadata=self.device._PRINT_METADATA, + filepaths=[self.mock_file_location], + ) - def test_Device_print_file_file_missing(self, mock_subprocess, mocker): + def test_Device_print_file_file_missing(self, mocker): device = Export() - warning_logger = mocker.patch("securedrop_client.export.logger.warning") + spy = QSignalSpy(device.export_state_changed) - log_msg = "File not found at specified filepath, skipping" + with mock.patch( + "securedrop_client.export.mkdtemp", + return_value=self.mock_tmpdir, + ), mock.patch("securedrop_client.export.QProcess") as mock_qprocess: + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() - device.print("some-missing-file-uuid") + device.print("some-missing-file-uuid") - assert log_msg in warning_logger.call_args[0] - mock_subprocess.assert_not_called() + mock_qproc.start.assert_not_called() + # TODO + # assert len(spy) == 1 and spy[0] == ExportError(ExportStatus.ERROR_MISSING_FILES) - def test_Device_run_export_preflight_checks(self, mock_subprocess): + def test_Device_run_export_preflight_checks(self): with mock.patch( - "securedrop_client.export.TemporaryDirectory", + "securedrop_client.export.mkdtemp", return_value=self.mock_tmpdir, - ): + ), mock.patch("securedrop_client.export.QProcess") as mock_qprocess: + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() + self.device.run_export_preflight_checks() - mock_subprocess.assert_called_once() + # mock_qproc.start.call_args[0] + # '/usr/bin/qrexec-client-vm', ['--', 'sd-devices', 'qubes.OpenInVM', '/usr/lib/qubes/qopen-in-vm', '--view-only', '--', '/tmp/archive-pretend'] + + mock_qproc.start.assert_called_once() + assert mock_qproc.start.call_args[0] == _QREXEC_EXPORT_COMMAND + # Call args: call(archive_dir=, archive_fn='usb-test.sd-export', metadata={'device': 'usb-test'}) self.device._create_archive.assert_called_once_with( - archive_dir=_MOCK_FILEDIR, + archive_dir=self.mock_tmpdir, archive_fn=self.device._USB_TEST_FN, metadata=self.device._USB_TEST_METADATA, ) - mock_subprocess.assert_called_once() - assert _QREXEC_EXPORT_COMMAND in mock_subprocess.call_args[0] - def test_Device_run_export_preflight_checks_with_error(self, mock_sp): - mock_sp.return_value = b"Houston, we have a problem\n" + def test_Device_run_export_preflight_checks_with_error(self): + spy = QSignalSpy(self.device.export_state_changed) + + with mock.patch( + "securedrop_client.export.mkdtemp", + return_value=self.mock_tmpdir, + ), mock.patch.object(self.device, "_create_archive"), mock.patch( + "securedrop_client.export.QProcess" + ) as mock_qprocess: + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() + mock_qproc.start.side_effect = ( + lambda proc, args: self.device._on_export_process_complete() + ) + mock_qproc.readAllStandardError = mock.MagicMock() + mock_qproc.readAllStandardError.data.return_value = b"Houston, we have a problem\n" - with mock.patch("securedrop_client.export.logger.error") as err: self.device.run_export_preflight_checks() - assert "Export preflight failed" in err.call_args[0] + assert len(spy) == 1 and spy[0] == ExportStatus.UNEXPECTED_RETURN_STATUS - def test_Device_export_file_missing(self, mock_subprocess, mocker): + def test_Device_export_file_missing(self, mocker): device = Export() warning_logger = mocker.patch("securedrop_client.export.logger.warning") @@ -136,53 +175,67 @@ def test_Device_export_file_missing(self, mock_subprocess, mocker): "securedrop_client.export.tarfile.open", return_value=mock.MagicMock(), ), mock.patch( - "securedrop_client.export.TemporaryDirectory", + "securedrop_client.export.mkdtemp", return_value=self.mock_tmpdir, - ): + ), mock.patch( + "securedrop_client.export.QProcess" + ) as mock_qprocess: device.export(["/not/a/real/location"], "mock passphrase") + mock_qprocess.assert_not_called() + warning_logger.assert_called_once() - mock_subprocess.assert_not_called() # Todo: could get more specific about looking for the emitted failure signal - def test_Device_export(self, mock_subprocess): + def test_Device_export(self): filepath = "some/file/path" passphrase = "passphrase" + expected_metadata = self.device._DISK_METADATA.copy() + expected_metadata[self.device._DISK_ENCRYPTION_KEY_NAME] = passphrase + with mock.patch( - "securedrop_client.export.TemporaryDirectory", + "securedrop_client.export.mkdtemp", return_value=self.mock_tmpdir, - ): + ), mock.patch("securedrop_client.export.QProcess") as mock_qprocess: + mock_qproc = mock_qprocess.return_value + mock_qproc.start = mock.MagicMock() self.device.export([filepath], passphrase) - expected_metadata = self.device._DISK_METADATA.copy() - expected_metadata[self.device._DISK_ENCRYPTION_KEY_NAME] = passphrase + mock_qproc.start.assert_called_once() + assert mock_qproc.start.call_args[0] == _QREXEC_EXPORT_COMMAND self.device._create_archive.assert_called_once_with( - archive_dir=_MOCK_FILEDIR, + archive_dir=self.mock_tmpdir, archive_fn=self.device._DISK_FN, metadata=expected_metadata, filepaths=[filepath], ) - mock_subprocess.assert_called_once() - assert _QREXEC_EXPORT_COMMAND in mock_subprocess.call_args[0] @pytest.mark.parametrize("status", [i.value for i in ExportStatus]) - def test__run_qrexec_success(self, mocked_subprocess, status): - mocked_subprocess.readAllStandardError.return_value = f"{status}\n".encode("utf-8") + def test__run_qrexec_success(self, status): enum = ExportStatus(status) - with mock.patch.object(self.device, "_on_export_process_finished") as mock_finished: + with mock.patch("securedrop_client.export.QProcess") as mock_qprocess, mock.patch.object( + self.device, "_on_export_process_complete" + ) as mock_callback: + mock_qproc = mock_qprocess.return_value + mock_qproc.finished = mock.MagicMock() + mock_qproc.start = mock.MagicMock() + mock_qproc.start.side_effect = ( + lambda proc, args: self.device._on_export_process_complete() + ) + mock_qproc.readAllStandardError.return_value = f"{status}\n".encode("utf-8") + self.device._run_qrexec_export( _PATH_TO_PRETEND_ARCHIVE, - self.device._on_export_process_finished, + mock_callback, self.device._on_export_process_error, ) - mock_finished.assert_called_once() - assert status in mock_finished.call_args[0] + mock_qproc.start.assert_called_once() @mock.patch("securedrop_client.export.tarfile") - def test__add_virtual_file_to_archive(self, mock_tarfile, mock_sp): + def test__add_virtual_file_to_archive(self, mock_tarfile): mock_tarinfo = mock.MagicMock(spec=tarfile.TarInfo) mock_tarfile.TarInfo.return_value = mock_tarinfo @@ -211,7 +264,7 @@ def test__create_archive(self, mocker): assert not os.path.exists(archive_path) - def test__create_archive_with_an_export_file(self, mocker): + def test__create_archive_with_an_export_file(self): device = Export() archive_path = None with TemporaryDirectory() as temp_dir, NamedTemporaryFile() as export_file: @@ -223,7 +276,7 @@ def test__create_archive_with_an_export_file(self, mocker): assert not os.path.exists(archive_path) - def test__create_archive_with_multiple_export_files(self, mocker): + def test__create_archive_with_multiple_export_files(self): device = Export() archive_path = None with TemporaryDirectory() as tmpdir, NamedTemporaryFile() as f1, NamedTemporaryFile() as f2: @@ -240,14 +293,19 @@ def test__create_archive_with_multiple_export_files(self, mocker): assert not os.path.exists(archive_path) - def test__tmpdir_cleaned_up_on_exception(self, mock_sp): + def test__tmpdir_cleaned_up_on_exception(self): """ Sanity check. If we encounter an error after archive has been built, ensure the tmpdir directory cleanup happens. """ - with TemporaryDirectory() as tmpdir, pytest.raises(ExportError): - print(f"{tmpdir} created") - - raise ExportError("Something bad happened!") - - assert not os.path.exists(tmpdir) + with mock.patch( + "securedrop_client.export.mkdtemp", return_value=self.mock_tmpdir + ) as tmpdir, mock.patch("securedrop_client.export.QProcess") as qprocess, mock.patch.object( + self.device, "_cleanup_tmpdir" + ) as mock_cleanup: + mock_qproc = qprocess.return_value + mock_qproc.readAllStandardError.data.return_value = b"Something awful happened!\n" + mock_qproc.start = lambda proc, args: self.device._on_export_process_error() + self.device.run_printer_preflight_checks() + assert self.device.tmpdir == self.mock_tmpdir + mock_cleanup.assert_called()