Work on module path + schema file connection

Author: maddenp-cu

src/uwtools/api/driver.py

@@ -61,8 +61,10 @@ def execute(
     :param stdin_ok: OK to read from stdin?
     :return: ``True`` if task completes without raising an exception.
     """
-    if not (class_ := _get_driver_class(module, classname)):
+    class_, module_path = _get_driver_class(module, classname)
+    if not class_:
         return False
+    assert module_path is not None
     args = dict(locals())
     accepted = set(getfullargspec(class_).args)
     non_optional = {STR.cycle, STR.leadtime}
@@ -78,7 +80,7 @@ def execute(
         config=ensure_data_source(config, bool(stdin_ok)),
         dry_run=dry_run,
         key_path=key_path,
-        schema_file=schema_file or Path(module).with_suffix(".jsonschema"),
+        schema_file=schema_file or module_path.with_suffix(".jsonschema"),
     )
     required = non_optional & accepted
     for arg in sorted([STR.batch, *required]):
@@ -93,46 +95,50 @@ def execute(
     return True
 
 
-def tasks(module: str, classname: str) -> dict[str, str]:
+def tasks(module: Union[Path, str], classname: str) -> dict[str, str]:
     """
     Returns a mapping from task names to their one-line descriptions.
 
     :param module: Name of driver module.
     :param classname: Name of driver class to instantiate.
     """
-    if not (class_ := _get_driver_class(module, classname)):
+    class_, _ = _get_driver_class(module, classname)
+    if not class_:
         log.error("Could not get tasks from class %s in module %s", classname, module)
         return {}
     return _tasks(class_)
 
 
-def _get_driver_class(module: Union[Path, str], classname: str) -> Optional[Type]:
+def _get_driver_class(
+    module: Union[Path, str], classname: str
+) -> tuple[Optional[Type], Optional[Path]]:
     """
     Returns the driver class.
 
     :param module: Name of driver module to load.
     :param classname: Name of driver class to instantiate.
     """
-    module = str(module)
-    if not (m := _get_driver_module_explicit(module)):
-        if not (m := _get_driver_module_implicit(module)):
+    if not (m := _get_driver_module_explicit(Path(module))):
+        if not (m := _get_driver_module_implicit(str(module))):
             log.error("Could not load module %s", module)
-            return None
+            return None, None
+    assert m.__file__ is not None
+    module_path = Path(m.__file__)
     if hasattr(m, classname):
         c: Type = getattr(m, classname)
-        return c
+        return c, module_path
     log.error("Module %s has no class %s", module, classname)
-    return None
+    return None, module_path
 
 
-def _get_driver_module_explicit(module: str) -> Optional[ModuleType]:
+def _get_driver_module_explicit(module: Path) -> Optional[ModuleType]:
     """
     Returns the named module found via explicit lookup of given path.
 
     :param module: Name of driver module to load.
     """
     log.debug("Loading module %s", module)
-    if spec := spec_from_file_location(Path(module).name, module):
+    if spec := spec_from_file_location(module.name, module):
         m = module_from_spec(spec)
         if loader := spec.loader:
             try:
@@ -150,8 +156,8 @@ def _get_driver_module_implicit(module: str) -> Optional[ModuleType]:
 
     :param module: Name of driver module to load.
     """
+    log.debug("Loading module %s from sys.path", module)
     try:
-        log.debug("Loading module %s from sys.path", module)
         return import_module(module)
     except Exception:  # pylint: disable=broad-exception-caught
         return None

src/uwtools/tests/api/test_driver.py

@@ -117,82 +117,86 @@ def test_tasks_fail_no_cycle(args, caplog, kwargs):
     assert logged(caplog, "%s requires argument '%s'" % (args.classname, "cycle"))
 
 
-def test_tasks_pass(args):
-    tasks = driver_api.tasks(classname=args.classname, module=args.module)
+@mark.parametrize("f", [Path, str])
+def test_tasks_pass(args, f):
+    tasks = driver_api.tasks(classname=args.classname, module=f(args.module))
     assert tasks["eighty_eight"] == "88"
 
 
 def test__get_driver_class_explicit_fail_bad_class(caplog, args):
     log.setLevel(logging.DEBUG)
     bad_class = "BadClass"
-    c = driver_api._get_driver_class(classname=bad_class, module=args.module)
+    c, module_path = driver_api._get_driver_class(classname=bad_class, module=args.module)
     assert c is None
+    assert module_path == args.module
     assert logged(caplog, "Module %s has no class %s" % (args.module, bad_class))
 
 
 def test__get_driver_class_explicit_fail_bad_name(caplog, args):
     log.setLevel(logging.DEBUG)
-    bad_name = "bad_name"
-    c = driver_api._get_driver_class(classname=args.classname, module=bad_name)
+    bad_name = Path("bad_name")
+    c, module_path = driver_api._get_driver_class(classname=args.classname, module=bad_name)
     assert c is None
+    assert module_path is None
     assert logged(caplog, "Could not load module %s" % bad_name)
 
 
 def test__get_driver_class_explicit_fail_bad_path(caplog, args, tmp_path):
     log.setLevel(logging.DEBUG)
     module = tmp_path / "not.py"
-    c = driver_api._get_driver_class(classname=args.classname, module=module)
+    c, module_path = driver_api._get_driver_class(classname=args.classname, module=module)
     assert c is None
+    assert module_path is None
     assert logged(caplog, "Could not load module %s" % module)
 
 
 def test__get_driver_class_explicit_fail_bad_spec(caplog, args):
     log.setLevel(logging.DEBUG)
     with patch.object(driver_api, "spec_from_file_location", return_value=None):
-        c = driver_api._get_driver_class(classname=args.classname, module=args.module)
+        c, module_path = driver_api._get_driver_class(classname=args.classname, module=args.module)
     assert c is None
+    assert module_path is None
     assert logged(caplog, "Could not load module %s" % args.module)
 
 
 def test__get_driver_class_explicit_pass(args):
     log.setLevel(logging.DEBUG)
-    c = driver_api._get_driver_class(classname=args.classname, module=args.module)
+    c, module_path = driver_api._get_driver_class(classname=args.classname, module=args.module)
     assert c
     assert c.__name__ == "TestDriver"
+    assert module_path == args.module
 
 
 def test__get_driver_class_implicit_pass(args):
     log.setLevel(logging.DEBUG)
     with patch.object(Path, "cwd", return_value=fixture_path()):
-        c = driver_api._get_driver_class(classname=args.classname, module=args.module)
-        assert c
-        assert c.__name__ == "TestDriver"
+        c, module_path = driver_api._get_driver_class(classname=args.classname, module=args.module)
+    assert c
+    assert c.__name__ == "TestDriver"
+    assert module_path == args.module
 
 
 def test__get_driver_module_explicit_absolute_fail(args):
     assert args.module.is_absolute()
-    module = str(args.module.with_suffix(".bad"))
+    module = args.module.with_suffix(".bad")
     assert not driver_api._get_driver_module_explicit(module=module)
 
 
 def test__get_driver_module_explicit_absolute_pass(args):
     assert args.module.is_absolute()
-    module = str(args.module)
-    assert driver_api._get_driver_module_explicit(module=module)
+    assert driver_api._get_driver_module_explicit(module=args.module)
 
 
 def test__get_driver_module_explicit_relative_fail(args):
     args.module = Path(os.path.relpath(args.module)).with_suffix(".bad")
     assert not args.module.is_absolute()
-    module = str(args.module)
-    assert not driver_api._get_driver_module_explicit(module=module)
+    assert not driver_api._get_driver_module_explicit(module=args.module)
 
 
 def test__get_driver_module_explicit_relative_pass(args):
     args.module = Path(os.path.relpath(args.module))
     assert not args.module.is_absolute()
-    module = str(args.module)
-    assert driver_api._get_driver_module_explicit(module=module)
+    assert driver_api._get_driver_module_explicit(module=args.module)
 
 
 def test__get_driver_module_implicit_pass_full_package():