fix: keep user client config when switching to direct connect endpoint

README.md

@@ -327,6 +327,10 @@ driver = webdriver.Remote(
 )
 ```
 
+Only the endpoint the client talks to is replaced. The rest of the given `client_config`,
+such as `timeout`, `init_args_for_pool_manager`, proxy, certificates and authentication
+settings, keeps being applied to the requests sent to the direct connect endpoint.
+
 ## Relax SSL validation
 
 `strict_ssl` option allows you to send commands to an invalid certificate host like a self-signed one.

appium/webdriver/webdriver.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
 
@@ -318,11 +319,19 @@ def _update_command_executor(self, keep_alive: bool) -> None:
         executor = f'{protocol}://{hostname}:{port}{path}'
 
         logger.debug('Updated request endpoint to %s', executor)
+
         # Override command executor.
+        # The client configuration given by a user, e.g. the read timeout, the proxy or
+        # the authentication credentials, must be kept as-is. Only the endpoint the client
+        # talks to changes, thus a copy of the current configuration is reused instead of
+        # building a brand-new one out of the endpoint URL.
+        client_config = copy.copy(self.command_executor.client_config)
+        client_config.remote_server_addr = executor
+        client_config.keep_alive = keep_alive
         if isinstance(self.command_executor, AppiumConnection):  # type: ignore
-            self.command_executor = AppiumConnection(executor, keep_alive=keep_alive)
+            self.command_executor = AppiumConnection(client_config=client_config)
         else:
-            self.command_executor = RemoteConnection(executor, keep_alive=keep_alive)
+            self.command_executor = RemoteConnection(client_config=client_config)
         self._add_commands()
 
     # https://github.com/SeleniumHQ/selenium/blob/06fdf2966df6bca47c0ae45e8201cd30db9b9a49/py/selenium/webdriver/remote/webdriver.py#L277

test/unit/webdriver/webdriver_test.py

@@ -136,6 +136,63 @@ def test_create_session_register_uridirect(self):
         assert driver.contexts == ['NATIVE_APP', 'CHROMIUM']
         assert isinstance(driver.command_executor, AppiumConnection)
 
+    @httpretty.activate
+    def test_create_session_register_uridirect_keeps_client_config(self):
+        """The client configuration given by a user must survive the direct connect switch.
+        https://github.com/appium/python-client/issues/855
+        """
+        httpretty.register_uri(
+            httpretty.POST,
+            f'{SERVER_URL_BASE}/session',
+            body=json.dumps(
+                {
+                    'sessionId': 'session-id',
+                    'capabilities': {
+                        'deviceName': 'Android Emulator',
+                        'directConnectProtocol': 'http',
+                        'directConnectHost': 'localhost2',
+                        'directConnectPort': 4800,
+                        'directConnectPath': '/special/path/wd/hub',
+                    },
+                }
+            ),
+        )
+
+        desired_caps = {
+            'platformName': 'Android',
+            'deviceName': 'Android Emulator',
+            'app': 'path/to/app',
+            'automationName': 'UIAutomator2',
+        }
+        client_config = AppiumClientConfig(
+            remote_server_addr=SERVER_URL_BASE,
+            direct_connection=True,
+            timeout=5,
+            username='user',
+            password='pass',
+            user_agent='custom-agent',
+            init_args_for_pool_manager={'init_args_for_pool_manager': {'retries': 3}},
+        )
+        driver = webdriver.Remote(
+            SERVER_URL_BASE,
+            options=UiAutomator2Options().load_capabilities(desired_caps),
+            client_config=client_config,
+        )
+
+        new_client_config = driver.command_executor.client_config
+        assert isinstance(new_client_config, AppiumClientConfig)
+        assert new_client_config.remote_server_addr == 'http://localhost2:4800/special/path/wd/hub'
+        assert new_client_config.timeout == 5
+        assert new_client_config.username == 'user'
+        assert new_client_config.password == 'pass'
+        assert new_client_config.user_agent == 'custom-agent'
+        assert new_client_config.init_args_for_pool_manager == {'init_args_for_pool_manager': {'retries': 3}}
+        assert new_client_config.direct_connection
+        assert new_client_config.keep_alive == client_config.keep_alive
+
+        # the configuration instance given by a user must not be modified in-place
+        assert client_config.remote_server_addr == SERVER_URL_BASE
+
     @httpretty.activate
     def test_create_session_register_uridirect_no_direct_connect_path(self):
         httpretty.register_uri(