dangerzone/tests/isolation_provider/base.py
deeplow 0a099540c8
Stream pages in containers: merge isolation providers
Merge Qubes and Containers isolation providers core code into the class
parent IsolationProviders abstract class.

This is done by streaming pages in containers for exclusively in first
conversion process. The commit is rather large due to the multiple
interdependencies of the code, making it difficult to split into various
commits.

The main conversion method (_convert) now in the superclass simply calls
two methods:
  - doc_to_pixels()
  - pixels_to_pdf()

Critically, doc_to_pixels is implemented in the superclass, diverging
only in a specialized method called "start_doc_to_pixels_proc()". This
method obtains the process responsible that communicates with the
isolation provider (container / disp VM) via `podman/docker` and qrexec
on Containers and Qubes respectively.

Known regressions:
  - progress reports stopped working on containers

Fixes #443
2024-02-06 19:42:33 +00:00

71 lines
2.7 KiB
Python

import os
import pytest
from colorama import Style
from pytest_mock import MockerFixture
from dangerzone.conversion import errors
from dangerzone.document import Document
from dangerzone.isolation_provider import base
from dangerzone.isolation_provider.qubes import running_on_qubes
from .. import pdf_11k_pages, sanitized_text, uncommon_text
@pytest.mark.skipif(
os.environ.get("DUMMY_CONVERSION", False),
reason="dummy conversions not supported",
)
@pytest.mark.skipif(not running_on_qubes(), reason="Not on a Qubes system")
class IsolationProviderTest:
def test_print_progress(
self,
provider: base.IsolationProvider,
uncommon_text: str,
sanitized_text: str,
mocker: MockerFixture,
) -> None:
"""Test that the print_progress() method of our isolation providers sanitizes text.
Iterate our isolation providers and make sure that their print_progress() methods
sanitizes the provided text, before passing it to the logging functions and other
callbacks.
"""
d = Document()
provider.progress_callback = mocker.MagicMock()
log_info_spy = mocker.spy(base.log, "info")
log_error_spy = mocker.spy(base.log, "error")
_print_progress_spy = mocker.spy(provider, "_print_progress")
for error, untrusted_text, sanitized_text in [
(True, "normal text", "UNTRUSTED> normal text"),
(False, "normal text", "UNTRUSTED> normal text"),
(True, uncommon_text, "UNTRUSTED> " + sanitized_text),
(False, uncommon_text, "UNTRUSTED> " + sanitized_text),
]:
log_info_spy.reset_mock()
log_error_spy.reset_mock()
provider.print_progress(d, error, untrusted_text, 0)
provider.progress_callback.assert_called_with(error, sanitized_text, 0) # type: ignore [union-attr]
_print_progress_spy.assert_called_with(d, error, sanitized_text, 0)
if error:
assert log_error_spy.call_args[0][0].endswith(
sanitized_text + Style.RESET_ALL
)
log_info_spy.assert_not_called()
else:
assert log_info_spy.call_args[0][0].endswith(sanitized_text)
log_error_spy.assert_not_called()
def test_max_pages_received(
self,
pdf_11k_pages: str,
provider: base.IsolationProvider,
mocker: MockerFixture,
) -> None:
provider.progress_callback = mocker.MagicMock()
doc = Document(pdf_11k_pages)
with pytest.raises(errors.MaxPagesException):
success = provider._convert(doc, ocr_lang=None)
assert not success