Fix Process.terminal() returning None for PTYs opened after first call

docs/changelog.rst

@@ -248,6 +248,10 @@ Others:
 
 **Bug fixes**
 
+- :gh:`2830`, [Linux, macOS, BSD]: :meth:`Process.terminal` could return
+  ``None`` for PTYs allocated after the first call to ``get_terminal_map()``.
+  The ``lru_cache`` was never refreshed, so newly opened terminals were
+  invisible. On a cache miss the cache is now cleared and rebuilt once.
 - :gh:`1007`, [Windows]: :func:`boot_time` no longer fluctuates by ~1 second
   across calls or across processes. It is now read atomically from the kernel
   via ``NtQuerySystemInformation(SystemTimeOfDayInformation)``, replacing the

psutil/_psbsd.py

@@ -536,10 +536,12 @@ def environ(self):
     def terminal(self):
         tty_nr = self.oneshot()["ttynr"]
         tmap = _psposix.get_terminal_map()
-        try:
-            return tmap[tty_nr]
-        except KeyError:
-            return None
+        if tty_nr not in tmap:
+            # The PTY may have been allocated after the initial cache build;
+            # clear the cache and rebuild once to pick up new /dev/pts entries.
+            _psposix.get_terminal_map.cache_clear()
+            tmap = _psposix.get_terminal_map()
+        return tmap.get(tty_nr)
 
     @wrap_exceptions
     def ppid(self):

psutil/_pslinux.py

@@ -1743,10 +1743,12 @@ def environ(self):
     def terminal(self):
         tty_nr = int(self._parse_stat_file()['ttynr'])
         tmap = _psposix.get_terminal_map()
-        try:
-            return tmap[tty_nr]
-        except KeyError:
-            return None
+        if tty_nr not in tmap:
+            # The PTY may have been allocated after the initial cache build;
+            # clear the cache and rebuild once to pick up new /dev/pts entries.
+            _psposix.get_terminal_map.cache_clear()
+            tmap = _psposix.get_terminal_map()
+        return tmap.get(tty_nr)
 
     # May not be available on old kernels.
     if os.path.exists(f"/proc/{os.getpid()}/io"):

psutil/_psosx.py

@@ -397,10 +397,12 @@ def gids(self):
     def terminal(self):
         tty_nr = self._oneshot_kinfo()["ttynr"]
         tmap = _psposix.get_terminal_map()
-        try:
-            return tmap[tty_nr]
-        except KeyError:
-            return None
+        if tty_nr not in tmap:
+            # The PTY may have been allocated after the initial cache build;
+            # clear the cache and rebuild once to pick up new /dev/pts entries.
+            _psposix.get_terminal_map.cache_clear()
+            tmap = _psposix.get_terminal_map()
+        return tmap.get(tty_nr)
 
     @wrap_exceptions
     def memory_info(self):

tests/test_process.py

@@ -187,6 +187,41 @@ def test_terminal(self):
             else:
                 assert terminal == tty
 
+    @skipif(not POSIX, reason="POSIX only")
+    def test_terminal_stale_cache(self):
+        # Regression test for https://github.com/giampaolo/psutil/issues/2830.
+        # terminal() must refresh the get_terminal_map() cache on a miss so
+        # PTYs allocated after the first call are resolved correctly.
+        import functools
+
+        from psutil import _psposix
+
+        p = psutil.Process()
+        _psposix.get_terminal_map.cache_clear()
+        real_map = dict(_psposix.get_terminal_map())
+
+        call_count = [0]
+
+        @functools.lru_cache
+        def fake_get_terminal_map():
+            call_count[0] += 1
+            # First call returns an empty map (simulates stale cache built
+            # before any PTY existed); second call returns the real map.
+            if call_count[0] == 1:
+                return {}
+            return real_map
+
+        orig = _psposix.get_terminal_map
+        _psposix.get_terminal_map = fake_get_terminal_map
+        try:
+            p.terminal()
+            assert (
+                call_count[0] == 2
+            ), "terminal() did not refresh the stale get_terminal_map() cache"
+        finally:
+            _psposix.get_terminal_map = orig
+            orig.cache_clear()
+
     @skipif(not HAS_PROC_IO_COUNTERS, reason="not supported")
     @skip_on_not_implemented(only_if=LINUX)
     def test_io_counters(self):