fix(base): cpu_runtime 在非 Linux 主机上的类型检查与测试修复

src/unilab/base/cpu_runtime.py

@@ -42,11 +42,15 @@ def _confine_existing_threads(ids: set[int]) -> None:
     with sibling ranks' pinned CPUs. Threads may exit between listing and
     pinning; those races are ignored.
     """
-    if not os.path.isdir(_PROC_TASK_DIR):
+    # Resolved at call time (not import) so tests can monkeypatch the seam;
+    # ``getattr`` keeps this checkable on platforms where typeshed hides the
+    # Linux-only symbol (mypy ``attr-defined`` on darwin).
+    sched_setaffinity = getattr(os, "sched_setaffinity", None)
+    if sched_setaffinity is None or not os.path.isdir(_PROC_TASK_DIR):
         return
     for entry in os.listdir(_PROC_TASK_DIR):
         try:
-            os.sched_setaffinity(int(entry), ids)
+            sched_setaffinity(int(entry), ids)
         except OSError:
             continue
 
@@ -66,15 +70,17 @@ def apply_env_cpu_runtime(cpu_ids: Sequence[int] | None) -> None:
         return
     ids = {int(cpu_id) for cpu_id in cpu_ids}
 
-    if hasattr(os, "sched_setaffinity") and hasattr(os, "sched_getaffinity"):
-        available = set(os.sched_getaffinity(0))
+    sched_setaffinity = getattr(os, "sched_setaffinity", None)
+    sched_getaffinity = getattr(os, "sched_getaffinity", None)
+    if sched_setaffinity is not None and sched_getaffinity is not None:
+        available = set(sched_getaffinity(0))
         missing = sorted(ids - available)
         if missing:
             raise ValueError(
                 f"EnvCfg.cpu_ids entries {missing} are not available to this process "
                 f"(sched_getaffinity={sorted(available)})"
             )
-        os.sched_setaffinity(0, ids)
+        sched_setaffinity(0, ids)
         _confine_existing_threads(ids)
     else:
         warnings.warn(

tests/base/test_cpu_runtime.py

@@ -29,8 +29,12 @@
 
 def _record_affinity(monkeypatch: pytest.MonkeyPatch, available: set[int]) -> list[tuple]:
     calls: list[tuple] = []
-    monkeypatch.setattr(os, "sched_getaffinity", lambda _pid: set(available))
-    monkeypatch.setattr(os, "sched_setaffinity", lambda pid, ids: calls.append((pid, set(ids))))
+    # raising=False: sched_*affinity is Linux-only, so the attribute may not
+    # exist on the host running the tests (e.g. macOS dev machines).
+    monkeypatch.setattr(os, "sched_getaffinity", lambda _pid: set(available), raising=False)
+    monkeypatch.setattr(
+        os, "sched_setaffinity", lambda pid, ids: calls.append((pid, set(ids))), raising=False
+    )
     return calls
 
 
@@ -103,8 +107,8 @@ def test_unavailable_cpu_ids_fail_closed(monkeypatch: pytest.MonkeyPatch):
 
 def test_platform_without_affinity_warns_and_caps_numba(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.delenv("NUMBA_NUM_THREADS", raising=False)
-    monkeypatch.delattr(os, "sched_setaffinity")
-    monkeypatch.delattr(os, "sched_getaffinity")
+    monkeypatch.delattr(os, "sched_setaffinity", raising=False)
+    monkeypatch.delattr(os, "sched_getaffinity", raising=False)
     confine_calls = _record_confine(monkeypatch)
     numba_calls = _record_numba(monkeypatch)
 
@@ -125,7 +129,7 @@ def fake_setaffinity(pid, ids):
             raise ProcessLookupError
         calls.append((pid, set(ids)))
 
-    monkeypatch.setattr(os, "sched_setaffinity", fake_setaffinity)
+    monkeypatch.setattr(os, "sched_setaffinity", fake_setaffinity, raising=False)
     monkeypatch.setattr(os.path, "isdir", lambda path: path == cpu_runtime._PROC_TASK_DIR)
     monkeypatch.setattr(os, "listdir", lambda path: ["123", "456", "789"])
 
@@ -136,7 +140,9 @@ def fake_setaffinity(pid, ids):
 
 def test_confine_existing_threads_without_proc_is_noop(monkeypatch: pytest.MonkeyPatch):
     calls: list[tuple] = []
-    monkeypatch.setattr(os, "sched_setaffinity", lambda pid, ids: calls.append((pid, set(ids))))
+    monkeypatch.setattr(
+        os, "sched_setaffinity", lambda pid, ids: calls.append((pid, set(ids))), raising=False
+    )
     monkeypatch.setattr(os.path, "isdir", lambda path: False)
 
     cpu_runtime._confine_existing_threads({0})