User defined mapping: add option and handling for multiple mapping files

fawltydeps/cli_parser.py

@@ -166,6 +166,8 @@ def populate_parser_paths_options(parser: argparse._ActionsContainer) -> None:
     )
     parser.add_argument(
         "--custom-mapping-file",
+        nargs="+",
+        action="union",
         type=Path,
         metavar="FILE_PATH",
         help=(

fawltydeps/main.py

@@ -109,7 +109,7 @@ def resolved_deps(self) -> Dict[str, Package]:
         """The resolved mapping of dependency names to provided import names."""
         return resolve_dependencies(
             (dep.name for dep in self.declared_deps),
-            custom_mapping_file=self.settings.custom_mapping_file,
+            custom_mapping_files=self.settings.custom_mapping_file,
             custom_mapping=self.settings.custom_mapping,
             pyenv_path=self.settings.pyenv,
             install_deps=self.settings.install_deps,

fawltydeps/packages.py

@@ -9,6 +9,7 @@
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from enum import Enum
+from itertools import chain
 from pathlib import Path
 from typing import Dict, Iterable, Iterator, List, Optional, Set
 
@@ -116,69 +117,71 @@ class UserDefinedMapping(BasePackageResolver):
 
     def __init__(
         self,
-        mapping_path: Optional[Path] = None,
+        mapping_paths: Optional[Set[Path]] = None,
         custom_mapping: Optional[CustomMapping] = None,
     ) -> None:
-        self.mapping_path = mapping_path
-        if self.mapping_path and not self.mapping_path.is_file():
-            raise UnparseablePathException(
-                ctx="Given mapping path is not a file.", path=self.mapping_path
-            )
+        self.mapping_paths = mapping_paths or set()
+        for path in self.mapping_paths:
+            if not path.is_file():
+                raise UnparseablePathException(
+                    ctx="Given mapping path is not a file.", path=path
+                )
         self.custom_mapping = custom_mapping
         # We enumerate packages declared in the mapping _once_ and cache the result here:
         self._packages: Optional[Dict[str, Package]] = None
 
+    @staticmethod
+    def accumulate_mappings(custom_mappings: Iterable[CustomMapping]) -> CustomMapping:
+        """Merge mapping dictionaries and normalise key (package) names."""
+        result: CustomMapping = {}
+        for name, imports in chain.from_iterable(cm.items() for cm in custom_mappings):
+            normalised_name = Package.normalize_name(name)
+            if normalised_name in result:
+                logger.info(
+                    "Mapping for %s already found. Import names "
+                    "from the second mapping are appended to ones "
+                    "found in the first mapping.",
+                    normalised_name,
+                )
+            result.setdefault(normalised_name, []).extend(imports)
+        return result
+
     @property
     @calculated_once
     def packages(self) -> Dict[str, Package]:
         """Gather a custom mapping given by a user.
 
         Mapping may come from two sources:
-        * _mapping_path_ which is a file in a given path, which is parsed to a ditionary
-        * _custom_mapping_ which is a dictionary of package to imports mapping.
+        * custom_mapping: an already-parsed CustomMapping, i.e. a dict mapping
+          package names to imports.
+        * mapping_paths: set of file paths, where each file contains a TOML-
+          formatted CustomMapping.
 
         This enumerates the available packages  _once_, and caches the result for
         the remainder of this object's life in _packages.
         """
-        custom_mapping_from_file: Dict[str, List[str]] = {}
 
-        if self.mapping_path is not None:
-            logger.debug(f"Loading user-defined mapping from {self.mapping_path}")
-            with open(self.mapping_path, "rb") as mapping_file:
-                custom_mapping_from_file = tomllib.load(mapping_file)
+        def _custom_mappings() -> Iterator[CustomMapping]:
+            if self.custom_mapping is not None:
+                logger.debug("Applying user-defined mapping from settings.")
+                yield self.custom_mapping
 
-        mapping = {
-            Package.normalize_name(name): Package(
+            if self.mapping_paths is not None:
+                for path in self.mapping_paths:
+                    logger.debug(f"Loading user-defined mapping from {path}")
+                    with open(path, "rb") as mapping_file:
+                        yield tomllib.load(mapping_file)
+
+        custom_mapping = self.accumulate_mappings(_custom_mappings())
+
+        return {
+            name: Package(
                 name,
                 {DependenciesMapping.USER_DEFINED: set(imports)},
             )
-            for name, imports in custom_mapping_from_file.items()
+            for name, imports in custom_mapping.items()
         }
 
-        if self.custom_mapping is not None:
-            logger.debug("Applying user-defined mapping from settings.")
-
-            for name, imports in self.custom_mapping.items():
-                normalised_name = Package.normalize_name(name)
-                if normalised_name in mapping:
-                    logger.info(
-                        "Mapping for %s already found in %s. Import names "
-                        "from the configuration file are appended to ones "
-                        "found in the mapping file.",
-                        normalised_name,
-                        self.mapping_path,
-                    )
-                    mapping[normalised_name].add_import_names(
-                        *imports, mapping=DependenciesMapping.USER_DEFINED
-                    )
-                else:
-                    mapping[normalised_name] = Package(
-                        name,
-                        {DependenciesMapping.USER_DEFINED: set(imports)},
-                    )
-
-        return mapping
-
     def lookup_packages(self, package_names: Set[str]) -> Dict[str, Package]:
         """Convert package names to locally available Package objects."""
         return {
@@ -354,7 +357,7 @@ def lookup_packages(self, package_names: Set[str]) -> Dict[str, Package]:
 
 def resolve_dependencies(
     dep_names: Iterable[str],
-    custom_mapping_file: Optional[Path] = None,
+    custom_mapping_files: Optional[Set[Path]] = None,
     custom_mapping: Optional[CustomMapping] = None,
     pyenv_path: Optional[Path] = None,
     install_deps: bool = False,
@@ -375,12 +378,11 @@ def resolve_dependencies(
     # each resolver in order until one of them returns a Package object. At
     # that point we are happy, and don't consult any of the later resolvers.
     resolvers: List[BasePackageResolver] = []
-    if custom_mapping_file or custom_mapping:
-        resolvers.append(
-            UserDefinedMapping(
-                mapping_path=custom_mapping_file, custom_mapping=custom_mapping
-            )
+    resolvers.append(
+        UserDefinedMapping(
+            mapping_paths=custom_mapping_files or set(), custom_mapping=custom_mapping
         )
+    )
 
     resolvers.append(LocalPackageResolver(pyenv_path))
     if install_deps:

fawltydeps/settings.py

@@ -110,7 +110,7 @@ class Settings(BaseSettings):  # type: ignore
     code: Set[PathOrSpecial] = {Path(".")}
     deps: Set[Path] = {Path(".")}
     pyenv: Optional[Path] = None
-    custom_mapping_file: Optional[Path] = None
+    custom_mapping_file: Set[Path] = set()
     custom_mapping: Optional[CustomMapping] = None
     ignore_undeclared: Set[str] = set()
     ignore_unused: Set[str] = set()

tests/test_cmdline.py

@@ -36,7 +36,7 @@ def make_json_settings_dict(**kwargs):
         "code": ["."],
         "deps": ["."],
         "pyenv": None,
-        "custom_mapping_file": None,
+        "custom_mapping_file": [],
         "custom_mapping": None,
         "output_format": "human_summary",
         "ignore_undeclared": [],
@@ -848,7 +848,7 @@ def test_cmdline_on_ignored_undeclared_option(
                 # code = ['.']
                 deps = ['foobar']
                 # pyenv = ...
-                # custom_mapping_file = ...
+                # custom_mapping_file = []
                 # ignore_undeclared = []
                 # ignore_unused = []
                 # deps_parser_choice = ...
@@ -872,7 +872,7 @@ def test_cmdline_on_ignored_undeclared_option(
                 # code = ['.']
                 # deps = ['.']
                 pyenv = 'None'
-                # custom_mapping_file = ...
+                # custom_mapping_file = []
                 # ignore_undeclared = []
                 # ignore_unused = []
                 # deps_parser_choice = ...

tests/test_packages.py

@@ -146,27 +146,82 @@ def test_package__both_mappings():
     assert p.import_names == {"foobar", "foo", "bar", "baz"}
 
 
-def test_user_defined_mapping__well_formated_input_file__parses_correctly(
-    tmp_path,
-):
-    custom_mapping_file = tmp_path / "mapping.toml"
-    tmp_path = custom_mapping_file.write_text(
-        dedent(
-            """\
+@pytest.mark.parametrize(
+    "mapping_files_content,custom_mapping,expect",
+    [
+        pytest.param(
+            [
+                """\
                 apache-airflow = ["airflow"]
                 attrs = ["attr", "attrs"]
             """
-        )
-    )
+            ],
+            None,
+            {"apache_airflow": ["airflow"], "attrs": ["attr", "attrs"]},
+            id="well_formated_input_file__parses_correctly",
+        ),
+        pytest.param(
+            [
+                """\
+                apache-airflow = ["airflow"]
+                attrs = ["attr", "attrs"]
+                """,
+                """\
+                apache-airflow = ["baz"]
+                foo = ["bar"]
+                """,
+            ],
+            None,
+            {
+                "apache_airflow": ["airflow", "baz"],
+                "attrs": ["attr", "attrs"],
+                "foo": ["bar"],
+            },
+            id="well_formated_input_2files__parses_correctly",
+        ),
+        pytest.param(
+            [
+                """\
+                apache-airflow = ["airflow"]
+                attrs = ["attr", "attrs"]
+                """,
+                """\
+                apache-airflow = ["baz"]
+                foo = ["bar"]
+                """,
+            ],
+            {"apache-airflow": ["unicorn"]},
+            {
+                "apache_airflow": ["airflow", "baz", "unicorn"],
+                "attrs": ["attr", "attrs"],
+                "foo": ["bar"],
+            },
+            id="well_formated_input_2files_and_config__parses_correctly",
+        ),
+    ],
+)
+def test_user_defined_mapping__well_formated_input_file__parses_correctly(
+    mapping_files_content,
+    custom_mapping,
+    expect,
+    tmp_path,
+):
+    custom_mapping_files = set()
+    for i, mapping in enumerate(mapping_files_content):
+        custom_mapping_file = tmp_path / f"mapping{i}.toml"
+        custom_mapping_file.write_text(dedent(mapping))
+        custom_mapping_files.add(custom_mapping_file)
 
-    udm = UserDefinedMapping(custom_mapping_file)
-    mapped_packages = udm.packages
-    assert set(mapped_packages.keys()) == {"apache_airflow", "attrs"}
+    udm = UserDefinedMapping(
+        mapping_paths=custom_mapping_files, custom_mapping=custom_mapping
+    )
+    mapped_packages = {k: sorted(list(v.import_names)) for k, v in udm.packages.items()}
+    assert mapped_packages == expect
 
 
 def test_user_defined_mapping__input_is_no_file__raises_unparsable_path_exeption():
     with pytest.raises(UnparseablePathException):
-        UserDefinedMapping(SAMPLE_PROJECTS_DIR)
+        UserDefinedMapping({SAMPLE_PROJECTS_DIR})
 
 
 def test_user_defined_mapping__no_input__returns_empty_mapping():
@@ -278,7 +333,7 @@ def test_LocalPackageResolver_lookup_packages(dep_name, expect_import_names):
             },
             {
                 "apache_airflow": Package(
-                    "apache-airflow", {DependenciesMapping.USER_DEFINED: {"airflow"}}
+                    "apache_airflow", {DependenciesMapping.USER_DEFINED: {"airflow"}}
                 ),
                 "pip": Package("pip", {DependenciesMapping.LOCAL_ENV: {"pip"}}),
                 "pandas": Package("pandas", {DependenciesMapping.IDENTITY: {"pandas"}}),
@@ -299,7 +354,7 @@ def test_LocalPackageResolver_lookup_packages(dep_name, expect_import_names):
             {"configuration": {"apache-airflow": ["airflow"]}},
             {
                 "apache_airflow": Package(
-                    "apache-airflow", {DependenciesMapping.USER_DEFINED: {"airflow"}}
+                    "apache_airflow", {DependenciesMapping.USER_DEFINED: {"airflow"}}
                 ),
                 "pip": Package("pip", {DependenciesMapping.LOCAL_ENV: {"pip"}}),
                 "pandas": Package("pandas", {DependenciesMapping.IDENTITY: {"pandas"}}),
@@ -314,7 +369,7 @@ def test_LocalPackageResolver_lookup_packages(dep_name, expect_import_names):
             },
             {
                 "apache_airflow": Package(
-                    "apache-airflow",
+                    "apache_airflow",
                     {DependenciesMapping.USER_DEFINED: {"airflow", "foo", "bar"}},
                 ),
                 "pip": Package("pip", {DependenciesMapping.LOCAL_ENV: {"pip"}}),
@@ -327,18 +382,19 @@ def test_LocalPackageResolver_lookup_packages(dep_name, expect_import_names):
 def test_resolve_dependencies__focus_on_mappings(
     dep_names, user_mapping, expected, tmp_path
 ):
-    custom_mapping_file = None
+    custom_mapping_files = set()
     custom_mapping = None
     if user_mapping is not None:
         custom_mapping = user_mapping.get("configuration")
         if "file" in user_mapping:
             custom_mapping_file = tmp_path / "mapping.toml"
             custom_mapping_file.write_text(dedent(user_mapping["file"]))
+            custom_mapping_files = {custom_mapping_file}
 
     assert (
         resolve_dependencies(
             dep_names,
-            custom_mapping_file=custom_mapping_file,
+            custom_mapping_files=custom_mapping_files,
             custom_mapping=custom_mapping,
         )
         == expected

tests/test_settings.py

@@ -28,7 +28,7 @@
     code={Path(".")},
     deps={Path(".")},
     pyenv=None,
-    custom_mapping_file=None,
+    custom_mapping_file=set(),
     custom_mapping=None,
     output_format=OutputFormat.HUMAN_SUMMARY,
     ignore_undeclared=set(),
@@ -231,12 +231,12 @@ class SettingsTestVector:
         config=dict(
             actions=["list_deps"],
             deps=["my_requirements.txt"],
-            custom_mapping_file="mapping.toml",
+            custom_mapping_file=["mapping.toml"],
         ),
         expect=make_settings_dict(
             actions={Action.LIST_DEPS},
             deps={Path("my_requirements.txt")},
-            custom_mapping_file=Path("mapping.toml"),
+            custom_mapping_file={Path("mapping.toml")},
         ),
     ),
     SettingsTestVector(
@@ -259,20 +259,20 @@ class SettingsTestVector:
             deps=["my_requirements.txt"],
             custom_mapping={"package": ["foo", "bar"]},
         ),
-        cmdline=dict(custom_mapping_file="mapping.toml"),
+        cmdline=dict(custom_mapping_file=["mapping.toml"]),
         expect=make_settings_dict(
             actions={Action.LIST_DEPS},
             deps={Path("my_requirements.txt")},
             custom_mapping={"package": ["foo", "bar"]},
-            custom_mapping_file=Path("mapping.toml"),
+            custom_mapping_file={Path("mapping.toml")},
         ),
     ),
     SettingsTestVector(
         "config_file_with_mapping_and_cli__cli_mapping_overrides_config",
-        config=dict(custom_mapping_file="foo.toml"),
-        cmdline=dict(custom_mapping_file="mapping.toml"),
+        config=dict(custom_mapping_file=["foo.toml"]),
+        cmdline=dict(custom_mapping_file=["mapping.toml"]),
         expect=make_settings_dict(
-            custom_mapping_file=Path("mapping.toml"),
+            custom_mapping_file={Path("mapping.toml")},
         ),
     ),
     SettingsTestVector(