(self, plugin_name)
| 1152 | return pkt |
| 1153 | |
| 1154 | def _get_auth_plugin_handler(self, plugin_name): |
| 1155 | plugin_class = self._auth_plugin_map.get(plugin_name) |
| 1156 | if not plugin_class and isinstance(plugin_name, bytes): |
| 1157 | plugin_class = self._auth_plugin_map.get(plugin_name.decode("ascii")) |
| 1158 | if plugin_class: |
| 1159 | try: |
| 1160 | handler = plugin_class(self) |
| 1161 | except TypeError: |
| 1162 | raise err.OperationalError( |
| 1163 | CR.CR_AUTH_PLUGIN_CANNOT_LOAD, |
| 1164 | f"Authentication plugin '{plugin_name}'" |
| 1165 | f" not loaded: - {plugin_class!r} cannot be constructed with connection object", |
| 1166 | ) |
| 1167 | else: |
| 1168 | handler = None |
| 1169 | return handler |
| 1170 | |
| 1171 | # _mysql support |
| 1172 | def thread_id(self): |
no test coverage detected