diff --git a/pyproject.toml b/pyproject.toml index 709b7879d..8f969c1c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ classifiers = [ ] dependencies = [ "click>=8.0.6", + "distro>=1.0.0; sys_platform == 'linux'", "hatchling>=1.27.0", "httpx>=0.22.0", "hyperlink>=21.0.0", diff --git a/src/hatch/index/core.py b/src/hatch/index/core.py index bc300ffc6..a0cc68821 100644 --- a/src/hatch/index/core.py +++ b/src/hatch/index/core.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sys from functools import cached_property from typing import TYPE_CHECKING @@ -52,13 +51,10 @@ def __init__(self, repo: str, *, user="", auth="", ca_cert=None, client_cert=Non def client(self) -> httpx.Client: import httpx + from hatch.utils.linehaul import get_linehaul_component from hatch.utils.network import DEFAULT_TIMEOUT - user_agent = ( - f"Hatch/{__version__} " - f"{sys.implementation.name}/{'.'.join(map(str, sys.version_info[:3]))} " - f"HTTPX/{httpx.__version__}" - ) + user_agent = f"Hatch/{__version__} {get_linehaul_component()} HTTPX/{httpx.__version__}" return httpx.Client( headers={"User-Agent": user_agent}, transport=httpx.HTTPTransport(retries=3, verify=self.__verify, cert=self.__cert), diff --git a/src/hatch/utils/linehaul.py b/src/hatch/utils/linehaul.py new file mode 100644 index 000000000..282fc61fd --- /dev/null +++ b/src/hatch/utils/linehaul.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import json +import os +import platform +import sys +from functools import lru_cache +from typing import Any + +from hatch._version import __version__ + + +def get_linehaul_data() -> dict[str, Any]: + data: dict[str, Any] = { + "installer": {"name": "hatch", "version": __version__}, + "python": platform.python_version(), + "implementation": { + "name": platform.python_implementation(), + "version": _get_implementation_version(), + }, + "system": { + "name": platform.system(), + "release": platform.release(), + }, + "cpu": platform.machine(), + "ci": _looks_like_ci() or None, + } + + if sys.platform.startswith("linux"): + distro_info = _get_linux_distro_info() + if distro_info: + data["distro"] = distro_info + elif sys.platform == "darwin": + mac_version = platform.mac_ver()[0] + if mac_version: + data["distro"] = {"name": "macOS", "version": mac_version} + + openssl_version = _get_openssl_version() + if openssl_version: + data["openssl_version"] = openssl_version + + return data + + +def _get_implementation_version() -> str: + implementation_name = platform.python_implementation() + + if implementation_name == "PyPy": + pypy_version_info = sys.pypy_version_info # type: ignore[attr-defined] + if pypy_version_info.releaselevel == "final": + pypy_version_info = pypy_version_info[:3] + return ".".join(str(x) for x in pypy_version_info) + + return platform.python_version() + + +def _get_linux_distro_info() -> dict[str, Any] | None: + import distro + + name = distro.name() + if not name: + return None + + distro_info: dict[str, Any] = {"name": name} + + version = distro.version() + if version: + distro_info["version"] = version + + codename = distro.codename() + if codename: + distro_info["id"] = codename + + libc_info = _get_libc_info() + if libc_info: + distro_info["libc"] = libc_info + + return distro_info + + +def _get_libc_info() -> dict[str, str] | None: + libc, version = _get_libc_version() + if libc and version: + return {"lib": libc, "version": version} + return None + + +def _get_libc_version() -> tuple[str | None, str | None]: + import ctypes + import ctypes.util + + libc_name = ctypes.util.find_library("c") + if not libc_name: + return None, None + + try: + libc = ctypes.CDLL(libc_name) + libc.gnu_get_libc_version.restype = ctypes.c_char_p + version = libc.gnu_get_libc_version() + if version: + return "glibc", version.decode() + except (OSError, AttributeError, UnicodeDecodeError): + pass + + return None, None + + +def _get_openssl_version() -> str | None: + try: + import ssl + except (ImportError, AttributeError): + return None + else: + return ssl.OPENSSL_VERSION + + +def _looks_like_ci() -> bool: + # INFO: Matches pip's CI detection for linehaul statistics. + # Uses existence check (not value check) because variables like BUILD_BUILDID + # and BUILD_ID are set to build IDs, not boolean strings. + ci_env_vars = ( + "BUILD_BUILDID", + "BUILD_ID", + "CI", + "GITHUB_ACTIONS", + "GITLAB_CI", + "PIP_IS_CI", + "TRAVIS", + ) + return any(env_var in os.environ for env_var in ci_env_vars) + + +@lru_cache(maxsize=1) +def get_linehaul_component() -> str: + data = get_linehaul_data() + return json.dumps(data, separators=(",", ":"), sort_keys=True) diff --git a/tests/index/test_core.py b/tests/index/test_core.py index 3b8b17fdf..5828a6a2c 100644 --- a/tests/index/test_core.py +++ b/tests/index/test_core.py @@ -1,11 +1,12 @@ +import json import platform -import sys import httpx import pytest from hatch._version import __version__ from hatch.index.core import PackageIndex +from hatch.utils.linehaul import get_linehaul_component class TestRepo: @@ -80,7 +81,50 @@ def test_user_agent_header_format(self): user_agent = client.headers["User-Agent"] - expected = ( - f"Hatch/{__version__} {sys.implementation.name}/{platform.python_version()} HTTPX/{httpx.__version__}" - ) - assert user_agent == expected + assert user_agent.startswith(f"Hatch/{__version__} ") + assert user_agent.endswith(f" HTTPX/{httpx.__version__}") + + def _extract_json_from_user_agent(self, user_agent: str) -> dict: + json_start = user_agent.index("{") + json_end = user_agent.rindex("}") + 1 + return json.loads(user_agent[json_start:json_end]) + + def test_user_agent_contains_linehaul_json(self): + index = PackageIndex("https://foo.internal/a/b/") + client = index.client + + user_agent = client.headers["User-Agent"] + data = self._extract_json_from_user_agent(user_agent) + + assert data["installer"]["name"] == "hatch" + assert data["installer"]["version"] == __version__ + assert data["python"] == platform.python_version() + assert data["implementation"]["name"] == platform.python_implementation() + assert data["system"]["name"] == platform.system() + assert data["cpu"] == platform.machine() + + def test_user_agent_linehaul_ci_detection(self, monkeypatch): + get_linehaul_component.cache_clear() + monkeypatch.setenv("CI", "true") + + index = PackageIndex("https://foo.internal/a/b/") + client = index.client + + user_agent = client.headers["User-Agent"] + data = self._extract_json_from_user_agent(user_agent) + + assert data["ci"] is True + + def test_user_agent_linehaul_macos_distro(self, mocker): + get_linehaul_component.cache_clear() + mocker.patch("hatch.utils.linehaul.sys.platform", "darwin") + mocker.patch("hatch.utils.linehaul.platform.mac_ver", return_value=("14.0", "", "arm64")) + + index = PackageIndex("https://foo.internal/a/b/") + client = index.client + + user_agent = client.headers["User-Agent"] + data = self._extract_json_from_user_agent(user_agent) + + assert data["distro"]["name"] == "macOS" + assert data["distro"]["version"] == "14.0" diff --git a/tests/utils/test_linehaul.py b/tests/utils/test_linehaul.py new file mode 100644 index 000000000..69ef04ac7 --- /dev/null +++ b/tests/utils/test_linehaul.py @@ -0,0 +1,165 @@ +import json +import platform + +import pytest + +from hatch._version import __version__ +from hatch.utils.linehaul import get_linehaul_component, get_linehaul_data + + +class TestGetLinehaulData: + def test_installer_info(self): + data = get_linehaul_data() + + assert data["installer"]["name"] == "hatch" + assert data["installer"]["version"] == __version__ + + def test_python_version(self): + data = get_linehaul_data() + + assert data["python"] == platform.python_version() + + def test_implementation_info(self): + data = get_linehaul_data() + + assert data["implementation"]["name"] == platform.python_implementation() + assert "version" in data["implementation"] + + def test_system_info(self): + data = get_linehaul_data() + + assert data["system"]["name"] == platform.system() + assert data["system"]["release"] == platform.release() + + def test_cpu(self): + data = get_linehaul_data() + + assert data["cpu"] == platform.machine() + + def test_openssl_version_present(self): + data = get_linehaul_data() + + # ssl module should be available in standard CPython + assert "openssl_version" in data + assert data["openssl_version"].startswith(("OpenSSL", "LibreSSL")) + + +class TestCIDetection: + _CI_VARS = ("BUILD_BUILDID", "BUILD_ID", "CI", "GITHUB_ACTIONS", "GITLAB_CI", "PIP_IS_CI", "TRAVIS") + + def _clear_ci_env(self, monkeypatch): + for var in self._CI_VARS: + monkeypatch.delenv(var, raising=False) + + def test_ci_true_when_env_var_set(self, monkeypatch): + self._clear_ci_env(monkeypatch) + monkeypatch.setenv("CI", "true") + + data = get_linehaul_data() + + assert data["ci"] is True + + def test_ci_true_when_env_var_has_any_value(self, monkeypatch): + self._clear_ci_env(monkeypatch) + monkeypatch.setenv("BUILD_BUILDID", "12345") + + data = get_linehaul_data() + + assert data["ci"] is True + + def test_ci_none_when_no_env_vars(self, monkeypatch): + self._clear_ci_env(monkeypatch) + + data = get_linehaul_data() + + assert data["ci"] is None + + @pytest.mark.parametrize( + "env_var", + [ + "BUILD_BUILDID", + "BUILD_ID", + "CI", + "GITHUB_ACTIONS", + "GITLAB_CI", + "PIP_IS_CI", + "TRAVIS", + ], + ) + def test_each_ci_env_var_detected(self, monkeypatch, env_var): + self._clear_ci_env(monkeypatch) + monkeypatch.setenv(env_var, "1") + + data = get_linehaul_data() + + assert data["ci"] is True + + +class TestDistro: + def test_macos_distro(self, mocker): + mocker.patch("hatch.utils.linehaul.sys.platform", "darwin") + mocker.patch("hatch.utils.linehaul.platform.mac_ver", return_value=("14.0", "", "arm64")) + + data = get_linehaul_data() + + assert data["distro"]["name"] == "macOS" + assert data["distro"]["version"] == "14.0" + + def test_macos_no_distro_when_mac_ver_empty(self, mocker): + mocker.patch("hatch.utils.linehaul.sys.platform", "darwin") + mocker.patch("hatch.utils.linehaul.platform.mac_ver", return_value=("", ("", "", ""), "")) + + data = get_linehaul_data() + + assert "distro" not in data + + def test_no_distro_on_windows(self, mocker): + mocker.patch("hatch.utils.linehaul.sys.platform", "win32") + + data = get_linehaul_data() + + assert "distro" not in data + + @pytest.mark.requires_linux + def test_linux_distro(self): + data = get_linehaul_data() + + # On actual Linux, distro info should be present if distro name is non-empty + import distro + + if distro.name(): + assert "distro" in data + assert data["distro"]["name"] == distro.name() + + +class TestGetLinehaulComponent: + def test_returns_valid_json(self): + component = get_linehaul_component() + + data = json.loads(component) + assert isinstance(data, dict) + + def test_compact_json_format(self): + component = get_linehaul_component() + + # Compact JSON should have no spaces after separators + assert ": " not in component + assert ", " not in component + + def test_keys_are_sorted(self): + component = get_linehaul_component() + + data = json.loads(component) + keys = list(data.keys()) + assert keys == sorted(keys) + + def test_contains_required_fields(self): + component = get_linehaul_component() + + data = json.loads(component) + assert "installer" in data + assert "python" in data + assert "implementation" in data + assert "system" in data + assert "cpu" in data + assert "ci" in data