Require TaskDef.taskName to be fully-qualified.

Documentation and code that tried to permit it to not be
fully-qualified was inconsistent with code that calls doImport on it,
and with round-tripping TaskDef objects through pipeline I/O.

python/lsst/pipe/base/pipeline.py

@@ -43,6 +43,7 @@
 #  Imports for other modules --
 from lsst.daf.butler import DatasetType, NamedValueSet, Registry, SkyPixDimension, ButlerURI
 from lsst.utils import doImport
+from lsst.utils.introspection import get_full_type_name
 from .configOverrides import ConfigOverrides
 from .connections import iterConnections
 from .pipelineTask import PipelineTask
@@ -96,10 +97,8 @@ class TaskDef:
     Attributes
     ----------
     taskName : `str`, optional
-        `PipelineTask` class name, currently it is not specified whether this
-        is a fully-qualified name or partial name (e.g. ``module.TaskClass``).
-        Framework should be prepared to handle all cases.  If not provided,
-        ``taskClass`` must be, and ``taskClass.__name__`` is used.
+        The fully-qualified `PipelineTask` class name.  If not provided,
+        ``taskClass`` must be.
     config : `lsst.pex.config.Config`, optional
         Instance of the configuration class corresponding to this task class,
         usually with all overrides applied. This config will be frozen.  If
@@ -117,7 +116,7 @@ def __init__(self, taskName=None, config=None, taskClass=None, label=None):
         if taskName is None:
             if taskClass is None:
                 raise ValueError("At least one of `taskName` and `taskClass` must be provided.")
-            taskName = taskClass.__name__
+            taskName = get_full_type_name(taskClass)
         if config is None:
             if taskClass is None:
                 raise ValueError("`taskClass` must be provided if `config` is not.")
@@ -606,7 +605,7 @@ def _buildTaskDef(self, label: str) -> TaskDef:
         if (taskIR := self._pipelineIR.tasks.get(label)) is None:
             raise NameError(f"Label {label} does not appear in this pipeline")
         taskClass = doImport(taskIR.klass)
-        taskName = taskClass.__qualname__
+        taskName = get_full_type_name(taskClass)
         config = taskClass.ConfigClass()
         overrides = ConfigOverrides()
         if self._pipelineIR.instrument is not None:

tests/test_pipeline.py

@@ -59,8 +59,8 @@ def testInitial(self):
         pipeline = makeSimplePipeline(2)
         self.assertEqual(len(pipeline), 2)
         expandedPipeline = list(pipeline.toExpandedPipeline())
-        self.assertEqual(expandedPipeline[0].taskName, "AddTask")
-        self.assertEqual(expandedPipeline[1].taskName, "AddTask")
+        self.assertEqual(expandedPipeline[0].taskName, "lsst.pipe.base.tests.simpleQGraph.AddTask")
+        self.assertEqual(expandedPipeline[1].taskName, "lsst.pipe.base.tests.simpleQGraph.AddTask")
         self.assertEqual(expandedPipeline[0].taskClass, AddTask)
         self.assertEqual(expandedPipeline[1].taskClass, AddTask)
         self.assertEqual(expandedPipeline[0].label, "task0")