fix race conditions with symlink and apply format

n-burger · Jun 22, 2022 · 4107a9b · 4107a9b
1 parent 3e79ca9
commit 4107a9b
Show file tree

Hide file tree

Showing 22 changed files with 270 additions and 280 deletions.
diff --git a/noxfile.py b/noxfile.py
@@ -1,7 +1,7 @@
 import nox
 from nox_poetry import session
 
-PYTHON_VERSIONS = ["3.6", "3.8", "3.9.7"]
+PYTHON_VERSIONS = ["3.6", "3.8", "3.9.7", "3.10.4"]
 SPHINX_VERSIONS = ["4.1", "4.2", "4.3", "4.4", "4.5"]
 TEST_DEPENDENCIES = [
     "pytest",
@@ -11,7 +11,7 @@
 
 
 def is_supported(python: str, sphinx: str) -> bool:
-    return not (python == "3.6" and sphinx not in ["3.2"])
+    return not (python in ["3.10.4"] and sphinx in ["4.1"])
 
 
 def run_tests(session, sphinx):

diff --git a/poetry.lock b/poetry.lock
diff --git a/sphinxcontrib/__init__.py b/sphinxcontrib/__init__.py
@@ -1 +1 @@
-__import__('pkg_resources').declare_namespace(__name__)
+__import__("pkg_resources").declare_namespace(__name__)
diff --git a/sphinxcontrib/collections/api.py b/sphinxcontrib/collections/api.py
@@ -4,12 +4,12 @@
 
 def register_driver(name, driver_class):
     if not issubclass(driver_class, Driver):
-        raise SphinxCollectionsApiError('Given driver class must be a subclass of the main Driver class.')
+        raise SphinxCollectionsApiError("Given driver class must be a subclass of the main Driver class.")
 
     try:
         DRIVERS[name] = driver_class
     except KeyError:
-        raise SphinxCollectionsApiError('Driver with name {} already exists.'.format(name))
+        raise SphinxCollectionsApiError("Driver with name {} already exists.".format(name))
 
 
 class SphinxCollectionsApiError(BaseException):

diff --git a/sphinxcontrib/collections/collections.py b/sphinxcontrib/collections/collections.py
@@ -1,16 +1,16 @@
-import sphinx
 import os
 
+import sphinx
 from pkg_resources import parse_version
 
-from sphinxcontrib.collections.drivers.copy_folder import CopyFolderDriver
 from sphinxcontrib.collections.drivers.copy_file import CopyFileDriver
-from sphinxcontrib.collections.drivers.string import StringDriver
+from sphinxcontrib.collections.drivers.copy_folder import CopyFolderDriver
 from sphinxcontrib.collections.drivers.function import FunctionDriver
+from sphinxcontrib.collections.drivers.git import GitDriver
+from sphinxcontrib.collections.drivers.jinja import JinjaDriver
 from sphinxcontrib.collections.drivers.report import ReportDriver
+from sphinxcontrib.collections.drivers.string import StringDriver
 from sphinxcontrib.collections.drivers.symlink import SymlinkDriver
-from sphinxcontrib.collections.drivers.jinja import JinjaDriver
-from sphinxcontrib.collections.drivers.git import GitDriver
 
 sphinx_version = sphinx.__version__
 if parse_version(sphinx_version) >= parse_version("1.6"):
@@ -24,41 +24,42 @@
 COLLECTIONS = []
 
 DRIVERS = {
-    'copy_folder': CopyFolderDriver,
-    'copy_file': CopyFileDriver,
-    'string': StringDriver,
-    'function': FunctionDriver,
-    'report': ReportDriver,
-    'symlink': SymlinkDriver,
-    'jinja': JinjaDriver,
-    'git': GitDriver,
+    "copy_folder": CopyFolderDriver,
+    "copy_file": CopyFileDriver,
+    "string": StringDriver,
+    "function": FunctionDriver,
+    "report": ReportDriver,
+    "symlink": SymlinkDriver,
+    "jinja": JinjaDriver,
+    "git": GitDriver,
 }
 
 
 def collect_collections(app, config):
-    LOG.info('Read in collections ...')
+    LOG.info("Read in collections ...")
     for name, collection in config.collections.items():
         COLLECTIONS.append(Collection(app, name, **collection))
 
 
 def clean_collections(app, config):
-    LOG.info('Clean collections ...')
+    LOG.info("Clean collections ...")
     for collection in COLLECTIONS:
         collection.clean()
 
 
 def execute_collections(app, config):
-    LOG.info('Executing collections ...')
+    LOG.info("Executing collections ...")
     for collection in COLLECTIONS:
         try:
             collection.run()
         except Exception as e:
-            LOG.error('Error executing driver {} for collection {}. {}'.format(collection.driver.name,
-                                                                               collection.name, e))
+            LOG.error(
+                "Error executing driver {} for collection {}. {}".format(collection.driver.name, collection.name, e)
+            )
 
 
 def final_clean_collections(app, exception):
-    LOG.info('Final clean of collections ...')
+    LOG.info("Final clean of collections ...")
     for collection in COLLECTIONS:
         collection.final_clean()
 
@@ -70,72 +71,76 @@ def __init__(self, app, name, **kwargs):
         self.executed = False
         self._log = LOG
 
-        self.active = bool(kwargs.get('active', True))
-        tags = kwargs.get('tags', [])
+        self.active = bool(kwargs.get("active", True))
+        tags = kwargs.get("tags", [])
 
         # Check if tags are set and change active to True if this is the case
         self.tags = tags
         for tag in tags:
             if tag in self.app.tags.tags.keys() and self.app.tags.tags[tag]:
                 self.active = True
 
-        self._prefix = '  {}: '.format(self.name)
+        self._prefix = "  {}: ".format(self.name)
 
-        collection_main_folder = os.path.join(app.confdir, app.config['collections_target'])
+        collection_main_folder = os.path.join(app.confdir, app.config["collections_target"])
 
-        target = kwargs.get('target', None)
+        target = kwargs.get("target", None)
         if target is None:
             target = self.name
         if not os.path.isabs(target):
             if not os.path.exists(collection_main_folder):
                 os.makedirs(collection_main_folder, exist_ok=True)
             target = os.path.join(collection_main_folder, target)
 
-        # Check if we manipulate data only in documentation folder.
-        # Any other location is not allowed.
-        target_inside_confdir = ( 
-            os.path.abspath(target).startswith(os.path.abspath(app.confdir)) if os.path.islink(target)
-            else os.path.realpath(target).startswith(os.path.realpath(app.confdir)) )
-
-
-        if not target_inside_confdir:
-            raise CollectionsException(
-                'Target path is not part of documentation folder\n'
-                'Target path: {}\n'
-                'Sphinx app conf path: {}'.format(os.path.realpath(target),
-                                                  os.path.realpath(app.confdir)))
-
         if not os.path.exists(os.path.dirname(target)):
             os.makedirs(os.path.dirname(target), exist_ok=True)
 
         self.target = target
 
-        clean = kwargs.get('clean', None)
+        clean = kwargs.get("clean", None)
         if clean is None:
-            clean = app.config['collections_clean']
+            clean = app.config["collections_clean"]
         self.needs_clean = clean
 
-        final_clean = kwargs.get('final_clean', None)
+        final_clean = kwargs.get("final_clean", None)
         if final_clean is None:
-            final_clean = app.config['collections_final_clean']
+            final_clean = app.config["collections_final_clean"]
         self.needs_final_clean = final_clean
 
         self.config = kwargs
-        self.config['name'] = self.name
-        self.config['confdir'] = self.app.confdir
-        self.config['target'] = target
-        if 'safe' not in self.config.keys():
-            self.config['safe'] = True
+        self.config["name"] = self.name
+        self.config["confdir"] = self.app.confdir
+        self.config["target"] = target
+        if "safe" not in self.config.keys():
+            self.config["safe"] = True
 
         # Driver init
-        driver_name = kwargs.get('driver', None)
+        driver_name = kwargs.get("driver", None)
         if driver_name is None or driver_name not in DRIVERS.keys():
-            raise Exception('Unknown driver: {}'.format(driver_name))
-        self.driver = DRIVERS[kwargs['driver']](self.name, self.config)
+            raise Exception("Unknown driver: {}".format(driver_name))
+        self.driver = DRIVERS[kwargs["driver"]](self.name, self.config)
+
+        # Check if we manipulate data only in documentation folder.
+        # Any other location is not allowed.
+        target_inside_confdir = (
+            os.path.abspath(target).startswith(os.path.abspath(app.confdir))
+            if driver_name == "symlink"
+            else os.path.realpath(target).startswith(os.path.realpath(app.confdir))
+        )
+
+        if not target_inside_confdir:
+            raise CollectionsException(
+                "Target path is not part of documentation folder\n"
+                "Target path abs: {}\n"
+                "Target path: {}\n"
+                "Sphinx app conf path: {}\n".format(
+                    os.path.abspath(target), os.path.realpath(target), os.path.realpath(app.confdir)
+                )
+            )
 
         self.result = None
 
-        self.info('Initialised')
+        self.info("Initialised")
 
     def __repr__(self):
         return self.name
@@ -155,15 +160,15 @@ def final_clean(self):
             self.driver.clean()
 
     def info(self, message):
-        self._log.info('{}{}'.format(self._prefix, message))
+        self._log.info("{}{}".format(self._prefix, message))
 
     def warn(self, message):
-        self._log.warn('{}{}'.format(self._prefix, message))
+        self._log.warn("{}{}".format(self._prefix, message))
 
     def error(self, message):
-        if self.config['safe']:
-            raise CollectionsException('{}{}'.format(self._prefix, message))
-        self._log.info('{}{}'.format(self._prefix, message))
+        if self.config["safe"]:
+            raise CollectionsException("{}{}".format(self._prefix, message))
+        self._log.info("{}{}".format(self._prefix, message))
 
 
 class CollectionsException(BaseException):

diff --git a/sphinxcontrib/collections/directives/if_collection.py b/sphinxcontrib/collections/directives/if_collection.py
@@ -1,7 +1,6 @@
 import sphinx
-
 from docutils import nodes
-from docutils.parsers.rst import directives, Directive
+from docutils.parsers.rst import Directive, directives
 from docutils.statemachine import ViewList
 from pkg_resources import parse_version
 from sphinx.util.nodes import nested_parse_with_titles
@@ -24,6 +23,7 @@ class CollectionsIfDirective(Directive):
     """
     Directive to add content based on executed collections.
     """
+
     has_content = True
     required_arguments = 1
     optional_arguments = 0
@@ -44,11 +44,11 @@ def env(self):
 
     def run(self):
         collections = sphinxcontrib.collections.collections.COLLECTIONS
-        search_cols = [x.strip() for x in self.arguments[0].split(',')]
+        search_cols = [x.strip() for x in self.arguments[0].split(",")]
 
         found = False
         for search_col in search_cols:
-            if search_col == '':
+            if search_col == "":
                 continue
 
             for collection in collections:
@@ -70,4 +70,3 @@ def run(self):
             collection_node += node_collection_content.children
 
         return [collection_node]
-
diff --git a/sphinxcontrib/collections/drivers/__init__.py b/sphinxcontrib/collections/drivers/__init__.py
@@ -1,6 +1,6 @@
 import os
-import sphinx
 
+import sphinx
 from pkg_resources import parse_version
 
 sphinx_version = sphinx.__version__
@@ -13,13 +13,12 @@
 
 
 class Driver:
-
     def __init__(self, collection, config=None):
         self._log = logging.getLogger(__name__)
         self.name = self.__class__.__name__
         self.collection = collection
 
-        self._prefix = '  {}: ({}) '.format(self.collection, self.name)
+        self._prefix = "  {}: ({}) ".format(self.collection, self.name)
 
         if config is None:
             config = {}
@@ -31,7 +30,7 @@ def run(self):
 
         Must be implement by the parent driver class.
         """
-        raise NotImplementedError('run() function must be implemented by driver {} itself.'.format(self.name))
+        raise NotImplementedError("run() function must be implemented by driver {} itself.".format(self.name))
 
     def clean(self):
         """
@@ -41,7 +40,7 @@ def clean(self):
 
         Must be implement by the parent driver class.
         """
-        raise NotImplementedError('clean() function must be implemented by driver {} itself.'.format(self.name))
+        raise NotImplementedError("clean() function must be implemented by driver {} itself.".format(self.name))
 
     def error(self, message, e=None):
         """
@@ -53,13 +52,13 @@ def error(self, message, e=None):
         :return: None
         """
         if e is not None and isinstance(e, BaseException):
-            if self.config['safe']:
-                raise ColectionsDriverError('{}{}'.format(self._prefix, message)) from e
-            self._log.error(('{}{} - {}'.format(self._prefix, message, e)))
+            if self.config["safe"]:
+                raise ColectionsDriverError("{}{}".format(self._prefix, message)) from e
+            self._log.error(("{}{} - {}".format(self._prefix, message, e)))
         else:
-            if self.config['safe']:
-                raise ColectionsDriverError('{}{}'.format(self._prefix, message))
-            self._log.error(('{}{}'.format(self._prefix, message)))
+            if self.config["safe"]:
+                raise ColectionsDriverError("{}{}".format(self._prefix, message))
+            self._log.error(("{}{}".format(self._prefix, message)))
 
     def info(self, message):
         """
@@ -70,7 +69,7 @@ def info(self, message):
         :param message: string
         :return: None
         """
-        self._log.info('{}{}'.format(self._prefix, message))
+        self._log.info("{}{}".format(self._prefix, message))
 
     def debug(self, message):
         """
@@ -81,7 +80,7 @@ def debug(self, message):
         :param message: string
         :return: None
         """
-        self._log.debug('{}{}'.format(self._prefix, message))
+        self._log.debug("{}{}".format(self._prefix, message))
 
     def get_source_path(self):
         """
@@ -91,11 +90,11 @@ def get_source_path(self):
 
         :return: path string
         """
-        source = self.config.get('source', None)
+        source = self.config.get("source", None)
         if source is None:
-            self.error('Source must be defined')
+            self.error("Source must be defined")
         if not os.path.isabs(source):
-            source = os.path.join(self.config['confdir'], source)
+            source = os.path.join(self.config["confdir"], source)
         return source
 
     def get_path(self, path):
@@ -107,12 +106,12 @@ def get_path(self, path):
         :return: path string
         """
         if path is None:
-            self.error('Path must be defined')
+            self.error("Path must be defined")
         if not isinstance(path, str):
-            self.debug('This functions makes mostly sense for string source only.')
+            self.debug("This functions makes mostly sense for string source only.")
             return path
         if not os.path.isabs(path):
-            path = os.path.join(self.config['confdir'], path)
+            path = os.path.join(self.config["confdir"], path)
         return path
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		__import__('pkg_resources').declare_namespace(__name__)
		__import__("pkg_resources").declare_namespace(__name__)