infer auth method without request

vespa/application.py

@@ -8,7 +8,6 @@
 from concurrent.futures import ThreadPoolExecutor, Future, as_completed
 from queue import Queue, Empty
 import threading
-import requests
 from requests import Session
 from requests.models import Response
 from requests.exceptions import ConnectionError, HTTPError, JSONDecodeError
@@ -125,7 +124,7 @@ def __init__(
             token = environ.get(VESPA_CLOUD_SECRET_TOKEN, None)
             if token is not None:
                 self.vespa_cloud_secret_token = token
-        self.auth_method = None
+        self.auth_method = self._get_valid_auth_method()
 
     def asyncio(
         self, connections: Optional[int] = 8, total_timeout: int = 10
@@ -256,55 +255,20 @@ def _get_valid_auth_method(self) -> Optional[str]:
 
         :return: Auth method used for Vespa connection. Either 'token','mtls_key_cert','mtls_cert' or 'http'. None if not able to authenticate.
         """
-        endpoint = f"{self.end_point}/ApplicationStatus"
-
-        if self.auth_method:
-            return self.auth_method
-
-        # Plain HTTP
-        response = requests.get(endpoint, headers=self.base_headers)
-        if response.status_code == 200:
-            print(
-                f"Using plain HTTP to connect to Vespa endpoint {self.end_point}",
-                file=self.output_file,
-            )
-            return "http"
-
         # Vespa Cloud Secret Token
         if self.vespa_cloud_secret_token is not None:
-            headers = {"Authorization": f"Bearer {self.vespa_cloud_secret_token}"}
-            response = requests.get(endpoint, headers={**self.base_headers, **headers})
-            if response.status_code == 200:
-                print(
-                    f"Using Vespa Cloud Secret Token to connect to Vespa endpoint {self.end_point}",
-                    file=self.output_file,
-                )
-                return "token"
+            return "token"
 
         # Mutual TLS with key and cert
-        if self.key and self.cert:
-            response = requests.get(
-                endpoint, headers=self.base_headers, cert=(self.cert, self.key)
-            )
-            if response.status_code == 200:
-                print(
-                    f"Using Mutual TLS with key and cert to connect to Vespa endpoint {self.end_point}",
-                    file=self.output_file,
-                )
-                return "mtls_key_cert"
+        elif self.key and self.cert:
+            return "mtls_key_cert"
 
         # Mutual TLS with cert
-        if self.cert:
-            response = requests.get(endpoint, headers=self.base_headers, cert=self.cert)
-            if response.status_code == 200:
-                print(
-                    f"Using Mutual TLS with cert to connect to Vespa endpoint {self.end_point}",
-                    file=self.output_file,
-                )
-                return "mtls_cert"
-
-        # There may be some cases where ApplicationStatus is not available, such as http://api.cord19.vespa.ai
-        return None
+        elif self.cert:
+            return "mtls_cert"
+        # Plain HTTP
+        else:
+            return "http"
 
     def get_application_status(self) -> Optional[Response]:
         """
@@ -1054,7 +1018,6 @@ def __init__(
             self.cert = (self.app.cert, self.app.key)
         else:
             self.cert = self.app.cert
-        self.app.auth_method = self.app._get_valid_auth_method()
         self.headers = self.app.base_headers.copy()
         if self.app.auth_method == "token" and self.app.vespa_cloud_secret_token:
             # Bearer and user-agent
@@ -1472,7 +1435,6 @@ def __init__(
         self.httpx_client = None
         self.connections = connections
         self.total_timeout = total_timeout
-        self.app.auth_method = self.app._get_valid_auth_method()
         self.headers = self.app.base_headers.copy()
         if self.app.auth_method == "token" and self.app.vespa_cloud_secret_token:
             # Bearer and user-agent