diff --git a/tests/test_export.py b/tests/test_export.py index a289a44d53..85f0ae90ca 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -40,17 +40,15 @@ def test_run_printer_preflight(mocker): Ensure TemporaryDirectory is used when creating and sending the archives during the preflight checks and that the success signal is emitted by Export. """ - mock_temp_dir = mocker.MagicMock() - mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") - mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) + export = Export() + mocker.patch.object( + export, "_build_archive_and_export", return_value=ExportStatus.PREFLIGHT_SUCCESS + ) export.printer_preflight_success = mocker.MagicMock() export.printer_preflight_success.emit = mocker.MagicMock() - _run_printer_preflight = mocker.patch.object(export, "_run_printer_preflight") export.run_printer_preflight() - - _run_printer_preflight.assert_called_once_with("mock_temp_dir") export.printer_preflight_success.emit.assert_called_once_with() @@ -59,70 +57,36 @@ def test_run_printer_preflight_error(mocker): Ensure TemporaryDirectory is used when creating and sending the archives during the preflight checks and that the failure signal is emitted by Export. """ - mock_temp_dir = mocker.MagicMock() - mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") - mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) + export = Export() + error = ExportError("bang!") + mocker.patch.object(export, "_build_archive_and_export", side_effect=error) + export.printer_preflight_failure = mocker.MagicMock() export.printer_preflight_failure.emit = mocker.MagicMock() - error = ExportError("bang!") - _run_print_preflight = mocker.patch.object(export, "_run_printer_preflight", side_effect=error) export.run_printer_preflight() - _run_print_preflight.assert_called_once_with("mock_temp_dir") export.printer_preflight_failure.emit.assert_called_once_with(error) -def test__run_printer_preflight(mocker): - """ - Ensure _export_archive and _create_archive are called with the expected parameters, - _export_archive is called with the return value of _create_archive, and - _run_disk_test returns without error if 'USB_CONNECTED' is the return value of _export_archive. - """ +def test_print(mocker): export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="") - - export._run_printer_preflight("mock_archive_dir") - export._export_archive.assert_called_once_with("mock_archive_path") - export._create_archive.assert_called_once_with( - "mock_archive_dir", "printer-preflight.sd-export", {"device": "printer-preflight"} + mock_qrexec_call = mocker.patch.object( + export, "_build_archive_and_export", return_value=ExportStatus.PRINT_SUCCESS ) - -def test__run_printer_preflight_raises_ExportError_if_not_empty_string(mocker): - """ - Ensure ExportError is raised if _run_disk_test returns anything other than 'USB_CONNECTED'. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="SOMETHING_OTHER_THAN_EMPTY_STRING") - - with pytest.raises(ExportError): - export._run_printer_preflight("mock_archive_dir") - - -def test_print(mocker): - """ - Ensure TemporaryDirectory is used when creating and sending the archive containing the file to - print and that the success signal is emitted. - """ - mock_temp_dir = mocker.MagicMock() - mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") - mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) - 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) export.print(["path1", "path2"]) - _run_print.assert_called_once_with("mock_temp_dir", ["path1", "path2"]) + mock_qrexec_call.assert_called_once_with( + metadata=export.PRINT_METADATA, filename=export.PRINT_FN, filepaths=["path1", "path2"] + ) export.print_call_success.emit.assert_called_once_with() export.export_completed.emit.assert_called_once_with(["path1", "path2"]) @@ -135,51 +99,25 @@ def test_print_error(mocker): mock_temp_dir = mocker.MagicMock() mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) + 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) + error = ExportError("oh no!") + _run_print = mocker.patch.object(export, "_build_archive_and_export", side_effect=error) mocker.patch("os.path.exists", return_value=True) export.print(["path1", "path2"]) - _run_print.assert_called_once_with("mock_temp_dir", ["path1", "path2"]) + _run_print.assert_called_once_with( + metadata=export.PRINT_METADATA, filename=export.PRINT_FN, filepaths=["path1", "path2"] + ) export.print_call_failure.emit.assert_called_once_with(error) export.export_completed.emit.assert_called_once_with(["path1", "path2"]) -def test__run_print(mocker): - """ - Ensure _export_archive and _create_archive are called with the expected parameters and - _export_archive is called with the return value of _create_archive. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="") - - export._run_print("mock_archive_dir", ["mock_filepath"]) - - export._export_archive.assert_called_once_with("mock_archive_path") - export._create_archive.assert_called_once_with( - "mock_archive_dir", "print_archive.sd-export", {"device": "printer"}, ["mock_filepath"] - ) - - -def test__run_print_raises_ExportError_if_not_empty_string(mocker): - """ - Ensure ExportError is raised if _run_print returns anything other than ''. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="SOMETHING_OTHER_THAN_EMPTY_STRING") - - with pytest.raises(ExportError): - export._run_print("mock_archive_dir", ["mock_filepath"]) - - def test_send_file_to_usb_device(mocker): """ Ensure TemporaryDirectory is used when creating and sending the archive containing the export @@ -188,17 +126,23 @@ def test_send_file_to_usb_device(mocker): mock_temp_dir = mocker.MagicMock() mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) + 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") + _run_disk_export = mocker.patch.object(export, "_build_archive_and_export") mocker.patch("os.path.exists", return_value=True) + metadata = export.DISK_METADATA + metadata[export.DISK_ENCRYPTION_KEY_NAME] = "mock passphrase" + export.send_file_to_usb_device(["path1", "path2"], "mock passphrase") - _run_disk_export.assert_called_once_with("mock_temp_dir", ["path1", "path2"], "mock passphrase") + _run_disk_export.assert_called_once_with( + metadata=metadata, filename=export.DISK_FN, filepaths=["path1", "path2"] + ) export.export_usb_call_success.emit.assert_called_once_with() export.export_completed.emit.assert_called_once_with(["path1", "path2"]) @@ -208,26 +152,32 @@ def test_send_file_to_usb_device_error(mocker): Ensure TemporaryDirectory is used when creating and sending the archive containing the export file and that the failure signal is emitted. """ + export = Export() + mock_temp_dir = mocker.MagicMock() mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) - 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) + error = ExportError("ohno") + _run_disk_export = mocker.patch.object(export, "_build_archive_and_export", side_effect=error) + + metadata = export.DISK_METADATA + metadata[export.DISK_ENCRYPTION_KEY_NAME] = "mock passphrase" export.send_file_to_usb_device(["path1", "path2"], "mock passphrase") - _run_disk_export.assert_called_once_with("mock_temp_dir", ["path1", "path2"], "mock passphrase") + _run_disk_export.assert_called_once_with( + metadata=metadata, filename=export.DISK_FN, filepaths=["path1", "path2"] + ) export.export_usb_call_failure.emit.assert_called_once_with(error) export.export_completed.emit.assert_called_once_with(["path1", "path2"]) -def test_run_preflight_checks(mocker): +def test_run_usb_preflight_checks(mocker): """ Ensure TemporaryDirectory is used when creating and sending the archives during the preflight checks and that the success signal is emitted by Export. @@ -236,131 +186,93 @@ def test_run_preflight_checks(mocker): mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) export = Export() + export.preflight_check_call_success = mocker.MagicMock() export.preflight_check_call_success.emit = mocker.MagicMock() - _run_usb_export = mocker.patch.object(export, "_run_usb_test") - _run_disk_export = mocker.patch.object(export, "_run_disk_test") + _run_export = mocker.patch.object(export, "_build_archive_and_export") export.run_preflight_checks() - _run_usb_export.assert_called_once_with("mock_temp_dir") - _run_disk_export.assert_called_once_with("mock_temp_dir") + _run_export.assert_called_once_with( + metadata=export.USB_TEST_METADATA, filename=export.USB_TEST_FN + ) export.preflight_check_call_success.emit.assert_called_once_with() -def test_run_preflight_checks_error(mocker): +def test_run_usb_preflight_checks_error(mocker): """ Ensure TemporaryDirectory is used when creating and sending the archives during the preflight checks and that the failure signal is emitted by Export. """ + mock_temp_dir = mocker.MagicMock() mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) + export = Export() export.preflight_check_call_failure = mocker.MagicMock() export.preflight_check_call_failure.emit = mocker.MagicMock() error = ExportError("bang!") - _run_usb_export = mocker.patch.object(export, "_run_usb_test") - _run_disk_export = mocker.patch.object(export, "_run_disk_test", side_effect=error) + _run_export = mocker.patch.object(export, "_build_archive_and_export", side_effect=error) export.run_preflight_checks() - _run_usb_export.assert_called_once_with("mock_temp_dir") - _run_disk_export.assert_called_once_with("mock_temp_dir") + _run_export.assert_called_once_with( + metadata=export.USB_TEST_METADATA, filename=export.USB_TEST_FN + ) export.preflight_check_call_failure.emit.assert_called_once_with(error) -def test__run_disk_export(mocker): +@pytest.mark.parametrize("success_qrexec", [e.value for e in ExportStatus]) +def test__build_archive_and_export_success(mocker, success_qrexec): """ - Ensure _export_archive and _create_archive are called with the expected parameters, - _export_archive is called with the return value of _create_archive, and - _run_disk_test returns without error if '' is the output status of _export_archive. + Test the command that calls out to underlying qrexec service. """ export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="") - export._run_disk_export("mock_archive_dir", ["mock_filepath"], "mock_passphrase") + mock_temp_dir = mocker.MagicMock() + mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") + mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) - export._export_archive.assert_called_once_with("mock_archive_path") - export._create_archive.assert_called_once_with( - "mock_archive_dir", - "archive.sd-export", - {"encryption_key": "mock_passphrase", "device": "disk", "encryption_method": "luks"}, - ["mock_filepath"], + mock_qrexec_call = mocker.patch.object( + export, "_run_qrexec_export", return_value=bytes(success_qrexec, "utf-8") ) + mocker.patch.object(export, "_create_archive", return_value="mock_archive_path") + metadata = {"device": "pretend", "encryption_method": "transparent"} -def test__run_disk_export_raises_ExportError_if_not_empty_string(mocker): - """ - Ensure ExportError is raised if _run_disk_test returns anything other than ''. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="SOMETHING_OTHER_THAN_EMPTY_STRING") - - with pytest.raises(ExportError): - export._run_disk_export("mock_archive_dir", ["mock_filepath"], "mock_passphrase") - - -def test__run_disk_test(mocker): - """ - Ensure _export_archive and _create_archive are called with the expected parameters, - _export_archive is called with the return value of _create_archive, and - _run_disk_test returns without error if 'USB_ENCRYPTED' is the output status of _export_archive. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value=ExportStatus("USB_ENCRYPTED")) - - export._run_disk_test("mock_archive_dir") - - export._export_archive.assert_called_once_with("mock_archive_path") - export._create_archive.assert_called_once_with( - "mock_archive_dir", "disk-test.sd-export", {"device": "disk-test"} + result = export._build_archive_and_export( + metadata=metadata, filename="mock_filename", filepaths=["mock_filepath"] ) + mock_qrexec_call.assert_called_once() - -def test__run_disk_test_raises_ExportError_if_not_USB_ENCRYPTED(mocker): - """ - Ensure ExportError is raised if _run_disk_test returns anything other than 'USB_ENCRYPTED'. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="SOMETHING_OTHER_THAN_USB_ENCRYPTED") - - with pytest.raises(ExportError): - export._run_disk_test("mock_archive_dir") + assert result == bytes(success_qrexec, "utf-8") -def test__run_usb_test(mocker): +def test__build_archive_and_export_error(mocker): """ - Ensure _export_archive and _create_archive are called with the expected parameters, - _export_archive is called with the return value of _create_archive, and - _run_disk_test returns without error if 'USB_CONNECTED' is the return value of _export_archive. + Test the command that calls out to underlying qrexec service. """ export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value=ExportStatus("USB_CONNECTED")) + mock_temp_dir = mocker.MagicMock() + mock_temp_dir.__enter__ = mocker.MagicMock(return_value="mock_temp_dir") + mocker.patch("securedrop_client.export.TemporaryDirectory", return_value=mock_temp_dir) - export._run_usb_test("mock_archive_dir") + mocker.patch.object(export, "_create_archive", return_value="mock_archive_path") - export._export_archive.assert_called_once_with("mock_archive_path") - export._create_archive.assert_called_once_with( - "mock_archive_dir", "usb-test.sd-export", {"device": "usb-test"} + mock_qrexec_call = mocker.patch.object( + export, "_run_qrexec_export", side_effect=ExportError(ExportStatus.UNEXPECTED_RETURN_STATUS) ) - -def test__run_usb_test_raises_ExportError_if_not_USB_CONNECTED(mocker): - """ - Ensure ExportError is raised if _run_disk_test returns anything other than 'USB_CONNECTED'. - """ - export = Export() - export._create_archive = mocker.MagicMock(return_value="mock_archive_path") - export._export_archive = mocker.MagicMock(return_value="SOMETHING_OTHER_THAN_USB_CONNECTED") + metadata = {"device": "pretend", "encryption_method": "transparent"} with pytest.raises(ExportError): - export._run_usb_test("mock_archive_dir") + result = export._build_archive_and_export( + metadata=metadata, filename="mock_filename", filepaths=["mock_filepath"] + ) + assert result == ExportStatus.UNEXPECTED_RETURN_STATUS + + mock_qrexec_call.assert_called_once() def test__create_archive(mocker): @@ -370,7 +282,9 @@ def test__create_archive(mocker): export = Export() archive_path = None with TemporaryDirectory() as temp_dir: - archive_path = export._create_archive(temp_dir, "mock.sd-export", {}) + archive_path = export._create_archive( + archive_dir=temp_dir, archive_fn="mock.sd-export", metadata={}, filepaths=[] + ) assert archive_path == os.path.join(temp_dir, "mock.sd-export") assert os.path.exists(archive_path) # sanity check @@ -381,7 +295,12 @@ def test__create_archive_with_an_export_file(mocker): export = Export() archive_path = None with TemporaryDirectory() as temp_dir, NamedTemporaryFile() as export_file: - archive_path = export._create_archive(temp_dir, "mock.sd-export", {}, [export_file.name]) + archive_path = export._create_archive( + archive_dir=temp_dir, + archive_fn="mock.sd-export", + metadata={}, + filepaths=[export_file.name], + ) assert archive_path == os.path.join(temp_dir, "mock.sd-export") assert os.path.exists(archive_path) # sanity check @@ -409,21 +328,63 @@ def test__create_archive_with_multiple_export_files(mocker): assert not os.path.exists(archive_path) -def test__export_archive(mocker): +@pytest.mark.parametrize("qrexec_return_value_success", [e.value for e in ExportStatus]) +def test__run_qrexec_export(mocker, qrexec_return_value_success): """ Ensure the subprocess call returns the expected output. """ export = Export() - mocker.patch("subprocess.check_output", return_value=b"USB_CONNECTED") - status = export._export_archive("mock.sd-export") - assert status == ExportStatus.USB_CONNECTED + qrexec_mocker = mocker.patch( + "subprocess.check_output", return_value=bytes(qrexec_return_value_success, "utf-8") + ) + result = export._run_qrexec_export("mock.sd-export") + + qrexec_mocker.assert_called_once_with( + [ + "qrexec-client-vm", + "--", + "sd-devices", + "qubes.OpenInVM", + "/usr/lib/qubes/qopen-in-vm", + "--view-only", + "--", + "mock.sd-export", + ], + stderr=-2, + ) + + assert ExportStatus(result) + + +@pytest.mark.parametrize( + "qrexec_return_value_error", [b"", b"qrexec not connected", b"DEVICE_UNLOCKED\nERROR_WRITE"] +) +def test__run_qrexec_export_returns_bad_data(mocker, qrexec_return_value_error): + """ + Ensure the subprocess call returns the expected output. + """ + export = Export() + qrexec_mocker = mocker.patch("subprocess.check_output", return_value=qrexec_return_value_error) - mocker.patch("subprocess.check_output", return_value=b"mock") with pytest.raises(ExportError, match="UNEXPECTED_RETURN_STATUS"): - export._export_archive("mock.sd-export") + export._run_qrexec_export("mock.sd-export") + + qrexec_mocker.assert_called_once_with( + [ + "qrexec-client-vm", + "--", + "sd-devices", + "qubes.OpenInVM", + "/usr/lib/qubes/qopen-in-vm", + "--view-only", + "--", + "mock.sd-export", + ], + stderr=-2, + ) -def test__export_archive_does_not_raise_ExportError_when_CalledProcessError(mocker): +def test__run_qrexec_export_does_not_raise_ExportError_when_CalledProcessError(mocker): """ Ensure ExportError is raised if a CalledProcessError is encountered. """ @@ -433,10 +394,10 @@ def test__export_archive_does_not_raise_ExportError_when_CalledProcessError(mock export = Export() with pytest.raises(ExportError, match="CALLED_PROCESS_ERROR"): - export._export_archive("mock.sd-export") + export._run_qrexec_export("mock.sd-export") -def test__export_archive_with_evil_command(mocker): +def test__run_qrexec_export_with_evil_command(mocker): """ Ensure shell command is shell-escaped. """ @@ -444,7 +405,7 @@ def test__export_archive_with_evil_command(mocker): check_output = mocker.patch("subprocess.check_output", return_value=b"ERROR_FILE_NOT_FOUND") with pytest.raises(ExportError, match="UNEXPECTED_RETURN_STATUS"): - export._export_archive("somefile; rm -rf ~") + export._run_qrexec_export("somefile; ls -la ~") check_output.assert_called_once_with( [ @@ -455,24 +416,21 @@ def test__export_archive_with_evil_command(mocker): "/usr/lib/qubes/qopen-in-vm", "--view-only", "--", - "'somefile; rm -rf ~'", + "'somefile; ls -la ~'", ], stderr=-2, ) -def test__export_archive_success_on_empty_return_value(mocker): +def test__run_qrexec_export_error_on_empty_return_value(mocker): """ - Ensure an error is not raised when qrexec call returns empty string, - (success state for `disk`, `print`, `printer-test`). - - When export behaviour changes so that all success states return a status - string, this test will no longer pass and should be rewritten. + Ensure an error is raised when qrexec call returns empty string, """ export = Export() check_output = mocker.patch("subprocess.check_output", return_value=b"") - result = export._export_archive("somefile.sd-export") + with pytest.raises(ExportError, match="UNEXPECTED_RETURN_STATUS"): + export._run_qrexec_export("somefile.sd-export") check_output.assert_called_once_with( [ @@ -487,5 +445,3 @@ def test__export_archive_success_on_empty_return_value(mocker): ], stderr=-2, ) - - assert result is None