Add test coverage for intern_atom and atom cache lifecycle

src/tests/test_xcb.py

@@ -12,10 +12,13 @@
     sizeof,
 )
 from types import SimpleNamespace
-from typing import Any, Callable
+from typing import TYPE_CHECKING, Any, Callable
 from unittest.mock import Mock
 from weakref import finalize
 
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
 import pytest
 
 from mss.exception import ScreenShotError
@@ -224,6 +227,77 @@ def visual_validation_env(monkeypatch: pytest.MonkeyPatch) -> _VisualValidationH
     return _VisualValidationHarness(monkeypatch)
 
 
+#### intern_atom tests
+
+
+class TestInternAtom:
+    """Tests for xcb.intern_atom and the _ATOM_CACHE mechanism."""
+
+    @pytest.fixture(autouse=True)
+    def setup_intern_atom(self) -> Generator[None, None, None]:
+        self.conn, _ = xcb.connect()
+        yield
+        xcb.disconnect(self.conn)
+
+    def _mock_xcb_intern_atom(self, monkeypatch: pytest.MonkeyPatch, atom_value: int) -> Mock:
+        """Patch LIB.xcb.xcb_intern_atom to return a fake reply with the given atom value."""
+        fake_reply = SimpleNamespace(atom=SimpleNamespace(value=atom_value))
+        fake_cookie = Mock()
+        fake_cookie.reply.return_value = fake_reply
+        mock = Mock(return_value=fake_cookie)
+        monkeypatch.setattr(xcb.LIB.xcb, "xcb_intern_atom", mock)
+        return mock
+
+    def test_predefined_atom_skips_xcb(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        mock = self._mock_xcb_intern_atom(monkeypatch, 0)
+        atom = xcb.intern_atom(self.conn, "PRIMARY")
+        assert atom == xcb.Atom(1)
+        mock.assert_not_called()
+
+    def test_cache_miss_calls_xcb_and_caches_result(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        mock = self._mock_xcb_intern_atom(monkeypatch, 100)
+        cache_key = addressof(self.conn)
+        atom = xcb.intern_atom(self.conn, "_NET_WM_NAME")
+        assert atom == xcb.Atom(100)
+        mock.assert_called_once()
+        assert xcb._ATOM_CACHE[cache_key]["_NET_WM_NAME"] == xcb.Atom(100)
+
+    def test_cache_hit_skips_xcb(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        mock = self._mock_xcb_intern_atom(monkeypatch, 0)
+        # xcb.connect() in setup_intern_atom guarantees a cache entry for self.conn.
+        xcb._ATOM_CACHE[addressof(self.conn)]["_NET_WM_NAME"] = xcb.Atom(100)
+        atom = xcb.intern_atom(self.conn, "_NET_WM_NAME")
+        assert atom == xcb.Atom(100)
+        mock.assert_not_called()
+
+    def test_only_if_exists_returns_none_when_missing(self) -> None:
+        atom = xcb.intern_atom(self.conn, "_MSS_TEST_NONEXISTENT_ATOM_12345", only_if_exists=True)
+        assert atom is None
+
+    def test_raises_when_missing_and_not_only_if_exists(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Exercises the "shouldn't be possible" code path where the server returns 0 with only_if_exists=False.
+        self._mock_xcb_intern_atom(monkeypatch, 0)
+        with pytest.raises(xcb.XError, match="X server failed to intern atom"):
+            xcb.intern_atom(self.conn, "_NET_NONEXISTENT")
+
+    def test_pointer_connection_uses_correct_cache_key(self) -> None:
+        atom = xcb.intern_atom(pointer(self.conn), "_NET_WM_NAME")
+        assert atom is not None
+        assert addressof(self.conn) in xcb._ATOM_CACHE
+
+
+def test_atom_cache_lifecycle() -> None:
+    """connect() initializes and disconnect() clears the per-connection atom cache entry."""
+    before = set(xcb._ATOM_CACHE)
+    conn, _ = xcb.connect()
+    cache_key = addressof(conn)
+    assert cache_key in xcb._ATOM_CACHE
+    assert xcb._ATOM_CACHE[cache_key] == {}
+    xcb.disconnect(conn)
+    assert cache_key not in xcb._ATOM_CACHE
+    assert set(xcb._ATOM_CACHE) == before
+
+
 def test_xgetimage_visual_validation_accepts_default_setup(visual_validation_env: _VisualValidationHarness) -> None:
     visual_validation_env.reset()
     mss_instance = xgetimage.MSS()