Rely on getpass for cache path expansion

servicex/configuration.py

@@ -26,6 +26,7 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 import os
+from getpass import getuser
 import tempfile
 from pathlib import Path, PurePath
 from typing import List, Optional, Dict
@@ -55,7 +56,7 @@ class Configuration(BaseModel):
     config_file: Optional[str] = Field(default=None, exclude=True)
 
     @model_validator(mode="after")
-    def expand_cache_path(self):
+    def expand_cache_path(self) -> "Configuration":
         """
         Expand the cache path to a full path, and create it if it doesn't exist.
         Expand ${USER} to be the user name on the system. Works for windows, too.
@@ -68,12 +69,13 @@ def expand_cache_path(self):
 
         s_path = os.path.expanduser(self.cache_path)
 
-        # If they have tried to use the USER or UserName as an expansion, and it has failed, then
-        # translate it to maintain harmony across platforms.
-        if "${USER}" in s_path and "UserName" in os.environ:
-            s_path = s_path.replace("${USER}", os.environ["UserName"])
-        if "${USER}" in s_path and "USER" in os.environ:
-            s_path = s_path.replace("${USER}", os.environ["USER"])
+        if "${USER}" in s_path:
+            try:
+                username = getuser()
+            except Exception:  # pragma: no cover - getpass failures are unexpected
+                username = ""
+
+            s_path = s_path.replace("${USER}", username)
 
         p_p = PurePath(s_path)
         if len(p_p.parts) > 1 and p_p.parts[1] == "tmp":

tests/test_config.py

@@ -27,46 +27,47 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 import os
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 import pytest
 
 from servicex.configuration import Configuration
 
 
+@patch("servicex.configuration.getuser", return_value="cache_user")
 @patch("servicex.configuration.tempfile.gettempdir", return_value="./mytemp")
-def test_config_read(tempdir):
-    # Windows style user name
+def test_config_read(tempdir: MagicMock, getuser_mock: MagicMock) -> None:
+    # Windows style user name should not override the getpass result.
     os.environ["UserName"] = "p_higgs"
     c = Configuration.read(config_path="tests/example_config.yaml")
-    assert c.cache_path == "mytemp/servicex_p_higgs"
+    assert c.cache_path == "mytemp/servicex_cache_user"
+    os.environ.pop("UserName")
 
-    # Reset environment
-    del os.environ["UserName"]
-
-    # Linux style user name
+    # Linux style user name should also not override the getpass result.
     os.environ["USER"] = "p_higgs2"
     c = Configuration.read(config_path="tests/example_config.yaml")
-    assert c.cache_path == "mytemp/servicex_p_higgs2"
+    assert c.cache_path == "mytemp/servicex_cache_user"
+    os.environ.pop("USER")
 
     # but what if there is no file at all?
     with pytest.raises(NameError):
         Configuration.read(config_path="invalid.yaml")
 
 
+@patch("servicex.configuration.getuser", return_value="cache_user")
 @patch("servicex.configuration.tempfile.gettempdir", return_value="./mytemp")
-def test_default_cache_path(tempdir):
+def test_default_cache_path(tempdir: MagicMock, getuser_mock: MagicMock) -> None:
 
-    # Windows style user name
+    # Windows style user name should not override the getpass result.
     os.environ["UserName"] = "p_higgs"
     c = Configuration.read(config_path="tests/example_config_no_cache_path.yaml")
-    assert c.cache_path == "mytemp/servicex_p_higgs"
-    del os.environ["UserName"]
+    assert c.cache_path == "mytemp/servicex_cache_user"
+    os.environ.pop("UserName")
 
-    # Linux style user name
+    # Linux style user name should also not override the getpass result.
     os.environ["USER"] = "p_higgs"
     c = Configuration.read(config_path="tests/example_config_no_cache_path.yaml")
-    assert c.cache_path == "mytemp/servicex_p_higgs"
-    del os.environ["USER"]
+    assert c.cache_path == "mytemp/servicex_cache_user"
+    os.environ.pop("USER")
 
 
 def test_read_from_home(monkeypatch, tmp_path):