connection plugins: extras fix (#83353)

Currently we match the load name, which can be an fqcn, but most users expect the 'naked' name
Now plugins can declare that name by setting _extras_prefix property or fallback to 'non fqcn' if no extras prefix

changelogs/fragments/extras_fix.yml

@@ -0,0 +1,4 @@
+bugfixes:
+  - connection plugins using the 'extras' option feature would need variables to match the plugin's loaded name,
+    sometimes requiring fqcn, which is not the same as the documented/declared/expected variables.
+    Now we fall back to the 'basename' of the fqcn, but plugin authors can still set the expected value directly.

lib/ansible/executor/task_executor.py

@@ -1068,7 +1068,7 @@ class TaskExecutor:
         # add extras if plugin supports them
         if getattr(self._connection, 'allow_extras', False):
             for k in variables:
-                if k.startswith('ansible_%s_' % self._connection._load_name) and k not in options:
+                if k.startswith('ansible_%s_' % self._connection.extras_prefix) and k not in options:
                     options['_extras'][k] = templar.template(variables[k])
 
         task_keys = self._task.dump_attrs()

lib/ansible/plugins/__init__.py

@@ -50,16 +50,23 @@ def get_plugin_class(obj):
 
 class AnsiblePlugin(ABC):
 
-    # allow extra passthrough parameters
-    allow_extras = False
-
     # Set by plugin loader
     _load_name: str
 
+    # allow extra passthrough parameters
+    allow_extras: bool = False
+    _extras_prefix: str | None = None
+
     def __init__(self):
         self._options = {}
         self._defs = None
 
+    @property
+    def extras_prefix(self):
+        if not self._extras_prefix:
+            self._extras_prefix = self._load_name.split('.')[-1]
+        return self._extras_prefix
+
     def matches_name(self, possible_names):
         possible_fqcns = set()
         for name in possible_names:

test/units/plugins/connection/test_psrp.py

@@ -218,7 +218,8 @@ class TestConnectionPSRP(object):
         pc = PlayContext()
         new_stdin = StringIO()
 
-        conn = connection_loader.get('psrp', pc, new_stdin)
+        for conn_name in ('psrp', 'ansible.legacy.psrp'):
+            conn = connection_loader.get(conn_name, pc, new_stdin)
             conn.set_options(var_options={'_extras': {'ansible_psrp_mock_test3': True}})
 
             mock_display = MagicMock()