Improve volumespec parsing on windows platforms

Signed-off-by: Joffrey F <[email protected]>

compose/config/config.py

@@ -3,7 +3,6 @@ from __future__ import unicode_literals
 
 import functools
 import logging
-import ntpath
 import os
 import string
 import sys
@@ -16,6 +15,7 @@ from cached_property import cached_property
 from ..const import COMPOSEFILE_V1 as V1
 from ..const import COMPOSEFILE_V2_0 as V2_0
 from ..utils import build_string_dict
+from ..utils import splitdrive
 from .environment import env_vars_from_file
 from .environment import Environment
 from .environment import split_env
@@ -942,13 +942,7 @@ def split_path_mapping(volume_path):
     path. Using splitdrive so windows absolute paths won't cause issues with
     splitting on ':'.
     """
-    # splitdrive is very naive, so handle special cases where we can be sure
-    # the first character is not a drive.
-    if (volume_path.startswith('.') or volume_path.startswith('~') or
-            volume_path.startswith('/')):
-        drive, volume_config = '', volume_path
-    else:
-        drive, volume_config = ntpath.splitdrive(volume_path)
+    drive, volume_config = splitdrive(volume_path)
 
     if ':' in volume_config:
         (host, container) = volume_config.split(':', 1)

compose/config/types.py

@@ -12,6 +12,7 @@ import six
 from compose.config.config import V1
 from compose.config.errors import ConfigurationError
 from compose.const import IS_WINDOWS_PLATFORM
+from compose.utils import splitdrive
 
 
 class VolumeFromSpec(namedtuple('_VolumeFromSpec', 'source mode type')):
@@ -114,41 +115,23 @@ def parse_extra_hosts(extra_hosts_config):
         return extra_hosts_dict
 
 
-def normalize_paths_for_engine(external_path, internal_path):
+def normalize_path_for_engine(path):
     """Windows paths, c:\my\path\shiny, need to be changed to be compatible with
     the Engine. Volume paths are expected to be linux style /c/my/path/shiny/
     """
-    if not IS_WINDOWS_PLATFORM:
-        return external_path, internal_path
+    drive, tail = splitdrive(path)
 
-    if external_path:
-        drive, tail = os.path.splitdrive(external_path)
+    if drive:
+        path = '/' + drive.lower().rstrip(':') + tail
 
-        if drive:
-            external_path = '/' + drive.lower().rstrip(':') + tail
-
-        external_path = external_path.replace('\\', '/')
-
-    return external_path, internal_path.replace('\\', '/')
+    return path.replace('\\', '/')
 
 
 class VolumeSpec(namedtuple('_VolumeSpec', 'external internal mode')):
 
     @classmethod
-    def parse(cls, volume_config):
-        """Parse a volume_config path and split it into external:internal[:mode]
-        parts to be returned as a valid VolumeSpec.
-        """
-        if IS_WINDOWS_PLATFORM:
-            # relative paths in windows expand to include the drive, eg C:\
-            # so we join the first 2 parts back together to count as one
-            drive, tail = os.path.splitdrive(volume_config)
-            parts = tail.split(":")
-
-            if drive:
-                parts[0] = drive + parts[0]
-        else:
-            parts = volume_config.split(':')
+    def _parse_unix(cls, volume_config):
+        parts = volume_config.split(':')
 
         if len(parts) > 3:
             raise ConfigurationError(
@@ -156,13 +139,11 @@ class VolumeSpec(namedtuple('_VolumeSpec', 'external internal mode')):
                 "external:internal[:mode]" % volume_config)
 
         if len(parts) == 1:
-            external, internal = normalize_paths_for_engine(
-                None,
-                os.path.normpath(parts[0]))
+            external = None
+            internal = os.path.normpath(parts[0])
         else:
-            external, internal = normalize_paths_for_engine(
-                os.path.normpath(parts[0]),
-                os.path.normpath(parts[1]))
+            external = os.path.normpath(parts[0])
+            internal = os.path.normpath(parts[1])
 
         mode = 'rw'
         if len(parts) == 3:
@@ -170,6 +151,48 @@ class VolumeSpec(namedtuple('_VolumeSpec', 'external internal mode')):
 
         return cls(external, internal, mode)
 
+    @classmethod
+    def _parse_win32(cls, volume_config):
+        # relative paths in windows expand to include the drive, eg C:\
+        # so we join the first 2 parts back together to count as one
+        mode = 'rw'
+
+        def separate_next_section(volume_config):
+            drive, tail = splitdrive(volume_config)
+            parts = tail.split(':', 1)
+            if drive:
+                parts[0] = drive + parts[0]
+            return parts
+
+        parts = separate_next_section(volume_config)
+        if len(parts) == 1:
+            internal = normalize_path_for_engine(os.path.normpath(parts[0]))
+            external = None
+        else:
+            external = parts[0]
+            parts = separate_next_section(parts[1])
+            external = normalize_path_for_engine(os.path.normpath(external))
+            internal = normalize_path_for_engine(os.path.normpath(parts[0]))
+            if len(parts) > 1:
+                if ':' in parts[1]:
+                    raise ConfigurationError(
+                        "Volume %s has incorrect format, should be "
+                        "external:internal[:mode]" % volume_config
+                    )
+                mode = parts[1]
+
+        return cls(external, internal, mode)
+
+    @classmethod
+    def parse(cls, volume_config):
+        """Parse a volume_config path and split it into external:internal[:mode]
+        parts to be returned as a valid VolumeSpec.
+        """
+        if IS_WINDOWS_PLATFORM:
+            return cls._parse_win32(volume_config)
+        else:
+            return cls._parse_unix(volume_config)
+
     def repr(self):
         external = self.external + ':' if self.external else ''
         return '{ext}{v.internal}:{v.mode}'.format(ext=external, v=self)

compose/utils.py

@@ -6,6 +6,7 @@ import hashlib
 import json
 import json.decoder
 import logging
+import ntpath
 
 import six
 
@@ -108,3 +109,11 @@ def microseconds_from_time_nano(time_nano):
 
 def build_string_dict(source_dict):
     return dict((k, str(v if v is not None else '')) for k, v in source_dict.items())
+
+
+def splitdrive(path):
+    if len(path) == 0:
+        return ('', '')
+    if path[0] in ['.', '\\', '/', '~']:
+        return ('', path)
+    return ntpath.splitdrive(path)

tests/unit/config/types_test.py

@@ -9,7 +9,6 @@ from compose.config.errors import ConfigurationError
 from compose.config.types import parse_extra_hosts
 from compose.config.types import VolumeFromSpec
 from compose.config.types import VolumeSpec
-from compose.const import IS_WINDOWS_PLATFORM
 
 
 def test_parse_extra_hosts_list():
@@ -64,15 +63,38 @@ class TestVolumeSpec(object):
             VolumeSpec.parse('one:two:three:four')
         assert 'has incorrect format' in exc.exconly()
 
-    @pytest.mark.xfail((not IS_WINDOWS_PLATFORM), reason='does not have a drive')
     def test_parse_volume_windows_absolute_path(self):
         windows_path = "c:\\Users\\me\\Documents\\shiny\\config:\\opt\\shiny\\config:ro"
-        assert VolumeSpec.parse(windows_path) == (
+        assert VolumeSpec._parse_win32(windows_path) == (
             "/c/Users/me/Documents/shiny/config",
             "/opt/shiny/config",
             "ro"
         )
 
+    def test_parse_volume_windows_internal_path(self):
+        windows_path = 'C:\\Users\\reimu\\scarlet:C:\\scarlet\\app:ro'
+        assert VolumeSpec._parse_win32(windows_path) == (
+            '/c/Users/reimu/scarlet',
+            '/c/scarlet/app',
+            'ro'
+        )
+
+    def test_parse_volume_windows_just_drives(self):
+        windows_path = 'E:\\:C:\\:ro'
+        assert VolumeSpec._parse_win32(windows_path) == (
+            '/e/',
+            '/c/',
+            'ro'
+        )
+
+    def test_parse_volume_windows_mixed_notations(self):
+        windows_path = '/c/Foo:C:\\bar'
+        assert VolumeSpec._parse_win32(windows_path) == (
+            '/c/Foo',
+            '/c/bar',
+            'rw'
+        )
+
 
 class TestVolumesFromSpec(object):