Move token server check to validation

Author: kzscisoft

simvue/config/parameters.py

@@ -3,9 +3,12 @@
 import pydantic
 import typing
 import pathlib
+import http
 
 import simvue.models as sv_models
 from simvue.utilities import get_expiry
+from simvue.version import __version__
+from simvue.api import get
 
 CONFIG_FILE_NAMES: list[str] = ["simvue.toml", ".simvue.toml"]
 
@@ -27,12 +30,34 @@ def url_to_str(cls, v: typing.Any) -> str:
         return f"{v}"
 
     @pydantic.field_validator("token")
-    def check_token(cls, v: pydantic.SecretStr) -> str:
-        if not (expiry := get_expiry(v.get_secret_value())):
+    def check_token(cls, v: typing.Any) -> str:
+        if not (expiry := get_expiry(v)):
             raise AssertionError("Failed to parse Simvue token - invalid token form")
         if time.time() - expiry > 0:
             raise AssertionError("Simvue token has expired")
-        return v.get_secret_value()
+        return v
+
+    @pydantic.model_validator(mode="after")
+    def check_valid_server(cls, values: dict) -> bool:
+        url = values["url"]
+        token = values["token"]
+        headers: dict[str, str] = {
+            "Authorization": f"Bearer {token}",
+            "User-Agent": f"Simvue Python client {__version__}",
+        }
+        try:
+            response = get(f"{url}/api/version", headers)
+
+            if response.status_code != http.HTTPStatus.OK or not response.json().get(
+                "version"
+            ):
+                raise AssertionError
+
+            if response.status_code == http.HTTPStatus.UNAUTHORIZED:
+                raise AssertionError("Unauthorised token")
+
+        except Exception as err:
+            raise AssertionError(f"Exception retrieving server version: {str(err)}")
 
 
 class DefaultRunSpecifications(pydantic.BaseModel):

simvue/factory/proxy/remote.py

@@ -1,14 +1,13 @@
 import logging
 import typing
 import http
-import time
 
 if typing.TYPE_CHECKING:
     from simvue.config import SimvueConfiguration
 
 from simvue.api import get, post, put
 from simvue.factory.proxy.base import SimvueBaseClass
-from simvue.utilities import prepare_for_api, skip_if_failed, get_expiry
+from simvue.utilities import prepare_for_api, skip_if_failed
 from simvue.version import __version__
 
 logger = logging.getLogger(__name__)
@@ -483,36 +482,6 @@ def send_heartbeat(self) -> typing.Optional[dict[str, typing.Any]]:
         self._error(f"Got status code {response.status_code} when sending heartbeat")
         return None
 
-    @skip_if_failed("_aborted", "_suppress_errors", False)
-    def check_token(self) -> bool:
-        """
-        Check token
-        """
-        if not (expiry := get_expiry(self._token)):
-            self._error("Failed to parse user token")
-            return False
-
-        if time.time() - expiry > 0:
-            self._error("Token has expired")
-            return False
-
-        try:
-            response = get(f"{self._url}/api/version", self._headers)
-
-            if response.status_code != http.HTTPStatus.OK or not response.json().get(
-                "version"
-            ):
-                raise AssertionError
-
-            if response.status_code == http.HTTPStatus.UNAUTHORIZED:
-                self._error("Unauthorised token")
-                return False
-
-        except Exception as err:
-            self._error(f"Exception retrieving server version: {str(err)}")
-            return False
-        return True
-
     @skip_if_failed("_aborted", "_suppress_errors", False)
     def get_abort_status(self) -> bool:
         logger.debug("Retrieving alert status")