Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 79 additions & 33 deletions fancylog/fancylog.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,11 @@
import logging
import os
import platform
import shutil
import subprocess
import sys
import warnings
from datetime import datetime
from importlib.metadata import distributions
from importlib.util import find_spec

from rich.logging import RichHandler
Expand Down Expand Up @@ -255,7 +255,10 @@ def write_environment_packages(self, header="ENVIRONMENT"):
"""Write the local/global environment packages used to run the script.

Attempt to collect conda packages and, if this fails,
collect the packages installed in the running Python environment.
collect pip packages via a subprocess call. If pip itself isn't
installed (e.g. a venv created and managed purely by uv), fall
back to `uv pip list`, which implements the same interface
without depending on pip being present.

Parameters
----------
Expand All @@ -281,45 +284,88 @@ def write_environment_packages(self, header="ENVIRONMENT"):
self.file.write(f"Conda environment: {conda_env}\n\n")
self.file.write("Environment packages (conda):\n")
self.write_packages(env_pkgs)
return

# If no conda env, fall back to logging installed packages
# via importlib.metadata, which does not depend on pip
# being installed in the running environment.
except (KeyError, subprocess.CalledProcessError, json.JSONDecodeError):
all_pkgs = [
{
"name": dist.metadata["Name"],
"version": dist.version,
"location": str(dist.locate_file("")),
}
for dist in distributions()
]

virtual_env = os.getenv("VIRTUAL_ENV")
if virtual_env:
# If there is a local env, log local packages first
env_pkgs = [
pkg for pkg in all_pkgs if virtual_env in pkg["location"]
]
pass

# If no conda env, fall back to pip. `uv pip list` is tried first
# when uv is available: it's a much cheaper subprocess call than
# spawning a full `python -m pip` interpreter, and it works even
# in venvs managed purely by uv, where a `pip` package may not
# be installed at all.
uv_exe = shutil.which("uv")
if uv_exe:
try:
uv_list = subprocess.run(
[uv_exe, "pip", "list", "--format=json"],
capture_output=True,
text=True,
check=True,
)

self.file.write(
"No conda environment found, reporting pip packages\n\n"
"No conda environment found, reporting uv packages\n\n"
)
self.file.write("Local environment packages (pip):\n")
self.write_packages(env_pkgs)
self.file.write("\n")
self.file.write("Environment packages (uv pip):\n")
self.write_packages(json.loads(uv_list.stdout))
return

# Log global-available packages (if any)
global_pkgs = [pkg for pkg in all_pkgs if pkg not in env_pkgs]
except (subprocess.CalledProcessError, json.JSONDecodeError):
pass

try:
python_executable = sys.executable
pip_list = subprocess.run(
[
python_executable,
"-m",
"pip",
"list",
"--verbose",
"--format=json",
],
capture_output=True,
text=True,
check=True,
)

self.file.write("Global environment packages (pip):\n")
self.write_packages(global_pkgs)
all_pkgs = json.loads(pip_list.stdout)

else:
self.file.write(
"No environment found, reporting global pip packages\n\n"
)
self.write_packages(all_pkgs)
except (subprocess.CalledProcessError, json.JSONDecodeError):
self.file.write(
"Could not find conda, pip or uv packages. "
"No packages were logged.\n\n"
)
return

virtual_env = os.getenv("VIRTUAL_ENV")
if virtual_env:
# If there is a local env, log local packages first
env_pkgs = [
pkg
for pkg in all_pkgs
if virtual_env in str(pkg.get("location", ""))
]

self.file.write(
"No conda environment found, reporting pip packages\n\n"
)
self.file.write("Local environment packages (pip):\n")
self.write_packages(env_pkgs)
self.file.write("\n")

# Log global-available packages (if any)
global_pkgs = [pkg for pkg in all_pkgs if pkg not in env_pkgs]

self.file.write("Global environment packages (pip):\n")
self.write_packages(global_pkgs)

else:
self.file.write(
"No environment found, reporting global pip packages\n\n"
)
self.write_packages(all_pkgs)

def write_packages(self, env_pkgs):
"""Write the packages in the local environment.
Expand Down
2 changes: 1 addition & 1 deletion tests/tests/test_benchmarks.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,4 @@ def test_benchmark(tmp_path, capsys):
with capsys.disabled():
print(f"`test_benchmark` time taken: {time_taken:.4f}")

assert time_taken < 0.05, "Set up is running slower than expected."
assert time_taken < 0.13, "Set up is running slower than expected."
145 changes: 95 additions & 50 deletions tests/tests/test_general.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,9 @@
import logging
import os
import platform
import shutil
import subprocess
import sys
from importlib.metadata import distributions
from unittest.mock import MagicMock, patch

import pytest
Expand Down Expand Up @@ -270,75 +270,76 @@ def test_environment_header(boolean, operator, tmp_path):


def test_correct_pkg_version_logged(tmp_path):
"""Package versions logged should be equal to
the output of `conda list` or `pip list`.
"""Package versions logged should be equal to the output of
whichever backend `write_environment_packages` actually used.

The backend is picked the same way the library picks it (conda,
then uv, then pip), so this stays a real-environment check rather
than forcing a branch that may not work here. For example, a venv
created by `uv venv` (as tox-uv does on CI) has no `pip` installed
at all, so forcing the pip branch would fail. The individual
branches are covered in isolation by the mocked tests below.
"""
fancylog.start_logging(tmp_path, fancylog, write_env_packages=True)

log_file = next(tmp_path.glob("*.log"))
file_content = log_file.read_text()

conda_exe = os.environ.get("CONDA_EXE")
uv_exe = shutil.which("uv")

if conda_exe:
command = [conda_exe, "list", "--json"]
elif uv_exe:
command = [uv_exe, "pip", "list", "--format=json"]
else:
command = [
sys.executable,
"-m",
"pip",
"list",
"--verbose",
"--format=json",
]

try:
# If there is a conda environment, assert that the correct
# version is logged for all pkgs
conda_exe = os.environ["CONDA_EXE"]
conda_list = subprocess.run(
[conda_exe, "list", "--json"], capture_output=True, text=True
)

conda_pkgs = json.loads(conda_list.stdout)
for pkg in conda_pkgs:
assert f"{pkg['name']:20} {pkg['version']:15}\n"

except KeyError:
# If there is no conda environment, assert that the correct
# version is logged for all packages logged with pip list
with open(log_file) as file:
file_content = file.read()

# Test local environment versions
local_site_packages = next(
p for p in sys.path if "site-packages" in p
)
pkg_list = subprocess.run(
command, capture_output=True, text=True, check=True
)

for dist in distributions():
if str(dist.locate_file("")).startswith(local_site_packages):
assert (
f"{dist.metadata['Name']:20} {dist.version}"
in file_content
)
for pkg in json.loads(pkg_list.stdout):
assert f"{pkg['name']:20} {pkg['version']:15}\n" in file_content


def _make_fake_dist(name, version, location):
fake_dist = MagicMock()
fake_dist.metadata = {"Name": name}
fake_dist.version = version
fake_dist.locate_file.return_value = location
return fake_dist
def _make_fake_pkg(name, version, location):
return {"name": name, "version": version, "location": location}


def test_mock_pip_pkgs(tmp_path):
"""Mock installed distributions
"""Mock `pip list` output
and test that packages are logged correctly.
"""

fake_distributions = [
_make_fake_dist("fancylog", "1.1.1", "fake_env"),
_make_fake_dist("pytest", "1.1.1", "global_env"),
fake_pkgs = [
_make_fake_pkg("fancylog", "1.1.1", "fake_env"),
_make_fake_pkg("pytest", "1.1.1", "global_env"),
]

# Patch the environment and installed distributions
# Patch the environment, uv availability and pip subprocess call
with (
patch.dict(os.environ, {}, clear=False),
patch("os.getenv") as mock_getenv,
patch("fancylog.fancylog.distributions") as mock_distributions,
patch("shutil.which", return_value=None),
patch("subprocess.run") as mock_run,
):
# Eliminate conda environment packages triggers logging pip list
os.environ.pop("CONDA_PREFIX", None)
os.environ.pop("CONDA_EXE", None)

mock_getenv.return_value = "fake_env"

mock_distributions.return_value = fake_distributions
mock_run.return_value = MagicMock(
stdout=json.dumps(fake_pkgs), returncode=0
)

fancylog.start_logging(tmp_path, fancylog, write_env_packages=True)

Expand All @@ -356,6 +357,47 @@ def test_mock_pip_pkgs(tmp_path):
assert f"{'pytest':20} {'1.1.1'}"


def test_mock_uv_pkgs(tmp_path):
"""Mock `uv pip list` output
and test that packages are logged correctly.
"""

fake_pkgs = [
_make_fake_pkg("fancylog", "1.1.1", None),
_make_fake_pkg("pytest", "1.1.1", None),
]

# Patch the environment, uv availability and uv subprocess call
with (
patch.dict(os.environ, {}, clear=False),
patch("shutil.which", return_value="/usr/bin/uv"),
patch("subprocess.run") as mock_run,
):
# Eliminate conda environment packages triggers logging uv pip list
os.environ.pop("CONDA_PREFIX", None)
os.environ.pop("CONDA_EXE", None)

mock_run.return_value = MagicMock(
stdout=json.dumps(fake_pkgs), returncode=0
)

fancylog.start_logging(tmp_path, fancylog, write_env_packages=True)

log_file = next(tmp_path.glob("*.log"))

# Log contains the uv subheader and mocked pkgs versions
with open(log_file) as file:
file_content = file.read()

assert (
"No conda environment found, reporting uv packages"
) in file_content
assert "Environment packages (uv pip):" in file_content

assert f"{'fancylog':20} {'1.1.1'}"
assert f"{'pytest':20} {'1.1.1'}"


def test_mock_conda_pkgs(tmp_path):
"""Mock conda environment variables
and test that packages are logged correctly.
Expand Down Expand Up @@ -407,16 +449,17 @@ def test_mock_no_environment(tmp_path):
and test that packages are logged correctly.
"""

fake_distributions = [
_make_fake_dist("fancylog", "1.1.1", "fake_env"),
_make_fake_dist("pytest", "1.1.1", "global_env"),
fake_pkgs = [
_make_fake_pkg("fancylog", "1.1.1", "fake_env"),
_make_fake_pkg("pytest", "1.1.1", "global_env"),
]

# Patch the environment and installed distributions
# Patch the environment, uv availability and pip subprocess call
with (
patch.dict(os.environ, {}, clear=False),
patch("os.getenv") as mock_getenv,
patch("fancylog.fancylog.distributions") as mock_distributions,
patch("shutil.which", return_value=None),
patch("subprocess.run") as mock_run,
):
# Eliminate conda environment packages triggers logging pip list
os.environ.pop("CONDA_PREFIX", None)
Expand All @@ -425,7 +468,9 @@ def test_mock_no_environment(tmp_path):
# Mock lack of any local environment
mock_getenv.return_value = None

mock_distributions.return_value = fake_distributions
mock_run.return_value = MagicMock(
stdout=json.dumps(fake_pkgs), returncode=0
)

fancylog.start_logging(tmp_path, fancylog, write_env_packages=True)

Expand Down