Skip to content
Open
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
3 changes: 2 additions & 1 deletion fades/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import os
import time

from pathlib import Path
from fades import REPO_VCS
from fades.multiplatform import filelock
from fades.parsing import VCSDependency, NameVerDependency
Expand Down Expand Up @@ -92,7 +93,7 @@ def _match_by_uuid(self, current_venvs, uuid):
for venv_str in current_venvs:
venv = json.loads(venv_str)
env_path = venv.get('metadata', {}).get('env_path')
_, env_uuid = os.path.split(env_path)
env_uuid = Path(env_path).name
if env_uuid == uuid:
return venv

Expand Down
11 changes: 6 additions & 5 deletions fades/envbuilder.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import pathlib
import shutil

from pathlib import Path
from datetime import datetime, timezone
from venv import EnvBuilder
from uuid import uuid4
Expand All @@ -46,7 +47,7 @@ class _FadesEnvBuilder(EnvBuilder):

def __init__(self):
basedir = helpers.get_basedir()
self.env_path = os.path.join(basedir, str(uuid4()))
self.env_path = Path(basedir) / str(uuid4())
self.env_bin_path = ''
logger.debug("Env will be created at: %s", self.env_path)

Expand Down Expand Up @@ -105,9 +106,9 @@ def create_env(self, interpreter, is_current, options):
logger.debug("env_bin_path: %s", self.env_bin_path)

# Re check if pip was installed (supporting both binary and .exe for Windows)
pip_bin = os.path.join(self.env_bin_path, "pip")
pip_exe = os.path.join(self.env_bin_path, "pip.exe")
if not (os.path.exists(pip_bin) or os.path.exists(pip_exe)):
pip_bin = Path(self.env_bin_path) / "pip"
pip_exe = Path(self.env_bin_path) / "pip.exe"
if not (pip_bin.exists() or pip_exe.exists()):
logger.debug("pip isn't installed in the venv, setting pip_installed=False")
self.pip_installed = False

Expand Down Expand Up @@ -201,7 +202,7 @@ def _create_initial_usage_file_if_not_exists(self):
self._write_venv_usage(f, venv_data)

def _write_venv_usage(self, file_, venv_data):
_, uuid = os.path.split(venv_data['env_path'])
uuid = Path(venv_data['env_path']).name
file_.write('{} {}\n'.format(uuid, self._datetime_to_str(datetime.now(UTC))))

def _datetime_to_str(self, datetime_):
Expand Down
10 changes: 5 additions & 5 deletions fades/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
"""A collection of utilities for fades."""

import os
from pathlib import Path
import sys
import json
import logging
Expand Down Expand Up @@ -96,20 +97,19 @@ def _get_specific_dir(dir_type):
"""Get a specific directory, using some XDG base, with sensible default."""
if SNAP_BASEDIR_NAME in os.environ:
logger.debug("Getting base dir information from SNAP_BASEDIR_NAME env var.")
direct = os.path.join(os.environ[SNAP_BASEDIR_NAME], dir_type)
direct = Path(os.environ[SNAP_BASEDIR_NAME]) / dir_type
else:
try:
basedirectory = _get_basedirectory()
except ImportError:
logger.debug("Using last resort base dir: ~/.fades")
from os.path import expanduser
direct = os.path.join(expanduser("~"), ".fades")
direct = Path.home() / ".fades"
else:
xdg_attrib = 'xdg_{}_home'.format(dir_type)
base = getattr(basedirectory, xdg_attrib)
direct = os.path.join(base, 'fades')
direct = Path(base) / 'fades'

if not os.path.exists(direct):
if not direct.exists():
os.makedirs(direct)
return direct

Expand Down
13 changes: 7 additions & 6 deletions fades/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
pkgnamesdb,
)
from fades.logger import set_up as logger_set_up

from pathlib import Path

# Get the logger here; it will be properly setup at bootstrap, but can be used from
# the rest of the module just fine
Expand Down Expand Up @@ -346,10 +346,11 @@ def go():
logger.warning("Overriding 'quiet' option ('verbose' also requested)")

# start the virtualenvs manager
venvscache = cache.VEnvsCache(os.path.join(helpers.get_basedir(), 'venvs.idx'))
venvscache = cache.VEnvsCache(helpers.get_basedir() / 'venvs.idx')
# start usage manager
usage_manager = envbuilder.UsageManager(
os.path.join(helpers.get_basedir(), 'usage_stats'), venvscache)
helpers.get_basedir() / 'usage_stats', venvscache)


if args.clean_unused_venvs:
try:
Expand Down Expand Up @@ -405,7 +406,7 @@ def go():
if venv_data:
env_path = venv_data['env_path']
# A venv was found in the cache check if its valid or re-generate it.
if not os.path.exists(env_path):
if not Path(env_path).exists():
logger.warning("Missing directory (the virtualenv will be re-created): %r", env_path)
venvscache.remove(env_path)
create_venv = True
Expand Down Expand Up @@ -440,7 +441,7 @@ def go():

# run forest run!!
python_exe = 'ipython' if args.ipython else 'python'
python_exe = os.path.join(venv_data['env_bin_path'], python_exe)
python_exe = Path(venv_data['env_bin_path']) / python_exe

# add the virtualenv /bin path to the child PATH.
environ_path = venv_data['env_bin_path']
Expand Down Expand Up @@ -468,7 +469,7 @@ def go():
# Build the exec path relative to 'bin' dir; note that if child_program's path
# is absolute (starting with '/') the resulting exec_path will be just it,
# which is something fades supports
exec_path = os.path.join(venv_data['env_bin_path'], child_program)
exec_path = Path(venv_data['env_bin_path']) / child_program
cmd = [exec_path]
elif args.module:
cmd = [python_exe, '-m'] + python_options + [child_program]
Expand Down
4 changes: 2 additions & 2 deletions fades/parsing.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import os
import re

from pathlib import Path
from packaging.requirements import Requirement
from packaging.version import Version

Expand Down Expand Up @@ -276,8 +277,7 @@ def _read_lines(filepath):
logger.warning(
"Invalid format to indicate a nested requirements file: '%r'", line)
else:
nested_filepath = os.path.join(
os.path.dirname(filepath), nested_filename)
nested_filepath = Path(filepath).parent / nested_filename
yield from _read_lines(nested_filepath)
else:
yield line
Expand Down
12 changes: 7 additions & 5 deletions fades/pipmanager.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
import shutil
import contextlib

from pathlib import Path

from urllib import request

from fades import helpers
Expand All @@ -43,9 +45,9 @@ def __init__(self, env_bin_path, pip_installed=False, options=None, avoid_pip_up
self.env_bin_path = env_bin_path
self.pip_installed = pip_installed
self.options = options
self.pip_exe = os.path.join(self.env_bin_path, "pip")
self.pip_exe = Path(self.env_bin_path) / "pip"
basedir = helpers.get_basedir()
self.pip_installer_fname = os.path.join(basedir, "get-pip.py")
self.pip_installer_fname = Path(basedir) / "get-pip.py"
self.avoid_pip_upgrade = avoid_pip_upgrade

def install(self, dependency):
Expand All @@ -58,7 +60,7 @@ def install(self, dependency):
# Always update pip to get latest behaviours (specially regarding security); this has
# the nice side effect of getting logged the pip version that is used.
if not self.avoid_pip_upgrade:
python_exe = os.path.join(self.env_bin_path, "python")
python_exe = Path(self.env_bin_path) / "python"
helpers.logged_exec([python_exe, '-m', 'pip', 'install', 'pip', '--upgrade'])

# split to pass several tokens on multiword dependency (this is very specific for '-e' on
Expand Down Expand Up @@ -104,15 +106,15 @@ def _download_pip_installer(self):

def _brute_force_install_pip(self):
"""Check a brute force install of pip itself."""
if os.path.exists(self.pip_installer_fname):
if self.pip_installer_fname.exists():
logger.debug("Using pip installer from %r", self.pip_installer_fname)
else:
logger.debug(
"Installer for pip not found in %r, downloading it", self.pip_installer_fname)
self._download_pip_installer()

logger.debug("Installing PIP manually in the virtualenv")
python_exe = os.path.join(self.env_bin_path, "python")
python_exe = Path(self.env_bin_path) / "python"
helpers.logged_exec([python_exe, self.pip_installer_fname, '-I'])
self.pip_installed = True

Expand Down
13 changes: 7 additions & 6 deletions tests/test_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import sys
import tempfile
import unittest
from pathlib import Path
from http.server import HTTPStatus
from unittest.mock import patch
from urllib.error import HTTPError
Expand Down Expand Up @@ -233,7 +234,7 @@ class GetDirsTestCase(unittest.TestCase):

def test_basedir_xdg(self):
direct = helpers.get_basedir()
self.assertEqual(direct, os.path.join(BaseDirectory.xdg_data_home, 'fades'))
self.assertEqual(direct, Path(BaseDirectory.xdg_data_home) / 'fades')

def _fake_snap_env_dir(self, direct):
"""Fake Snap's environment variable."""
Expand All @@ -244,13 +245,13 @@ def test_basedir_snap(self):
with tempfile.TemporaryDirectory() as dirname:
self._fake_snap_env_dir(dirname)
direct = helpers.get_basedir()
self.assertEqual(direct, os.path.join(dirname, 'data'))
self.assertEqual(direct, Path(dirname) / 'data')

def test_basedir_default(self):
with patch.object(helpers, "_get_basedirectory") as mock:
mock.side_effect = ImportError()
direct = helpers.get_basedir()
self.assertEqual(direct, os.path.join(self._home, '.fades'))
self.assertEqual(direct, Path(self._home) / '.fades')

def test_basedir_xdg_nonexistant(self):
with patch("xdg.BaseDirectory") as mock:
Expand All @@ -267,19 +268,19 @@ def test_basedir_snap_nonexistant(self):

def test_confdir_xdg(self):
direct = helpers.get_confdir()
self.assertEqual(direct, os.path.join(BaseDirectory.xdg_config_home, 'fades'))
self.assertEqual(direct, Path(BaseDirectory.xdg_config_home) / 'fades')

def test_confdir_snap(self):
with tempfile.TemporaryDirectory() as dirname:
self._fake_snap_env_dir(dirname)
direct = helpers.get_confdir()
self.assertEqual(direct, os.path.join(dirname, 'config'))
self.assertEqual(direct, Path(dirname) / 'config')

def test_confdir_default(self):
with patch.object(helpers, "_get_basedirectory") as mock:
mock.side_effect = ImportError()
direct = helpers.get_confdir()
self.assertEqual(direct, os.path.join(self._home, '.fades'))
self.assertEqual(direct, Path(self._home) / '.fades')

def test_confdir_xdg_nonexistant(self):
with patch("xdg.BaseDirectory") as mock:
Expand Down
25 changes: 13 additions & 12 deletions tests/test_pipmanager.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import pytest
from unittest.mock import patch, call

from pathlib import Path
from fades.pipmanager import PipManager
from fades import pipmanager
from fades import helpers
Expand Down Expand Up @@ -74,44 +75,44 @@ def test_real_case_levenshtein():

def test_install():
mgr = PipManager(BIN_PATH, pip_installed=True)
pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
with patch.object(helpers, "logged_exec") as mock:
mgr.install("foo")

# check it always upgrades pip, and then the proper install
python_path = os.path.join(BIN_PATH, "python")
python_path = Path(BIN_PATH) / "python"
c1 = call([python_path, "-m", "pip", "install", "pip", "--upgrade"])
c2 = call([pip_path, "install", "foo"])
assert mock.call_args_list == [c1, c2]


def test_install_without_pip_upgrade():
mgr = PipManager(BIN_PATH, pip_installed=True, avoid_pip_upgrade=True)
pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
with patch.object(helpers, "logged_exec") as mock:
mgr.install("foo")
mock.assert_called_with([pip_path, "install", "foo"])


def test_install_multiword_dependency():
mgr = PipManager(BIN_PATH, pip_installed=True)
pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
with patch.object(helpers, "logged_exec") as mock:
mgr.install("foo bar")
mock.assert_called_with([pip_path, "install", "foo", "bar"])


def test_install_with_options():
mgr = PipManager(BIN_PATH, pip_installed=True, options=["--bar baz"])
pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
with patch.object(helpers, "logged_exec") as mock:
mgr.install("foo")
mock.assert_called_with([pip_path, "install", "foo", "--bar", "baz"])


def test_install_with_options_using_equal():
mgr = PipManager(BIN_PATH, pip_installed=True, options=["--bar=baz"])
pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
with patch.object(helpers, "logged_exec") as mock:
mgr.install("foo")
mock.assert_called_with([pip_path, "install", "foo", "--bar=baz"])
Expand All @@ -128,7 +129,7 @@ def test_install_raise_error(logs):

def test_install_without_pip():
mgr = PipManager(BIN_PATH, pip_installed=False)
pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
with patch.object(helpers, "logged_exec") as mocked_exec:
with patch.object(mgr, "_brute_force_install_pip") as mocked_install_pip:
mgr.install("foo")
Expand All @@ -139,11 +140,11 @@ def test_install_without_pip():
def test_brute_force_install_pip_installer_exists(tmp_path):
tmp_file = str(tmp_path / "hello.txt")
mgr = PipManager(BIN_PATH, pip_installed=False)
python_path = os.path.join(BIN_PATH, "python")
python_path = Path(BIN_PATH) / "python"

# get the tempfile but leave it there to be found
open(tmp_file, 'wt', encoding='utf8').close()
mgr.pip_installer_fname = tmp_file
mgr.pip_installer_fname = Path(tmp_file)

with patch.object(helpers, "logged_exec") as mocked_exec:
with patch.object(mgr, "_download_pip_installer") as download_installer:
Expand All @@ -157,9 +158,9 @@ def test_brute_force_install_pip_installer_exists(tmp_path):
def test_brute_force_install_pip_no_installer(tmp_path):
tmp_file = str(tmp_path / "hello.txt")
mgr = PipManager(BIN_PATH, pip_installed=False)
python_path = os.path.join(BIN_PATH, "python")
python_path = Path(BIN_PATH) / "python"

mgr.pip_installer_fname = tmp_file
mgr.pip_installer_fname = Path(tmp_file)
with patch.object(helpers, "logged_exec") as mocked_exec:
with patch.object(mgr, "_download_pip_installer") as download_installer:
mgr._brute_force_install_pip()
Expand Down Expand Up @@ -189,7 +190,7 @@ def test_freeze(tmp_path):
mock.return_value = ['moño>11', 'foo==1.2'] # "bad" order, on purpose
mgr.freeze(tmp_file)

pip_path = os.path.join(BIN_PATH, "pip")
pip_path = Path(BIN_PATH) / "pip"
mock.assert_called_with([pip_path, "freeze", "--all", "--local"])

# check results were stored properly
Expand Down