def get_supported_protocol_versions():
"""
1.2 -> 1
2.0 -> 2, 1
2.1 -> 3, 2, 1
2.2 -> 4, 3, 2, 1
3.X -> 4, 3
3.10 -> 5(beta),4,3
` """
if Version(CASSANDRA_VERSION) >= Version('3.10'):
return (3, 4, 5)
elif Version(CASSANDRA_VERSION) >= Version('3.0'):
return (3, 4)
elif Version(CASSANDRA_VERSION) >= Version('2.2'):
return (1, 2, 3, 4)
elif Version(CASSANDRA_VERSION) >= Version('2.1'):
return (1, 2, 3)
elif Version(CASSANDRA_VERSION) >= Version('2.0'):
return (1, 2)
else:
return (1)
python类Version()的实例源码
def _fetch_index_package_info(self, package_name, current_version):
"""
:type package_name: str
:type current_version: version.Version
"""
try:
package_canonical_name = package_name
if self.PYPI_API_TYPE == 'simple_html':
package_canonical_name = canonicalize_name(package_name)
response = requests.get(self.PYPI_API_URL.format(package=package_canonical_name), timeout=15)
except HTTPError as e: # pragma: nocover
return False, e.message
if not response.ok: # pragma: nocover
return False, 'API error: {}'.format(response.reason)
if self.PYPI_API_TYPE == 'pypi_json':
return self._parse_pypi_json_package_info(package_name, current_version, response)
elif self.PYPI_API_TYPE == 'simple_html':
return self._parse_simple_html_package_info(package_name, current_version, response)
else: # pragma: nocover
raise NotImplementedError('This type of PYPI_API_TYPE type is not supported')
def test_class_get_latest_major_versions_method(self, mocker, instance):
"""Test class `get_latest_major_versions()` method"""
mocker.patch.object(
instance, 'get_versions',
return_value=[Version(v) for v in self.versions],
)
assert instance.get_latest_major_versions() == {
'17': '17.3.1',
'1': '1.3.1',
'0': '0.6.1',
}
# Unsorted result from `get_version()`
unsorted_versions = self.versions.copy()
random.shuffle(unsorted_versions)
mocker.patch.object(
instance, 'get_versions',
return_value=[Version(v) for v in unsorted_versions],
)
with pytest.raises(ValueError):
instance.get_latest_major_versions()
def test_class_get_versions_method(self, mocker, instance):
"""Test class `get_versions()` method"""
versions = [
Version(v) for v in
['v1.3.2', 'v1.3.1', 'v1.3.0', '3.2.8']
]
mocked_get_github_tags = mocker.patch.object(
instance, '_get_github_tags', return_value=versions,
)
sorted_versions = sorted(
versions,
key=operator.attrgetter('base_version'),
reverse=True,
)
assert instance.get_versions() == sorted_versions
mocked_get_github_tags.assert_called_once_with(
normalize_func=instance._normalize_tag_name,
)
def test_versions(self):
with Mocker(CondaforgeRepoMock('package_a')):
src_a = Condaforge(Package('package_a'))
self.assertSetEqual(
{Version('1.0'), Version('1.2'), Version('2.0')},
src_a.versions
)
with Mocker(CondaforgeRepoMock('package_b')):
src_b = Condaforge(Package('package_b'))
self.assertSetEqual(
{Version('1.0')},
src_b.versions
)
with Mocker(CondaforgeRepoMock('package_z')):
src_z = Condaforge(Package('package_z'))
self.assertSetEqual(
{Version('0.1')},
src_z.versions
)
def versions(self) -> Set[Version]:
if self._versions is not None:
return self._versions
if self._data_versions is None:
return None
versions = set()
for version_str in self._data_versions:
try:
version = Version(version_str)
except InvalidVersion:
logging.info("Got invalid version for {}: {}".format(self.package.get_name(), version_str))
continue
versions.add(version)
self._versions = versions
return self._versions
def _check_version(self, config):
"""
Raises a VersionIncompatibleException when the current sceptre version
does not comply with the configured version requirement.
:raises: sceptre.exceptions.VersionIncompatibleException
"""
sceptre_version = __version__
if 'require_version' in config:
require_version = config['require_version']
if Version(sceptre_version) not in SpecifierSet(require_version):
raise VersionIncompatibleError(
"Current sceptre version ({0}) does not meet version "
"requirements: {1}".format(
sceptre_version, require_version
)
)
def get_supported_protocol_versions():
"""
1.2 -> 1
2.0 -> 2, 1
2.1 -> 3, 2, 1
2.2 -> 4, 3, 2, 1
3.X -> 4, 3
3.10(C*) -> 5(beta),4,3
3.10(DSE) -> DSE_V1,4,3
` """
if Version(CASSANDRA_VERSION) >= Version('3.10') and DSE_VERSION:
return (3, 4, ProtocolVersion.DSE_V1)
elif Version(CASSANDRA_VERSION) >= Version('3.10'):
return (3, 4, 5)
elif Version(CASSANDRA_VERSION) >= Version('3.0'):
return (3, 4)
elif Version(CASSANDRA_VERSION) >= Version('2.2'):
return (3, 4)
elif Version(CASSANDRA_VERSION) >= Version('2.1'):
return (3, )
elif Version(CASSANDRA_VERSION) >= Version('2.0'):
raise Exception("Cassandra Version 2.0 not supported anymore")
else:
raise Exception("Cassandra Version not supported anymore")
def get_default_protocol():
if Version(CASSANDRA_VERSION) >= Version('2.2'):
return 4
elif Version(CASSANDRA_VERSION) >= Version('2.1'):
return 3
elif Version(CASSANDRA_VERSION) >= Version('2.0'):
return 2
else:
return 1
def get_unsupported_lower_protocol():
"""
This is used to determine the lowest protocol version that is NOT
supported by the version of C* running
"""
if Version(CASSANDRA_VERSION) >= Version('3.0'):
return 2
else:
return None
def get_unsupported_upper_protocol():
"""
This is used to determine the highest protocol version that is NOT
supported by the version of C* running
"""
if Version(CASSANDRA_VERSION) >= Version('2.2'):
return None
if Version(CASSANDRA_VERSION) >= Version('2.1'):
return 4
elif Version(CASSANDRA_VERSION) >= Version('2.0'):
return 3
else:
return None
def _parse_pypi_json_package_info(self, package_name, current_version, response):
"""
:type package_name: str
:type current_version: version.Version
:type response: requests.models.Response
"""
data = response.json()
all_versions = [version.parse(vers) for vers in data['releases'].keys()]
filtered_versions = [vers for vers in all_versions if not vers.is_prerelease and not vers.is_postrelease]
if not filtered_versions: # pragma: nocover
return False, 'error while parsing version'
latest_version = max(filtered_versions)
# even if user did not choose prerelease, if the package from requirements is pre/post release, use it
if self._prerelease or current_version.is_postrelease or current_version.is_prerelease:
prerelease_versions = [vers for vers in all_versions if vers.is_prerelease or vers.is_postrelease]
if prerelease_versions:
latest_version = max(prerelease_versions)
try:
try:
latest_version_info = data['releases'][str(latest_version)][0]
except KeyError: # pragma: nocover
# non-RFC versions, get the latest from pypi response
latest_version = version.parse(data['info']['version'])
latest_version_info = data['releases'][str(latest_version)][0]
except Exception: # pragma: nocover
return False, 'error while parsing version'
upload_time = latest_version_info['upload_time'].replace('T', ' ')
return {
'name': package_name,
'current_version': current_version,
'latest_version': latest_version,
'upgrade_available': current_version < latest_version,
'upload_time': upload_time
}, 'success'
def _parse_simple_html_package_info(self, package_name, current_version, response):
"""
:type package_name: str
:type current_version: version.Version
:type response: requests.models.Response
"""
pattern = r'<a.*>.*{name}-([A-z0-9\.-]*)(?:-py|\.tar).*<\/a>'.format(name=re.escape(package_name))
versions_match = re.findall(pattern, response.content.decode('utf-8'), flags=re.IGNORECASE)
all_versions = [version.parse(vers) for vers in versions_match]
filtered_versions = [vers for vers in all_versions if not vers.is_prerelease and not vers.is_postrelease]
if not filtered_versions: # pragma: nocover
return False, 'error while parsing version'
latest_version = max(filtered_versions)
# even if user did not choose prerelease, if the package from requirements is pre/post release, use it
if self._prerelease or current_version.is_postrelease or current_version.is_prerelease:
prerelease_versions = [vers for vers in all_versions if vers.is_prerelease or vers.is_postrelease]
if prerelease_versions:
latest_version = max(prerelease_versions)
return {
'name': package_name,
'current_version': current_version,
'latest_version': latest_version,
'upgrade_available': current_version < latest_version,
'upload_time': '-'
}, 'success'
def parse_args(args=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('version', help="Version to tag")
parser.add_argument('-r', '--remote', default='upstream')
parser.add_argument('--no-push', action='store_false')
return parser.parse_args(args)
def check(version):
v = parse(version)
assert isinstance(v, Version), f'Invalid version: {version}'
assert not version.startswith('v')
def _adapt(v: Version) -> PEP440Adapter:
return PEP440Adapter(str(v))
def can_execute_action(self, version: VersionContainer, action: str) -> bool:
v = version.version # type: Version
if action == "bump-pre" and v._version.post or v._version.dev:
return False
elif action == "bump-post" and v._version.dev or v._version.pre:
return False
elif action == "bump-dev" and v._version.pre or v._version.post:
return False
else:
return True
def print_help(self) -> None:
print("%s\n"
"=====================\n"
"\n"
"The %s Version Parser should be used with PIP/PEP-440 version strings.\n"
"This is by far the easiest choice if you're uploading your source code to the\n"
"cheeseshop (pypi) or a local pypi installation through setup.py anyway, since\n"
"that means that the version string used by your setup.py is probably already\n"
"pep440-compatible anyway.\n"
"\n"
"The PEP440 Version Parser does not require any additional configuration.\n" %
(highlight("PEP440 Version Parser"), highlight("pep440")))
test_fortiosapi_virsh.py 文件源码
项目:fortiosapi
作者: fortinet-solutions-cse
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def test_02getsystemglobal(self):
resp = fgt.get('system','global', vdom="global")
fortiversion = resp['version']
self.assertEqual(resp['status'], 'success')
#should put a test on version to disable if less than 5.6 don't work decoration
#@unittest.skipIf(Version(fgt.get_version()) < Version('5.6'),
# "not supported with fortios before 5.6")
test_fortiosapi_virsh.py 文件源码
项目:fortiosapi
作者: fortinet-solutions-cse
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def test_is_license_valid(self):
if Version(fgt.get_version()) > Version('5.6'):
self.assertTrue(fgt.license()['results']['vm']['status'] == "vm_valid" or "vm_eval")
else:
self.assertTrue(True, "not supported before 5.6")
azure_rm.py 文件源码
项目:f5-automation-workflows-multicloud
作者: f5devcentral
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def main():
if not HAS_AZURE:
sys.exit("The Azure python sdk is not installed (try `pip install 'azure>=2.0.0rc5' --upgrade`) - {0}".format(HAS_AZURE_EXC))
if Version(azure_compute_version) < Version(AZURE_MIN_VERSION):
sys.exit("Expecting azure.mgmt.compute.__version__ to be {0}. Found version {1} "
"Do you have Azure >= 2.0.0rc5 installed? (try `pip install 'azure>=2.0.0rc5' --upgrade`)".format(AZURE_MIN_VERSION, azure_compute_version))
AzureInventory()
def main():
if not HAS_AZURE:
sys.exit("The Azure python sdk is not installed (try `pip install 'azure>=2.0.0rc5' --upgrade`) - {0}".format(HAS_AZURE_EXC))
if Version(azure_compute_version) < Version(AZURE_MIN_VERSION):
sys.exit("Expecting azure.mgmt.compute.__version__ to be {0}. Found version {1} "
"Do you have Azure >= 2.0.0rc5 installed? (try `pip install 'azure>=2.0.0rc5' --upgrade`)".format(AZURE_MIN_VERSION, azure_compute_version))
AzureInventory()
def get_latest_version(self):
"""
Returns latest project version
:returns: latest version
:rtype: packaging.version.Version
"""
for version in self.get_versions():
if not version.is_postrelease and not version.is_prerelease:
return str(version)
def _get_github_tags(self, github_url=None, normalize_func=None):
"""
Yields GitHub tag names, serialized to `Version` object
:param github_url: project GitHub repository URL
:type github_url: str
:param normalize_func: function that normalizes tag name
:type normalize_func: callable
:returns: generator of project versions
:rtype: generator of packaging.version.Version
:raises ValueError: when passed URL is not a GitHub repository
"""
if not github_url:
github_url = self.repository
owner, name = deconstruct_github_url(github_url)
tags = github_client.repository(owner, name).iter_tags()
for tag in tags:
if normalize_func:
name = normalize_func(tag.name)
else:
name = tag.name
# Add a minor version if it's not there, i.e. v1 -> v1.0
if name and len(name.split('.')) == 1:
name += '.0'
try:
version = Version(name)
except InvalidVersion:
continue
yield version
def test_class_get_latest_version_method(self, mocker, instance):
"""Test class `get_latest_version()` method"""
mocker.patch.object(
instance, 'get_versions',
return_value=[Version(v) for v in self.versions],
)
assert instance.get_latest_version() == '17.3.1'
def test_class_get_latest_minor_versions_method(self, mocker, instance):
"""Test class `get_latest_minor_versions()` method"""
mocker.patch.object(
instance, 'get_versions',
return_value=[Version(v) for v in self.versions],
)
assert instance.get_latest_minor_versions() == {
'17.3': '17.3.1',
'1.3': '1.3.1',
'1.2': '1.2.2',
'1.0': '1.0',
'0.6': '0.6.1',
'0.5': '0.5',
'0.1': '0.1.1',
}
# Unsorted result from `get_version()`
unsorted_versions = self.versions.copy()
random.shuffle(unsorted_versions)
mocker.patch.object(
instance, 'get_versions',
return_value=[Version(v) for v in unsorted_versions],
)
with pytest.raises(ValueError):
instance.get_latest_minor_versions()
def test_class_get_latest_version(self, mocker, instance):
"""Test class `get_latest_version()` method"""
versions = [
'v1.3.2', 'v1.3.1', 'v1.3.0', '3.2.8',
]
mocker.patch.object(
instance, 'get_versions',
return_value=[Version(v) for v in versions],
)
assert instance.get_latest_version() == '3.2.8'
def _verify_version_compatibility(self):
current_version = get_edi_version()
required_version = str(self._get_general_item('edi_required_minimal_edi_version', current_version))
if Version(get_stripped_version(current_version)) < Version(get_stripped_version(required_version)):
raise FatalError(('The current configuration requires a newer version of edi (>={}).\n'
'Please update your edi installation!'
).format(get_stripped_version(required_version)))
def normalize(cls, v):
# TODO: Trailing '.0's should be ignored - i.e., "1.2.0" == "1.2" - however,
# `packaging.version` does not seem to expose any functions to deal with that.
return str(PackagingVersion(v))
def sort_key(self):
return PackagingVersion(self.version)