mirror of
https://github.com/Niko1221/Strata.git
synced 2026-11-07 06:15:58 -10:00
Control: the OS and architecture are read without WMI on Windows, so a stuck winmgmt cannot hang models list or a load (#4975, #5185 as reference)
This commit is contained in:
1 parent
a3a10b90af
commit
eb4f41fb9a
14 files changed
+291
-32
No files matched your search
+2
-1
@@ -16,6 +16,7 @@ import collections
|
||||
import ctypes
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -478,7 +479,7 @@ def _cpu_name():
|
||||
for line in fh:
|
||||
if line.startswith("model name"):
|
||||
return line.split(":", 1)[1].strip()
|
||||
return platform.processor() or None
|
||||
return sysinfo.processor() or None
|
||||
|
||||
|
||||
class _CpuRamFallback:
|
||||
|
||||
@@ -70,6 +70,7 @@ from strata_control import atomic
|
||||
from strata_control import draft_layer
|
||||
from strata_control import dsv41_context
|
||||
from strata_control import engine_component
|
||||
from strata_control import sysinfo
|
||||
from strata_control.discovery import in_trash
|
||||
from strata_control import user_args
|
||||
|
||||
@@ -1501,7 +1502,7 @@ def vision_threads(logical: int | None = None) -> int:
|
||||
def cpu_info():
|
||||
"""(name, avx2, avx512): avx512 means everything Strata's fast AVX-512 kernels use (F, BW, VL, VNNI, VBMI),
|
||||
the same test the engine makes (cpu_avx512_ok), not just AVX-512F."""
|
||||
name, avx2, avx512 = platform.processor() or "unknown CPU", False, False
|
||||
name, avx2, avx512 = sysinfo.processor() or "unknown CPU", False, False
|
||||
if WIN:
|
||||
pf = ctypes.windll.kernel32.IsProcessorFeaturePresent
|
||||
avx2 = bool(pf(40)) or _cpuid_avx2() # PF_AVX2_INSTRUCTIONS_AVAILABLE, else the CPU itself (#159)
|
||||
@@ -1516,7 +1517,7 @@ def cpu_info():
|
||||
# aarch64 lists Features, not the x86 flags line, and no "model name": lscpu names the cores (a DGX Spark:
|
||||
# Cortex-X925 + Cortex-A725). No AVX of any kind; the arm64 engine needs none.
|
||||
cores = re.findall(r"^\s*Model name:\s*(.+)$", out(["lscpu"]), re.M)
|
||||
name = " + ".join(dict.fromkeys(c.strip() for c in cores)) or (platform.processor() or "arm64 CPU")
|
||||
name = " + ".join(dict.fromkeys(c.strip() for c in cores)) or (sysinfo.processor() or "arm64 CPU")
|
||||
return name, False, False
|
||||
flags = set(mflags.group(1).split())
|
||||
avx2 = "avx2" in flags
|
||||
@@ -1551,7 +1552,7 @@ def _cpuid_floor() -> str:
|
||||
|
||||
def is_arm64() -> bool:
|
||||
"""An ARM64 CPU (DGX Spark / GB10, Grace): there is no AVX here, and no ready-made engine for it."""
|
||||
return platform.machine().lower() in ("aarch64", "arm64")
|
||||
return sysinfo.machine().lower() in ("aarch64", "arm64")
|
||||
|
||||
|
||||
def is_gb10(g) -> bool:
|
||||
@@ -5553,7 +5554,7 @@ def get_prebuilt(url_base, gpu, vision, updating=False, toolkit=13) -> Path | No
|
||||
# package when the release carries one; without it the caller compiles (spark/build_engine.sh does the same by hand)
|
||||
got = engine_from_feed(eng, updating) if int(toolkit) != 12 else None
|
||||
if got is None:
|
||||
warn(f"no ready-made engine for {platform.machine()} in the update feed" + ("" if updating else ": compiling it here, once"))
|
||||
warn(f"no ready-made engine for {sysinfo.machine()} in the update feed" + ("" if updating else ": compiling it here, once"))
|
||||
return got
|
||||
if not url_base:
|
||||
return None
|
||||
|
||||
@@ -12,6 +12,7 @@ import hashlib
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import re
|
||||
import socket
|
||||
import subprocess
|
||||
@@ -57,7 +58,7 @@ def redactor(key: str = ""):
|
||||
except Exception: # noqa: BLE001 - no user name is not an error
|
||||
pass
|
||||
add(socket.gethostname(), "<host>", True)
|
||||
add(platform.node(), "<host>", True)
|
||||
add(sysinfo.node(), "<host>", True)
|
||||
add(key, "<key>")
|
||||
subs.sort(key=lambda item: -len(item[0].pattern))
|
||||
path_user = re.compile(r"(?i)([A-Za-z]:)?([\\/])(Users|home)([\\/])(?!<)[^\\/\s\"':;,]+")
|
||||
@@ -143,7 +144,7 @@ def cpu_name() -> str | None:
|
||||
out = powershell("(Get-CimInstance Win32_Processor | Select-Object -First 1).Name").strip()
|
||||
if out:
|
||||
return out
|
||||
return platform.processor() or None
|
||||
return sysinfo.processor() or None
|
||||
|
||||
|
||||
def nvidia_gpus() -> tuple[list[dict], str | None, str | None]:
|
||||
@@ -373,9 +374,9 @@ def hardware(why: Why, *, selected: list[int], served_gpu: str | None, model_dir
|
||||
# ------------------------------------------------------------------------------------------------------------ software
|
||||
def os_info() -> dict:
|
||||
if os.name == "nt":
|
||||
return {"name": "Windows", "version": platform.version()}
|
||||
return {"name": "Windows", "version": sysinfo.version()}
|
||||
if sys.platform == "darwin":
|
||||
return {"name": "macOS", "version": platform.mac_ver()[0] or platform.release()}
|
||||
return {"name": "macOS", "version": platform.mac_ver()[0] or sysinfo.release()}
|
||||
name = None
|
||||
try:
|
||||
for line in Path("/etc/os-release").read_text().splitlines():
|
||||
@@ -383,7 +384,7 @@ def os_info() -> dict:
|
||||
name = line.split("=", 1)[1].strip().strip('"')
|
||||
except OSError:
|
||||
pass
|
||||
return {"name": name or platform.system(), "version": platform.release()}
|
||||
return {"name": name or sysinfo.system(), "version": sysinfo.release()}
|
||||
|
||||
|
||||
def git_commit(folder: Path = ROOT) -> str | None:
|
||||
@@ -426,7 +427,7 @@ def software(why: Why, machine: dict, *, engine_version: str | None, build_optio
|
||||
if machine.get("rocm") and not any(r.get("name") == "rocm" for r in runtime):
|
||||
runtime.append({"name": "rocm", "version": machine["rocm"]})
|
||||
if sys.platform == "darwin":
|
||||
runtime.append({"name": "metal", "version": platform.mac_ver()[0] or platform.release()})
|
||||
runtime.append({"name": "metal", "version": platform.mac_ver()[0] or sysinfo.release()})
|
||||
drivers = [{"name": "gpu-driver", "version": machine["driver"]}] if machine.get("driver") else []
|
||||
block = {"os": os_info(), "drivers": drivers or None, "runtime": runtime or None,
|
||||
"strata_commit": commit, "engine_version": engine_version,
|
||||
|
||||
@@ -453,9 +453,9 @@ def apply_defaults(s: S.LaunchSettings, catalog) -> tuple[S.LaunchSettings, list
|
||||
|
||||
|
||||
def _host_platform() -> tuple[str, str]:
|
||||
"""(sys.platform, platform.machine()) of this computer, in one place the tests can replace."""
|
||||
"""(sys.platform, sysinfo.machine()) of this computer, in one place the tests can replace."""
|
||||
import platform
|
||||
return sys.platform, platform.machine()
|
||||
return sys.platform, sysinfo.machine()
|
||||
|
||||
|
||||
def parallel_notices(s: S.LaunchSettings, comp) -> list[S.Notice]:
|
||||
|
||||
@@ -7,6 +7,7 @@ import io
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
@@ -110,7 +111,7 @@ def _nvidia_cards() -> tuple[list[dict] | None, str | None]:
|
||||
if len(row) != 5 or len(cards) >= 64:
|
||||
raise ValueError("invalid field count")
|
||||
index_raw, name, uuid, memory_raw, driver = [part.strip() for part in row]
|
||||
unified = memory_raw == "[N/A]" and platform.machine().lower() in ("aarch64", "arm64") # a DGX Spark's GB10: no separate graphics memory
|
||||
unified = memory_raw == "[N/A]" and sysinfo.machine().lower() in ("aarch64", "arm64") # a DGX Spark's GB10: no separate graphics memory
|
||||
if not (re.fullmatch(r"[0-9]{1,4}", index_raw)
|
||||
and (unified or re.fullmatch(r"[0-9]{1,9}", memory_raw))
|
||||
and 0 <= int(index_raw) <= 1023 and (unified or int(memory_raw) > 0)
|
||||
|
||||
@@ -21,6 +21,7 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
@@ -781,10 +782,10 @@ AARCH64_NOT_YET_LINE = "Unsloth's 1-bit DeepSeek sets were not tested on a DGX S
|
||||
|
||||
def not_here(model: str | None) -> str | None:
|
||||
"""The plain line when `model` is not offered on THIS computer (a Mac: MAC_NOT_YET; linux-aarch64, the DGX Spark: AARCH64_NOT_YET), else None.
|
||||
Read at call time from sys.platform and platform.machine(), so one rule serves the model lists, setup, `strata launch` and the app's file check."""
|
||||
Read at call time from sys.platform and sysinfo.machine(), so one rule serves the model lists, setup, `strata launch` and the app's file check."""
|
||||
if sys.platform == "darwin" and model in MAC_NOT_YET:
|
||||
return MAC_NOT_YET_LINE
|
||||
if sys.platform.startswith("linux") and platform.machine().lower() in ("aarch64", "arm64") and model in AARCH64_NOT_YET:
|
||||
if sys.platform.startswith("linux") and sysinfo.machine().lower() in ("aarch64", "arm64") and model in AARCH64_NOT_YET:
|
||||
return AARCH64_NOT_YET_LINE
|
||||
return None
|
||||
MAC_SMALL_RAM_GB = 64.0 # below this the measured M5 Max numbers do not apply (nothing smaller than 32 GB was ever started)
|
||||
|
||||
@@ -77,7 +77,7 @@ def explain(n: int, context: int | None) -> list[str]:
|
||||
def warnings(n: int | None, *, cards: list[dict] | None, context: int | None = None, os_name: str | None = None, machine: str | None = None,
|
||||
vendor: str | None = None, unified: bool = False) -> list[tuple[str, str]]:
|
||||
"""(what, what to do) for a pick that goes beyond what was run: a warning, never a refusal. `cards` are the usable cards of the run;
|
||||
`os_name` is sys.platform, `machine` platform.machine(). Nothing for an unset or one-at-a-time pick."""
|
||||
`os_name` is sys.platform, `machine` sysinfo.machine(). Nothing for an unset or one-at-a-time pick."""
|
||||
out: list[tuple[str, str]] = []
|
||||
if n is None or int(n) <= 1:
|
||||
return out
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
"""The OS and machine facts Control needs, answered without asking WMI on Windows (#4975, #5185).
|
||||
|
||||
`platform.system()`, `.machine()`, `.release()`, `.version()`, `.processor()` and `.node()` all go through
|
||||
`platform.uname()`. On Windows with Python 3.12 that calls `win32_ver()` and `_get_machine_win32()`, which query WMI
|
||||
(`platform._wmi_query`) with no timeout: a stuck `winmgmt` service froze every Control command that resolves the engine
|
||||
component ("Strata Control did not answer within 60 s (models list)"), although the interpreter, the environment and the
|
||||
registry know the same facts. On Windows these functions read only those. Everywhere else they are the `platform` calls
|
||||
(an `os.uname()` there, not a service), looked up at call time so a test can replace them.
|
||||
|
||||
Pure and tiny on purpose, like targets.py: the start path imports it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import platform
|
||||
|
||||
_WINDOWS_MACHINES = {"amd64": "AMD64", "x86_64": "AMD64", "arm64": "ARM64", "aarch64": "ARM64", "x86": "x86", "i386": "x86",
|
||||
"i686": "x86", "ia64": "ia64"}
|
||||
|
||||
|
||||
def is_windows() -> bool:
|
||||
return sys.platform.startswith("win")
|
||||
|
||||
|
||||
def system() -> str:
|
||||
"""`platform.system()`: "Windows", "Linux", "Darwin" ..."""
|
||||
return "Windows" if is_windows() else platform.system()
|
||||
|
||||
|
||||
def _windows_machine() -> str:
|
||||
"""Windows names its architecture in the environment the loader gives every process: PROCESSOR_ARCHITEW6432 is the real
|
||||
one inside a 32-bit process on 64-bit Windows, PROCESSOR_ARCHITECTURE the process's own (which is also what an x64
|
||||
program emulated on ARM64 sees: the engine that fits is the x64 one). `platform.uname()` itself read these first on
|
||||
Python 3.11 and before. Without them (a stripped environment) the interpreter's own build tag answers."""
|
||||
for key in ("PROCESSOR_ARCHITEW6432", "PROCESSOR_ARCHITECTURE"):
|
||||
value = os.environ.get(key, "").strip()
|
||||
if value:
|
||||
return _WINDOWS_MACHINES.get(value.lower(), value)
|
||||
try:
|
||||
import sysconfig
|
||||
tag = sysconfig.get_platform().lower()
|
||||
except Exception: # noqa: BLE001 - a fact lookup never fails a start
|
||||
tag = ""
|
||||
if "arm64" in tag:
|
||||
return "ARM64"
|
||||
if "amd64" in tag or "x86_64" in tag:
|
||||
return "AMD64"
|
||||
return "x86" if tag in ("win32", "") else tag
|
||||
|
||||
|
||||
def machine() -> str:
|
||||
"""`platform.machine()`: "AMD64", "ARM64" on Windows; "x86_64", "arm64", "aarch64" elsewhere."""
|
||||
return _windows_machine() if is_windows() else platform.machine()
|
||||
|
||||
|
||||
def release() -> str:
|
||||
"""`platform.release()`: "10" or "11" on Windows (11 from build 22000), the kernel release elsewhere."""
|
||||
if is_windows():
|
||||
info = _windows_version()
|
||||
if info is None:
|
||||
return ""
|
||||
return "11" if info[0] == 10 and info[2] >= 22000 else str(info[0])
|
||||
return platform.release()
|
||||
|
||||
|
||||
def version() -> str:
|
||||
"""`platform.version()`: "10.0.19045" on Windows, the kernel version elsewhere."""
|
||||
if is_windows():
|
||||
info = _windows_version()
|
||||
return "" if info is None else f"{info[0]}.{info[1]}.{info[2]}"
|
||||
return platform.version()
|
||||
|
||||
|
||||
def _windows_version() -> tuple[int, int, int] | None:
|
||||
try:
|
||||
info = sys.getwindowsversion() # type: ignore[attr-defined]
|
||||
return int(info.major), int(info.minor), int(info.build)
|
||||
except Exception: # noqa: BLE001
|
||||
return None
|
||||
|
||||
|
||||
def processor() -> str:
|
||||
"""`platform.processor()`: the CPU's description (PROCESSOR_IDENTIFIER on Windows; the empty string when unknown)."""
|
||||
if is_windows():
|
||||
return os.environ.get("PROCESSOR_IDENTIFIER", "").strip()
|
||||
return platform.processor()
|
||||
|
||||
|
||||
def node() -> str:
|
||||
"""`platform.node()`: the computer's name."""
|
||||
if is_windows():
|
||||
try:
|
||||
return socket.gethostname()
|
||||
except OSError:
|
||||
return os.environ.get("COMPUTERNAME", "")
|
||||
return platform.node()
|
||||
|
||||
|
||||
def platform_string() -> str:
|
||||
"""A one-line description of the OS for reports (`platform.platform()` asks the same questions as `uname`)."""
|
||||
if is_windows():
|
||||
return f"Windows-{release()}-{version()}"
|
||||
return platform.platform()
|
||||
@@ -6,7 +6,9 @@ start path must not import the component fetcher (components.py) to get them.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import platform
|
||||
import platform # noqa: F401 - kept: tests replace targets.platform's functions
|
||||
|
||||
from strata_control import sysinfo
|
||||
|
||||
|
||||
class ComponentError(RuntimeError):
|
||||
@@ -30,10 +32,12 @@ BACKEND_MATRIX: dict[str, tuple[str, ...]] = {
|
||||
|
||||
def host_platform() -> str:
|
||||
"""This machine as an `os-arch` key, matching `BACKEND_MATRIX`."""
|
||||
system = {"Darwin": "macos", "Linux": "linux", "Windows": "windows"}.get(platform.system())
|
||||
# sysinfo, not platform: on Windows `platform.system()` and `.machine()` query WMI with no timeout, and a stuck winmgmt
|
||||
# froze every command that resolves the engine (#4975)
|
||||
system = {"Darwin": "macos", "Linux": "linux", "Windows": "windows"}.get(sysinfo.system())
|
||||
if system is None:
|
||||
raise ComponentError(f"unsupported operating system: {platform.system()}")
|
||||
machine = platform.machine().lower()
|
||||
raise ComponentError(f"unsupported operating system: {sysinfo.system()}")
|
||||
machine = sysinfo.machine().lower()
|
||||
arch = {"arm64": "arm64", "aarch64": "arm64",
|
||||
"x86_64": "x86_64", "amd64": "x86_64"}.get(machine, machine)
|
||||
if system == "linux" and arch == "arm64":
|
||||
|
||||
@@ -28,6 +28,7 @@ import hashlib
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import time
|
||||
import uuid
|
||||
from dataclasses import asdict, dataclass, field
|
||||
@@ -88,10 +89,10 @@ def machine_identity() -> dict[str, Any]:
|
||||
is the same machine), so it is deliberately absent.
|
||||
"""
|
||||
return {
|
||||
"system": platform.system(),
|
||||
"release": platform.release(),
|
||||
"machine": platform.machine(),
|
||||
"processor": platform.processor(),
|
||||
"system": sysinfo.system(),
|
||||
"release": sysinfo.release(),
|
||||
"machine": sysinfo.machine(),
|
||||
"processor": sysinfo.processor(),
|
||||
"cpu_count": os.cpu_count(),
|
||||
}
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ import hashlib
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
@@ -132,7 +133,7 @@ def gpu_used_mib(indexes: list[int] | None = None) -> int | None:
|
||||
def cpu() -> dict[str, Any]:
|
||||
"""``model``, ``logical`` and ``physical`` thread counts, and ``hybrid`` (performance plus efficiency cores)."""
|
||||
logical = os.cpu_count() or 1
|
||||
model = platform.processor() or platform.machine()
|
||||
model = sysinfo.processor() or sysinfo.machine()
|
||||
physical: int | None = None
|
||||
hybrid = False
|
||||
try:
|
||||
@@ -272,9 +273,9 @@ def hardware_identity(*, pcie: str | None = None) -> dict[str, Any]:
|
||||
"gpus": [{"name": g["name"], "vram_mib": g["vram_total_mib"]} for g in cards],
|
||||
"cpu": c["model"], "logical": c["logical"], "physical": c["physical"], "hybrid": c["hybrid"],
|
||||
"ram_gib": _round_down((mem["total_mib"] or 0) // 1024 or None, 2),
|
||||
"pcie": pcie, "os": platform.system(),
|
||||
"pcie": pcie, "os": sysinfo.system(),
|
||||
},
|
||||
"soft": {"driver": sorted({g["driver"] for g in cards if g.get("driver")}), "os_release": platform.release()},
|
||||
"soft": {"driver": sorted({g["driver"] for g in cards if g.get("driver")}), "os_release": sysinfo.release()},
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -49,6 +49,7 @@ from __future__ import annotations
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
from strata_control import sysinfo
|
||||
import statistics
|
||||
import subprocess
|
||||
import time
|
||||
@@ -803,9 +804,9 @@ def machine_identity() -> dict[str, Any]:
|
||||
comparable — which is why this is recorded with every report rather than assumed stable.
|
||||
"""
|
||||
return {
|
||||
"platform": platform.platform(),
|
||||
"machine": platform.machine(),
|
||||
"processor": platform.processor(),
|
||||
"platform": sysinfo.platform_string(),
|
||||
"machine": sysinfo.machine(),
|
||||
"processor": sysinfo.processor(),
|
||||
"python": platform.python_version(),
|
||||
"cpu_count": os.cpu_count(),
|
||||
}
|
||||
@@ -22,7 +22,13 @@ import package_terminal # noqa: E402
|
||||
|
||||
|
||||
def host(system: str, machine: str) -> str:
|
||||
with mock.patch.object(targets.platform, "system", return_value=system), mock.patch.object(targets.platform, "machine", return_value=machine):
|
||||
# the name and the architecture come from strata_control.sysinfo: on Windows from sys.platform and the environment (no WMI, #4975),
|
||||
# elsewhere from the platform module
|
||||
win = system == "Windows"
|
||||
env = {"PROCESSOR_ARCHITECTURE": machine, "PROCESSOR_ARCHITEW6432": ""} if win else {}
|
||||
with mock.patch.object(targets.platform, "system", return_value=system), mock.patch.object(targets.platform, "machine", return_value=machine), \
|
||||
mock.patch.object(sys, "platform", "win32" if win else ("darwin" if system == "Darwin" else "linux")), \
|
||||
mock.patch.dict("os.environ", env):
|
||||
return targets.host_platform()
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""#4975 (PR #5185 is the reference): the OS and architecture are known without WMI on Windows.
|
||||
|
||||
On Windows with Python 3.12 `platform.uname()` (so `system()`, `machine()`, `release()`, `version()`, `processor()` and `node()`) asks WMI
|
||||
(`platform._wmi_query`) with no timeout. A stuck `winmgmt` froze every Control command that resolves the engine component. A stuck WMI
|
||||
cannot be produced from a test, so these tests make every `platform` call that could reach it fail loudly, and check that the Windows
|
||||
answers still come out of the interpreter and the environment.
|
||||
|
||||
python -m pytest tests/terminal/test_sysinfo.py
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from collections import namedtuple
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from strata_control import engine_component, sysinfo, targets
|
||||
|
||||
WinVer = namedtuple("WinVer", "major minor build")
|
||||
|
||||
|
||||
def _stuck(name):
|
||||
def hang(*a, **k):
|
||||
raise AssertionError(f"platform.{name} reached WMI")
|
||||
return hang
|
||||
|
||||
|
||||
def _windows(env, build=19045):
|
||||
"""sys.platform is win32, the environment is `env`, and every platform call that asks WMI fails."""
|
||||
stack = [mock.patch.object(sys, "platform", "win32"),
|
||||
mock.patch.object(sys, "getwindowsversion", lambda: WinVer(10, 0, build), create=True),
|
||||
mock.patch.dict(os.environ, env, clear=False)]
|
||||
for name in ("uname", "system", "machine", "release", "version", "processor", "node", "platform", "win32_ver", "_wmi_query"):
|
||||
stack.append(mock.patch.object(sysinfo.platform, name, _stuck(name), create=True))
|
||||
return stack
|
||||
|
||||
|
||||
class Stuck(unittest.TestCase):
|
||||
def run_in(self, env, fn, build=19045):
|
||||
stack = _windows(env, build)
|
||||
for item in stack:
|
||||
item.start()
|
||||
try:
|
||||
return fn()
|
||||
finally:
|
||||
for item in reversed(stack):
|
||||
item.stop()
|
||||
|
||||
def test_windows_x64_host_platform_while_wmi_is_stuck(self):
|
||||
env = {"PROCESSOR_ARCHITECTURE": "AMD64", "PROCESSOR_ARCHITEW6432": ""}
|
||||
self.assertEqual(self.run_in(env, targets.host_platform), "windows-x86_64")
|
||||
|
||||
def test_windows_arm64_native(self):
|
||||
env = {"PROCESSOR_ARCHITECTURE": "ARM64", "PROCESSOR_ARCHITEW6432": ""}
|
||||
self.assertEqual(self.run_in(env, targets.host_platform), "windows-arm64")
|
||||
|
||||
def test_x64_python_emulated_on_arm64_sees_amd64_and_gets_the_x64_engine(self):
|
||||
env = {"PROCESSOR_ARCHITECTURE": "AMD64", "PROCESSOR_ARCHITEW6432": ""}
|
||||
self.assertEqual(self.run_in(env, sysinfo.machine), "AMD64")
|
||||
|
||||
def test_a_32_bit_process_on_64_bit_windows_reads_the_real_architecture(self):
|
||||
for real, expect in (("AMD64", "windows-x86_64"), ("ARM64", "windows-arm64")):
|
||||
env = {"PROCESSOR_ARCHITECTURE": "x86", "PROCESSOR_ARCHITEW6432": real}
|
||||
self.assertEqual(self.run_in(env, targets.host_platform), expect)
|
||||
|
||||
def test_without_the_environment_the_interpreter_build_answers(self):
|
||||
stack = _windows({})
|
||||
for item in stack:
|
||||
item.start()
|
||||
try:
|
||||
with mock.patch.dict(os.environ, {}, clear=True), mock.patch("sysconfig.get_platform", return_value="win-arm64"):
|
||||
self.assertEqual(sysinfo.machine(), "ARM64")
|
||||
with mock.patch.dict(os.environ, {}, clear=True), mock.patch("sysconfig.get_platform", return_value="win-amd64"):
|
||||
self.assertEqual(sysinfo.machine(), "AMD64")
|
||||
finally:
|
||||
for item in reversed(stack):
|
||||
item.stop()
|
||||
|
||||
def test_release_version_processor_and_node_need_no_wmi(self):
|
||||
env = {"PROCESSOR_ARCHITECTURE": "AMD64", "PROCESSOR_IDENTIFIER": "Intel64 Family 6 Model 183"}
|
||||
|
||||
def facts():
|
||||
return (sysinfo.system(), sysinfo.release(), sysinfo.version(), sysinfo.processor(), bool(sysinfo.node()),
|
||||
sysinfo.platform_string())
|
||||
self.assertEqual(self.run_in(env, facts, 22631),
|
||||
("Windows", "11", "10.0.22631", "Intel64 Family 6 Model 183", True, "Windows-11-10.0.22631"))
|
||||
self.assertEqual(self.run_in(env, facts, 19045)[1], "10")
|
||||
|
||||
def test_the_engine_lookup_of_a_run_config_does_not_reach_wmi(self):
|
||||
# the very path of the report: engine_component.resolve() -> host_platform()
|
||||
env = {"PROCESSOR_ARCHITECTURE": "AMD64", "PROCESSOR_ARCHITEW6432": ""}
|
||||
|
||||
def look():
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
try:
|
||||
engine_component.resolve(Path(folder), backend="cpu")
|
||||
except engine_component.EngineMissing:
|
||||
return "missing"
|
||||
self.assertEqual(self.run_in(env, look), "missing")
|
||||
|
||||
|
||||
class OtherSystemsKeepTheirAnswers(unittest.TestCase):
|
||||
def test_macos_and_linux_use_the_platform_module(self):
|
||||
with mock.patch.object(sys, "platform", "darwin"), mock.patch.object(sysinfo.platform, "system", return_value="Darwin"), \
|
||||
mock.patch.object(sysinfo.platform, "machine", return_value="arm64"):
|
||||
self.assertEqual(targets.host_platform(), "macos-arm64")
|
||||
with mock.patch.object(sys, "platform", "linux"), mock.patch.object(sysinfo.platform, "system", return_value="Linux"), \
|
||||
mock.patch.object(sysinfo.platform, "machine", return_value="aarch64"):
|
||||
self.assertEqual(targets.host_platform(), "linux-aarch64")
|
||||
|
||||
def test_the_answer_is_a_key_of_the_backend_matrix(self):
|
||||
self.assertIn(targets.host_platform(), {t for ts in targets.BACKEND_MATRIX.values() for t in ts})
|
||||
|
||||
def test_no_control_module_calls_the_uname_based_platform_functions_directly(self):
|
||||
# the callers that can run on Windows were moved to sysinfo; a new direct call would bring the hang back
|
||||
root = Path(__file__).resolve().parents[2]
|
||||
bad = []
|
||||
for rel in ("strata_control", "strata_cli", "serve"):
|
||||
for path in (root / rel).rglob("*.py"):
|
||||
# mac_fit.py is macOS only; engine_component's one call is behind sys.platform.startswith("linux") on the same line
|
||||
if path.name.startswith("test_") or path.name in ("sysinfo.py", "mac_fit.py"):
|
||||
continue
|
||||
for number, line in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), 1):
|
||||
if line.lstrip().startswith("#") or "`platform." in line:
|
||||
continue
|
||||
if re.search(r"(?<![\w.])platform\.(system|machine|release|version|processor|node|platform|uname)\(\)", line) \
|
||||
and "linux" not in line:
|
||||
bad.append(f"{path.relative_to(root)}:{number}: {line.strip()}")
|
||||
self.assertEqual(bad, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user