Make Dummy isolation provider more realistic

Make the Dummy isolation provider follow the rest of the isolation
providers and perform the second part of the conversion on the host. The
first part of the conversion is just a dummy script that reads a file
from stdin and prints pixels to stdout.
This commit is contained in:
Alex Pyrgiotis 2024-10-08 18:47:50 +03:00
parent 703bb0e42a
commit 6e55e43fef
No known key found for this signature in database
GPG key ID: B6C15EBA0357C9AA
2 changed files with 22 additions and 80 deletions

View file

@ -1,18 +1,25 @@
import logging import logging
import os
import shutil
import subprocess import subprocess
import sys import sys
import time
from typing import Callable, Optional
from ..conversion.common import DangerzoneConverter
from ..document import Document from ..document import Document
from ..util import get_resource_path
from .base import IsolationProvider, terminate_process_group from .base import IsolationProvider, terminate_process_group
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
def dummy_script() -> None:
sys.stdin.buffer.read()
pages = 2
width = height = 9
DangerzoneConverter._write_int(pages)
for page in range(pages):
DangerzoneConverter._write_int(width)
DangerzoneConverter._write_int(height)
DangerzoneConverter._write_bytes(width * height * 3 * b"A")
class Dummy(IsolationProvider): class Dummy(IsolationProvider):
"""Dummy Isolation Provider (FOR TESTING ONLY) """Dummy Isolation Provider (FOR TESTING ONLY)
@ -32,51 +39,15 @@ class Dummy(IsolationProvider):
def install(self) -> bool: def install(self) -> bool:
return True return True
def convert(
self,
document: Document,
ocr_lang: Optional[str],
progress_callback: Optional[Callable] = None,
) -> None:
self.progress_callback = None
log.debug("Dummy converter started:")
log.debug(
f" - document: {os.path.basename(document.input_filename)} ({document.id})"
)
log.debug(f" - ocr : {ocr_lang}")
log.debug("\n(simulating conversion)")
success = True
progress = [
[False, "Converting to PDF using GraphicsMagick", 0.0],
[False, "Separating document into pages", 3.0],
[False, "Converting page 1/1 to pixels", 5.0],
[False, "Converted document to pixels", 50.0],
[False, "Converting page 1/1 from pixels to PDF", 50.0],
[False, "Merging 1 pages into a single PDF", 95.0],
[False, "Compressing PDF", 97.0],
[False, "Safe PDF created", 100.0],
]
for error, text, percentage in progress:
self.print_progress(document, error, text, percentage) # type: ignore [arg-type]
if error:
success = False
if success:
shutil.copy(
get_resource_path("dummy_document.pdf"), document.output_filename
)
document.mark_as_safe()
if document.archive_after_conversion:
document.archive()
def pixels_to_pdf(
self, document: Document, tempdir: str, ocr_lang: Optional[str]
) -> None:
pass
def start_doc_to_pixels_proc(self, document: Document) -> subprocess.Popen: def start_doc_to_pixels_proc(self, document: Document) -> subprocess.Popen:
dummy_cmd = ["python3", "-c", "print('The cake is a lie')"] cmd = [
sys.executable,
"-c",
"from dangerzone.isolation_provider.dummy import dummy_script;"
" dummy_script()",
]
return subprocess.Popen( return subprocess.Popen(
dummy_cmd, cmd,
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=self.proc_stderr, stderr=self.proc_stderr,

View file

@ -1,11 +1,9 @@
import os import os
import subprocess
import pytest import pytest
from pytest_mock import MockerFixture from pytest_mock import MockerFixture
from dangerzone.conversion import errors from dangerzone.conversion import errors
from dangerzone.document import Document
from dangerzone.isolation_provider.base import IsolationProvider from dangerzone.isolation_provider.base import IsolationProvider
from dangerzone.isolation_provider.dummy import Dummy from dangerzone.isolation_provider.dummy import Dummy
@ -16,24 +14,6 @@ if not os.environ.get("DUMMY_CONVERSION", False):
pytest.skip("Dummy conversion is not enabled", allow_module_level=True) pytest.skip("Dummy conversion is not enabled", allow_module_level=True)
class DummyWait(Dummy):
"""Dummy isolation provider that spawns a blocking process."""
def start_doc_to_pixels_proc(self, document: Document) -> subprocess.Popen:
return subprocess.Popen(
["python3"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=True,
)
@pytest.fixture
def provider_wait() -> DummyWait:
return DummyWait()
@pytest.fixture @pytest.fixture
def provider() -> Dummy: def provider() -> Dummy:
return Dummy() return Dummy()
@ -42,21 +22,12 @@ def provider() -> Dummy:
class TestDummyTermination(IsolationProviderTermination): class TestDummyTermination(IsolationProviderTermination):
def test_failed( def test_failed(
self, self,
provider_wait: IsolationProvider, provider: IsolationProvider,
mocker: MockerFixture, mocker: MockerFixture,
) -> None: ) -> None:
mocker.patch.object( mocker.patch.object(
provider_wait, provider,
"get_proc_exception", "get_proc_exception",
return_value=errors.DocFormatUnsupported(), return_value=errors.DocFormatUnsupported(),
) )
super().test_failed(provider_wait, mocker) super().test_failed(provider, mocker)
def test_linger_unkillable(
self,
provider_wait: IsolationProvider,
mocker: MockerFixture,
) -> None:
# We have to spawn a blocking process here, else we can't imitate an
# "unkillable" process.
super().test_linger_unkillable(provider_wait, mocker)