py: fix access token parsing (#341)

* py: fix access token parsing

Co-authored-by: Isabella do Amaral <[email protected]>
Signed-off-by: Alessio Pragliola <[email protected]>

* py: added regression test

Signed-off-by: Alessio Pragliola <[email protected]>

* py: apply commit suggestion from review

Co-authored-by: Matteo Mortari <[email protected]>
Signed-off-by: Alessio Pragliola <[email protected]>

---------

Signed-off-by: Alessio Pragliola <[email protected]>
Co-authored-by: Isabella do Amaral <[email protected]>
Co-authored-by: Matteo Mortari <[email protected]>

clients/python/src/model_registry/_client.py

@@ -55,7 +55,7 @@ def __init__(
             # /var/run/secrets/kubernetes.io/serviceaccount/token
             sa_token = os.environ.get("KF_PIPELINES_SA_TOKEN_PATH")
             if sa_token:
-                user_token = Path(sa_token).read_bytes()
+                user_token = Path(sa_token).read_text()
             else:
                 warn("User access token is missing", stacklevel=2)
 

clients/python/tests/conftest.py

@@ -2,6 +2,7 @@
 import inspect
 import os
 import subprocess
+import tempfile
 import time
 from contextlib import asynccontextmanager
 from pathlib import Path
@@ -133,3 +134,18 @@ def event_loop():
 @cleanup
 def client() -> ModelRegistry:
     return ModelRegistry(REGISTRY_HOST, REGISTRY_PORT, author="author", is_secure=False)
+
+@pytest.fixture(scope="module")
+def setup_env_user_token():
+    with tempfile.NamedTemporaryFile(delete=False) as token_file:
+        token_file.write(b"Token")
+    old_token_path = os.getenv("KF_PIPELINES_SA_TOKEN_PATH")
+    os.environ["KF_PIPELINES_SA_TOKEN_PATH"] = token_file.name
+
+    yield token_file.name
+
+    if old_token_path is None:
+        del os.environ["KF_PIPELINES_SA_TOKEN_PATH"]
+    else:
+        os.environ["KF_PIPELINES_SA_TOKEN_PATH"] = old_token_path
+    os.remove(token_file.name)

clients/python/tests/regression_test.py

@@ -22,3 +22,25 @@ def test_create_tagged_version(client: ModelRegistry):
     mv = client.get_model_version(name, version)
     assert mv
     assert mv.id
+
+@pytest.mark.e2e
+def test_get_model_without_user_token(setup_env_user_token, client):
+    """Test regression for using client methods without an user_token in the init arguments.
+
+    Reported on: https://github.com/kubeflow/model-registry/issues/340
+    """
+    assert setup_env_user_token != ""
+    name = "test_model"
+    version = "1.0.0"
+    metadata = {"a": 1, "b": "2"}
+    rm = client.register_model(
+        name,
+        "s3",
+        model_format_name="test_format",
+        model_format_version="test_version",
+        version=version,
+        metadata=metadata,
+    )
+    assert rm.id
+    assert (_rm := client.get_registered_model(name))
+    assert rm.id == _rm.id