Merge pull request #770 from ulb-sachsen-anhalt/feat/rfct/pytest3

Feat/rfct/pytest3

Makefile

@@ -152,8 +152,7 @@ assets-server:
 test: assets
 	HOME=$(CURDIR)/ocrd_utils $(PYTHON) -m pytest --continue-on-collection-errors -k TestLogging $(TESTDIR)
 	HOME=$(CURDIR) $(PYTHON) -m pytest --continue-on-collection-errors -k TestLogging $(TESTDIR)
-	HOME=$(CURDIR) $(PYTHON) -m pytest --continue-on-collection-errors $(TESTDIR)/test_resource_manager.py
-	$(PYTHON) -m pytest --continue-on-collection-errors --ignore=$(TESTDIR)/test_logging.py --ignore=$(TESTDIR)/test_resource_manager.py $(TESTDIR)
+	$(PYTHON) -m pytest --continue-on-collection-errors --ignore=$(TESTDIR)/test_logging.py $(TESTDIR)
 
 test-profile:
 	$(PYTHON) -m cProfile -o profile $$(which pytest)

ocrd/ocrd/resource_manager.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 from os.path import join
-from os import environ, listdir, getcwd
+from os import environ, listdir, getcwd, path
 import re
 from shutil import copytree
 from datetime import datetime
@@ -13,27 +13,56 @@
 from ocrd_validators import OcrdResourceListValidator
 from ocrd_utils import getLogger
 from ocrd_utils.os import list_all_resources, pushd_popd
-from ocrd_utils.constants import XDG_CONFIG_HOME, XDG_DATA_HOME
-
 from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT
 
 class OcrdResourceManager():
 
     """
     Managing processor resources
     """
-    def __init__(self):
+    def __init__(self, userdir=None, xdg_config_home=None, xdg_data_home=None):
         self.log = getLogger('ocrd.resource_manager')
         self.database = {}
+
+        self._xdg_data_home = xdg_data_home
+        self._xdg_config_home = xdg_config_home
+        self._userdir = userdir
+        self.user_list = Path(self.xdg_config_home, 'ocrd', 'resources.yml')
+
         self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
-        self.user_list = Path(XDG_CONFIG_HOME, 'ocrd', 'resources.yml')
         if not self.user_list.exists():
             if not self.user_list.parent.exists():
                 self.user_list.parent.mkdir(parents=True)
             with open(str(self.user_list), 'w', encoding='utf-8') as f:
                 f.write(RESOURCE_USER_LIST_COMMENT)
         self.load_resource_list(self.user_list)
 
+    @property
+    def userdir(self):
+        if not self._userdir:
+            self._userdir = path.expanduser('~')
+            if 'HOME' in environ and environ['HOME'] != path.expanduser('~'):
+                self._userdir = environ['HOME']
+        return self._userdir
+
+    @property
+    def xdg_data_home(self):
+        if not self._xdg_data_home:
+            if 'XDG_DATA_HOME' in environ:
+                self._xdg_data_home = environ['XDG_DATA_HOME']
+            else:
+                self._xdg_data_home = join(self.userdir, '.local', 'share')
+        return self._xdg_data_home
+
+    @property
+    def xdg_config_home(self):
+        if not self._xdg_config_home:
+            if 'XDG_CONFIG_HOME' in environ:
+                self._xdg_config_home = environ['XDG_CONFIG_HOME']
+            else:
+                self._xdg_config_home = join(self.userdir, '.config')
+        return self._xdg_config_home
+
     def load_resource_list(self, list_filename, database=None):
         if not database:
             database = self.database
@@ -57,7 +86,7 @@ def list_available(self, executable=None):
         """
         if executable:
             return [(executable, self.database[executable])]
-        return [(x, y) for x, y in self.database.items()]
+        return self.database.items()
 
     def list_installed(self, executable=None):
         """
@@ -70,7 +99,7 @@ def list_installed(self, executable=None):
             # resources we know about
             all_executables = list(self.database.keys())
             # resources in the file system
-            parent_dirs = [join(x, 'ocrd-resources') for x in [XDG_DATA_HOME, '/usr/local/share']]
+            parent_dirs = [join(x, 'ocrd-resources') for x in [self.xdg_data_home, '/usr/local/share']]
             for parent_dir in parent_dirs:
                 if Path(parent_dir).exists():
                     all_executables += [x for x in listdir(parent_dir) if x.startswith('ocrd-')]
@@ -80,7 +109,7 @@ def list_installed(self, executable=None):
                 res_name = Path(res_filename).name
                 resdict = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
                 if not resdict:
-                    self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'" % (this_executable, res_name, res_filename, self.user_list))
+                    self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'", this_executable, res_name, res_filename, self.user_list)
                     resdict = [self.add_to_user_database(this_executable, res_filename)]
                 resdict[0]['path'] = res_filename
                 reslist.append(resdict[0])
@@ -113,6 +142,7 @@ def add_to_user_database(self, executable, res_filename, url=None):
             f.write(RESOURCE_USER_LIST_COMMENT)
             f.write('\n')
             f.write(safe_dump(user_database))
+        self.load_resource_list(self.user_list)
         return resdict
 
     def find_resources(self, executable=None, name=None, url=None, database=None):
@@ -140,21 +170,22 @@ def default_resource_dir(self):
 
     def location_to_resource_dir(self, location):
         return '/usr/local/share/ocrd-resources' if location == 'system' else \
-                join(XDG_DATA_HOME, 'ocrd-resources') if location == 'data' else \
+                join(self.xdg_data_home, 'ocrd-resources') if location == 'data' else \
                 getcwd()
 
     def resource_dir_to_location(self, resource_path):
         resource_path = str(resource_path)
         return 'system' if resource_path.startswith('/usr/local/share/ocrd-resources') else \
-               'data' if resource_path.startswith(join(XDG_DATA_HOME, 'ocrd-resources')) else \
+               'data' if resource_path.startswith(join(self.xdg_data_home, 'ocrd-resources')) else \
                'cwd' if resource_path.startswith(getcwd()) else \
                resource_path
 
     def parameter_usage(self, name, usage='as-is'):
         if usage == 'as-is':
             return name
-        if usage == 'without-extension':
+        elif usage == 'without-extension':
             return Path(name).stem
+        raise ValueError("No such usage '%s'" % usage)
 
     def _download_impl(self, url, filename, progress_cb=None, size=None):
         log = getLogger('ocrd.resource_manager._download_impl')

ocrd_utils/ocrd_utils/__init__.py

@@ -114,8 +114,7 @@
     LOG_FORMAT,
     LOG_TIMEFMT,
     VERSION,
-    XDG_CONFIG_HOME,
-    XDG_DATA_HOME)
+    )
 
 from .deprecate import (
     deprecated_alias)

tests/test_resource_manager.py

@@ -1,40 +1,111 @@
-from contextlib import contextmanager
-from pathlib import Path
-from tests.base import TestCase, main # pylint: disable=import-error,no-name-in-module
-
-from pytest import fixture
-
-from ocrd_utils import pushd_popd, initLogging
-import ocrd_utils.constants
-
-@contextmanager
-def monkey_patch_temp_xdg():
-    with pushd_popd(tempdir=True) as tempdir:
-        old_config = ocrd_utils.constants.XDG_CONFIG_HOME
-        old_data = ocrd_utils.constants.XDG_DATA_HOME
-        ocrd_utils.constants.XDG_CONFIG_HOME = tempdir
-        ocrd_utils.constants.XDG_DATA_HOME = tempdir
-        from ocrd.resource_manager import OcrdResourceManager
-        yield tempdir, OcrdResourceManager()
-        ocrd_utils.constants.XDG_CONFIG_HOME = old_config
-        ocrd_utils.constants.XDG_DATA_HOME = old_data
-
-def test_config_created():
-    with monkey_patch_temp_xdg() as (tempdir, mgr):
-        f = Path(tempdir, 'ocrd', 'resources.yml')
-        assert f.exists()
-        assert f == mgr.user_list
-        ret = mgr.add_to_user_database('ocrd-foo', f)
-        ret = mgr.add_to_user_database('ocrd-foo', f)
-        assert ret
-        mgr.list_installed()
-        proc = 'ocrd-anybaseocr-layout-analysis'
-        url = 'https://ocr-d-repo.scc.kit.edu/models/dfki/layoutAnalysis/mapping_densenet.pickle'
-        fpath = mgr.download(proc, url, mgr.location_to_resource_dir('data'))
-        assert fpath.exists()
-        ret = mgr.add_to_user_database(proc, fpath)
-        ret = mgr.add_to_user_database(proc, fpath)
-        assert ret
+# -*- coding: utf-8 -*-
+
+import os
+import pathlib
+
+from ocrd.resource_manager import OcrdResourceManager
+
+from pytest import raises
+from tests.base import main
+
+CONST_RESOURCE_YML = 'resources.yml'
+CONST_RESOURCE_URL_LAYOUT = 'https://ocr-d-repo.scc.kit.edu/models/dfki/layoutAnalysis/mapping_densenet.pickle'
+
+
+def test_resources_manager_config_default():
+
+    # act
+    mgr = OcrdResourceManager()
+
+    # assert
+    default_config_dir = os.path.join(os.environ['HOME'], '.config', 'ocrd')
+    f = pathlib.Path(default_config_dir) / CONST_RESOURCE_YML
+    assert f.exists()
+    assert f == mgr.user_list
+    assert mgr.add_to_user_database('ocrd-foo', f)
+    mgr.list_installed()
+    proc = 'ocrd-anybaseocr-layout-analysis'
+    # TODO mock request
+    fpath = mgr.download(proc, CONST_RESOURCE_URL_LAYOUT, mgr.location_to_resource_dir('data'))
+    assert fpath.exists()
+    assert mgr.add_to_user_database(proc, fpath)
+
+
+def test_resources_manager_from_environment(tmp_path, monkeypatch):
+
+    # arrange
+    monkeypatch.setenv('XDG_CONFIG_HOME', str(tmp_path))
+    monkeypatch.setenv('XDG_DATA_HOME', str(tmp_path))
+    monkeypatch.setenv('HOME', str(tmp_path))
+
+    # act
+    mgr = OcrdResourceManager()
+
+    # assert
+    f = tmp_path / 'ocrd' / CONST_RESOURCE_YML
+    assert f.exists()
+    assert f == mgr.user_list
+    assert mgr.add_to_user_database('ocrd-foo', f)
+    mgr.list_installed()
+    proc = 'ocrd-anybaseocr-layout-analysis'
+    fpath = mgr.download(proc, CONST_RESOURCE_URL_LAYOUT, mgr.location_to_resource_dir('data'))
+    assert fpath.exists()
+    assert mgr.add_to_user_database(proc, fpath)
+    assert mgr.userdir == str(tmp_path)
+
+
+def test_resources_manager_config_explicite(tmp_path):
+
+    # act
+    mgr = OcrdResourceManager(xdg_config_home=str(tmp_path))
+
+    # assert
+    f = tmp_path / 'ocrd' / CONST_RESOURCE_YML
+    assert f.exists()
+    assert f == mgr.user_list
+    assert mgr.add_to_user_database('ocrd-foo', f)
+    mgr.list_installed()
+    proc = 'ocrd-anybaseocr-layout-analysis'
+    fpath = mgr.download(proc, CONST_RESOURCE_URL_LAYOUT, mgr.location_to_resource_dir('data'))
+    assert fpath.exists()
+    assert mgr.add_to_user_database(proc, fpath)
+
+def test_resources_manager_config_explicit_invalid(tmp_path):
+
+    # act
+    (tmp_path / 'ocrd').mkdir()
+    (tmp_path / 'ocrd' / CONST_RESOURCE_YML).write_text('::INVALID::')
+
+    # assert
+    with raises(ValueError, match='is invalid'):
+        OcrdResourceManager(xdg_config_home=tmp_path)
+
+def test_find_resources(tmp_path):
+
+    # act
+    f = tmp_path / 'ocrd-foo' / 'foo.bar'
+    f.parent.mkdir()
+    f.write_text('foobar')
+    mgr = OcrdResourceManager(xdg_config_home=tmp_path)
+
+    # assert
+    assert mgr.find_resources(executable='ocrd-foo') == []
+    assert mgr.add_to_user_database('ocrd-foo', f, url='http://foo/bar')
+    assert 'ocrd-foo' in [x for x, _ in mgr.find_resources()]
+    assert 'ocrd-foo' in [x for x, _ in mgr.find_resources(url='http://foo/bar')]
+
+def test_parameter_usage(tmp_path):
+    mgr = OcrdResourceManager(xdg_config_home=tmp_path)
+    assert mgr.parameter_usage('foo.bar') == 'foo.bar'
+    assert mgr.parameter_usage('foo.bar', 'without-extension') == 'foo'
+    with raises(ValueError, match='No such usage'):
+        mgr.parameter_usage('foo.bar', 'baz')
+
+def test_default_resource_dir(tmp_path):
+    mgr = OcrdResourceManager(xdg_data_home=tmp_path)
+    assert mgr.xdg_config_home != mgr.xdg_data_home
+    assert mgr.default_resource_dir == str(mgr.xdg_data_home / 'ocrd-resources')
+
 
 if __name__ == "__main__":
     main(__file__)