diff --git a/.gitignore b/.gitignore index c30900c..b613465 100644 --- a/.gitignore +++ b/.gitignore @@ -24,6 +24,7 @@ var/ *.egg-info/ .installed.cfg *.egg +.mypy_cache/ # PyInstaller # Usually these files are written by a python script from a template @@ -59,3 +60,6 @@ target/ # archlinux !pkg/archlinux/PKGBUILD pkg/archlinux/* + +# Ides +.vscode/ diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..29ec3c7 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,11 @@ +import shutil + +from pytest import fixture + + +@fixture(scope="function") +def tmp_file(tmpdir_factory): + """ Fixture for a unique tmpfile for each test.""" + dir_path = tmpdir_factory.mktemp("test") + yield str(dir_path.join("testfile")) # Converted to str to support python <3.6 versions + shutil.rmtree(str(dir_path)) diff --git a/tests/test_cache.py b/tests/test_cache.py deleted file mode 100644 index 3afc8af..0000000 --- a/tests/test_cache.py +++ /dev/null @@ -1,710 +0,0 @@ -# Copyright 2015-2017 Facundo Batista, Nicolás Demarchi -# -# This program is free software: you can redistribute it and/or modify it -# under the terms of the GNU General Public License version 3, as published -# by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranties of -# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR -# PURPOSE. See the GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program. If not, see . -# -# For further info, check https://github.com/PyAr/fades - -"""Tests for the helpers.""" - -import json -import os -import time -import unittest -import uuid - -from threading import Thread -from unittest.mock import patch - -from pkg_resources import parse_requirements, Distribution - -from fades import cache, helpers, parsing -from tests import get_tempfile - - -def get_req(text): - """Transform a text requirement into the pkg_resources object.""" - return list(parse_requirements(text)) - - -def get_distrib(*dep_ver_pairs): - """Build some Distributions with indicated info.""" - return [Distribution(project_name=dep, version=ver) for dep, ver in dep_ver_pairs] - - -class GetTestCase(unittest.TestCase): - """A shallow 'get'.""" - - def setUp(self): - super().setUp() - self.tempfile = get_tempfile(self) - - def test_missing_file(self): - os.remove(self.tempfile) - venvscache = cache.VEnvsCache(self.tempfile) - with patch.object(venvscache, '_select') as mock: - mock.return_value = None - resp = venvscache.get_venv('requirements', 'interpreter', uuid='', options='options') - mock.assert_called_with([], 'requirements', 'interpreter', uuid='', options='options') - self.assertEqual(resp, None) - - def test_empty_file(self): - open(self.tempfile, 'wt', encoding='utf8').close() - venvscache = cache.VEnvsCache(self.tempfile) - with patch.object(venvscache, '_select') as mock: - mock.return_value = None - resp = venvscache.get_venv('requirements', 'interpreter') - mock.assert_called_with([], 'requirements', 'interpreter', uuid='', options=None) - self.assertEqual(resp, None) - - def test_some_file_content(self): - with open(self.tempfile, 'wt', encoding='utf8') as fh: - fh.write('foo\nbar\n') - venvscache = cache.VEnvsCache(self.tempfile) - with patch.object(venvscache, '_select') as mock: - mock.return_value = 'resp' - resp = venvscache.get_venv('requirements', 'interpreter', uuid='', options='options') - mock.assert_called_with(['foo', 'bar'], 'requirements', 'interpreter', uuid='', - options='options') - self.assertEqual(resp, 'resp') - - def test_get_by_uuid(self): - with open(self.tempfile, 'wt', encoding='utf8') as fh: - fh.write('foo\nbar\n') - venvscache = cache.VEnvsCache(self.tempfile) - with patch.object(venvscache, '_select') as mock: - mock.return_value = 'resp' - resp = venvscache.get_venv(uuid='uuid') - mock.assert_called_with(['foo', 'bar'], None, '', uuid='uuid', options=None) - self.assertEqual(resp, 'resp') - - -class StoreTestCase(unittest.TestCase): - """Store what received.""" - - def setUp(self): - super().setUp() - self.tempfile = get_tempfile(self) - - def test_missing_file(self): - venvscache = cache.VEnvsCache(self.tempfile) - venvscache.store('installed', 'metadata', 'interpreter', 'options') - - with open(self.tempfile, 'rt', encoding='utf8') as fh: - data = json.loads(fh.readline()) - self.assertTrue('timestamp' in data) - self.assertEqual(data['installed'], 'installed') - self.assertEqual(data['metadata'], 'metadata') - self.assertEqual(data['interpreter'], 'interpreter') - self.assertEqual(data['options'], 'options') - - def test_with_previous_content(self): - with open(self.tempfile, 'wt', encoding='utf8') as fh: - fh.write(json.dumps({'foo': 'bar'}) + '\n') - - venvscache = cache.VEnvsCache(self.tempfile) - venvscache.store('installed', 'metadata', 'interpreter', 'options') - - with open(self.tempfile, 'rt', encoding='utf8') as fh: - data = json.loads(fh.readline()) - self.assertEqual(data, {'foo': 'bar'}) - - data = json.loads(fh.readline()) - self.assertTrue('timestamp' in data) - self.assertEqual(data['installed'], 'installed') - self.assertEqual(data['metadata'], 'metadata') - self.assertEqual(data['interpreter'], 'interpreter') - self.assertEqual(data['options'], 'options') - - -class RemoveTestCase(unittest.TestCase): - """Remove virtualenv from cache.""" - - def setUp(self): - super().setUp() - self.tempfile = get_tempfile(self) - - def test_missing_file(self): - venvscache = cache.VEnvsCache(self.tempfile) - venvscache.remove('missing/path') - - lines = venvscache._read_cache() - self.assertEqual(lines, []) - - def test_missing_env_in_cache(self): - venvscache = cache.VEnvsCache(self.tempfile) - options = {'foo': 'bar'} - venvscache.store('installed', {'env_path': 'some/path'}, 'interpreter', options=options) - lines = venvscache._read_cache() - assert len(lines) == 1 - - venvscache.remove('some/path') - - lines = venvscache._read_cache() - self.assertEqual(lines, []) - - def test_preserve_cache_data_ordering(self): - venvscache = cache.VEnvsCache(self.tempfile) - # store 3 venvs - options = {'foo': 'bar'} - venvscache.store('installed1', {'env_path': 'path/env1'}, 'interpreter', options=options) - venvscache.store('installed2', {'env_path': 'path/env2'}, 'interpreter', options=options) - venvscache.store('installed3', {'env_path': 'path/env3'}, 'interpreter', options=options) - - venvscache.remove('path/env2') - - lines = venvscache._read_cache() - self.assertEqual(len(lines), 2) - self.assertEqual( - json.loads(lines[0]).get('metadata').get('env_path'), 'path/env1') - self.assertEqual( - json.loads(lines[1]).get('metadata').get('env_path'), 'path/env3') - - def test_lock_cache_for_remove(self): - venvscache = cache.VEnvsCache(self.tempfile) - # store 3 venvs - options = {'foo': 'bar'} - venvscache.store('installed1', {'env_path': 'path/env1'}, 'interpreter', options=options) - venvscache.store('installed2', {'env_path': 'path/env2'}, 'interpreter', options=options) - venvscache.store('installed3', {'env_path': 'path/env3'}, 'interpreter', options=options) - - # patch _write_cache so it emulates a slow write during which - # another process managed to modify the cache file before the - # first process finished writing the modified cache data - original_write_cache = venvscache._write_cache - p = patch('fades.cache.VEnvsCache._write_cache') - mock_write_cache = p.start() - - t1 = Thread(target=venvscache.remove, args=('path/env1',)) - - def slow_write_cache(*args, **kwargs): - p.stop() - t1.start() - # wait to ensure t1 thread must wait for lock to be released - time.sleep(0.01) - original_write_cache(*args, **kwargs) - - mock_write_cache.side_effect = slow_write_cache - - # just a sanity check - assert not os.path.exists(venvscache.filepath + '.lock') - # remove a virtualenv from the cache - venvscache.remove('path/env2') - t1.join() - - # when cache file is properly locked both virtualenvs - # will have been removed from the cache - lines = venvscache._read_cache() - self.assertEqual(len(lines), 1) - self.assertEqual( - json.loads(lines[0]).get('metadata').get('env_path'), 'path/env3') - self.assertFalse(os.path.exists(venvscache.filepath + '.lock')) - - -class SelectionTestCase(unittest.TestCase): - """The venv selection.""" - - def setUp(self): - super().setUp() - tempfile = get_tempfile(self) - self.venvscache = cache.VEnvsCache(tempfile) - - def test_empty(self): - resp = self.venvscache._select([], {}, 'pythonX.Y', 'options') - self.assertEqual(resp, None) - - def test_nomatch_repo(self): - reqs = {'repoloco': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_nomatch_pypi_dependency(self): - reqs = {'pypi': get_req('dep1 == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep2': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_nomatch_vcs_dependency(self): - reqs = {'vcs': [parsing.VCSDependency('someurl')]} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'vcs': {'otherurl': None}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_nomatch_version(self): - reqs = {'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep': '7'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_simple_pypi_match(self): - reqs = {'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_simple_vcs_match(self): - reqs = {'vcs': [parsing.VCSDependency('someurl')]} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'vcs': {'someurl': None}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_match_mixed_single(self): - reqs = {'vcs': [parsing.VCSDependency('someurl')], 'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv1 = json.dumps({ - 'metadata': 'foobar1', - 'installed': {'vcs': {'someurl': None}, 'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - venv2 = json.dumps({ - 'metadata': 'foobar2', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - venv3 = json.dumps({ - 'metadata': 'foobar3', - 'installed': {'vcs': {'someurl': None}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select( - [venv1, venv2, venv3], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar1') - - def test_match_mixed_multiple(self): - reqs = {'vcs': [parsing.VCSDependency('url1'), parsing.VCSDependency('url2')], - 'pypi': get_req(['dep1 == 5', 'dep2'])} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'vcs': {'url1': None, 'url2': None}, - 'pypi': {'dep1': '5', 'dep2': '7'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_match_noversion(self): - reqs = {'pypi': get_req('dep')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_middle_match(self): - reqs = {'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv1 = json.dumps({ - 'metadata': 'venv1', - 'installed': {'pypi': {'dep': '3'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - venv2 = json.dumps({ - 'metadata': 'venv2', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - venv3 = json.dumps({ - 'metadata': 'venv3', - 'installed': {'pypi': {'dep': '7'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv1, venv2, venv3], reqs, interpreter, uuid='', - options=options) - self.assertEqual(resp, 'venv2') - - def test_multiple_match_bigger_version(self): - reqs = {'pypi': get_req('dep')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv1 = json.dumps({ - 'metadata': 'venv1', - 'installed': {'pypi': {'dep': '3'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - venv2 = json.dumps({ - 'metadata': 'venv2', - 'installed': {'pypi': {'dep': '7'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - venv3 = json.dumps({ - 'metadata': 'venv3', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv1, venv2, venv3], reqs, interpreter, uuid='', - options=options) - # matches venv2 because it has the bigger version for 'dep' (even if it's not the - # latest virtualenv created) - self.assertEqual(resp, 'venv2') - - def test_multiple_deps_ok(self): - reqs = {'pypi': get_req(['dep1 == 5', 'dep2 == 7'])} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep1': '5', 'dep2': '7'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_multiple_deps_just_one(self): - reqs = {'pypi': get_req(['dep1 == 5', 'dep2 == 7'])} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep1': '5', 'dep2': '2'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_not_too_crowded(self): - reqs = {'pypi': get_req(['dep1'])} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep1': '5', 'dep2': '2'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_same_quantity_different_deps(self): - reqs = {'pypi': get_req(['dep1', 'dep2'])} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep1': '5', 'dep3': '2'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_no_requirements_some_installed(self): - reqs = {} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep1': '5', 'dep3': '2'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_no_requirements_empty_venv(self): - reqs = {} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_simple_match_empty_options(self): - reqs = {'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'foobar') - - def test_no_match_due_to_options(self): - reqs = {'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'foobar', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, None) - - def test_match_due_to_options(self): - reqs = {'pypi': get_req('dep == 5')} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv1 = json.dumps({ - 'metadata': 'venv1', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {} - }) - venv2 = json.dumps({ - 'metadata': 'venv2', - 'installed': {'pypi': {'dep': '5'}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv1, venv2], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'venv2') - - def test_no_deps_but_options(self): - reqs = {} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv1 = json.dumps({ - 'metadata': 'venv1', - 'installed': {}, - 'interpreter': 'pythonX.Y', - 'options': {} - }) - venv2 = json.dumps({ - 'metadata': 'venv2', - 'installed': {}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv1, venv2], reqs, interpreter, uuid='', options=options) - self.assertEqual(resp, 'venv2') - - def test_match_uuid(self): - venv_uuid = str(uuid.uuid4()) - metadata = { - 'env_path': os.path.join(helpers.get_basedir(), venv_uuid), - } - venv = json.dumps({ - 'metadata': metadata, - 'installed': {}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], uuid=venv_uuid) - self.assertEqual(resp, metadata) - - -class ComparisonsTestCase(unittest.TestCase): - """The comparison in the selection.""" - - def setUp(self): - super().setUp() - tempfile = get_tempfile(self) - self.venvscache = cache.VEnvsCache(tempfile) - - def check(self, req, installed): - """Check if the requirement is satisfied with what is installed.""" - reqs = {'pypi': get_req('dep' + req)} - interpreter = 'pythonX.Y' - options = {'foo': 'bar'} - venv = json.dumps({ - 'metadata': 'ok', - 'installed': {'pypi': {'dep': installed}}, - 'interpreter': 'pythonX.Y', - 'options': {'foo': 'bar'} - }) - resp = self.venvscache._select([venv], reqs, interpreter, uuid='', options=options) - return resp - - def test_comp_eq(self): - self.assertEqual(self.check('==5', '5'), 'ok') - self.assertEqual(self.check('==5', '2'), None) - - def test_comp_gt(self): - self.assertEqual(self.check('>5', '4'), None) - self.assertEqual(self.check('>5', '5'), None) - self.assertEqual(self.check('>5', '6'), 'ok') - - def test_comp_ge(self): - self.assertEqual(self.check('>=5', '4'), None) - self.assertEqual(self.check('>=5', '5'), 'ok') - self.assertEqual(self.check('>=5', '6'), 'ok') - - def test_comp_lt(self): - self.assertEqual(self.check('<5', '4'), 'ok') - self.assertEqual(self.check('<5', '5'), None) - self.assertEqual(self.check('<5', '6'), None) - - def test_comp_le(self): - self.assertEqual(self.check('<=5', '4'), 'ok') - self.assertEqual(self.check('<=5', '5'), 'ok') - self.assertEqual(self.check('<=5', '6'), None) - - def test_complex_cases(self): - self.assertEqual(self.check('== 2.5', '2.5.0'), 'ok') - self.assertEqual(self.check('> 2.7', '2.12'), 'ok') - self.assertEqual(self.check('> 2.7a0', '2.7'), 'ok') - self.assertEqual(self.check('> 2.7', '2.7a0'), None) - - def test_crazy_picky(self): - self.assertEqual(self.check('>1.6,<1.9,!=1.9.6', '1.5.0'), None) - self.assertEqual(self.check('>1.6,<1.9,!=1.9.6', '1.6.7'), 'ok') - self.assertEqual(self.check('>1.6,<1.9,!=1.8.6', '1.8.7'), 'ok') - self.assertEqual(self.check('>1.6,<1.9,!=1.9.6', '1.9.6'), None) - - -class BestFitTestCase(unittest.TestCase): - """Check the venv best fitting decissor.""" - - def setUp(self): - super().setUp() - tempfile = get_tempfile(self) - self.venvscache = cache.VEnvsCache(tempfile) - - def check(self, possible_venvs): - """Assert that the selected venv is the best fit one.""" - self.assertEqual(self.venvscache._select_better_fit(possible_venvs), 'venv_best_fit') - - def test_one_simple(self): - self.check([ - (get_distrib(('dep', '3')), 'venv_best_fit'), - ]) - - def test_one_double(self): - self.check([ - (get_distrib(('dep1', '3'), ('dep2', '3')), 'venv_best_fit'), - ]) - - def test_two_simple(self): - self.check([ - (get_distrib(('dep', '5')), 'venv_best_fit'), - (get_distrib(('dep', '3')), 'venv_1'), - ]) - - def test_two_double_evident(self): - self.check([ - (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_best_fit'), - (get_distrib(('dep1', '3'), ('dep2', '6')), 'venv_1'), - ]) - - def test_two_double_mixed_1(self): - # tie! the one chosen is the last one - self.check([ - (get_distrib(('dep1', '3'), ('dep2', '9')), 'venv_1'), - (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_best_fit'), - ]) - - def test_two_double_mixed_2(self): - # tie! the one chosen is the last one - self.check([ - (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_1'), - (get_distrib(('dep1', '3'), ('dep2', '9')), 'venv_best_fit'), - ]) - - def test_two_triple(self): - self.check([ - (get_distrib(('dep1', '3'), ('dep2', '9'), ('dep3', '4')), 'venv_best_fit'), - (get_distrib(('dep1', '5'), ('dep2', '7'), ('dep3', '2')), 'venv_1'), - ]) - - def test_unordered(self): - # assert it compares each dependency with its equivalent - self.check([ - (get_distrib(('dep2', '3'), ('dep1', '2'), ('dep3', '8')), 'venv_best_fit'), - (get_distrib(('dep1', '7'), ('dep3', '5'), ('dep2', '2')), 'venv_1'), - ]) - - def test_big(self): - self.check([ - (get_distrib(('dep1', '3'), ('dep2', '2')), 'venv_1'), - (get_distrib(('dep1', '4'), ('dep2', '2')), 'venv_2'), - (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_best_fit'), - (get_distrib(('dep1', '5'), ('dep2', '6')), 'venv_3'), - ]) - - def test_vcs_alone(self): - self.check([ - ([parsing.VCSDependency('someurl')], 'venv_best_fit'), - ]) - - def test_vcs_mixed_simple(self): - self.check([ - ([parsing.VCSDependency('someurl')] + get_distrib(('dep', '3')), 'venv_best_fit'), - ]) - - def test_vcs_mixed_multiple(self): - self.check([ - ([parsing.VCSDependency('someurl')] + get_distrib(('dep', '3')), 'venv_best_fit'), - ([parsing.VCSDependency('someurl')] + get_distrib(('dep', '1')), 'venv_1'), - ]) diff --git a/tests/test_cache/conftest.py b/tests/test_cache/conftest.py new file mode 100644 index 0000000..e4b5ab0 --- /dev/null +++ b/tests/test_cache/conftest.py @@ -0,0 +1,25 @@ +import shutil +from pkg_resources import parse_requirements, Distribution + +from pytest import fixture + +from fades import cache + + +@fixture(scope="function") +def venvscache(tmpdir_factory): + """Fixture for a cache file for virtualenvs.""" + dir_path = tmpdir_factory.mktemp("test") + venvs_cache = cache.VEnvsCache(dir_path.join("test_venv_cache")) + yield venvs_cache + shutil.rmtree(str(dir_path)) + + +def get_req(text): + """Transform a text requirement into the pkg_resources object.""" + return list(parse_requirements(text)) + + +def get_distrib(*dep_ver_pairs): + """Build some Distributions with indicated info.""" + return [Distribution(project_name=dep, version=ver) for dep, ver in dep_ver_pairs] diff --git a/tests/test_cache/test_caches.py b/tests/test_cache/test_caches.py new file mode 100644 index 0000000..40900b5 --- /dev/null +++ b/tests/test_cache/test_caches.py @@ -0,0 +1,40 @@ +from fades import cache + + +def test_missing_file_pytest(tmp_file, mocker): + venvscache = cache.VEnvsCache(str(tmp_file)) + mock = mocker.patch.object(venvscache, '_select') + mock.return_value = None + resp = venvscache.get_venv('requirements', 'interpreter', uuid='', options='options') + mock.assert_called_with([], 'requirements', 'interpreter', uuid='', options='options') + assert not resp + + +def test_empty_file_pytest(tmp_file, mocker): + open(tmp_file, 'wt', encoding='utf8').close() + venvscache = cache.VEnvsCache(tmp_file) + mock = mocker.patch.object(venvscache, '_select', return_value=None) + resp = venvscache.get_venv('requirements', 'interpreter') + mock.assert_called_with([], 'requirements', 'interpreter', uuid='', options=None) + assert not resp + + +def test_some_file_content_pytest(tmp_file, mocker): + with open(tmp_file, 'wt', encoding='utf8') as fh: + fh.write('foo\nbar\n') + venvscache = cache.VEnvsCache(tmp_file) + mock = mocker.patch.object(venvscache, '_select', return_value="resp") + resp = venvscache.get_venv('requirements', 'interpreter', uuid='', options='options') + mock.assert_called_with(['foo', 'bar'], 'requirements', 'interpreter', uuid='', + options='options') + assert resp == 'resp' + + +def test_get_by_uuid_pytest(tmp_file, mocker): + with open(tmp_file, 'wt', encoding='utf8') as fh: + fh.write('foo\nbar\n') + venvscache = cache.VEnvsCache(tmp_file) + mock = mocker.patch.object(venvscache, '_select', return_value='resp') + resp = venvscache.get_venv(uuid='uuid') + mock.assert_called_with(['foo', 'bar'], None, '', uuid='uuid', options=None) + assert resp == 'resp' diff --git a/tests/test_cache/test_comparisons.py b/tests/test_cache/test_comparisons.py new file mode 100644 index 0000000..4ee8b74 --- /dev/null +++ b/tests/test_cache/test_comparisons.py @@ -0,0 +1,105 @@ +import json +import pytest + +from fades import parsing + +from conftest import get_req, get_distrib + + +@pytest.mark.parametrize("req,installed,expected", [ + # Equal + ("==5", "5", "ok"), + ("==5", "2", None), + # Greater than + (">5", "4", None), + (">5", "5", None), + (">5", "6", "ok"), + # Greater than or equal + (">=5", "4", None), + (">=5", "5", "ok"), + (">=5", "6", "ok"), + # Less than + ("<5", "4", "ok"), + ("<5", "5", None), + ("<5", "6", None), + # Less than or equal + ("<=5", "4", "ok"), + ("<=5", "5", "ok"), + ("<=5", "6", None), + # Complex cases + ("== 2.5", "2.5.0", "ok"), + ("> 2.7", "2.12", "ok"), + ("> 2.7a0", "2.7", "ok"), + ("> 2.7", "2.7a0", None), + # Crazy picky + (">1.6,<1.9,!=1.9.6", "1.5.0", None), + (">1.6,<1.9,!=1.9.6", "1.6.7", "ok"), + (">1.6,<1.9,!=1.8.6", "1.8.7", "ok"), + (">1.6,<1.9,!=1.9.6", "1.9.6", None), +]) +def test_check_versions(venvscache, req, installed, expected): + """The comparison in the selection.""" + reqs = {"pypi": get_req("dep" + req)} + interpreter = "pythonX.Y" + options = {"foo": "bar"} + venv = json.dumps({ + "metadata": "ok", + "installed": {"pypi": {"dep": installed}}, + "interpreter": "pythonX.Y", + "options": {"foo": "bar"} + }) + resp = venvscache._select([venv], reqs, interpreter, uuid="", options=options) + assert resp == expected + + +@pytest.mark.parametrize("possible_venvs", [ + [ + (get_distrib(('dep', '3')), 'venv_best_fit'), + ], + [ + (get_distrib(('dep1', '3'), ('dep2', '3')), 'venv_best_fit'), + ], + [ + (get_distrib(('dep', '5')), 'venv_best_fit'), + (get_distrib(('dep', '3')), 'venv_1'), + ], + [ + (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_best_fit'), + (get_distrib(('dep1', '3'), ('dep2', '6')), 'venv_1'), + ], + [ + (get_distrib(('dep1', '3'), ('dep2', '9')), 'venv_1'), + (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_best_fit'), + ], + [ + (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_1'), + (get_distrib(('dep1', '3'), ('dep2', '9')), 'venv_best_fit'), + ], + [ + (get_distrib(('dep1', '3'), ('dep2', '9'), ('dep3', '4')), 'venv_best_fit'), + (get_distrib(('dep1', '5'), ('dep2', '7'), ('dep3', '2')), 'venv_1'), + ], + [ + (get_distrib(('dep2', '3'), ('dep1', '2'), ('dep3', '8')), 'venv_best_fit'), + (get_distrib(('dep1', '7'), ('dep3', '5'), ('dep2', '2')), 'venv_1'), + ], + [ + (get_distrib(('dep1', '3'), ('dep2', '2')), 'venv_1'), + (get_distrib(('dep1', '4'), ('dep2', '2')), 'venv_2'), + (get_distrib(('dep1', '5'), ('dep2', '7')), 'venv_best_fit'), + (get_distrib(('dep1', '5'), ('dep2', '6')), 'venv_3'), + ], + [ + ([parsing.VCSDependency('someurl')], 'venv_best_fit'), + ], + [ + ([parsing.VCSDependency('someurl')] + get_distrib(('dep', '3')), 'venv_best_fit'), + ], + [ + ([parsing.VCSDependency('someurl')] + get_distrib(('dep', '3')), 'venv_best_fit'), + ([parsing.VCSDependency('someurl')] + get_distrib(('dep', '1')), 'venv_1'), + ], +]) +def test_best_fit(venvscache, possible_venvs): + """Check the venv best fitting decissor.""" + assert venvscache._select_better_fit(possible_venvs) == 'venv_best_fit' diff --git a/tests/test_cache/test_remove.py b/tests/test_cache/test_remove.py new file mode 100644 index 0000000..42fc2ec --- /dev/null +++ b/tests/test_cache/test_remove.py @@ -0,0 +1,85 @@ +import json +import os +import time + +from threading import Thread + +from fades import cache + + +def test_missing_file(tmp_file): + venvscache = cache.VEnvsCache(tmp_file) + venvscache.remove('missing/path') + + lines = venvscache._read_cache() + assert lines == [] + + +def test_missing_env_in_cache(tmp_file): + venvscache = cache.VEnvsCache(tmp_file) + options = {'foo': 'bar'} + venvscache.store('installed', {'env_path': 'some/path'}, 'interpreter', options=options) + lines = venvscache._read_cache() + assert len(lines) == 1 + + venvscache.remove('some/path') + + lines = venvscache._read_cache() + assert lines == [] + + +def test_preserve_cache_data_ordering(tmp_file): + venvscache = cache.VEnvsCache(tmp_file) + # store 3 venvs + options = {'foo': 'bar'} + venvscache.store('installed1', {'env_path': 'path/env1'}, 'interpreter', options=options) + venvscache.store('installed2', {'env_path': 'path/env2'}, 'interpreter', options=options) + venvscache.store('installed3', {'env_path': 'path/env3'}, 'interpreter', options=options) + + venvscache.remove('path/env2') + + lines = venvscache._read_cache() + assert len(lines) == 2 + assert json.loads(lines[0]).get('metadata').get('env_path') == 'path/env1' + assert json.loads(lines[1]).get('metadata').get('env_path') == 'path/env3' + + +def test_lock_cache_for_remove(tmp_file): + venvscache = cache.VEnvsCache(tmp_file) + # store 3 venvs + options = {'foo': 'bar'} + venvscache.store('installed1', {'env_path': 'path/env1'}, 'interpreter', options=options) + venvscache.store('installed2', {'env_path': 'path/env2'}, 'interpreter', options=options) + venvscache.store('installed3', {'env_path': 'path/env3'}, 'interpreter', options=options) + + # patch _write_cache so it emulates a slow write during which + # another process managed to modify the cache file before the + # first process finished writing the modified cache data + original_write_cache = venvscache._write_cache + + other_process = Thread(target=venvscache.remove, args=('path/env1',)) + + def slow_write_cache(*args, **kwargs): + venvscache._write_cache = original_write_cache + + # start "other process" and wait a little to ensure it must wait + # for the lock to be released + other_process.start() + time.sleep(0.01) + + original_write_cache(*args, **kwargs) + + venvscache._write_cache = slow_write_cache + + # just a sanity check + assert not os.path.exists(venvscache.filepath + '.lock') + # remove a virtualenv from the cache + venvscache.remove('path/env2') + other_process.join() + + # when cache file is properly locked both virtualenvs + # will have been removed from the cache + lines = venvscache._read_cache() + assert len(lines) == 1 + assert json.loads(lines[0]).get('metadata').get('env_path') == 'path/env3' + assert not os.path.exists(venvscache.filepath + '.lock') diff --git a/tests/test_cache/test_selection.py b/tests/test_cache/test_selection.py new file mode 100644 index 0000000..8775a32 --- /dev/null +++ b/tests/test_cache/test_selection.py @@ -0,0 +1,376 @@ +import os +import json +import uuid + +from fades import helpers, parsing +from conftest import get_req + + +def test_empty(venvscache): + resp = venvscache._select([], {}, 'pythonX.Y', 'options') + assert resp is None + + +def test_nomatch_repo_dependency(venvscache): + reqs = {"repoloco": get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_nomatch_pypi_dependency(venvscache): + reqs = {'pypi': get_req('dep1 == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep2': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache._select([venv], reqs, interpreter, uuid='', options=options) + resp is None + + +def test_nomatch_vcs_dependency(venvscache): + reqs = {'vcs': [parsing.VCSDependency('someurl')]} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'vcs': {'otherurl': None}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_nomatch_version(venvscache): + reqs = {'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep': '7'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_simple_pypi_match(venvscache): + reqs = {'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == 'foobar' + + +def test_simple_vcs_match(venvscache): + reqs = {'vcs': [parsing.VCSDependency('someurl')]} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'vcs': {'someurl': None}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == 'foobar' + + +def test_match_mixed_single(venvscache): + reqs = {'vcs': [parsing.VCSDependency('someurl')], 'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv1 = json.dumps({ + 'metadata': 'foobar1', + 'installed': {'vcs': {'someurl': None}, 'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + venv2 = json.dumps({ + 'metadata': 'foobar2', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + venv3 = json.dumps({ + 'metadata': 'foobar3', + 'installed': {'vcs': {'someurl': None}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select( + [venv1, venv2, venv3], reqs, interpreter, uuid='', options=options) + assert resp == 'foobar1' + + +def test_match_mixed_multiple(venvscache): + reqs = {'vcs': [parsing.VCSDependency('url1'), parsing.VCSDependency('url2')], + 'pypi': get_req(['dep1 == 5', 'dep2'])} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': { + 'vcs': {'url1': None, 'url2': None}, + 'pypi': {'dep1': '5', 'dep2': '7'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == 'foobar' + + +def test_match_noversion(venvscache): + reqs = {'pypi': get_req('dep')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == 'foobar' + + +def test_middle_match(venvscache): + reqs = {'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv1 = json.dumps({ + 'metadata': 'venv1', + 'installed': {'pypi': {'dep': '3'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + venv2 = json.dumps({ + 'metadata': 'venv2', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + venv3 = json.dumps({ + 'metadata': 'venv3', + 'installed': {'pypi': {'dep': '7'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv1, venv2, venv3], reqs, interpreter, uuid='', + options=options) + assert resp == 'venv2' + + +def test_multiple_match_bigger_version(venvscache): + reqs = {'pypi': get_req('dep')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv1 = json.dumps({ + 'metadata': 'venv1', + 'installed': {'pypi': {'dep': '3'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + venv2 = json.dumps({ + 'metadata': 'venv2', + 'installed': {'pypi': {'dep': '7'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + venv3 = json.dumps({ + 'metadata': 'venv3', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv1, venv2, venv3], reqs, interpreter, uuid='', + options=options) + # matches venv2 because it has the bigger version for 'dep' (even if it's not the + # latest virtualenv created) + assert resp == 'venv2' + + +def test_multiple_deps_ok(venvscache): + reqs = {'pypi': get_req(['dep1 == 5', 'dep2 == 7'])} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep1': '5', 'dep2': '7'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == 'foobar' + + +def test_multiple_deps_just_one(venvscache): + reqs = {'pypi': get_req(['dep1 == 5', 'dep2 == 7'])} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep1': '5', 'dep2': '2'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_not_too_crowded(venvscache): + reqs = {'pypi': get_req(['dep1'])} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep1': '5', 'dep2': '2'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_same_quantity_different_deps(venvscache): + reqs = {'pypi': get_req(['dep1', 'dep2'])} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep1': '5', 'dep3': '2'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_no_requirements_some_installed(venvscache): + reqs = {} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep1': '5', 'dep3': '2'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_no_requirements_empty_venv(venvscache): + reqs = {} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == "foobar" + + +def test_simple_match_empty_options(venvscache): + reqs = {'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp == "foobar" + + +def test_no_match_due_to_options(venvscache): + reqs = {'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv = json.dumps({ + 'metadata': 'foobar', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {} + }) + resp = venvscache ._select([venv], reqs, interpreter, uuid='', options=options) + assert resp is None + + +def test_match_due_to_options(venvscache): + reqs = {'pypi': get_req('dep == 5')} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv1 = json.dumps({ + 'metadata': 'venv1', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {} + }) + venv2 = json.dumps({ + 'metadata': 'venv2', + 'installed': {'pypi': {'dep': '5'}}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv1, venv2], reqs, interpreter, uuid='', options=options) + assert resp == "venv2" + + +def test_no_deps_but_options(venvscache): + reqs = {} + interpreter = 'pythonX.Y' + options = {'foo': 'bar'} + venv1 = json.dumps({ + 'metadata': 'venv1', + 'installed': {}, + 'interpreter': 'pythonX.Y', + 'options': {} + }) + venv2 = json.dumps({ + 'metadata': 'venv2', + 'installed': {}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv1, venv2], reqs, interpreter, uuid='', options=options) + assert resp == "venv2" + + +def test_match_uuid(venvscache): + venv_uuid = str(uuid.uuid4()) + metadata = { + 'env_path': os.path.join(helpers.get_basedir(), venv_uuid), + } + venv = json.dumps({ + 'metadata': metadata, + 'installed': {}, + 'interpreter': 'pythonX.Y', + 'options': {'foo': 'bar'} + }) + resp = venvscache ._select([venv], uuid=venv_uuid) + assert resp == metadata diff --git a/tests/test_cache/test_store.py b/tests/test_cache/test_store.py new file mode 100644 index 0000000..74c61c8 --- /dev/null +++ b/tests/test_cache/test_store.py @@ -0,0 +1,35 @@ +import json + +from fades import cache + + +def test_missing_file(tmp_file): + venvscache = cache.VEnvsCache(tmp_file) + venvscache.store('installed', 'metadata', 'interpreter', 'options') + + with open(tmp_file, 'rt', encoding='utf8') as fh: + data = json.loads(fh.readline()) + assert 'timestamp' in data + assert data['installed'], 'installed' + assert data['metadata'], 'metadata' + assert data['interpreter'], 'interpreter' + assert data['options'], 'options' + + +def test_with_previous_content(tmp_file): + with open(tmp_file, 'wt', encoding='utf8') as fh: + fh.write(json.dumps({'foo': 'bar'}) + '\n') + + venvscache = cache.VEnvsCache(tmp_file) + venvscache.store('installed', 'metadata', 'interpreter', 'options') + + with open(tmp_file, 'rt', encoding='utf8') as fh: + data = json.loads(fh.readline()) + assert data, {'foo': 'bar'} + + data = json.loads(fh.readline()) + assert 'timestamp' in data + assert data['installed'], 'installed' + assert data['metadata'], 'metadata' + assert data['interpreter'], 'interpreter' + assert data['options'], 'options'