Fix yaml2python convert_import for relative paths and prefixes

tests/test_dsl.py

@@ -10,6 +10,7 @@
 except ImportError:
     print(sys.path)
     raise
+
 from tosca.python2yaml import PythonToYaml, python_to_yaml, dump_yaml
 from toscaparser.elements.entity_type import EntityType
 from unfurl.yamlloader import ImportResolver, load_yaml, yaml
@@ -258,6 +259,85 @@ def create(self) -> tosca.artifacts.Root:
     }
 
 
+from unittest.mock import MagicMock, patch
+
+
+@pytest.mark.parametrize(
+    "test_input,exp_import,exp_path",
+    [
+        (dict(file="foo.yaml"), "from .foo import *", "/path/to/foo"),
+        (dict(file="../foo.yaml"), "from ..foo import *", "/path/to/../foo"),
+        (
+            dict(file="../../foo.yaml"),
+            "from ...foo import *",
+            "/path/to/../../foo",
+        ),
+    ],
+)
+# patch repo lookup so we don't need to write the whole template
+def test_convert_import(test_input, exp_import, exp_path):
+    c = tosca.yaml2python.Convert(MagicMock(path="/path/to/including_file.yaml"))
+
+    output = c.convert_import(test_input)
+
+    # generated import
+    assert output[0].strip() == exp_import
+    # import path
+    assert output[1] == exp_path
+
+
+@pytest.mark.parametrize(
+    "test_input,exp_import,exp_path",
+    [
+        (
+            dict(repository="repo", file="foo.yaml"),
+            "from tosca_repositories.repo.foo import *",
+            "tosca_repositories/repo/foo",
+        ),
+        (
+            dict(repository="repo", file="subdir/foo.yaml"),
+            "from tosca_repositories.repo.subdir.foo import *",
+            "tosca_repositories/repo/subdir/foo",
+        ),
+        (
+            dict(repository="repo", file="foo.yaml", namespace_prefix="tosca.ns"),
+            "from tosca_repositories.repo import foo as tosca.ns",
+            "tosca_repositories/repo/foo",
+        ),
+        (
+            dict(
+                repository="repo", file="subdir/foo.yaml", namespace_prefix="tosca.ns"
+            ),
+            "from tosca_repositories.repo.subdir import foo as tosca.ns",
+            "tosca_repositories/repo/subdir/foo",
+        ),
+        (
+            dict(repository="repo", file="foo.yaml", namespace_prefix="ns"),
+            "from tosca_repositories.repo import foo as ns",
+            "tosca_repositories/repo/foo",
+        ),
+    ],
+)
+# patch repo lookup so we don't need to write the whole template
+def test_convert_import_with_repo(test_input, exp_import, exp_path):
+    with patch.object(
+        tosca.yaml2python.Convert,
+        "find_repository",
+        return_value=(
+            f"tosca_repositories.{test_input.get('repository')}",
+            f"tosca_repositories/{test_input.get('repository')}",
+        ),
+    ):
+        c = tosca.yaml2python.Convert(MagicMock())
+
+        output = c.convert_import(test_input)
+
+        # generated import
+        assert output[0].strip() == exp_import
+        # import path
+        assert output[1] == exp_path
+
+
 if __name__ == "__main__":
     dump = True
     test_builtin_generation()

tosca-package/tosca/yaml2python.py

@@ -15,7 +15,7 @@
 from dataclasses import MISSING
 import logging
 import os
-from pathlib import Path
+from pathlib import Path, PurePath
 import pprint
 import re
 import string
@@ -281,46 +281,57 @@ def find_repository(self, name) -> Tuple[str, str]:
             return name, url
         return name, name
 
-    def convert_import(self, tpl) -> Tuple[str, str]:
-        repository = tpl.get("repository")
-        import_path = os.path.dirname(self.template.path or "")
-        if repository:
-            module_name, import_path = self.find_repository(repository)
+
+
+    def convert_import(self, imp: Dict[str, str]) -> Tuple[str, str]:
+        "converts tosca yaml import dict (as `imp`) to python import statement"
+        repo = imp.get("repository")
+        file = imp.get("file")
+        namespace = imp.get("namespace_prefix")
+
+        # file is required by TOSCA spec, so crash and burn if we don't have it
+        assert file, "file is required for TOSCA imports"
+
+        # figure out loading path
+        filepath = PurePath(file)
+        dirname = filepath.parent
+        filename, tosca_name = self._get_name(filepath.stem) # filename w/o ext
+
+        if repo:
+            # generate repo import if repository: key given
+            module_name, _import_path = self.find_repository(repo)
+            import_path = PurePath(_import_path)
+
         else:
-            absolute = os.path.isabs(import_path)
-            module_name = "." if not absolute else ""
-
-        dirname, filename = os.path.split(tpl.get("file"))
-        # strip out file extension if present
-        before, sep, remainder = filename.rpartition(".")
-        file_name, tosca_name = self._get_name(before or remainder)
-        if dirname:
-            modpath = dirname.strip("/").replace("/", ".")
-            if module_name and module_name[-1] != ".":
-                module_name += "." + modpath
-            else:
-                module_name += modpath
-
-        import_path = os.path.join(import_path, dirname, file_name)
-
-        uri_prefix = tpl.get("namespace_prefix")
-        if uri_prefix:
-            uri_prefix, tosca_name = self._get_name(uri_prefix)
-            tosca_prefix = tosca_name or uri_prefix
-            self.import_prefixes[tosca_prefix] = uri_prefix
-        if module_name:
-            if uri_prefix:
-                import_stmt = f"from {module_name} import {file_name} as {uri_prefix}"
-            elif module_name != ".":
-                import_stmt = f"from {module_name}.{file_name} import *"
-            else:
-                import_stmt = f"from .{file_name} import *"
+            # otherwise assume local path
+            import_path = PurePath(self.template.path).parent
+            # prefix module_name with . for relative path
+            module_name = ""
+
+        # import should be .path.to.file
+        module_name = ".".join(
+            ['' if d == '..' else d for d in [
+                module_name,
+                *dirname.parts
+            ]]
+        )
+
+        # handle tosca namespace prefixes
+        if namespace:
+            namespace, ns_tosca_name = self._get_name(namespace)
+            self.import_prefixes[ns_tosca_name or namespace] = namespace
+
+        # generate import statement
+        if not namespace:
+            import_stmt = f"from {module_name}.{filename} import *"
         else:
-            if uri_prefix:
-                import_stmt = f"import {file_name} as {uri_prefix}"
-            else:
-                import_stmt = f"from {file_name} import *"
-        return import_stmt + "\n", import_path
+            import_stmt = f"from {module_name} import {filename} as {ns_tosca_name or namespace}"
+
+        # add path to file in repo to repo path
+        import_path = import_path / dirname / filename
+
+        return import_stmt + "\n", str(import_path)
+
 
     def convert_types(
         self, node_types: dict, section: str, namespace_prefix="", indent=""
@@ -1126,7 +1137,7 @@ def yaml_to_python(
 
     Returns:
         str: The converted Python source code.
-    """    
+    """
     return convert_service_template(
         ToscaTemplate(
             path=yaml_path, yaml_dict_tpl=tosca_dict, import_resolver=import_resolver

unfurl/localenv.py

@@ -1386,18 +1386,27 @@ def link_repo(self, base_path: str, name: str, url: str, revision):
             base_path = get_base_dir(base_path)
         else:
             base_path = os.getcwd()
+
         assert name.isidentifier(), name
         repo = self.find_git_repo(url, revision)
         assert repo, url
-        repo_root = Path(base_path) / "tosca_repositories"
-        if not repo_root.exists():
-            os.mkdir(repo_root)
-            with open(repo_root / ".gitignore", "w") as gi:
+
+        tosca_repos_root = Path(base_path) / "tosca_repositories"
+        # ensure t_r and its gitignore exist
+        if not tosca_repos_root.exists():
+            os.mkdir(tosca_repos_root)
+            with open(tosca_repos_root / ".gitignore", "w") as gi:
                 gi.write("*")
-        target = repo_root / name
-        if os.path.exists(target):
-            os.unlink(target)
-        os.symlink(repo.working_dir, repo_root / name, True)
+
+        target = tosca_repos_root / name
+        # remove/recreate to ensure symlink is correct
+        if target.exists():
+            target.unlink()
+
+        # use os.path.relpath as Path.relative_to only accepts strict subpaths
+        rel_repo_path = os.path.relpath(repo.working_dir, tosca_repos_root)
+        target.symlink_to(rel_repo_path)
+
         return repo.working_dir
 
     def map_value(self, val: Any, env_rules: Optional[dict]) -> Any: