Add getent_cmd

.bumpversion.toml

@@ -1,6 +1,6 @@
 [tool.bumpversion]
   allow_dirty = true
-  current_version = "0.182.4"
+  current_version = "0.182.5"
 
   [[tool.bumpversion.files]]
     filename = "pyproject.toml"

pyproject.toml

@@ -25,7 +25,7 @@
     "coloredlogs>=15.0.1, <16",
     "coverage-conditional-plugin>=0.9.0, <1",
     "dycw-pytest-only>=2.1.1, <3",
-    "dycw-utilities[test]>=0.182.3, <1",
+    "dycw-utilities[test]>=0.182.4, <1",
     "pyright>=1.1.408, <2",
     "pytest-cov>=7.0.0, <8",
     "pytest-timeout>=2.4.0, <3",
@@ -128,7 +128,7 @@
   name = "dycw-utilities"
   readme = "README.md"
   requires-python = ">= 3.12"
-  version = "0.182.4"
+  version = "0.182.5"
 
   [project.entry-points.pytest11]
     pytest-randomly = "utilities.pytest_plugins.pytest_randomly"

src/tests/test_subprocess.py

@@ -16,14 +16,15 @@
 from utilities.pathlib import get_file_group, get_file_owner
 from utilities.permissions import Permissions
 from utilities.pwd import EFFECTIVE_USER_NAME
-from utilities.pytest import skipif_ci, skipif_mac, throttle_test
+from utilities.pytest import skipif_ci, skipif_mac, skipif_not_linux, throttle_test
 from utilities.shutil import which
 from utilities.subprocess import (
     BASH_LC,
     BASH_LS,
     KNOWN_HOSTS,
     ChownCmdError,
     CpError,
+    GetEntOutput,
     MvFileError,
     RsyncCmdNoSourcesError,
     RsyncCmdSourcesNotFoundError,
@@ -48,6 +49,9 @@
     echo_cmd,
     env_cmds,
     expand_path,
+    getent,
+    getent_cmd,
+    getent_ssh,
     git_branch_current,
     git_checkout,
     git_checkout_cmd,
@@ -420,6 +424,28 @@ def test_subs(self) -> None:
         assert result == expected
 
 
+class TestGetEnt:
+    @skipif_not_linux
+    def test_main(self) -> None:
+        result = getent("user")
+        assert isinstance(result, GetEntOutput)
+
+
+class TestGetEntCmd:
+    def test_main(self) -> None:
+        result = getent_cmd("user")
+        expected = ["getent", "passwd", "user"]
+        assert result == expected
+
+
+class TestGetEntSSH:
+    @skipif_ci
+    @throttle_test(duration=5 * MINUTE)
+    def test_main(self, *, ssh_user: str, ssh_hostname: str) -> None:
+        result = getent_ssh(ssh_user, ssh_hostname)
+        assert isinstance(result, GetEntOutput)
+
+
 class TestGitBranchCurrent:
     @throttle_test(duration=5 * MINUTE)
     def test_main(self, *, git_repo_url: str, tmp_path: Path) -> None:

src/utilities/subprocess.py

@@ -12,7 +12,7 @@
 from string import Template
 from subprocess import PIPE, CalledProcessError, Popen
 from threading import Thread
-from typing import IO, TYPE_CHECKING, Literal, assert_never, overload, override
+from typing import IO, TYPE_CHECKING, Literal, Self, assert_never, overload, override
 
 from utilities.atomicwrites import (
     _CopySourceNotFoundError,
@@ -548,6 +548,66 @@ def expand_path(
 ##
 
 
+def getent(user: str, /) -> GetEntOutput:
+    """Get an entry from Name Service Switch libraries."""
+    text = run(*getent_cmd(user), return_=True)  # skipif-not-linux
+    return GetEntOutput.parse(text)  # skipif-not-linux
+
+
+def getent_cmd(user: str, /) -> list[str]:
+    """Command to use 'getent' to get entries from Name Service Switch libraries."""
+    return ["getent", "passwd", user]
+
+
+def getent_ssh(
+    ssh_user: str,
+    hostname: str,
+    /,
+    *,
+    user: str | None = None,
+    retry: Retry | None = None,
+    logger: LoggerLike | None = None,
+) -> GetEntOutput:
+    """Get an entry from Name Service Switch libraries."""
+    user_use = ssh_user if user is None else user  # skipif-ci
+    text = ssh(  # skipif-ci
+        ssh_user,
+        hostname,
+        *getent_cmd(user_use),
+        return_=True,
+        retry=retry,
+        logger=logger,
+    )
+    return GetEntOutput.parse(text)  # skipif-ci
+
+
+@dataclass(order=True, unsafe_hash=True, kw_only=True, slots=True)
+class GetEntOutput:
+    username: str
+    passwd: str
+    uid: int
+    gid: int
+    gecos: str
+    home: Path
+    shell: Path
+
+    @classmethod
+    def parse(cls, text: str, /) -> Self:
+        username, passwd, uid, gid, gecos, home, shell = text.split(":")
+        return cls(
+            username=username,
+            passwd=passwd,
+            uid=int(uid),
+            gid=int(gid),
+            gecos=gecos,
+            home=Path(home),
+            shell=Path(shell),
+        )
+
+
+##
+
+
 def git_branch_current(path: PathLike, /) -> str:
     """Show the current a branch."""
     return run(*GIT_BRANCH_SHOW_CURRENT, cwd=path, return_=True)
@@ -2317,9 +2377,10 @@ def yield_ssh_temp_dir(
     keep: bool = False,
 ) -> Iterator[Path]:
     """Yield a temporary directory on a remote machine."""
-    path = Path(  # skipif-ci
-        ssh(user, hostname, *MKTEMP_DIR_CMD, return_=True, retry=retry, logger=logger)
+    text = ssh(  # skipif-ci
+        user, hostname, *MKTEMP_DIR_CMD, return_=True, retry=retry, logger=logger
     )
+    path = Path(text)  # skipif-ci
     try:  # skipif-ci
         yield path
     finally:  # skipif-ci
@@ -2345,6 +2406,8 @@ def yield_ssh_temp_dir(
     "UPDATE_CA_CERTIFICATES",
     "ChownCmdError",
     "CpError",
+    "GetEntOutput",
+    "GetEntOutput",
     "MvFileError",
     "RsyncCmdError",
     "RsyncCmdNoSourcesError",
@@ -2372,6 +2435,9 @@ def yield_ssh_temp_dir(
     "echo_cmd",
     "env_cmds",
     "expand_path",
+    "getent",
+    "getent_cmd",
+    "getent_ssh",
     "git_branch_current",
     "git_checkout",
     "git_checkout_cmd",