diff --git a/src/ocrmypdf/api.py b/src/ocrmypdf/api.py index 9a2cdb2f..cd8e2576 100644 --- a/src/ocrmypdf/api.py +++ b/src/ocrmypdf/api.py @@ -19,6 +19,7 @@ import logging import os import sys import warnings +from contextlib import suppress from enum import IntEnum from pathlib import Path from typing import Dict, List, Optional @@ -46,7 +47,7 @@ class TqdmConsole: tqdm.write(msg.rstrip(), end='\n', file=self.file) def flush(self): - if hasattr(self.file, "flush"): + with suppress(AttributeError): self.file.flush() diff --git a/src/ocrmypdf/helpers.py b/src/ocrmypdf/helpers.py index b719d4e7..70e884d9 100644 --- a/src/ocrmypdf/helpers.py +++ b/src/ocrmypdf/helpers.py @@ -104,28 +104,36 @@ def is_file_writable(test_file): can replace it atomically. Before doing the OCR work, make sure the location is writable. """ - p = Path(test_file) - - if p.is_symlink(): - p = p.resolve(strict=False) - - # p.is_file() throws an exception in some cases - if p.exists() and p.is_file(): - return os.access( - os.fspath(p), - os.W_OK, - effective_ids=(os.access in os.supports_effective_ids), - ) - else: - try: - fp = p.open('wb') - except OSError: - return False + try: + if not isinstance(test_file, Path): + p = Path(test_file) else: - fp.close() - with suppress(OSError): - p.unlink() - return True + p = test_file + + if p.is_symlink(): + p = p.resolve(strict=False) + + # p.is_file() throws an exception in some cases + if p.exists() and p.is_file(): + return os.access( + os.fspath(p), + os.W_OK, + effective_ids=(os.access in os.supports_effective_ids), + ) + else: + try: + fp = p.open('wb') + except OSError: + return False + else: + fp.close() + with suppress(OSError): + p.unlink() + return True + except (EnvironmentError, RuntimeError) as e: + log.debug(e) + log.error(str(e)) + return False def deprecated(func): diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 00000000..534e5959 --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,97 @@ +# © 2019 James R. Barlow: github.com/jbarlow83 +# +# This file is part of OCRmyPDF. +# +# OCRmyPDF is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# OCRmyPDF is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with OCRmyPDF. If not, see . + +import logging +import multiprocessing +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +import ocrmypdf.helpers as helpers + + +class TestSafeSymlink: + def test_safe_symlink_link_self(self, tmp_path, caplog): + helpers.safe_symlink(tmp_path / 'self', tmp_path / 'self') + assert caplog.record_tuples[0][1] == logging.WARNING + + def test_safe_symlink_overwrite(self, tmp_path): + (tmp_path / 'regular_file').touch() + with pytest.raises(FileExistsError): + helpers.safe_symlink(tmp_path / 'input', tmp_path / 'regular_file') + + def test_safe_symlink_relink(self, tmp_path): + (tmp_path / 'regular_file_a').touch() + (tmp_path / 'regular_file_b').write_bytes(b'ABC') + (tmp_path / 'link').symlink_to(tmp_path / 'regular_file_a') + helpers.safe_symlink(tmp_path / 'regular_file_b', tmp_path / 'link') + assert (tmp_path / 'link').samefile(tmp_path / 'regular_file_b') or ( + tmp_path / 'link' + ).read_bytes() == b'ABC' + + +def test_no_cpu_count(monkeypatch): + def cpu_count_raises(): + raise NotImplementedError() + + monkeypatch.setattr(multiprocessing, 'cpu_count', cpu_count_raises) + with pytest.warns(expected_warning=UserWarning): + assert helpers.available_cpu_count() == 1 + + +def test_deprecated(): + @helpers.deprecated + def old_function(): + return 42 + + with pytest.warns(expected_warning=DeprecationWarning): + assert old_function() == 42 + + +class TestFileIsWritable: + @pytest.fixture + def non_existent(self, tmp_path): + return tmp_path / 'nofile' + + @pytest.fixture + def basic_file(self, tmp_path): + basic = tmp_path / 'basic' + basic.touch() + return basic + + def test_plain(self, non_existent): + assert helpers.is_file_writable(non_existent) + + def test_symlink_loop(self, tmp_path): + loop = tmp_path / 'loop' + loop.symlink_to(loop) + assert not helpers.is_file_writable(loop) + + def test_chmod(self, basic_file): + assert helpers.is_file_writable(basic_file) + basic_file.chmod(0o400) + assert not helpers.is_file_writable(basic_file) + basic_file.chmod(0o000) + assert not helpers.is_file_writable(basic_file) + + def test_permission_error(self, basic_file): + pathmock = MagicMock(spec_set=basic_file) + pathmock.is_symlink.return_value = False + pathmock.exists.return_value = True + pathmock.is_file.side_effect = PermissionError + assert not helpers.is_file_writable(pathmock)