Commit 3f634370 authored by Cyril Roelandt's avatar Cyril Roelandt

Add upt.Archive.

This allows frontends to give upt information about the files (tarballs,
binaries...) available upstream. Cryptographic hashes may not be retrieved from
upstream or computed if needed.
parent b5dd6fd3
......@@ -3,6 +3,14 @@ All notable changes to this project will be documented in this file.
Based on [Keep a Changelog](
## [Unreleased]
### Added
- upt.Archive: a representation of archive files, such as source tarballs
### Deprecated
- upt.Package.download_urls has been deprecated. Users should use
upt.Package.archives instead.
## [0.5]
### Fixed
- Added the requirements.txt file
......@@ -263,6 +263,10 @@ class MyFrontendParser(upt.Parser):
upt.Archive('', size=1234,
md5='md5', rmd160='rmd160', sha256='sha256')
......@@ -2,9 +2,14 @@
# Licensed under the 3-clause BSD license. See the LICENSE file.
from . import licenses
from .upt import Archive
from .upt import ArchiveType
from .upt import ArchiveUnavailable
from .upt import Backend
from .upt import Frontend
from .upt import Package
from .upt import PackageRequirement
from .exceptions import InvalidPackageNameError
from .exceptions import UnhandledFrontendError
from .checksum import HashUnknown
from .checksum import HashUnavailable
# Copyright 2018 Cyril Roelandt
# Licensed under the 3-clause BSD license. See the LICENSE file.
from enum import Enum
import hashlib
class HashUnknown(Exception):
def __init__(self, hash_name):
self.hash_name = hash_name
def __str__(self):
return f'Unknown hash "{self.hash_name}"'
class HashUnavailable(Exception):
def __init__(self, hash_name):
self.hash_name = hash_name
def __str__(self):
return f'Hash "{self.hash_name}" is not available on your system'
def _compute_checksum(checksum_fn, filepath):
# TODO: Compute this chunk by chunk?
m = checksum_fn()
with open(filepath, 'rb') as f:
return m.hexdigest()
def _compute_md5_checksum(filepath):
return _compute_checksum(hashlib.md5, filepath)
def _compute_rmd160_checksum(filepath):
return _compute_checksum(lambda:'ripemd160'), filepath)
def _compute_sha256_checksum(filepath):
return _compute_checksum(hashlib.sha256, filepath)
def compute_checksum(filepath, hash_name):
hash_functions = {
'md5': _compute_md5_checksum,
'rmd160': _compute_rmd160_checksum,
'sha256': _compute_sha256_checksum,
return hash_functions[hash_name](filepath)
except KeyError:
raise HashUnknown(hash_name)
except ValueError:
raise HashUnavailable(hash_name)
# Copyright 2018 Cyril Roelandt
# Licensed under the 3-clause BSD license. See the LICENSE file.
import unittest
from unittest import mock
from upt import checksum
@mock.patch('', new_callable=mock.mock_open, read_data=b'foobar')
class TestChecksum(unittest.TestCase):
MD5 = '3858f62230ac3c915f300c664312c63f'
SHA256 = 'c3ab8ff13720e8ad9047dd39466b3c8974e592c2fa383d4a3960714caef0c4f2'
def setUp(self):
self.path = '/fake/path'
def test_md5_checksum(self, open_fn):
self.assertEqual(checksum._compute_md5_checksum(self.path), self.MD5)
def test_sha256_checksum(self, open_fn):
def test_compute_checksum(self, open_fn):
self.assertEqual(checksum.compute_checksum(self.path, 'md5'), self.MD5)
def test_compute_checksum_invalid_hash(self, open_fn):
error = 'Unknown hash "fake-hash"'
with self.assertRaisesRegex(checksum.HashUnknown, error):
checksum.compute_checksum(self.path, 'fake-hash')
@mock.patch('', side_effect=ValueError)
def test_compute_checksum_unavailable_hash(self, hash_fn, open_fn):
error = 'Hash "rmd160" is not available on your system'
with self.assertRaisesRegex(checksum.HashUnavailable, error):
checksum.compute_checksum(self.path, 'rmd160')
......@@ -17,6 +17,38 @@ class TestPackage(unittest.TestCase):
pkg = upt.Package('foo', '4.2')
self.assertNotEqual(str(pkg), 'foo@42')
def test_get_archive_none_available(self):
error = 'No such archive could be found'
pkg = upt.Package('foo', '4.2', archives=[])
with self.assertRaisesRegex(upt.ArchiveUnavailable, error):
def test_get_archive(self):
archives = [
pkg = upt.Package('foo', '4.2', archives=archives)
self.assertEqual(pkg.get_archive(), archives[0])
def test_clean_archive_downloaded(self, remove_fn):
archives = [
archives[0]._filepath = '/fake/path' # Archive 'downloaded'
pkg = upt.Package('foo', '4.2', archives=archives)
def test_clean_archive_not_downloaded(self, remove_fn):
archives = [
pkg = upt.Package('foo', '4.2', archives=archives)
class TestPackageRequirement(unittest.TestCase):
def test_str_with_specifier(self):
......@@ -206,3 +238,41 @@ class TestCommandLine(unittest.TestCase):
with self.assertRaises(SystemExit) as exit:
self.assertEqual(exit.exception.code, 1)
class TestArchive(unittest.TestCase):
def setUp(self):
@mock.patch('os.stat', return_value=mock.Mock(st_size=42))
def test_size(self, stat_fn):
archive = upt.Archive('url', size=0)
archive._filepath = '/fake/path'
self.assertEqual(archive.size, 42)
archive = upt.Archive('url', size=12)
archive._filepath = '/fake/path'
self.assertEqual(archive.size, 12)
@mock.patch('urllib.request.urlretrieve', return_value=('/path', None))
def test_filepath(self, urlretrieve_fn):
archive = upt.Archive('url')
self.assertEqual(archive.filepath, '/path')
@mock.patch('upt.checksum.compute_checksum', return_value='hash-output')
@mock.patch('urllib.request.urlretrieve', return_value=('/path', None))
def test_checksum(self, urlretrieve_fn, compute_checksum_fn):
archive = upt.Archive('url')
self.assertEqual(archive._checksum('fake-hash'), 'hash-output')
self.assertEqual(archive._checksum('fake-hash'), 'hash-output')
@mock.patch('upt.checksum.compute_checksum', return_value='hash-output')
def test_checksums_provided(self, compute_checksum_fn):
archive = upt.Archive('url', md5='md5', rmd160='rmd160',
self.assertEqual(archive.md5, 'md5')
self.assertEqual(archive.rmd160, 'rmd160')
self.assertEqual(archive.sha256, 'sha256')
......@@ -2,11 +2,15 @@
# Licensed under the 3-clause BSD license. See the LICENSE file.
import argparse
from enum import Enum
import logging
import os
import urllib.request
import sys
import pkg_resources
import upt.checksum
import upt.exceptions
import upt.log
......@@ -20,6 +24,79 @@ class Frontend(object):
class ArchiveType(Enum):
class ArchiveUnavailable(Exception):
def __str__(self):
return 'No such archive could be found'
class Archive(object):
'''An archive file.
This can be a source tarball, a Python wheel, a Ruby gem, a binary, etc.
def __init__(self, url, archive_type=ArchiveType.SOURCE_TARGZ, size=0,
md5=None, sha256=None, rmd160=None):
self.url = url
self.archive_type = archive_type
self._size = size
self._filepath = None
self._hashes = {}
if md5 is not None:
self.md5 = md5
if sha256 is not None:
self.sha256 = sha256
if rmd160 is not None:
self.rmd160 = rmd160
def filepath(self):
if self._filepath is None:
self._filepath, _ = urllib.request.urlretrieve(self.url)
return self._filepath
def size(self):
if self._size == 0:
self._size = os.stat(self.filepath).st_size
return self._size
def _checksum(self, hash_name):
return self._hashes[hash_name]
except KeyError:
value = upt.checksum.compute_checksum(self.filepath, hash_name)
self._hashes[hash_name] = value
return value
def md5(self):
return self._checksum('md5')
def md5(self, value):
self._hashes['md5'] = value
def rmd160(self):
return self._checksum('rmd160')
def rmd160(self, value):
self._hashes['rmd160'] = value
def sha256(self):
return self._checksum('sha256')
def sha256(self, value):
self._hashes['sha256'] = value
class PackageRequirement(object):
"""Package Requirement.
......@@ -55,7 +132,7 @@ class Package(object):
- homepage: a string, representing the official homepage of the package
- summary: a short summary of the package's purpose
- description: a long description of the package's purpose
- download_urls: a list of strings, representing possible download URLs
- download_urls: deprecated
- requirement: a dict with 3 keys:
- 'build': the build dependencies of the package;
- 'run': the runtime dependencies of the package;
......@@ -70,6 +147,7 @@ class Package(object):
self.homepage = kwargs.get('homepage', '')
self.summary = kwargs.get('summary', '')
self.description = kwargs.get('description', '')
# TODO: Remove download urls.
self.download_urls = kwargs.get('download_urls', [])
# Software requirements, as instances of PackageRequirement: {
# 'build': [pkg1, ...],
......@@ -78,10 +156,23 @@ class Package(object):
# }
self.requirements = kwargs.get('requirements', {})
self.licenses = kwargs.get('licenses', [])
self.archives = kwargs.get('archives', [])
def __str__(self):
return f'{}@{self.version}'
def get_archive(self, archive_type=ArchiveType.SOURCE_TARGZ):
for archive in self.archives:
if archive.archive_type == archive_type:
return archive
raise ArchiveUnavailable()
def _clean(self):
for archive in self.archives:
if archive._filepath is not None:
def create_parser():
parser = argparse.ArgumentParser(prog='upt')
......@@ -185,3 +276,5 @@ def main():
except upt.exceptions.InvalidPackageNameError as e:
print(e, file=sys.stderr)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment