diff --git a/.github/workflows/test.yml b/.github/workflows/ci.yml similarity index 100% rename from .github/workflows/test.yml rename to .github/workflows/ci.yml diff --git a/pyproject.toml b/pyproject.toml index 0c8ef70..ffc265a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,15 +49,14 @@ exclude = [ [tool.hatch.envs.default] dependencies = [ - "coverage[toml] >= 7.0", - "pytest >= 7.0", + "coverage[toml] >= 7.3", + "pytest >= 7.4", "pytest-sugar", - "pytest-cov", - "pytest-xdist", "pytest-httpx ~= 0.26; python_version >= '3.9'", "pytest-httpx ~= 0.22; python_version < '3.9'", - "pikepdf" - + "pikepdf", + "python-magic", + "brotli", ] [tool.hatch.envs.default.scripts] @@ -76,7 +75,8 @@ cov = [ "cov-clear", "test-cov", "cov-report", - "cov-json" + "cov-json", + "cov-html" ] pip-list = "pip list" diff --git a/src/gotenberg_client/base.py b/src/gotenberg_client/base.py new file mode 100644 index 0000000..a867ddc --- /dev/null +++ b/src/gotenberg_client/base.py @@ -0,0 +1,95 @@ +import logging +from contextlib import ExitStack +from importlib.util import find_spec +from pathlib import Path +from types import TracebackType +from typing import Dict +from typing import Optional +from typing import Type + +from httpx import Client +from httpx import Response +from httpx._types import RequestFiles + +from gotenberg_client.pdf_format import PdfAFormatOptions + +logger = logging.getLogger(__name__) + + +class BaseRoute: + def __init__(self, client: Client, api_route: str) -> None: + self._client = client + self._route = api_route + self._stack = ExitStack() + self._form_data: Dict[str, str] = {} + self._file_map: Dict[str, Path] = {} + + def __enter__(self) -> "BaseRoute": + self.reset() + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[TracebackType], + ) -> None: + self.reset() + + def reset(self) -> None: + self._stack.close() + self._form_data.clear() + self._file_map.clear() + + def run(self) -> Response: + resp = self._client.post(url=self._route, data=self._form_data, files=self.get_files()) + resp.raise_for_status() + return resp + + def get_files(self) -> RequestFiles: + files = {} + for filename in self._file_map: + file_path = self._file_map[filename] + # Gotenberg requires these to have the specific name + filepath_name = filename if filename in {"index.html", "header.html", "footer.html"} else file_path.name + + mime_type = guess_mime_type(file_path) + if mime_type is not None: + files.update( + {filepath_name: (filepath_name, self._stack.enter_context(file_path.open("rb")), mime_type)}, + ) + else: # pragma: no cover + files.update({filepath_name: (filepath_name, self._stack.enter_context(file_path.open("rb")))}) # type: ignore + return files + + def _add_file_map(self, filepath: Path, name: Optional[str] = None) -> None: + if name is None: + name = filepath.name + if name in self._file_map: # pragma: no cover + logger.warning(f"{name} has already been provided, overwriting anyway") + self._file_map[name] = filepath + + def pdf_format(self, pdf_format: PdfAFormatOptions) -> "BaseRoute": + self._form_data.update(pdf_format.to_form()) + return self + + +class BaseApi: + def __init__(self, client: Client) -> None: + self._client = client + + +def guess_mime_type_stdlib(url: Path) -> Optional[str]: + import mimetypes + + mime_type, _ = mimetypes.guess_type(url) + return mime_type + + +def guess_mime_type_magic(url: Path) -> Optional[str]: + import magic # type: ignore + + return magic.from_file(url, mime=True) # type: ignore + + +guess_mime_type = guess_mime_type_magic if find_spec("magic") is not None else guess_mime_type_stdlib diff --git a/src/gotenberg_client/client.py b/src/gotenberg_client/client.py index d39496d..b206d4b 100644 --- a/src/gotenberg_client/client.py +++ b/src/gotenberg_client/client.py @@ -1,4 +1,5 @@ import logging +from importlib.util import find_spec from types import TracebackType from typing import Dict from typing import Optional @@ -6,7 +7,11 @@ from httpx import Client -from gotenberg_client.convert.chromium import ChromiumRoutes +from gotenberg_client.convert.chromium import ChromiumApi +from gotenberg_client.convert.libre_office import LibreOfficeApi +from gotenberg_client.convert.pdfa import PdfAApi +from gotenberg_client.health import HealthCheckApi +from gotenberg_client.merge import MergeApi class GotenbergClient: @@ -16,7 +21,6 @@ def __init__( gotenerg_url: str, timeout: float = 30.0, log_level: int = logging.ERROR, - compress: bool = False, http2: bool = True, ): # Configure the client @@ -26,15 +30,18 @@ def __init__( logging.getLogger("httpx").setLevel(log_level) logging.getLogger("httpcore").setLevel(log_level) - # Only JSON responses supported - self._client.headers.update({"Accept": "application/json"}) - - if compress: - # TODO Brotli? - self._client.headers.update({"Accept-Encoding": "gzip"}) + # TODO Brotli? + if find_spec("brotli") is not None: + self._client.headers.update({"Accept-Encoding": "gzip,deflate,br"}) + else: + self._client.headers.update({"Accept-Encoding": "gzip,deflate"}) # Add the resources - self.chromium = ChromiumRoutes(self._client) + self.chromium = ChromiumApi(self._client) + self.libre_office = LibreOfficeApi(self._client) + self.pdf_a = PdfAApi(self._client) + self.merge = MergeApi(self._client) + self.health = HealthCheckApi(self._client) # TODO def add_headers(self, header: Dict[str, str]) -> None: # pragma: no cover diff --git a/src/gotenberg_client/convert/chromium.py b/src/gotenberg_client/convert/chromium.py index 2aa23ac..e5f1178 100644 --- a/src/gotenberg_client/convert/chromium.py +++ b/src/gotenberg_client/convert/chromium.py @@ -1,30 +1,29 @@ import dataclasses import enum import json -from contextlib import ExitStack +import logging from pathlib import Path +from typing import Dict from typing import Final +from typing import List from typing import Optional -from urllib.parse import quote - -from httpx import Client -from httpx import Response +from typing import Union +from gotenberg_client.base import BaseApi from gotenberg_client.convert.common import FORCE_MULTIPART -from gotenberg_client.convert.common import PageOrientationOptions -from gotenberg_client.convert.common import PageRangeType -from gotenberg_client.convert.common import PdfAFormatOptions -from gotenberg_client.convert.common import guess_mime_type -from gotenberg_client.convert.common import optional_page_ranges_to_form +from gotenberg_client.convert.common import ConvertBaseRoute +from gotenberg_client.convert.common import ForceMultipartDict from gotenberg_client.convert.common import optional_to_form +logger = logging.getLogger() + @dataclasses.dataclass class PageSize: - width: float | int | None = None - height: float | int | None = None + width: Optional[Union[float, int]] = None + height: Optional[Union[float, int]] = None - def to_form(self) -> dict[str, str]: + def to_form(self) -> Dict[str, str]: data = optional_to_form(self.width, "paperWidth") data.update(optional_to_form(self.height, "paperHeight")) return data @@ -46,12 +45,12 @@ def to_form(self) -> dict[str, str]: @dataclasses.dataclass class Margin: - top: float | int | None = None - bottom: float | int | None = None - left: float | int | None = None - right: float | int | None = None + top: Optional[Union[float, int]] = None + bottom: Optional[Union[float, int]] = None + left: Optional[Union[float, int]] = None + right: Optional[Union[float, int]] = None - def to_form(self) -> dict[str, str]: + def to_form(self) -> Dict[str, str]: data = optional_to_form(self.top, "marginTop") data.update(optional_to_form(self.bottom, "marginBottom")) data.update(optional_to_form(self.left, "marginLeft")) @@ -64,265 +63,147 @@ def to_form(self) -> dict[str, str]: Word_Narrow_Margins: Final = Margin(top=0.5, bottom=0.5, left=0.5, right=0.5) -@dataclasses.dataclass -class RenderControl: - delay: int | float | None = None - expression: str | None = None +@enum.unique +class EmulatedMediaTypeChoices(str, enum.Enum): + Print = enum.auto() + Screen = enum.auto() + + def to_form(self) -> Dict[str, str]: + if self.value == EmulatedMediaTypeChoices.Print.value: + return {"emulatedMediaType": "print"} + elif self.value == EmulatedMediaTypeChoices.Screen.value: + return {"emulatedMediaType": "screen"} + else: # pragma: no cover + raise NotImplementedError(self.value) - def to_form(self) -> dict[str, str]: - data = optional_to_form(self.delay, "waitDelay") - data.update(optional_to_form(self.expression, "waitForExpression")) - return data +class ChromiumBaseRoute(ConvertBaseRoute): + def header(self, header: Path) -> "ChromiumBaseRoute": + self._add_file_map(header, "header.html") + return self -@dataclasses.dataclass -class HttpOptions: - user_agent: str | None = None - headers: dict[str, str] | None = None - - def to_form(self) -> dict[str, str]: - data = optional_to_form(self.user_agent, "userAgent") - if self.headers is not None: - json_str = json.dumps(self.headers) - # TODO: Need to check this - data.update({"extraHttpHeaders": quote(json_str)}) - return data + def footer(self, footer: Path) -> "ChromiumBaseRoute": + self._add_file_map(footer, "footer.html") + return self + def resource(self, resource: Path) -> "ChromiumBaseRoute": + self._add_file_map(resource) + return self -@enum.unique -class EmulatedMediaTypeOptions(str, enum.Enum): - Print = "print" - Screen = "screen" + def resources(self, resources: List[Path]) -> "ChromiumBaseRoute": + for x in resources: + self.resource(x) + return self + + def size(self, size: PageSize) -> "ChromiumBaseRoute": + self._form_data.update(size.to_form()) + return self + + page_size = size + + def margins(self, margins: Margin) -> "ChromiumBaseRoute": + self._form_data.update(margins.to_form()) + return self + + def prefer_css_page_size(self) -> "ChromiumBaseRoute": + self._form_data.update({"preferCssPageSize": "true"}) + return self + + def prefer_set_page_size(self) -> "ChromiumBaseRoute": + self._form_data.update({"preferCssPageSize": "false"}) + return self + + def background_graphics(self) -> "ChromiumBaseRoute": + self._form_data.update({"printBackground": "true"}) + return self + + def no_background_graphics(self) -> "ChromiumBaseRoute": + self._form_data.update({"printBackground": "false"}) + return self + + def hide_background(self) -> "ChromiumBaseRoute": + self._form_data.update({"omitBackground": "true"}) + return self + + def show_background(self) -> "ChromiumBaseRoute": + self._form_data.update({"omitBackground": "false"}) + return self + + def scale(self, scale: Union[int, float]) -> "ChromiumBaseRoute": + self._form_data.update({"scale": str(scale)}) + return self + + def render_wait(self, wait: Union[int, float]) -> "ChromiumBaseRoute": + self._form_data.update({"waitDelay": str(wait)}) + return self - def to_form(self) -> dict[str, str]: - return {"emulatedMediaType": self.value} + def render_expr(self, expr: str) -> "ChromiumBaseRoute": + self._form_data.update({"waitForExpression": expr}) + return self + def media_type(self, media_type: EmulatedMediaTypeChoices) -> "ChromiumBaseRoute": + self._form_data.update(media_type.to_form()) + return self -class ChromiumRoutes: + def user_agent(self, agent: str) -> "ChromiumBaseRoute": + self._form_data.update({"userAgent": agent}) + return self + + def headers(self, headers: Dict[str, str]) -> "ChromiumBaseRoute": + json_str = json.dumps(headers) + # TODO: Need to check this + self._form_data.update({"extraHttpHeaders": json_str}) + return self + + def fail_on_exceptions(self) -> "ChromiumBaseRoute": + self._form_data.update({"failOnConsoleExceptions": "true"}) + return self + + def dont_fail_on_exceptions(self) -> "ChromiumBaseRoute": + self._form_data.update({"failOnConsoleExceptions": "false"}) + return self + + +class _FileBasedRoute(ChromiumBaseRoute): + def index(self, index: Path) -> "_FileBasedRoute": + self._add_file_map(index, "index.html") + return self + + +class HtmlRoute(_FileBasedRoute): + pass + + +class UrlRoute(ChromiumBaseRoute): + def url(self, url: str) -> "UrlRoute": + self._form_data["url"] = url + return self + + def get_files(self) -> ForceMultipartDict: + return FORCE_MULTIPART + + +class MarkdownRoute(_FileBasedRoute): + def markdown_file(self, markdown_file: Path) -> "MarkdownRoute": + self._add_file_map(markdown_file) + return self + + def markdown_files(self, markdown_files: List[Path]) -> "MarkdownRoute": + for x in markdown_files: + self.markdown_file(x) + return self + + +class ChromiumApi(BaseApi): _URL_CONVERT_ENDPOINT = "/forms/chromium/convert/url" _HTML_CONVERT_ENDPOINT = "/forms/chromium/convert/html" _MARKDOWN_CONVERT_ENDPOINT = "/forms/chromium/convert/markdown" - def __init__(self, client: Client) -> None: - self._client = client - - def convert_url( - self, - url_to_convert: str, - *, - page_size: Optional[PageSize] = None, - margins: Optional[Margin] = None, - prefer_css_page_size: Optional[bool] = None, - print_background: Optional[bool] = None, - omit_background: Optional[bool] = None, - orientation: Optional[PageOrientationOptions] = None, - scale: Optional[int | float] = None, - page_ranges: Optional[PageRangeType] = None, - header: Optional[Path] = None, # noqa: ARG002 - footer: Optional[Path] = None, # noqa: ARG002 - render_control: Optional[RenderControl] = None, - media_type_emulation: Optional[EmulatedMediaTypeOptions] = None, - http_control: Optional[HttpOptions] = None, - fail_on_console_exceptions: Optional[bool] = None, - pdf_a_output: Optional[PdfAFormatOptions] = None, - ) -> Response: - data = self._build_common_options_form_data( - page_size=page_size, - margins=margins, - prefer_css_page_size=prefer_css_page_size, - print_background=print_background, - omit_background=omit_background, - orientation=orientation, - scale=scale, - page_ranges=page_ranges, - render_control=render_control, - media_type_emulation=media_type_emulation, - http_control=http_control, - fail_on_console_exceptions=fail_on_console_exceptions, - pdf_a_output=pdf_a_output, - ) - data["url"] = url_to_convert - resp = self._client.post(url=self._URL_CONVERT_ENDPOINT, data=data, files=FORCE_MULTIPART) - resp.raise_for_status() - return resp - - def convert_html( # type: ignore - self, - index_file: Path, - *, - additional_files: Optional[list[Path]] = None, - page_size: Optional[PageSize] = None, - margins: Optional[Margin] = None, - prefer_css_page_size: Optional[bool] = None, - print_background: Optional[bool] = None, - omit_background: Optional[bool] = None, - orientation: Optional[PageOrientationOptions] = None, - scale: Optional[int | float] = None, - page_ranges: Optional[PageRangeType] = None, - header: Optional[Path] = None, - footer: Optional[Path] = None, - render_control: Optional[RenderControl] = None, - media_type_emulation: Optional[EmulatedMediaTypeOptions] = None, - http_control: Optional[HttpOptions] = None, - fail_on_console_exceptions: Optional[bool] = None, - pdf_a_output: Optional[PdfAFormatOptions] = None, - ) -> Response: - data = self._build_common_options_form_data( - page_size=page_size, - margins=margins, - prefer_css_page_size=prefer_css_page_size, - print_background=print_background, - omit_background=omit_background, - orientation=orientation, - scale=scale, - page_ranges=page_ranges, - render_control=render_control, - media_type_emulation=media_type_emulation, - http_control=http_control, - fail_on_console_exceptions=fail_on_console_exceptions, - pdf_a_output=pdf_a_output, - ) - # Open up all the file handles - files = {} - with ExitStack() as stack: - files.update({"index.html": ("index.html", stack.enter_context(index_file.open("rb")), "application/html")}) - if header is not None: - files.update( - {"header.html": ("header.html", stack.enter_context(header.open("rb")), "application/html")}, - ) - if footer is not None: - files.update( - {"footer.html": ("footer.html", stack.enter_context(footer.open("rb")), "application/html")}, - ) - if additional_files is not None: - for file in additional_files: - mime_type = guess_mime_type(file) - if mime_type is not None: - files.update({file.name: (file.name, stack.enter_context(file.open("rb")), mime_type)}) - else: - files.update( - { - file.name: ( # type: ignore - file.name, - stack.enter_context(file.open("rb")), - ), - }, - ) - - resp = self._client.post(url=self._HTML_CONVERT_ENDPOINT, data=data, files=files) - resp.raise_for_status() - return resp - - def convert_markdown( # type: ignore - self, - index_file: Path, - markdown_files: list[Path], - *, - additional_files: Optional[list[Path]] = None, - page_size: Optional[PageSize] = None, - margins: Optional[Margin] = None, - prefer_css_page_size: Optional[bool] = None, - print_background: Optional[bool] = None, - omit_background: Optional[bool] = None, - orientation: Optional[PageOrientationOptions] = None, - scale: Optional[int | float] = None, - page_ranges: Optional[PageRangeType] = None, - header: Optional[Path] = None, - footer: Optional[Path] = None, - render_control: Optional[RenderControl] = None, - media_type_emulation: Optional[EmulatedMediaTypeOptions] = None, - http_control: Optional[HttpOptions] = None, - fail_on_console_exceptions: Optional[bool] = None, - pdf_a_output: Optional[PdfAFormatOptions] = None, - ) -> Response: - data = self._build_common_options_form_data( - page_size=page_size, - margins=margins, - prefer_css_page_size=prefer_css_page_size, - print_background=print_background, - omit_background=omit_background, - orientation=orientation, - scale=scale, - page_ranges=page_ranges, - render_control=render_control, - media_type_emulation=media_type_emulation, - http_control=http_control, - fail_on_console_exceptions=fail_on_console_exceptions, - pdf_a_output=pdf_a_output, - ) - # Open up all the file handles - files = {} - with ExitStack() as stack: - files.update({"index.html": ("index.html", stack.enter_context(index_file.open("rb")), "application/html")}) - if header is not None: - files.update( - {"header.html": ("header.html", stack.enter_context(header.open("rb")), "application/html")}, - ) - if footer is not None: - files.update( - {"footer.html": ("footer.html", stack.enter_context(footer.open("rb")), "application/html")}, - ) - # Including the markdown files - # Us the index of the file to ensure the ordering - for file in markdown_files: - files.update({file.name: (file.name, stack.enter_context(file.open("rb")), "text/markdown")}) - if additional_files is not None: - for file in additional_files: - mime_type = guess_mime_type(file) - if mime_type is not None: - files.update({file.name: (file.name, stack.enter_context(file.open("rb")), mime_type)}) - else: - files.update( - { - file.name: ( # type: ignore - file.name, - stack.enter_context(file.open("rb")), - ), - }, - ) - - resp = self._client.post(url=self._MARKDOWN_CONVERT_ENDPOINT, data=data, files=files) - resp.raise_for_status() - return resp - - @staticmethod - def _build_common_options_form_data( - *, - page_size: Optional[PageSize] = None, - margins: Optional[Margin] = None, - prefer_css_page_size: Optional[bool] = None, - print_background: Optional[bool] = None, - omit_background: Optional[bool] = None, - orientation: Optional[PageOrientationOptions] = None, - scale: Optional[int | float] = None, - page_ranges: Optional[PageRangeType] = None, - render_control: Optional[RenderControl] = None, - media_type_emulation: Optional[EmulatedMediaTypeOptions] = None, - http_control: Optional[HttpOptions] = None, - fail_on_console_exceptions: Optional[bool] = None, - pdf_a_output: Optional[PdfAFormatOptions] = None, - ) -> dict[str, str]: - data = {} - if page_size is not None: - data.update(page_size.to_form()) - if margins is not None: - data.update(margins.to_form()) - data.update(optional_to_form(prefer_css_page_size, "preferCssPageSize")) - data.update(optional_to_form(print_background, "printBackground")) - data.update(optional_to_form(omit_background, "omitBackground")) - if orientation is not None: - data.update(orientation.to_form()) - data.update(optional_to_form(scale, "scale")) - data.update(optional_page_ranges_to_form(page_ranges, "nativePageRanges")) - # TODO page ranges - # TODO header & footer - if render_control is not None: - data.update(render_control.to_form()) - if media_type_emulation is not None: - data.update(media_type_emulation.to_form()) - if http_control is not None: - data.update(http_control.to_form()) - data.update(optional_to_form(fail_on_console_exceptions, "failOnConsoleExceptions")) - if pdf_a_output is not None: - data.update(pdf_a_output.to_form()) - return data + def html_to_pdf(self) -> HtmlRoute: + return HtmlRoute(self._client, self._HTML_CONVERT_ENDPOINT) + + def url_to_pdf(self) -> UrlRoute: + return UrlRoute(self._client, self._URL_CONVERT_ENDPOINT) + + def markdown_to_pdf(self) -> MarkdownRoute: + return MarkdownRoute(self._client, self._MARKDOWN_CONVERT_ENDPOINT) diff --git a/src/gotenberg_client/convert/common.py b/src/gotenberg_client/convert/common.py index 6746fb0..39db2b7 100644 --- a/src/gotenberg_client/convert/common.py +++ b/src/gotenberg_client/convert/common.py @@ -1,12 +1,18 @@ import enum +import logging from functools import lru_cache -from pathlib import Path +from typing import Dict from typing import Final from typing import Optional +from typing import Union + +from gotenberg_client.base import BaseRoute + +logger = logging.getLogger() # See https://github.com/psf/requests/issues/1081#issuecomment-428504128 -class ForceMultipartDict(dict): +class ForceMultipartDict(Dict): def __bool__(self) -> bool: return True @@ -19,7 +25,7 @@ class PageOrientationOptions(enum.Enum): Landscape = enum.auto() Potrait = enum.auto() - def to_form(self) -> dict[str, str]: + def to_form(self) -> Dict[str, str]: if self.value == PageOrientationOptions.Landscape.value: return {"landscape": "true"} elif self.value == PageOrientationOptions.Potrait.value: @@ -28,28 +34,18 @@ def to_form(self) -> dict[str, str]: raise NotImplementedError(self.value) -@enum.unique -class PdfAFormatOptions(enum.Enum): - A1a = enum.auto() - A2a = enum.auto() - A3b = enum.auto() +class ConvertBaseRoute(BaseRoute): + def orient(self, orient: PageOrientationOptions) -> "BaseRoute": + self._form_data.update(orient.to_form()) + return self - def to_form(self) -> dict[str, str]: - if self.value == PdfAFormatOptions.A1a.value: - return {"pdfFormat": "PDF/A-1a"} - elif self.value == PdfAFormatOptions.A2a.value: - return {"pdfFormat": "PDF/A-2b"} - elif self.value == PdfAFormatOptions.A3b.value: - return {"pdfFormat": "PDF/A-3b"} - else: # pragma: no cover - raise NotImplementedError(self.value) - - -PageRangeType = list[list[int]] | None + def page_ranges(self, ranges: str) -> "BaseRoute": + self._form_data.update({"nativePageRanges": ranges}) + return self @lru_cache # type: ignore -def optional_to_form(value: Optional[bool | int | float | str], name: str) -> dict[str, str]: +def optional_to_form(value: Optional[Union[bool, int, float, str]], name: str) -> Dict[str, str]: """ Quick helper to convert an optional type into a form data field with the given name or no changes if the value is None @@ -58,29 +54,3 @@ def optional_to_form(value: Optional[bool | int | float | str], name: str) -> di return {} else: return {name: str(value).lower()} - - -def optional_page_ranges_to_form(value: Optional[PageRangeType], name: str) -> dict[str, str]: - """ - Converts a list of lists of pages into the formatted strings Gotenberg expects - """ - if value is None: - return {} - else: - combined = [] - for range_value in value: - combined += range_value - combined.sort() - return {name: ",".join(str(x) for x in combined)} - - -def guess_mime_type(url: Path) -> Optional[str]: - try: - import magic # type: ignore - - return magic.from_file(url, mime=True) # type: ignore - except ImportError: - import mimetypes - - mime_type, _ = mimetypes.guess_type(url) - return mime_type diff --git a/src/gotenberg_client/convert/libre_office.py b/src/gotenberg_client/convert/libre_office.py index a269468..d98bddb 100644 --- a/src/gotenberg_client/convert/libre_office.py +++ b/src/gotenberg_client/convert/libre_office.py @@ -1,26 +1,31 @@ -import dataclasses +from pathlib import Path +from typing import List -from gotenberg_client.convert.common import PageOrientationOptions -from gotenberg_client.convert.common import PageRangeType -from gotenberg_client.convert.common import PdfAFormatOptions +from gotenberg_client.base import BaseApi +from gotenberg_client.convert.common import ConvertBaseRoute -@dataclasses.dataclass -class PageProperties: - """ - Defines possible page properties for the Libre Office routes, along - with the default values. +class LibreOfficeConvertRoute(ConvertBaseRoute): + def convert(self, file_path: Path) -> "LibreOfficeConvertRoute": + self._add_file_map(file_path) + return self - Documentation: - - https://gotenberg.dev/docs/routes#page-properties-libreoffice - """ + def convert_files(self, file_paths: List[Path]) -> "LibreOfficeConvertRoute": + for x in file_paths: + self.convert(x) + return self - orientation: PageOrientationOptions = PageOrientationOptions.Potrait - pages: PageRangeType = None + def merge(self) -> "LibreOfficeConvertRoute": + self._form_data.update({"merge": "true"}) + return self + def no_merge(self) -> "LibreOfficeConvertRoute": + self._form_data.update({"merge": "false"}) + return self -@dataclasses.dataclass -class LibreOfficeRouteOptions: - page_properties: PageProperties | None = None - merge: bool | None = None - pdf_a_format: PdfAFormatOptions | None = None + +class LibreOfficeApi(BaseApi): + _CONVERT_ENDPOINT = "/forms/libreoffice/convert" + + def to_pdf(self) -> LibreOfficeConvertRoute: + return LibreOfficeConvertRoute(self._client, self._CONVERT_ENDPOINT) diff --git a/src/gotenberg_client/convert/pdfa.py b/src/gotenberg_client/convert/pdfa.py index e69de29..3bfa88a 100644 --- a/src/gotenberg_client/convert/pdfa.py +++ b/src/gotenberg_client/convert/pdfa.py @@ -0,0 +1,23 @@ +from pathlib import Path +from typing import List + +from gotenberg_client.base import BaseApi +from gotenberg_client.convert.common import ConvertBaseRoute + + +class PdfAConvertRoute(ConvertBaseRoute): + def convert(self, file_path: Path) -> "PdfAConvertRoute": + self._add_file_map(file_path) + return self + + def convert_files(self, file_paths: List[Path]) -> "PdfAConvertRoute": + for x in file_paths: + self.convert(x) + return self + + +class PdfAApi(BaseApi): + _CONVERT_ENDPOINT = "/forms/pdfengines/convert" + + def to_pdfa(self) -> PdfAConvertRoute: + return PdfAConvertRoute(self._client, self._CONVERT_ENDPOINT) diff --git a/src/gotenberg_client/health.py b/src/gotenberg_client/health.py index e69de29..cb852cb 100644 --- a/src/gotenberg_client/health.py +++ b/src/gotenberg_client/health.py @@ -0,0 +1,116 @@ +import dataclasses +import datetime +import enum +import re +from typing import Optional +from typing import TypedDict +from typing import no_type_check + +from gotenberg_client.base import BaseApi + +_TIME_RE = re.compile( + r"(?P\d{4})-" + r"(?P\d{2})-" + r"(?P\d{2})" + r"[ tT]" + r"(?P\d{2}):" + r"(?P\d{2}):" + r"(?P\d{2})" + r"(?P\.\d+)?" + r"(?P[zZ]|[+-]\d{2}:\d{2})?", +) + + +class _ModuleStatusType(TypedDict): + status: str + timestamp: str + + +class _AllModulesType(TypedDict): + chromium: _ModuleStatusType + uno: _ModuleStatusType + + +class _HealthCheckApiResponseType(TypedDict): + status: str + details: _AllModulesType + + +class StatusOptions(str, enum.Enum): + Up = "up" + Down = "down" + + +class ModuleOptions(str, enum.Enum): + Chromium = "chromium" + Uno = "uno" + + +@dataclasses.dataclass +class ModuleStatus: + status: StatusOptions + timestamp: datetime.datetime + + +class HealthStatus: + def __init__(self, data: _HealthCheckApiResponseType) -> None: + self.data = data + self.overall = StatusOptions(data["status"]) + + self.chromium: Optional[ModuleStatus] = None + if ModuleOptions.Chromium.value in self.data["details"]: + self.chromium = self._extract_status(ModuleOptions.Chromium) + + self.uno: Optional[ModuleStatus] = None + if ModuleOptions.Uno.value in self.data["details"]: + self.uno = self._extract_status(ModuleOptions.Uno) + + def _extract_status(self, module: ModuleOptions) -> ModuleStatus: + status = StatusOptions(self.data["details"][module.value]["status"]) + + # mypy is quite wrong here, it's clearly marked as a datetime.datetime, not Any + timestamp = self._extract_datetime(self.data["details"][module.value]["timestamp"]) # type: ignore + # Also wrong here + return ModuleStatus(status, timestamp) # type: ignore + + @staticmethod + @no_type_check + def _extract_datetime(timestamp: str) -> datetime.datetime: + m = _TIME_RE.match(timestamp) + if not m: + msg = f"Unable to parse {timestamp}" + raise ValueError(msg) + + (year, month, day, hour, minute, second, frac_sec, timezone_str) = m.groups() + + microseconds = int(float(frac_sec) * 1000000.0) if frac_sec is not None else 0 + tzinfo = None + if timezone_str is not None: + if timezone_str.lower() == "z": + tzinfo = datetime.timezone.utc + else: + multi = -1 if timezone_str[0:1] == "-" else 1 + hours = int(timezone_str[1:3]) + minutes = int(timezone_str[4:]) + delta = datetime.timedelta(hours=hours, minutes=minutes) * multi + tzinfo = datetime.timezone(delta) + + return datetime.datetime( + year=int(year), + month=int(month), + day=int(day), + hour=int(hour), + minute=int(minute), + second=int(second), + microsecond=microseconds, + tzinfo=tzinfo, + ) + + +class HealthCheckApi(BaseApi): + _HEALTH_ENDPOINT = "/health" + + def health(self) -> HealthStatus: + resp = self._client.get(self._HEALTH_ENDPOINT, headers={"Accept": "application/json"}) + json_data: _HealthCheckApiResponseType = resp.raise_for_status().json() + return HealthStatus(json_data) diff --git a/src/gotenberg_client/merge.py b/src/gotenberg_client/merge.py index e69de29..50fa431 100644 --- a/src/gotenberg_client/merge.py +++ b/src/gotenberg_client/merge.py @@ -0,0 +1,20 @@ +from pathlib import Path +from typing import List + +from gotenberg_client.base import BaseApi +from gotenberg_client.base import BaseRoute + + +class MergeRoute(BaseRoute): + def merge(self, files: List[Path]) -> "MergeRoute": + for idx, filepath in enumerate(files): + # Include index to enforce ordering + self._add_file_map(filepath, f"{idx}_{filepath.name}") + return self + + +class MergeApi(BaseApi): + _MERGE_ENDPOINT = "/forms/pdfengines/merge" + + def merge(self) -> MergeRoute: + return MergeRoute(self._client, self._MERGE_ENDPOINT) diff --git a/src/gotenberg_client/pdf_format.py b/src/gotenberg_client/pdf_format.py new file mode 100644 index 0000000..db11bc9 --- /dev/null +++ b/src/gotenberg_client/pdf_format.py @@ -0,0 +1,19 @@ +import enum +from typing import Dict + + +@enum.unique +class PdfAFormatOptions(enum.Enum): + A1a = enum.auto() + A2b = enum.auto() + A3b = enum.auto() + + def to_form(self) -> Dict[str, str]: + if self.value == PdfAFormatOptions.A1a.value: + return {"pdfFormat": "PDF/A-1a"} + elif self.value == PdfAFormatOptions.A2b.value: + return {"pdfFormat": "PDF/A-2b"} + elif self.value == PdfAFormatOptions.A3b.value: + return {"pdfFormat": "PDF/A-3b"} + else: # pragma: no cover + raise NotImplementedError(self.value) diff --git a/tests/conftest.py b/tests/conftest.py index 9117dc5..7b2ba4a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ import logging import os +import shutil from pathlib import Path from typing import Final @@ -13,16 +14,11 @@ SAVE_DIR: Final[Path] = Path(__file__).parent.resolve() / "outputs" SAVE_OUTPUTS: Final[bool] = "SAVE_TEST_OUTPUT" in os.environ if SAVE_OUTPUTS: - SAVE_DIR.mkdir(exist_ok=True) + shutil.rmtree(SAVE_DIR, ignore_errors=True) + SAVE_DIR.mkdir() @pytest.fixture() def client() -> GotenbergClient: with GotenbergClient(gotenerg_url=GOTENBERG_URL, log_level=logging.INFO) as client: yield client - - -@pytest.fixture() -def client_compressed() -> GotenbergClient: - with GotenbergClient(gotenerg_url=GOTENBERG_URL, log_level=logging.INFO, compress=True) as client: - yield client diff --git a/tests/samples/sample.docx b/tests/samples/sample.docx new file mode 100755 index 0000000..894c73c Binary files /dev/null and b/tests/samples/sample.docx differ diff --git a/tests/samples/sample.ods b/tests/samples/sample.ods new file mode 100755 index 0000000..abd2760 Binary files /dev/null and b/tests/samples/sample.ods differ diff --git a/tests/samples/sample.odt b/tests/samples/sample.odt new file mode 100755 index 0000000..c613353 Binary files /dev/null and b/tests/samples/sample.odt differ diff --git a/tests/samples/sample.xlsx b/tests/samples/sample.xlsx new file mode 100755 index 0000000..c0d9631 Binary files /dev/null and b/tests/samples/sample.xlsx differ diff --git a/tests/samples/sample1.pdf b/tests/samples/sample1.pdf new file mode 100755 index 0000000..0a0b284 Binary files /dev/null and b/tests/samples/sample1.pdf differ diff --git a/tests/test_convert_chromium_html.py b/tests/test_convert_chromium_html.py index 11d7138..c22e5f7 100644 --- a/tests/test_convert_chromium_html.py +++ b/tests/test_convert_chromium_html.py @@ -2,15 +2,15 @@ from pathlib import Path import pikepdf +import pytest from httpx import codes from pytest_httpx import HTTPXMock from gotenberg_client.client import GotenbergClient from gotenberg_client.convert.chromium import A4 from gotenberg_client.convert.chromium import Margin -from gotenberg_client.convert.chromium import PageOrientationOptions -from gotenberg_client.convert.chromium import RenderControl -from gotenberg_client.convert.common import PdfAFormatOptions +from gotenberg_client.convert.common import PageOrientationOptions +from gotenberg_client.pdf_format import PdfAFormatOptions from tests.conftest import SAMPLE_DIR from tests.conftest import SAVE_DIR from tests.conftest import SAVE_OUTPUTS @@ -21,7 +21,8 @@ class TestConvertChromiumHtmlRoute: def test_basic_convert(self, client: GotenbergClient): test_file = SAMPLE_DIR / "basic.html" - resp = client.chromium.convert_html(test_file) + with client.chromium.html_to_pdf() as route: + resp = route.index(test_file).run() assert resp.status_code == codes.OK assert "Content-Type" in resp.headers @@ -34,7 +35,8 @@ def test_convert_with_header_footer(self, client: GotenbergClient): header_file = SAMPLE_DIR / "header.html" footer_file = SAMPLE_DIR / "footer.html" - resp = client.chromium.convert_html(test_file, header=header_file, footer=footer_file) + with client.chromium.html_to_pdf() as route: + resp = route.index(test_file).header(header_file).footer(footer_file).run() assert resp.status_code == codes.OK assert "Content-Type" in resp.headers @@ -46,7 +48,8 @@ def test_convert_additional_files(self, client: GotenbergClient): font = SAMPLE_DIR / "font.woff" style = SAMPLE_DIR / "style.css" - resp = client.chromium.convert_html(test_file, additional_files=[img, font, style]) + with client.chromium.html_to_pdf() as route: + resp = route.index(test_file).resource(img).resource(font).resource(style).run() assert resp.status_code == codes.OK assert "Content-Type" in resp.headers @@ -55,10 +58,15 @@ def test_convert_additional_files(self, client: GotenbergClient): if SAVE_OUTPUTS: (SAVE_DIR / "test_convert_additional_files.pdf").write_bytes(resp.content) - def test_convert_pdfa_format(self, client: GotenbergClient): + @pytest.mark.parametrize( + ("gt_format", "pike_format"), + [(PdfAFormatOptions.A1a, "1A"), (PdfAFormatOptions.A2b, "2B"), (PdfAFormatOptions.A3b, "3B")], + ) + def test_convert_pdfa_1a_format(self, client: GotenbergClient, gt_format: PdfAFormatOptions, pike_format: str): test_file = SAMPLE_DIR / "basic.html" - resp = client.chromium.convert_html(test_file, pdf_a_output=PdfAFormatOptions.A1a) + with client.chromium.html_to_pdf() as route: + resp = route.index(test_file).pdf_format(gt_format).run() assert resp.status_code == codes.OK assert "Content-Type" in resp.headers @@ -69,7 +77,7 @@ def test_convert_pdfa_format(self, client: GotenbergClient): output.write_bytes(resp.content) with pikepdf.open(output) as pdf: meta = pdf.open_metadata() - assert meta.pdfa_status == "1A" + assert meta.pdfa_status == pike_format class TestConvertChromiumHtmlRouteMocked: @@ -77,7 +85,8 @@ def test_convert_page_size(self, client: GotenbergClient, httpx_mock: HTTPXMock) httpx_mock.add_response(method="POST") test_file = SAMPLE_DIR / "basic.html" - _ = client.chromium.convert_html(test_file, page_size=A4) + with client.chromium.html_to_pdf() as route: + _ = route.index(test_file).size(A4).run() request = httpx_mock.get_request() verify_stream_contains("paperWidth", "8.5", request.stream) @@ -87,7 +96,8 @@ def test_convert_margin(self, client: GotenbergClient, httpx_mock: HTTPXMock): httpx_mock.add_response(method="POST") test_file = SAMPLE_DIR / "basic.html" - _ = client.chromium.convert_html(test_file, margins=Margin(1, 2, 3, 4)) + with client.chromium.html_to_pdf() as route: + _ = route.index(test_file).margins(Margin(1, 2, 3, 4)).run() request = httpx_mock.get_request() verify_stream_contains("marginTop", "1", request.stream) @@ -99,16 +109,31 @@ def test_convert_render_control(self, client: GotenbergClient, httpx_mock: HTTPX httpx_mock.add_response(method="POST") test_file = SAMPLE_DIR / "basic.html" - _ = client.chromium.convert_html(test_file, margins=RenderControl(500.0)) + with client.chromium.html_to_pdf() as route: + _ = route.index(test_file).render_wait(500.0).run() request = httpx_mock.get_request() verify_stream_contains("waitDelay", "500.0", request.stream) - def test_convert_orientation(self, client: GotenbergClient, httpx_mock: HTTPXMock): + @pytest.mark.parametrize( + ("orientation"), + [PageOrientationOptions.Landscape, PageOrientationOptions.Potrait], + ) + def test_convert_orientation( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + orientation: PageOrientationOptions, + ): httpx_mock.add_response(method="POST") test_file = SAMPLE_DIR / "basic.html" - _ = client.chromium.convert_html(test_file, orientation=PageOrientationOptions.Landscape) + with client.chromium.html_to_pdf() as route: + _ = route.index(test_file).orient(orientation).run() request = httpx_mock.get_request() - verify_stream_contains("landscape", "true", request.stream) + verify_stream_contains( + "landscape", + "true" if orientation == PageOrientationOptions.Landscape else "false", + request.stream, + ) diff --git a/tests/test_convert_chromium_markdown.py b/tests/test_convert_chromium_markdown.py index acb6caa..3c45521 100644 --- a/tests/test_convert_chromium_markdown.py +++ b/tests/test_convert_chromium_markdown.py @@ -11,11 +11,8 @@ def test_basic_convert(self, client: GotenbergClient): img = SAMPLE_DIR / "img.gif" font = SAMPLE_DIR / "font.woff" style = SAMPLE_DIR / "style.css" - resp = client.chromium.convert_markdown( - index_file=index, - markdown_files=md_files, - additional_files=[img, font, style], - ) + with client.chromium.markdown_to_pdf() as route: + resp = route.index(index).markdown_files(md_files).resources([img, font]).resource(style).run() assert resp.status_code == codes.OK assert "Content-Type" in resp.headers diff --git a/tests/test_convert_chromium_url.py b/tests/test_convert_chromium_url.py index dee7624..4826fe1 100644 --- a/tests/test_convert_chromium_url.py +++ b/tests/test_convert_chromium_url.py @@ -1,12 +1,241 @@ +import json + +import pytest from httpx import codes +from pytest_httpx import HTTPXMock from gotenberg_client.client import GotenbergClient +from gotenberg_client.convert.chromium import EmulatedMediaTypeChoices +from tests.utils import verify_stream_contains class TestConvertChromiumUrlRoute: def test_basic_convert(self, client: GotenbergClient): - resp = client.chromium.convert_url("https://en.wikipedia.org/wiki/William_Edward_Sanders") + with client.chromium.url_to_pdf() as route: + resp = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").run() assert resp.status_code == codes.OK assert "Content-Type" in resp.headers assert resp.headers["Content-Type"] == "application/pdf" + + @pytest.mark.parametrize( + ("emulation"), + [EmulatedMediaTypeChoices.Screen, EmulatedMediaTypeChoices.Print], + ) + def test_convert_orientation( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + emulation: EmulatedMediaTypeChoices, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").media_type(emulation).run() + + request = httpx_mock.get_request() + verify_stream_contains( + "emulatedMediaType", + "screen" if emulation == EmulatedMediaTypeChoices.Screen else "print", + request.stream, + ) + + @pytest.mark.parametrize( + ("method"), + ["prefer_css_page_size", "prefer_set_page_size"], + ) + def test_convert_css_or_not_size( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + method: str, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders") + getattr(route, method)() + _ = route.run() + + request = httpx_mock.get_request() + verify_stream_contains( + "preferCssPageSize", + "true" if method == "prefer_css_page_size" else "false", + request.stream, + ) + + @pytest.mark.parametrize( + ("method"), + ["background_graphics", "no_background_graphics"], + ) + def test_convert_background_graphics_or_not( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + method: str, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders") + getattr(route, method)() + _ = route.run() + + request = httpx_mock.get_request() + verify_stream_contains( + "printBackground", + "true" if method == "background_graphics" else "false", + request.stream, + ) + + @pytest.mark.parametrize( + ("method"), + ["hide_background", "show_background"], + ) + def test_convert_hide_background_or_not( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + method: str, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders") + getattr(route, method)() + _ = route.run() + + request = httpx_mock.get_request() + verify_stream_contains( + "omitBackground", + "true" if method == "hide_background" else "false", + request.stream, + ) + + @pytest.mark.parametrize( + ("method"), + ["fail_on_exceptions", "dont_fail_on_exceptions"], + ) + def test_convert_fail_exceptions( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + method: str, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders") + getattr(route, method)() + _ = route.run() + + request = httpx_mock.get_request() + verify_stream_contains( + "failOnConsoleExceptions", + "true" if method == "fail_on_exceptions" else "false", + request.stream, + ) + + def test_convert_scale( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").scale(1.5).run() + + request = httpx_mock.get_request() + verify_stream_contains( + "scale", + "1.5", + request.stream, + ) + + def test_convert_page_ranges( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").page_ranges("1-5").run() + + request = httpx_mock.get_request() + verify_stream_contains( + "nativePageRanges", + "1-5", + request.stream, + ) + + def test_convert_url_render_wait( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").render_wait(500).run() + + request = httpx_mock.get_request() + verify_stream_contains( + "waitDelay", + "500", + request.stream, + ) + + def test_convert_url_render_expression( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").render_expr("wait while false;").run() + + request = httpx_mock.get_request() + verify_stream_contains( + "waitForExpression", + "wait while false;", + request.stream, + ) + + def test_convert_url_user_agent( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + ): + httpx_mock.add_response(method="POST") + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").user_agent("Firefox").run() + + request = httpx_mock.get_request() + verify_stream_contains( + "userAgent", + "Firefox", + request.stream, + ) + + def test_convert_url_headers( + self, + client: GotenbergClient, + httpx_mock: HTTPXMock, + ): + httpx_mock.add_response(method="POST") + + headers = {"X-Auth-Token": "Secure"} + + with client.chromium.url_to_pdf() as route: + _ = route.url("https://en.wikipedia.org/wiki/William_Edward_Sanders").headers(headers).run() + + request = httpx_mock.get_request() + verify_stream_contains( + "extraHttpHeaders", + json.dumps(headers), + request.stream, + ) diff --git a/tests/test_convert_libre_office.py b/tests/test_convert_libre_office.py new file mode 100644 index 0000000..e0ba5d9 --- /dev/null +++ b/tests/test_convert_libre_office.py @@ -0,0 +1,127 @@ +import tempfile +from pathlib import Path +from unittest.mock import patch + +import pikepdf +import pytest +from httpx import codes + +from gotenberg_client.base import guess_mime_type_stdlib +from gotenberg_client.client import GotenbergClient +from gotenberg_client.pdf_format import PdfAFormatOptions +from tests.conftest import SAMPLE_DIR +from tests.conftest import SAVE_DIR +from tests.conftest import SAVE_OUTPUTS + + +class TestLibreOfficeConvert: + def test_libre_office_convert_docx_format(self, client: GotenbergClient): + test_file = SAMPLE_DIR / "sample.docx" + with client.libre_office.to_pdf() as route: + resp = route.convert(test_file).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_docx_format.pdf").write_bytes(resp.content) + + def test_libre_office_convert_odt_format(self, client: GotenbergClient): + test_file = SAMPLE_DIR / "sample.odt" + with client.libre_office.to_pdf() as route: + resp = route.convert(test_file).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_odt_format.pdf").write_bytes(resp.content) + + def test_libre_office_convert_xlsx_format(self, client: GotenbergClient): + test_file = SAMPLE_DIR / "sample.xlsx" + with client.libre_office.to_pdf() as route: + resp = route.convert(test_file).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_xlsx_format.pdf").write_bytes(resp.content) + + def test_libre_office_convert_ods_format(self, client: GotenbergClient): + test_file = SAMPLE_DIR / "sample.ods" + with client.libre_office.to_pdf() as route: + resp = route.convert(test_file).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_ods_format.pdf").write_bytes(resp.content) + + def test_libre_office_convert_multiples_format(self, client: GotenbergClient): + with client.libre_office.to_pdf() as route: + resp = route.convert_files([SAMPLE_DIR / "sample.docx", SAMPLE_DIR / "sample.odt"]).no_merge().run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/zip" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_multiples_format.zip").write_bytes(resp.content) + + def test_libre_office_convert_multiples_format_merged(self, client: GotenbergClient): + with client.libre_office.to_pdf() as route: + resp = route.convert_files([SAMPLE_DIR / "sample.docx", SAMPLE_DIR / "sample.odt"]).merge().run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_multiples_format.zip").write_bytes(resp.content) + + def test_libre_office_convert_std_lib_mime(self, client: GotenbergClient): + with patch("gotenberg_client.base.guess_mime_type") as mocked_guess_mime_type: + mocked_guess_mime_type.side_effect = guess_mime_type_stdlib + with client.libre_office.to_pdf() as route: + resp = route.convert_files([SAMPLE_DIR / "sample.docx", SAMPLE_DIR / "sample.odt"]).no_merge().run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/zip" + + if SAVE_OUTPUTS: + (SAVE_DIR / "test_libre_office_convert_multiples_format.zip").write_bytes(resp.content) + + @pytest.mark.parametrize( + ("gt_format", "pike_format"), + [(PdfAFormatOptions.A1a, "1A"), (PdfAFormatOptions.A2b, "2B"), (PdfAFormatOptions.A3b, "3B")], + ) + def test_libre_office_convert_xlsx_format_pdfa( + self, + client: GotenbergClient, + gt_format: PdfAFormatOptions, + pike_format: str, + ): + test_file = SAMPLE_DIR / "sample.xlsx" + with client.libre_office.to_pdf() as route: + resp = route.convert(test_file).pdf_format(gt_format).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + with tempfile.TemporaryDirectory() as temp_dir: + output = Path(temp_dir) / "test_libre_office_convert_xlsx_format_pdfa.pdf" + output.write_bytes(resp.content) + with pikepdf.open(output) as pdf: + meta = pdf.open_metadata() + assert meta.pdfa_status == pike_format + + if SAVE_OUTPUTS: + (SAVE_DIR / f"test_libre_office_convert_xlsx_format_{pike_format}.pdf").write_bytes(resp.content) diff --git a/tests/test_convert_pdf_a.py b/tests/test_convert_pdf_a.py new file mode 100644 index 0000000..3000012 --- /dev/null +++ b/tests/test_convert_pdf_a.py @@ -0,0 +1,59 @@ +import tempfile +from pathlib import Path + +import pikepdf +import pytest +from httpx import codes + +from gotenberg_client.client import GotenbergClient +from gotenberg_client.pdf_format import PdfAFormatOptions +from tests.conftest import SAMPLE_DIR +from tests.conftest import SAVE_DIR +from tests.conftest import SAVE_OUTPUTS + + +class TestPdfAConvert: + @pytest.mark.parametrize( + ("gt_format", "pike_format"), + [(PdfAFormatOptions.A1a, "1A"), (PdfAFormatOptions.A2b, "2B"), (PdfAFormatOptions.A3b, "3B")], + ) + def test_pdf_a_single_file( + self, + client: GotenbergClient, + gt_format: PdfAFormatOptions, + pike_format: str, + ): + test_file = SAMPLE_DIR / "sample1.pdf" + with client.pdf_a.to_pdfa() as route: + resp = route.convert(test_file).pdf_format(gt_format).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + with tempfile.TemporaryDirectory() as temp_dir: + output = Path(temp_dir) / "test_libre_office_convert_xlsx_format_pdfa.pdf" + output.write_bytes(resp.content) + with pikepdf.open(output) as pdf: + meta = pdf.open_metadata() + assert meta.pdfa_status == pike_format + + if SAVE_OUTPUTS: + (SAVE_DIR / f"test_libre_office_convert_xlsx_format_{pike_format}.pdf").write_bytes(resp.content) + + @pytest.mark.parametrize("gt_format", [PdfAFormatOptions.A1a, PdfAFormatOptions.A2b, PdfAFormatOptions.A3b]) + def test_pdf_a_multiple_file( + self, + client: GotenbergClient, + gt_format: PdfAFormatOptions, + ): + with tempfile.TemporaryDirectory() as temp_dir: + test_file = SAMPLE_DIR / "sample1.pdf" + other_test_file = Path(temp_dir) / "sample2.pdf" + other_test_file.write_bytes(test_file.read_bytes()) + with client.pdf_a.to_pdfa() as route: + resp = route.convert_files([test_file, other_test_file]).pdf_format(gt_format).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/zip" diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..5f26b4a --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,15 @@ +from gotenberg_client.client import GotenbergClient +from gotenberg_client.health import StatusOptions + + +class TestHealthStatus: + def test_health_endpoint( + self, + client: GotenbergClient, + ): + status = client.health.health() + assert status.overall == StatusOptions.Up + assert status.chromium is not None + assert status.chromium.status == StatusOptions.Up + assert status.uno is not None + assert status.uno.status == StatusOptions.Up diff --git a/tests/test_merge.py b/tests/test_merge.py new file mode 100644 index 0000000..008a5ab --- /dev/null +++ b/tests/test_merge.py @@ -0,0 +1,66 @@ +import shutil +import tempfile +from pathlib import Path +from typing import List + +import pikepdf +import pytest +from httpx import codes + +from gotenberg_client.client import GotenbergClient +from gotenberg_client.pdf_format import PdfAFormatOptions +from tests.conftest import SAMPLE_DIR +from tests.conftest import SAVE_DIR +from tests.conftest import SAVE_OUTPUTS + + +@pytest.fixture() +def create_files(): + temp_dir = Path(tempfile.mkdtemp()) + test_file = SAMPLE_DIR / "sample1.pdf" + other_test_file = temp_dir / "sample2.pdf" + other_test_file.write_bytes(test_file.read_bytes()) + yield [test_file, other_test_file] + shutil.rmtree(temp_dir, ignore_errors=True) + + +class TestMergePdfs: + @pytest.mark.parametrize( + ("gt_format", "pike_format"), + [(PdfAFormatOptions.A1a, "1A"), (PdfAFormatOptions.A2b, "2B"), (PdfAFormatOptions.A3b, "3B")], + ) + def test_merge_files_pdf_a( + self, + client: GotenbergClient, + create_files: List[Path], + gt_format: PdfAFormatOptions, + pike_format: str, + ): + with client.merge.merge() as route: + resp = route.merge(create_files).pdf_format(gt_format).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" + + with tempfile.TemporaryDirectory() as temp_dir: + output = Path(temp_dir) / "test_merge_files_pdf_a.pdf" + output.write_bytes(resp.content) + with pikepdf.open(output) as pdf: + meta = pdf.open_metadata() + assert meta.pdfa_status == pike_format + + if SAVE_OUTPUTS: + (SAVE_DIR / f"test_libre_office_convert_xlsx_format_{pike_format}.pdf").write_bytes(resp.content) + + def test_pdf_a_multiple_file( + self, + client: GotenbergClient, + create_files: List[Path], + ): + with client.merge.merge() as route: + resp = route.merge(create_files).run() + + assert resp.status_code == codes.OK + assert "Content-Type" in resp.headers + assert resp.headers["Content-Type"] == "application/pdf" diff --git a/tests/utils.py b/tests/utils.py index 5eed8df..021bfe3 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -11,5 +11,5 @@ def verify_stream_contains(key: str, value: str, stream: MultipartStream): assert item.value == value, f"Key {item.value} /= {value}" return - msg = f"Key {key} with value {value} not found in stream" + msg = f'Key "{key}" with value "{value}" not found in stream' raise AssertionError(msg)