Compare commits

...
Author SHA1 Message Date
rembo10 83398cb102 metadata: Fix typo in DISC_TOTAL 2022-02-18 15:04:42 +05:30
rembo10 61c2e1f821 Add option to ignore disc# for single disc albums (#3297) 2022-02-18 11:59:15 +05:30
rembo10 3e3047aef2 Use str(e) instead of e.message in searcher.py 2022-02-15 15:11:27 +05:30
rembo10 fff44e4631 Various fixes from last commit, fixes an issue where the hash couldn't be calculated from the torrent data 2022-02-15 15:06:26 +05:30
rembo10 0964371de8 Require python 3.7+ for dataclasses 2022-02-15 14:28:44 +05:30
rembo10 654f923a8d A little cleanup in searcher.py 2022-02-15 14:14:06 +05:30
rembo10 b91206c64a Remove tests from libs 2022-02-14 13:43:44 +05:30
rembo10 c9ba59ee9a Add zipp lib 2022-02-14 13:42:40 +05:30
rembo10 b7e35d5ff0 Decode the b64 data to utf-8 for nzbget (#3294) 2022-02-14 10:14:44 +05:30
rembo10 9d82143abe Remove errant print statements 2022-02-14 08:51:39 +05:30
rembo10 eaf2db6c59 Open blackhole nzb file as binary 2022-02-14 08:12:47 +05:30
rembo10 586b9ed3c8 Add importlib_resources to lib 2022-02-14 07:45:09 +05:30
rembo10 d89f4171da Disable interpolation in ConfigParser 2022-02-13 10:13:47 +05:30
rembo10 9f7be5348b Prevent accessing error.message in db.py action 2022-02-11 10:42:11 +05:30
rembo10 9c254ff222 Fix for trying to access e.message in findArtist 2022-02-11 10:34:44 +05:30
rembo10 ba969fd3b8 Fix for sending invalid dates to helpers.age 2022-02-10 07:37:49 +05:30
rembo10 c851d5ed1a Remove ordereddict 2022-02-09 21:24:20 +05:30
rembo10 2223928958 Fix sort by have in web ui, a little import cleanup in webserve.py 2022-02-09 21:10:14 +05:30
rembo10 164c3cacbc Disable last.fm getSimilar in importer 2022-02-09 17:51:12 +05:30
43 changed files with 1645 additions and 2479 deletions
+2 -2
View File
@@ -17,8 +17,8 @@
import os
import sys
if sys.version_info <= (3, 5):
sys.stdout.write("Headphones requires Python >= 3.6\n")
if sys.version_info <= (3, 6):
sys.stdout.write("Headphones requires Python >= 3.7\n")
sys.exit(1)
# Ensure lib added to path, before any other imports
+6 -3
View File
@@ -1370,17 +1370,20 @@
<div class="row">
<label>File Format</label>
<input type="text" name="file_format" value="${config['file_format']}" size="43">
<small>Use: $Disc/$disc (disc #), $Track/$track (track #), $Title/$title, $Artist/$artist, $Album/$album and $Year/$year. Put optional variables in curly braces, use single-quote marks to escape curly braces literally ('{', '}').</small>
<small>Use: In addition to the above, there is also $Title/$title (track title), $Track (track #), $Disc (disc #), $DiscTotal.</small>
</div>
<div class="checkbox row clearfix">
<div class="checkbox row left clearfix nopad">
<input type="checkbox" name="file_underscores" id="file_underscores" value="1" ${config['file_underscores']}/><label>Use underscores instead of spaces</label>
</div>
<div class="checkbox row left clearfix nopad">
<input type="checkbox" name="rename_single_disc_ignore" id="rename_single_disc_ignore" value="1" ${config['rename_single_disc_ignore']}/><label>Don't include disc# for single disc albums</label>
</div>
</fieldset>
<fieldset>
<legend>Re-Encoding Options</legend>
<small class="heading"><i class="fa fa-info-circle"></i> Note: this option requires the lame, ffmpeg or xld encoder</small>
<div class="checkbox row clearfix">
<div class="checkbox row left clearfix nopad">
<input type="checkbox" name="music_encoder" id="music_encoder" value="1" ${config['music_encoder']}/><label>Re-encode downloads during postprocessing</label>
</div>
<div id="encoderoptions" class="row clearfix checkbox">
+3 -2
View File
@@ -240,6 +240,7 @@ _CONFIG_DEFINITIONS = {
'QBITTORRENT_PASSWORD': (str, 'QBitTorrent', ''),
'QBITTORRENT_USERNAME': (str, 'QBitTorrent', ''),
'RENAME_FILES': (int, 'General', 0),
'RENAME_SINGLE_DISC_IGNORE': (int, 'General', 0),
'RENAME_UNPROCESSED': (bool_int, 'General', 1),
'RENAME_FROZEN': (bool_int, 'General', 1),
'REPLACE_EXISTING_FOLDERS': (int, 'General', 0),
@@ -327,7 +328,7 @@ class Config(object):
def __init__(self, config_file):
""" Initialize the config with values from a file """
self._config_file = config_file
self._config = ConfigParser()
self._config = ConfigParser(interpolation=None)
self._config.read(self._config_file)
for key in list(_CONFIG_DEFINITIONS.keys()):
self.check_setting(key)
@@ -376,7 +377,7 @@ class Config(object):
def write(self):
""" Make a copy of the stored config and write it to the configured file """
new_config = ConfigParser()
new_config = ConfigParser(interpolation=None)
# first copy over everything from the old config, even if it is not
# correctly defined to keep from losing data
+1 -1
View File
@@ -117,7 +117,7 @@ class DBConnection:
break
except sqlite3.OperationalError as e:
if "unable to open database file" in e.message or "database is locked" in e.message:
if "unable to open database file" in str(e) or "database is locked" in str(e):
dberror = e
if args is None:
logger.debug('Database error: %s. Query: %s', e, query)
+7
View File
@@ -1043,3 +1043,10 @@ def capture_beets_log(logger='beets'):
yield capture.messages
finally:
log.removeHandler(capture)
def have_pct_have_total(db_artist):
have_tracks = db_artist['HaveTracks'] or 0
total_tracks = db_artist['TotalTracks'] or 0
have_pct = have_tracks / total_tracks if total_tracks else 0
return (have_pct, total_tracks)
+8 -7
View File
@@ -102,12 +102,13 @@ def artistlist_to_mbids(artistlist, forced=False):
myDB.action('DELETE from newartists WHERE ArtistName=?', [artist])
# Update the similar artist tag cloud:
logger.info('Updating artist information from Last.fm')
# TODO: Fix last.fm api
# logger.info('Updating artist information from Last.fm')
try:
lastfm.getSimilar()
except Exception as e:
logger.warn('Failed to update artist information from Last.fm: %s' % e)
# try:
# lastfm.getSimilar()
# except Exception as e:
# logger.warn('Failed to update artist information from Last.fm: %s' % e)
def addArtistIDListToDB(artistidlist):
@@ -274,9 +275,9 @@ def addArtisttoDB(artistid, extrasonly=False, forcefull=False, type="artist"):
if len(check_release_date) == 10:
release_date = check_release_date
elif len(check_release_date) == 7:
release_date = check_release_date + "-31"
release_date = check_release_date + "-27"
elif len(check_release_date) == 4:
release_date = check_release_date + "-12-31"
release_date = check_release_date + "-12-27"
else:
release_date = today
if helpers.age(release_date) < ignore_age:
+5 -11
View File
@@ -14,20 +14,14 @@
# along with Headphones. If not, see <http://www.gnu.org/licenses/>.
from headphones import logger, db, helpers
from collections import OrderedDict
import musicbrainzngs
import headphones
import musicbrainzngs
import headphones.lock
from headphones import logger, db, helpers
try:
# pylint:disable=E0611
# ignore this error because we are catching the ImportError
from collections import OrderedDict
# pylint:enable=E0611
except ImportError:
# Python 2.6.x fallback, from libs
from ordereddict import OrderedDict
mb_lock = headphones.lock.TimedLock(0)
@@ -97,7 +91,7 @@ def findArtist(name, limit=1):
try:
artistResults = musicbrainzngs.search_artists(limit=limit, **criteria)['artist-list']
except ValueError as e:
if "at least one query term is required" in e.message:
if "at least one query term is required" in str(e):
logger.error(
"Tried to search without a term, or an empty one. Provided artist (probably emtpy): %s",
name)
+10 -2
View File
@@ -79,6 +79,7 @@ class Vars:
Metadata $variable names (only ones set explicitly by headphones).
"""
DISC = '$Disc'
DISC_TOTAL = '$DiscTotal'
TRACK = '$Track'
TITLE = '$Title'
ARTIST = '$Artist'
@@ -171,7 +172,7 @@ def _lower(s):
return None
def file_metadata(path, release):
def file_metadata(path, release, single_disc_ignore=False):
# type: (str,sqlite3.Row)->Tuple[Mapping[str,str],bool]
"""
Prepare metadata dictionary for path substitution, based on file name,
@@ -194,7 +195,13 @@ def file_metadata(path, release):
_row_to_dict(release, res)
date, year = _date_year(release)
if not f.disc:
if not f.disctotal or (f.disctotal == 1 and single_disc_ignore):
disc_total = ''
else:
disc_total = '%d' % f.disctotal
if not f.disc or (f.disctotal == 1 and single_disc_ignore):
disc_number = ''
else:
disc_number = '%d' % f.disc
@@ -226,6 +233,7 @@ def file_metadata(path, release):
album_title = release['AlbumTitle']
override_values = {
Vars.DISC: disc_number,
Vars.DISC_TOTAL: disc_total,
Vars.TRACK: track_number,
Vars.TITLE: title,
Vars.ARTIST: artist_name,
-1
View File
@@ -30,7 +30,6 @@ from . import getXldProfile
def encode(albumPath):
print(albumPath)
use_xld = headphones.CONFIG.ENCODER == 'xld'
# Return if xld details not found
+2 -1
View File
@@ -70,7 +70,8 @@ def sendNZB(nzb):
nzbcontent64 = None
if nzb.resultType == "nzbdata":
data = nzb.extraInfo[0]
nzbcontent64 = standard_b64encode(data)
# NZBGet needs a string, not bytes
nzbcontent64 = standard_b64encode(data).decode("utf-8")
logger.info("Sending NZB to NZBget")
logger.debug("URL: " + url)
+5 -3
View File
@@ -65,7 +65,6 @@ def checkFolder():
folder_name = torrent_folder_name
if folder_name:
print(folder_name)
album_path = os.path.join(download_dir, folder_name)
logger.debug("Checking if %s exists" % album_path)
@@ -80,7 +79,6 @@ def checkFolder():
def verify(albumid, albumpath, Kind=None, forced=False, keep_original_folder=False, single=False):
print(albumpath)
myDB = db.DBConnection()
release = myDB.action('SELECT * from albums WHERE AlbumID=?', [albumid]).fetchone()
tracks = myDB.select('SELECT * from tracks WHERE AlbumID=?', [albumid])
@@ -1087,7 +1085,11 @@ def renameFiles(albumpath, downloaded_track_list, release):
# Until tagging works better I'm going to rely on the already provided metadata
for downloaded_track in downloaded_track_list:
md, from_metadata = metadata.file_metadata(downloaded_track, release)
md, from_metadata = metadata.file_metadata(
downloaded_track,
release,
headphones.CONFIG.RENAME_SINGLE_DISC_IGNORE
)
if md is None:
# unable to parse media file, skip file
continue
+258 -199
View File
@@ -15,8 +15,8 @@
# NZBGet support added by CurlyMo <curlymoo1@gmail.com> as a part of XBian - XBMC on the Raspberry Pi
from base64 import b16encode, b32decode
from hashlib import sha1
import os
import re
import string
import random
import urllib.request, urllib.parse, urllib.error
@@ -24,19 +24,23 @@ import datetime
import subprocess
import unicodedata
import urllib.parse
from base64 import b16encode, b32decode
from hashlib import sha1
import os
import re
from bencode import encode as bencode
from bencode import decode as bdecode
from pygazelle import api as gazelleapi
from pygazelle import encoding as gazelleencoding
from pygazelle import format as gazelleformat
from pygazelle import release_type as gazellerelease_type
from unidecode import unidecode
import headphones
from headphones.common import USER_AGENT
from headphones.types import Result
from headphones import logger, db, helpers, classes, sab, nzbget, request
from headphones import utorrent, transmission, notifiers, rutracker, deluge, qbittorrent
from bencode import encode as bencode
from bencode import decode as bdecode
# Magnet to torrent services, for Black hole. Stolen from CouchPotato.
TORRENT_TO_MAGNET_SERVICES = [
@@ -52,6 +56,7 @@ ruobj = None
redobj = None
def fix_url(s, charset="utf-8"):
"""
Fix the URL so it is proper formatted and encoded.
@@ -77,8 +82,8 @@ def torrent_to_file(target_file, data):
fp.write(data)
except IOError as e:
logger.error(
"Could not write torrent file '%s': %s. Skipping.",
target_file, e.message)
f"Could not write `{target_file}`: {str(e)}"
)
return
# Try to change permissions
@@ -136,7 +141,7 @@ def calculate_torrent_hash(link, data=None):
if len(torrent_hash) == 32:
torrent_hash = b16encode(b32decode(torrent_hash)).lower()
elif data:
info = bdecode(data)["info"]
info = bdecode(data)[b"info"]
torrent_hash = sha1(bencode(info)).hexdigest()
else:
raise ValueError("Cannot calculate torrent hash without magnet link "
@@ -318,7 +323,7 @@ def do_sorted_search(album, new, losslessOnly, choose_specific_download=False):
return results
# Filter all results that do not comply
results = [result for result in results if result[5]]
results = [result for result in results if result.matches]
# Sort the remaining results
sorted_search_results = sort_search_results(results, album, new, albumlength)
@@ -326,11 +331,14 @@ def do_sorted_search(album, new, losslessOnly, choose_specific_download=False):
if not sorted_search_results:
return
logger.info("Making sure we can download the best result")
(data, bestqual) = preprocess(sorted_search_results)
logger.info(
"Making sure we can download the best result: "
f"{sorted_search_results[0].title} from {sorted_search_results[0].provider}"
)
(data, result) = preprocess(sorted_search_results)
if data and bestqual:
send_to_downloader(data, bestqual, album)
if data and result:
send_to_downloader(data, result, album)
def more_filtering(results, album, albumlength, new):
@@ -366,36 +374,46 @@ def more_filtering(results, album, albumlength, new):
for result in results:
if low_size_limit and (int(result[1]) < low_size_limit):
if low_size_limit and result.size < low_size_limit:
logger.info(
"%s from %s is too small for this album - not considering it. (Size: %s, Minsize: %s)",
result[0], result[3], helpers.bytes_to_mb(result[1]),
helpers.bytes_to_mb(low_size_limit))
f"{result.title} from {result.provider} is too small for this album. "
f"(Size: {result.size}, MinSize: {helpers.bytes_to_mb(low_size_limit)})"
)
continue
if high_size_limit and (int(result[1]) > high_size_limit):
if high_size_limit and result.size > high_size_limit:
logger.info(
"%s from %s is too large for this album - not considering it. (Size: %s, Maxsize: %s)",
result[0], result[3], helpers.bytes_to_mb(result[1]),
helpers.bytes_to_mb(high_size_limit))
f"{result.title} from {result.provider} is too large for this album. "
f"(Size: {result.size}, MaxSize: {helpers.bytes_to_mb(high_size_limit)})"
)
# Keep lossless results if there are no good lossy matches
if not (allow_lossless and 'flac' in result[0].lower()):
if not (allow_lossless and 'flac' in result.title.lower()):
continue
if new:
alreadydownloaded = myDB.select('SELECT * from snatched WHERE URL=?', [result[2]])
alreadydownloaded = myDB.select(
"SELECT * from snatched WHERE URL=?", [result.url]
)
if len(alreadydownloaded):
logger.info(
'%s has already been downloaded from %s. Skipping.' % (result[0], result[3]))
f"{result.title} has already been downloaded from "
f"{result.provider}. Skipping."
)
continue
newlist.append(result)
results = newlist
return newlist
return results
def sort_by_priority_then_size(rs):
return list(map(lambda x: x[0],
sorted(
rs,
key=lambda x: (x[0].matches, x[1], x[0].size),
reverse=True
)
))
def sort_search_results(resultlist, album, new, albumlength):
@@ -405,83 +423,74 @@ def sort_search_results(resultlist, album, new, albumlength):
return None
# Add a priority if it has any of the preferred words
temp_list = []
preferred_words = None
if headphones.CONFIG.PREFERRED_WORDS:
preferred_words = helpers.split_string(headphones.CONFIG.PREFERRED_WORDS)
results_with_priority = []
preferred_words = helpers.split_string(headphones.CONFIG.PREFERRED_WORDS)
for result in resultlist:
priority = 0
if preferred_words:
if any(word.lower() in result[0].lower() for word in preferred_words):
priority = 1
# add a search provider priority (weighted based on position)
i = next((i for i, word in enumerate(preferred_words) if word in result[3].lower()),
None)
if i is not None:
priority += round((len(preferred_words) - i) / float(len(preferred_words)), 2)
for word in preferred_words:
if word.lower() in [result.title.lower(), result.provider.lower()]:
priority += len(preferred_words) - preferred_words.index(word)
results_with_priority.append((result, priority))
temp_list.append((result[0], result[1], result[2], result[3], result[4], priority))
resultlist = temp_list
# if headphones.CONFIG.PREFERRED_QUALITY == 2 and headphones.CONFIG.PREFERRED_BITRATE and result[3] != 'Orpheus.network':
if headphones.CONFIG.PREFERRED_QUALITY == 2 and headphones.CONFIG.PREFERRED_BITRATE:
try:
targetsize = albumlength / 1000 * int(headphones.CONFIG.PREFERRED_BITRATE) * 128
if not targetsize:
logger.info('No track information for %s - %s. Defaulting to highest quality' % (
album['ArtistName'], album['AlbumTitle']))
finallist = sorted(resultlist, key=lambda title: (title[5], int(title[1])),
reverse=True)
logger.info(
f"No track information for {album['ArtistName']} - "
f"{album['AlbumTitle']}. Defaulting to highest quality"
)
return sort_by_priority_then_size(results_with_priority)
else:
newlist = []
flac_list = []
lossy_results_with_delta = []
lossless_results = []
for result in resultlist:
for result, priority in results_with_priority:
# Add lossless results to the "flac list" which we can use if there are no good lossy matches
if 'flac' in result[0].lower():
flac_list.append(
(result[0], result[1], result[2], result[3], result[4], result[5]))
continue
if 'flac' in result.title.lower():
lossless_results.append((result, priority))
else:
delta = abs(targetsize - result.size)
lossy_results_with_delta.append((result, priority, delta))
delta = abs(targetsize - int(result[1]))
newlist.append(
(result[0], result[1], result[2], result[3], result[4], result[5], delta))
return list(map(lambda x: x[0],
sorted(
lossy_results_with_delta,
key=lambda x: (-x[0].matches, -x[1], x[2])
)
))
finallist = sorted(newlist, key=lambda title: (-title[5], title[6]))
if not len(finallist) and len(
flac_list) and headphones.CONFIG.PREFERRED_BITRATE_ALLOW_LOSSLESS:
if (
not len(lossy_results_with_delta)
and len(lossless_results)
and headphones.CONFIG.PREFERRED_BITRATE_ALLOW_LOSSLESS
):
logger.info(
"Since there were no appropriate lossy matches (and at least one lossless match), going to use lossless instead")
finallist = sorted(flac_list, key=lambda title: (title[5], int(title[1])),
reverse=True)
"Since there were no appropriate lossy matches "
"(and at least one lossless match), going to use "
"lossless instead"
)
return sort_by_priority_then_size(results_with_priority)
except Exception:
logger.exception('Unhandled exception')
logger.info('No track information for %s - %s. Defaulting to highest quality',
album['ArtistName'], album['AlbumTitle'])
finallist = sorted(resultlist, key=lambda title: (title[5], int(title[1])),
reverse=True)
logger.info(
f"No track information for {album['ArtistName']} - "
f"{album['AlbumTitle']}. Defaulting to highest quality"
)
return sort_by_priority_then_size(results_with_priority)
else:
return sort_by_priority_then_size(results_with_priority)
finallist = sorted(resultlist, key=lambda title: (title[5], int(title[1])), reverse=True)
# keep number of seeders order for Orpheus.network
# if result[3] == 'Orpheus.network':
# finallist = resultlist
if not len(finallist):
logger.info('No appropriate matches found for %s - %s', album['ArtistName'],
album['AlbumTitle'])
return None
return finallist
logger.info(
f"No appropriate matches found for {album['ArtistName']} - "
f"{album['AlbumTitle']}"
)
return None
def get_year_from_release_date(release_date):
@@ -498,11 +507,22 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
reldate = album['ReleaseDate']
year = get_year_from_release_date(reldate)
dic = {'...': '', ' & ': ' ', ' = ': ' ', '?': '', '$': 's', ' + ': ' ', '"': '', ',': '',
'*': '', '.': '', ':': ''}
replacements = {
'...': '',
' & ': ' ',
' = ': ' ',
'?': '',
'$': 's',
' + ': ' ',
'"': '',
',': '',
'*': '',
'.': '',
':': ''
}
cleanalbum = helpers.latinToAscii(helpers.replace_all(album['AlbumTitle'], dic)).strip()
cleanartist = helpers.latinToAscii(helpers.replace_all(album['ArtistName'], dic)).strip()
cleanalbum = unidecode(helpers.replace_all(album['AlbumTitle'], replacements)).strip()
cleanartist = unidecode(helpers.replace_all(album['ArtistName'], replacements)).strip()
# Use the provided search term if available, otherwise build a search term
if album['SearchTerm']:
@@ -578,7 +598,7 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
title = item.title
size = int(item.links[1]['length'])
resultlist.append((title, size, url, provider, 'nzb', True))
resultlist.append(Result(title, size, url, provider, 'nzb', True))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
except Exception as e:
logger.error("An unknown error occurred trying to parse the feed: %s" % e)
@@ -651,7 +671,7 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
if all(word.lower() in title.lower() for word in term.split()):
logger.info(
'Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
resultlist.append((title, size, url, provider, 'nzb', True))
resultlist.append(Result(title, size, url, provider, 'nzb', True))
else:
logger.info('Skipping %s, not all search term words found' % title)
@@ -699,7 +719,7 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
title = item.title
size = int(item.links[1]['length'])
resultlist.append((title, size, url, provider, 'nzb', True))
resultlist.append(Result(title, size, url, provider, 'nzb', True))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
except Exception as e:
logger.exception("Unhandled exception while parsing feed")
@@ -746,7 +766,7 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
title = item['release']
size = int(item['sizebytes'])
resultlist.append((title, size, url, provider, 'nzb', True))
resultlist.append(Result(title, size, url, provider, 'nzb', True))
logger.info('Found %s. Size: %s', title, helpers.bytes_to_mb(size))
except Exception as e:
logger.exception("Unhandled exception")
@@ -758,7 +778,7 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
# Also will filter flac & remix albums if not specifically looking for it
# This code also checks the ignored words and required words
results = [result for result in resultlist if
verifyresult(result[0], artistterm, term, losslessOnly)]
verifyresult(result.title, artistterm, term, losslessOnly)]
# Additional filtering for size etc
if results and not choose_specific_download:
@@ -767,16 +787,18 @@ def searchNZB(album, new=False, losslessOnly=False, albumlength=None,
return results
def send_to_downloader(data, bestqual, album):
logger.info('Found best result from %s: <a href="%s">%s</a> - %s', bestqual[3], bestqual[2],
bestqual[0], helpers.bytes_to_mb(bestqual[1]))
def send_to_downloader(data, result, album):
logger.info(
f"Found best result from {result.provider}: <a href=\"{result.url}\">"
f"{result.title}</a> - {helpers.bytes_to_mb(result.size)}"
)
# Get rid of any dodgy chars here so we can prevent sab from renaming our downloads
kind = bestqual[4]
kind = result.kind
seed_ratio = None
torrentid = None
if kind == 'nzb':
folder_name = helpers.sab_sanitize_foldername(bestqual[0])
folder_name = helpers.sab_sanitize_foldername(result.title)
if headphones.CONFIG.NZB_DOWNLOADER == 1:
@@ -809,7 +831,7 @@ def send_to_downloader(data, bestqual, album):
try:
prev = os.umask(headphones.UMASK)
with open(download_path, 'w') as fp:
with open(download_path, 'wb') as fp:
fp.write(data)
os.umask(prev)
@@ -819,8 +841,8 @@ def send_to_downloader(data, bestqual, album):
return
else:
folder_name = '%s - %s [%s]' % (
helpers.latinToAscii(album['ArtistName']).replace('/', '_'),
helpers.latinToAscii(album['AlbumTitle']).replace('/', '_'),
unidecode(album['ArtistName']).replace('/', '_'),
unidecode(album['AlbumTitle']).replace('/', '_'),
get_year_from_release_date(album['ReleaseDate']))
# Blackhole
@@ -830,26 +852,26 @@ def send_to_downloader(data, bestqual, album):
torrent_name = helpers.replace_illegal_chars(folder_name) + '.torrent'
download_path = os.path.join(headphones.CONFIG.TORRENTBLACKHOLE_DIR, torrent_name)
if bestqual[2].lower().startswith("magnet:"):
if result.url.lower().startswith("magnet:"):
if headphones.CONFIG.MAGNET_LINKS == 1:
try:
if headphones.SYS_PLATFORM == 'win32':
os.startfile(bestqual[2])
os.startfile(result.url)
elif headphones.SYS_PLATFORM == 'darwin':
subprocess.Popen(["open", bestqual[2]], stdout=subprocess.PIPE,
subprocess.Popen(["open", result.url], stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
else:
subprocess.Popen(["xdg-open", bestqual[2]], stdout=subprocess.PIPE,
subprocess.Popen(["xdg-open", result.url], stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
# Gonna just take a guess at this..... Is there a better way to find this out?
folder_name = bestqual[0]
folder_name = result.title
except Exception as e:
logger.error("Error opening magnet link: %s" % str(e))
return
elif headphones.CONFIG.MAGNET_LINKS == 2:
# Procedure adapted from CouchPotato
torrent_hash = calculate_torrent_hash(bestqual[2])
torrent_hash = calculate_torrent_hash(result.url)
# Randomize list of services
services = TORRENT_TO_MAGNET_SERVICES[:]
@@ -863,8 +885,9 @@ def send_to_downloader(data, bestqual, album):
if not torrent_to_file(download_path, data):
return
# Extract folder name from torrent
folder_name = read_torrent_name(download_path,
bestqual[0])
folder_name = read_torrent_name(
download_path,
result.title)
# Break for loop
break
@@ -888,7 +911,7 @@ def send_to_downloader(data, bestqual, album):
return
# Extract folder name from torrent
folder_name = read_torrent_name(download_path, bestqual[0])
folder_name = read_torrent_name(download_path, result.title)
if folder_name:
logger.info('Torrent folder name: %s' % folder_name)
@@ -896,10 +919,10 @@ def send_to_downloader(data, bestqual, album):
logger.info("Sending torrent to Transmission")
# Add torrent
if bestqual[3] == 'rutracker.org':
if result.provider == 'rutracker.org':
torrentid = transmission.addTorrent('', data)
else:
torrentid = transmission.addTorrent(bestqual[2])
torrentid = transmission.addTorrent(result.url)
if not torrentid:
logger.error("Error sending torrent to Transmission. Are you sure it's running?")
@@ -913,7 +936,7 @@ def send_to_downloader(data, bestqual, album):
return
# Set Seed Ratio
seed_ratio = get_seed_ratio(bestqual[3])
seed_ratio = get_seed_ratio(result.provider)
if seed_ratio is not None:
transmission.setSeedRatio(torrentid, seed_ratio)
@@ -922,10 +945,10 @@ def send_to_downloader(data, bestqual, album):
try:
# Add torrent
if bestqual[3] == 'rutracker.org':
if result.provider == 'rutracker.org':
torrentid = deluge.addTorrent('', data)
else:
torrentid = deluge.addTorrent(bestqual[2])
torrentid = deluge.addTorrent(result.url)
if not torrentid:
logger.error("Error sending torrent to Deluge. Are you sure it's running? Maybe the torrent already exists?")
@@ -940,7 +963,7 @@ def send_to_downloader(data, bestqual, album):
deluge.setTorrentLabel({'hash': torrentid})
# Set Seed Ratio
seed_ratio = get_seed_ratio(bestqual[3])
seed_ratio = get_seed_ratio(result.provider)
if seed_ratio is not None:
deluge.setSeedRatio({'hash': torrentid, 'ratio': seed_ratio})
@@ -963,13 +986,13 @@ def send_to_downloader(data, bestqual, album):
logger.info("Sending torrent to uTorrent")
# Add torrent
if bestqual[3] == 'rutracker.org':
if result.provider == 'rutracker.org':
ruobj.utorrent_add_file(data)
else:
utorrent.addTorrent(bestqual[2])
utorrent.addTorrent(result.url)
# Get hash
torrentid = calculate_torrent_hash(bestqual[2], data)
torrentid = calculate_torrent_hash(result.url, data)
if not torrentid:
logger.error('Torrent id could not be determined')
return
@@ -987,23 +1010,23 @@ def send_to_downloader(data, bestqual, album):
utorrent.labelTorrent(torrentid)
# Set Seed Ratio
seed_ratio = get_seed_ratio(bestqual[3])
seed_ratio = get_seed_ratio(result.provider)
if seed_ratio is not None:
utorrent.setSeedRatio(torrentid, seed_ratio)
else: # if headphones.CONFIG.TORRENT_DOWNLOADER == 4:
logger.info("Sending torrent to QBiTorrent")
# Add torrent
if bestqual[3] == 'rutracker.org':
if result.provider == 'rutracker.org':
if qbittorrent.apiVersion2:
qbittorrent.addFile(data)
else:
ruobj.qbittorrent_add_file(data)
else:
qbittorrent.addTorrent(bestqual[2])
qbittorrent.addTorrent(result.url)
# Get hash
torrentid = calculate_torrent_hash(bestqual[2], data)
torrentid = calculate_torrent_hash(result.url, data)
torrentid = torrentid.lower()
if not torrentid:
logger.error('Torrent id could not be determined')
@@ -1018,29 +1041,33 @@ def send_to_downloader(data, bestqual, album):
return
# Set Seed Ratio
seed_ratio = get_seed_ratio(bestqual[3])
# Oh my god why is this repeated again for the 100th time
seed_ratio = get_seed_ratio(result.provider)
if seed_ratio is not None:
qbittorrent.setSeedRatio(torrentid, seed_ratio)
myDB = db.DBConnection()
myDB.action('UPDATE albums SET status = "Snatched" WHERE AlbumID=?', [album['AlbumID']])
myDB.action('INSERT INTO snatched VALUES( ?, ?, ?, ?, DATETIME("NOW", "localtime"), ?, ?, ?, ?)',
[album['AlbumID'], bestqual[0], bestqual[1], bestqual[2], "Snatched", folder_name,
kind, torrentid])
# Store the torrent id so we can check later if it's finished seeding and can be removed
if seed_ratio is not None and seed_ratio != 0 and torrentid:
myDB.action(
'INSERT INTO snatched VALUES( ?, ?, ?, ?, DATETIME("NOW", "localtime"), ?, ?, ?, ?)',
[album['AlbumID'], bestqual[0], bestqual[1], bestqual[2], "Seed_Snatched", folder_name,
kind, torrentid])
myDB.action(
"INSERT INTO snatched VALUES (?, ?, ?, ?, DATETIME('NOW', 'localtime'), "
"?, ?, ?, ?)", [
album['AlbumID'],
result.title,
result.size,
result.url,
"Seed_Snatched" if seed_ratio and torrentid else "Snatched",
folder_name,
kind,
torrentid
]
)
# notify
artist = album[1]
albumname = album[2]
rgid = album[6]
title = artist + ' - ' + albumname
provider = bestqual[3]
provider = result.provider
if provider.startswith(("http://", "https://")):
provider = provider.split("//")[1]
name = folder_name if folder_name else None
@@ -1209,13 +1236,22 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
year = get_year_from_release_date(reldate)
# MERGE THIS WITH THE TERM CLEANUP FROM searchNZB
dic = {'...': '', ' & ': ' ', ' = ': ' ', '?': '', '$': 's', ' + ': ' ', '"': '', ',': ' ',
'*': ''}
replacements = {
'...': '',
' & ': ' ',
' = ': ' ',
'?': '',
'$': 's',
' + ': ' ',
'"': '',
',': ' ',
'*': ''
}
semi_cleanalbum = helpers.replace_all(album['AlbumTitle'], dic)
cleanalbum = helpers.latinToAscii(semi_cleanalbum)
semi_cleanartist = helpers.replace_all(album['ArtistName'], dic)
cleanartist = helpers.latinToAscii(semi_cleanartist)
semi_cleanalbum = helpers.replace_all(album['AlbumTitle'], replacements)
cleanalbum = unidecode(semi_cleanalbum)
semi_cleanartist = helpers.replace_all(album['ArtistName'], replacements)
cleanartist = unidecode(semi_cleanartist)
# Use provided term if available, otherwise build our own (this code needs to be cleaned up since a lot
# of these torrent providers are just using cleanartist/cleanalbum terms
@@ -1350,7 +1386,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
if all(word.lower() in title.lower() for word in term.split()):
if size < maxsize and minimumseeders < seeders:
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
resultlist.append((title, size, url, provider, 'torrent', True))
resultlist.append(Result(title, size, url, provider, 'torrent', True))
else:
logger.info(
'%s is larger than the maxsize or has too little seeders for this category, '
@@ -1424,7 +1460,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
desc_match = re.search(r"Size: (\d+)<", item.description)
size = int(desc_match.group(1))
url = item.link
resultlist.append((title, size, url, provider, 'torrent', True))
resultlist.append(Result(title, size, url, provider, 'torrent', True))
logger.info('Found %s. Size: %s', title, helpers.bytes_to_mb(size))
except Exception as e:
logger.error(
@@ -1589,11 +1625,16 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
for torrent in match_torrents:
if not torrent.file_path:
torrent.group.update_group_data() # will load the file_path for the individual torrents
resultlist.append((torrent.file_path,
torrent.size,
orpheusobj.generate_torrent_link(torrent.id),
provider,
'torrent', True))
resultlist.append(
Result(
torrent.file_path,
torrent.size,
orpheusobj.generate_torrent_link(torrent.id),
provider,
'torrent',
True
)
)
# Redacted - Using same logic as What.CD as it's also Gazelle, so should really make this into something reusable
if headphones.CONFIG.REDACTED:
@@ -1690,11 +1731,16 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
if not torrent.file_path:
torrent.group.update_group_data() # will load the file_path for the individual torrents
use_token = headphones.CONFIG.REDACTED_USE_FLTOKEN and torrent.can_use_token
resultlist.append((torrent.file_path,
torrent.size,
redobj.generate_torrent_link(torrent.id, use_token),
provider,
'torrent', True))
resultlist.append(
Result(
torrent.file_path,
torrent.size,
redobj.generate_torrent_link(torrent.id, use_token),
provider,
'torrent',
True
)
)
# Pirate Bay
if headphones.CONFIG.PIRATEBAY:
@@ -1768,7 +1814,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
logger.info('%s is larger than the maxsize or has too little seeders for this category, '
'skipping. (Size: %i bytes, Seeders: %i)' % (title, size, int(seeds)))
resultlist.append((title, size, url, provider, "torrent", match))
resultlist.append(Result(title, size, url, provider, "torrent", match))
except Exception as e:
logger.error("An unknown error occurred in the Pirate Bay parser: %s" % e)
@@ -1822,7 +1868,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
logger.info('%s is larger than the maxsize or has too little seeders for this category, '
'skipping. (Size: %i bytes, Seeders: %i)' % (title, size, int(seeds)))
resultlist.append((title, size, url, provider, "torrent", match))
resultlist.append(Result(title, size, url, provider, "torrent", match))
except Exception as e:
logger.error(
"An unknown error occurred in the Old Pirate Bay parser: %s" % e)
@@ -1830,10 +1876,9 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
# attempt to verify that this isn't a substring result
# when looking for "Foo - Foo" we don't want "Foobar"
# this should be less of an issue when it isn't a self-titled album so we'll only check vs artist
results = [result for result in resultlist if verifyresult(result[0], artistterm, term, losslessOnly)]
results = [result for result in resultlist if verifyresult(result.title, artistterm, term, losslessOnly)]
# Additional filtering for size etc
# if results and not choose_specific_download and result[3] != 'Orpheus.network':
if results and not choose_specific_download:
results = more_filtering(results, album, albumlength, new)
@@ -1845,60 +1890,74 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
def preprocess(resultlist):
for result in resultlist:
if result[4] == 'torrent':
headers = {}
# rutracker always needs the torrent data
if result[3] == 'rutracker.org':
return ruobj.get_torrent_data(result[2]), result
# Jackett sometimes redirects
jackett_content = None
if result[3].startswith('Jackett_') or 'torznab' in result[3].lower():
r = request.request_response(url=result[2], headers=headers, allow_redirects=False)
if r:
jackett_content = r.content
link = r.headers.get('Location')
if link and link != result[2]:
if link.startswith('magnet:'):
result = (result[0], result[1], link, result[3], "magnet", result[5])
return "d10:magnet-uri%d:%se" % (len(link), link), result
else:
result = (result[0], result[1], link, result[3], result[4], result[5])
return True, result
if result.provider in ["The Pirate Bay", "Old Pirate Bay"]:
headers = {
'User-Agent':
'Mozilla/5.0 (Windows NT 6.3; Win64; x64) \
AppleWebKit/537.36 (KHTML, like Gecko) \
Chrome/41.0.2243.2 Safari/537.36'
}
else:
headers = {'User-Agent': USER_AGENT}
if result.kind == 'torrent':
# Get out of here if we're using Transmission or Deluge
# if not a magnet link still need the .torrent to generate hash... uTorrent support labeling
if headphones.CONFIG.TORRENT_DOWNLOADER in [1, 3]:
return True, result
# Get out of here if it's a magnet link
if result[2].lower().startswith("magnet:"):
if result.url.lower().startswith("magnet:"):
return True, result
# rutracker always needs the torrent data
if result.provider == 'rutracker.org':
return ruobj.get_torrent_data(result.url), result
# Jackett sometimes redirects
if result.provider.startswith('Jackett_') or 'torznab' in result.provider.lower():
r = request.request_response(url=result.url, headers=headers, allow_redirects=False)
if r:
link = r.headers.get('Location')
if link and link != result.url:
if link.startswith('magnet:'):
result = Result(
result.url,
result.size,
link,
result.provider,
"magnet",
result.matches
)
return "d10:magnet-uri%d:%se" % (len(link), link), result
else:
result = Result(
result.url,
result.size,
link,
result.provider,
result.kind,
result.matches
)
return True, result
else:
return r.content, result
# Download the torrent file
return request.request_content(url=result.url, headers=headers), result
if result[3] == 'Orpheus.network':
headers['User-Agent'] = 'Headphones'
elif result[3] == 'Redacted':
headers['User-Agent'] = 'Headphones'
elif result[3] == "The Pirate Bay" or result[3] == "Old Pirate Bay":
headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2243.2 Safari/537.36'
elif jackett_content:
return jackett_content, result
return request.request_content(url=result[2], headers=headers), result
if result[4] == 'magnet':
magnet_link = result[2]
if result.kind == 'magnet':
magnet_link = result.url
return "d10:magnet-uri%d:%se" % (len(magnet_link), magnet_link), result
else:
headers = {'User-Agent': USER_AGENT}
if result[3] == 'headphones':
return request.request_content(url=result[2], headers=headers,
auth=(headphones.CONFIG.HPUSER, headphones.CONFIG.HPPASS)), result
if result.provider == 'headphones':
return request.request_content(
url=result.url,
headers=headers,
auth=(headphones.CONFIG.HPUSER, headphones.CONFIG.HPPASS)
), result
else:
return request.request_content(url=result[2], headers=headers), result
return request.request_content(url=result.url, headers=headers), result
-1
View File
@@ -205,5 +205,4 @@ def torrentAction(method, arguments):
continue
resp_json = response.json()
print(resp_json)
return resp_json
+10
View File
@@ -0,0 +1,10 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class Result:
title: str
size: int
url: str
provider: str
kind: str
matches: bool
+60 -62
View File
@@ -15,34 +15,46 @@
# NZBGet support added by CurlyMo <curlymoo1@gmail.com> as a part of XBian - XBMC on the Raspberry Pi
from operator import itemgetter
import threading
import secrets
import random
import urllib.request, urllib.parse, urllib.error
import json
import time
import sys
from html import escape as html_escape
import urllib.request, urllib.error, urllib.parse
import os
import random
import re
from headphones import logger, searcher, db, importer, mb, lastfm, librarysync, helpers, notifiers, crier
from headphones.helpers import checked, radio, today, clean_name
from mako.lookup import TemplateLookup
from mako import exceptions
import headphones
import cherrypy
import secrets
import sys
import threading
import time
from collections import OrderedDict
from dataclasses import asdict
from html import escape as html_escape
from operator import itemgetter
from urllib import parse
try:
# pylint:disable=E0611
# ignore this error because we are catching the ImportError
from collections import OrderedDict
# pylint:enable=E0611
except ImportError:
# Python 2.6.x fallback, from libs
from ordereddict import OrderedDict
import cherrypy
from mako import exceptions
from mako.lookup import TemplateLookup
import headphones
from headphones import (
crier,
db,
importer,
lastfm,
librarysync,
logger,
mb,
notifiers,
searcher,
)
from headphones.helpers import (
checked,
clean_name,
have_pct_have_total,
pattern_substitute,
radio,
replace_illegal_chars,
today,
)
from headphones.types import Result
def serve_template(templatename, **kwargs):
@@ -326,9 +338,9 @@ class WebInterface(object):
'$first': firstchar.lower(),
}
folder = helpers.pattern_substitute(folder_format.strip(), values, normalize=True)
folder = pattern_substitute(folder_format.strip(), values, normalize=True)
folder = helpers.replace_illegal_chars(folder, type="folder")
folder = replace_illegal_chars(folder, type="folder")
folder = folder.replace('./', '_/').replace('/.', '/_')
if folder.endswith('.'):
@@ -440,40 +452,27 @@ class WebInterface(object):
@cherrypy.expose
@cherrypy.tools.json_out()
def choose_specific_download(self, AlbumID):
results = searcher.searchforalbum(AlbumID, choose_specific_download=True)
data = []
for result in results:
result_dict = {
'title': result[0],
'size': result[1],
'url': result[2],
'provider': result[3],
'kind': result[4],
'matches': result[5]
}
data.append(result_dict)
return data
results = searcher.searchforalbum(AlbumID, choose_specific_download=True) or []
return list(map(asdict, results))
@cherrypy.expose
@cherrypy.tools.json_out()
def download_specific_release(self, AlbumID, title, size, url, provider, kind, **kwargs):
# Handle situations where the torrent url contains arguments that are parsed
if kwargs:
url = urllib.parse.quote(url, safe=":?/=&") + '&' + urllib.parse.urlencode(kwargs)
url = parse.quote(url, safe=":?/=&") + '&' + parse.urlencode(kwargs)
try:
result = [(title, int(size), url, provider, kind)]
result = [Result(title, int(size), url, provider, kind, True)]
except ValueError:
result = [(title, float(size), url, provider, kind)]
result = [Result(title, float(size), url, provider, kind, True)]
logger.info("Making sure we can download the chosen result")
(data, bestqual) = searcher.preprocess(result)
data, result = searcher.preprocess(result)
if data and bestqual:
if data and result:
myDB = db.DBConnection()
album = myDB.action('SELECT * from albums WHERE AlbumID=?', [AlbumID]).fetchone()
searcher.send_to_downloader(data, bestqual, album)
searcher.send_to_downloader(data, result, album)
return {'result': 'success'}
else:
return {'result': 'failure'}
@@ -586,7 +585,7 @@ class WebInterface(object):
for albums in have_albums:
# Have to skip over manually matched tracks
if albums['ArtistName'] and albums['AlbumTitle'] and albums['TrackTitle']:
original_clean = helpers.clean_name(
original_clean = clean_name(
albums['ArtistName'] + " " + albums['AlbumTitle'] + " " + albums['TrackTitle'])
# else:
# original_clean = None
@@ -633,8 +632,8 @@ class WebInterface(object):
(artist, album))
elif action == "matchArtist":
existing_artist_clean = helpers.clean_name(existing_artist).lower()
new_artist_clean = helpers.clean_name(new_artist).lower()
existing_artist_clean = clean_name(existing_artist).lower()
new_artist_clean = clean_name(new_artist).lower()
if new_artist_clean != existing_artist_clean:
have_tracks = myDB.action(
'SELECT Matched, CleanName, Location, BitRate, Format FROM have WHERE ArtistName=?',
@@ -678,10 +677,10 @@ class WebInterface(object):
"Artist %s already named appropriately; nothing to modify" % existing_artist)
elif action == "matchAlbum":
existing_artist_clean = helpers.clean_name(existing_artist).lower()
new_artist_clean = helpers.clean_name(new_artist).lower()
existing_album_clean = helpers.clean_name(existing_album).lower()
new_album_clean = helpers.clean_name(new_album).lower()
existing_artist_clean = clean_name(existing_artist).lower()
new_artist_clean = clean_name(new_artist).lower()
existing_album_clean = clean_name(existing_album).lower()
new_album_clean = clean_name(new_album).lower()
existing_clean_string = existing_artist_clean + " " + existing_album_clean
new_clean_string = new_artist_clean + " " + new_album_clean
if existing_clean_string != new_clean_string:
@@ -737,7 +736,7 @@ class WebInterface(object):
'SELECT ArtistName, AlbumTitle, TrackTitle, CleanName, Matched from have')
for albums in manualalbums:
if albums['ArtistName'] and albums['AlbumTitle'] and albums['TrackTitle']:
original_clean = helpers.clean_name(
original_clean = clean_name(
albums['ArtistName'] + " " + albums['AlbumTitle'] + " " + albums['TrackTitle'])
if albums['Matched'] == "Ignored" or albums['Matched'] == "Manual" or albums[
'CleanName'] != original_clean:
@@ -778,7 +777,7 @@ class WebInterface(object):
[artist])
update_count = 0
for tracks in update_clean:
original_clean = helpers.clean_name(
original_clean = clean_name(
tracks['ArtistName'] + " " + tracks['AlbumTitle'] + " " + tracks[
'TrackTitle']).lower()
album = tracks['AlbumTitle']
@@ -810,7 +809,7 @@ class WebInterface(object):
(artist, album))
update_count = 0
for tracks in update_clean:
original_clean = helpers.clean_name(
original_clean = clean_name(
tracks['ArtistName'] + " " + tracks['AlbumTitle'] + " " + tracks[
'TrackTitle']).lower()
track_title = tracks['TrackTitle']
@@ -1018,9 +1017,7 @@ class WebInterface(object):
totalcount = myDB.select('SELECT COUNT(*) from artists')[0][0]
if sortbyhavepercent:
filtered.sort(key=lambda x: (
float(x['HaveTracks']) / x['TotalTracks'] if x['TotalTracks'] > 0 else 0.0,
x['HaveTracks'] if x['HaveTracks'] else 0.0), reverse=sSortDir_0 == "asc")
filtered.sort(key=have_pct_have_total, reverse=sSortDir_0 == "asc")
# can't figure out how to change the datatables default sorting order when its using an ajax datasource so ill
# just reverse it here and the first click on the "Latest Album" header will sort by descending release date
@@ -1271,6 +1268,7 @@ class WebInterface(object):
"cue_split_shntool_path": headphones.CONFIG.CUE_SPLIT_SHNTOOL_PATH,
"move_files": checked(headphones.CONFIG.MOVE_FILES),
"rename_files": checked(headphones.CONFIG.RENAME_FILES),
"rename_single_disc_ignore": checked(headphones.CONFIG.RENAME_SINGLE_DISC_IGNORE),
"correct_metadata": checked(headphones.CONFIG.CORRECT_METADATA),
"cleanup_files": checked(headphones.CONFIG.CLEANUP_FILES),
"keep_nfo": checked(headphones.CONFIG.KEEP_NFO),
@@ -1463,8 +1461,8 @@ class WebInterface(object):
"use_waffles", "use_rutracker",
"use_orpheus", "use_redacted", "redacted_use_fltoken", "preferred_bitrate_allow_lossless",
"detect_bitrate", "ignore_clean_releases", "freeze_db", "cue_split", "move_files",
"rename_files", "correct_metadata", "cleanup_files", "keep_nfo", "add_album_art",
"embed_album_art", "embed_lyrics",
"rename_files", "rename_single_disc_ignore", "correct_metadata", "cleanup_files",
"keep_nfo", "add_album_art", "embed_album_art", "embed_lyrics",
"replace_existing_folders", "keep_original_folder", "file_underscores",
"include_extras", "official_releases_only",
"wait_until_release_date", "autowant_upcoming", "autowant_all",
+36
View File
@@ -0,0 +1,36 @@
"""Read resources contained within a package."""
from ._common import (
as_file,
files,
Package,
)
from ._legacy import (
contents,
open_binary,
read_binary,
open_text,
read_text,
is_resource,
path,
Resource,
)
from importlib_resources.abc import ResourceReader
__all__ = [
'Package',
'Resource',
'ResourceReader',
'as_file',
'contents',
'files',
'is_resource',
'open_binary',
'open_text',
'path',
'read_binary',
'read_text',
]
+170
View File
@@ -0,0 +1,170 @@
from contextlib import suppress
from io import TextIOWrapper
from . import abc
class SpecLoaderAdapter:
"""
Adapt a package spec to adapt the underlying loader.
"""
def __init__(self, spec, adapter=lambda spec: spec.loader):
self.spec = spec
self.loader = adapter(spec)
def __getattr__(self, name):
return getattr(self.spec, name)
class TraversableResourcesLoader:
"""
Adapt a loader to provide TraversableResources.
"""
def __init__(self, spec):
self.spec = spec
def get_resource_reader(self, name):
return CompatibilityFiles(self.spec)._native()
def _io_wrapper(file, mode='r', *args, **kwargs):
if mode == 'r':
return TextIOWrapper(file, *args, **kwargs)
elif mode == 'rb':
return file
raise ValueError(
"Invalid mode value '{}', only 'r' and 'rb' are supported".format(mode)
)
class CompatibilityFiles:
"""
Adapter for an existing or non-existent resource reader
to provide a compatibility .files().
"""
class SpecPath(abc.Traversable):
"""
Path tied to a module spec.
Can be read and exposes the resource reader children.
"""
def __init__(self, spec, reader):
self._spec = spec
self._reader = reader
def iterdir(self):
if not self._reader:
return iter(())
return iter(
CompatibilityFiles.ChildPath(self._reader, path)
for path in self._reader.contents()
)
def is_file(self):
return False
is_dir = is_file
def joinpath(self, other):
if not self._reader:
return CompatibilityFiles.OrphanPath(other)
return CompatibilityFiles.ChildPath(self._reader, other)
@property
def name(self):
return self._spec.name
def open(self, mode='r', *args, **kwargs):
return _io_wrapper(self._reader.open_resource(None), mode, *args, **kwargs)
class ChildPath(abc.Traversable):
"""
Path tied to a resource reader child.
Can be read but doesn't expose any meaningful children.
"""
def __init__(self, reader, name):
self._reader = reader
self._name = name
def iterdir(self):
return iter(())
def is_file(self):
return self._reader.is_resource(self.name)
def is_dir(self):
return not self.is_file()
def joinpath(self, other):
return CompatibilityFiles.OrphanPath(self.name, other)
@property
def name(self):
return self._name
def open(self, mode='r', *args, **kwargs):
return _io_wrapper(
self._reader.open_resource(self.name), mode, *args, **kwargs
)
class OrphanPath(abc.Traversable):
"""
Orphan path, not tied to a module spec or resource reader.
Can't be read and doesn't expose any meaningful children.
"""
def __init__(self, *path_parts):
if len(path_parts) < 1:
raise ValueError('Need at least one path part to construct a path')
self._path = path_parts
def iterdir(self):
return iter(())
def is_file(self):
return False
is_dir = is_file
def joinpath(self, other):
return CompatibilityFiles.OrphanPath(*self._path, other)
@property
def name(self):
return self._path[-1]
def open(self, mode='r', *args, **kwargs):
raise FileNotFoundError("Can't open orphan path")
def __init__(self, spec):
self.spec = spec
@property
def _reader(self):
with suppress(AttributeError):
return self.spec.loader.get_resource_reader(self.spec.name)
def _native(self):
"""
Return the native reader if it supports files().
"""
reader = self._reader
return reader if hasattr(reader, 'files') else self
def __getattr__(self, attr):
return getattr(self._reader, attr)
def files(self):
return CompatibilityFiles.SpecPath(self.spec, self._reader)
def wrap_spec(package):
"""
Construct a package spec with traversable compatibility
on the spec/loader/reader.
"""
return SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)
+104
View File
@@ -0,0 +1,104 @@
import os
import pathlib
import tempfile
import functools
import contextlib
import types
import importlib
from typing import Union, Optional
from .abc import ResourceReader, Traversable
from ._compat import wrap_spec
Package = Union[types.ModuleType, str]
def files(package):
# type: (Package) -> Traversable
"""
Get a Traversable resource from a package
"""
return from_package(get_package(package))
def get_resource_reader(package):
# type: (types.ModuleType) -> Optional[ResourceReader]
"""
Return the package's loader if it's a ResourceReader.
"""
# We can't use
# a issubclass() check here because apparently abc.'s __subclasscheck__()
# hook wants to create a weak reference to the object, but
# zipimport.zipimporter does not support weak references, resulting in a
# TypeError. That seems terrible.
spec = package.__spec__
reader = getattr(spec.loader, 'get_resource_reader', None) # type: ignore
if reader is None:
return None
return reader(spec.name) # type: ignore
def resolve(cand):
# type: (Package) -> types.ModuleType
return cand if isinstance(cand, types.ModuleType) else importlib.import_module(cand)
def get_package(package):
# type: (Package) -> types.ModuleType
"""Take a package name or module object and return the module.
Raise an exception if the resolved module is not a package.
"""
resolved = resolve(package)
if wrap_spec(resolved).submodule_search_locations is None:
raise TypeError(f'{package!r} is not a package')
return resolved
def from_package(package):
"""
Return a Traversable object for the given package.
"""
spec = wrap_spec(package)
reader = spec.loader.get_resource_reader(spec.name)
return reader.files()
@contextlib.contextmanager
def _tempfile(reader, suffix=''):
# Not using tempfile.NamedTemporaryFile as it leads to deeper 'try'
# blocks due to the need to close the temporary file to work on Windows
# properly.
fd, raw_path = tempfile.mkstemp(suffix=suffix)
try:
try:
os.write(fd, reader())
finally:
os.close(fd)
del reader
yield pathlib.Path(raw_path)
finally:
try:
os.remove(raw_path)
except FileNotFoundError:
pass
@functools.singledispatch
def as_file(path):
"""
Given a Traversable object, return that object as a
path on the local file system in a context manager.
"""
return _tempfile(path.read_bytes, suffix=path.name)
@as_file.register(pathlib.Path)
@contextlib.contextmanager
def _(path):
"""
Degenerate behavior for pathlib.Path objects.
"""
yield path
+98
View File
@@ -0,0 +1,98 @@
# flake8: noqa
import abc
import sys
import pathlib
from contextlib import suppress
if sys.version_info >= (3, 10):
from zipfile import Path as ZipPath # type: ignore
else:
from zipp import Path as ZipPath # type: ignore
try:
from typing import runtime_checkable # type: ignore
except ImportError:
def runtime_checkable(cls): # type: ignore
return cls
try:
from typing import Protocol # type: ignore
except ImportError:
Protocol = abc.ABC # type: ignore
class TraversableResourcesLoader:
"""
Adapt loaders to provide TraversableResources and other
compatibility.
Used primarily for Python 3.9 and earlier where the native
loaders do not yet implement TraversableResources.
"""
def __init__(self, spec):
self.spec = spec
@property
def path(self):
return self.spec.origin
def get_resource_reader(self, name):
from . import readers, _adapters
def _zip_reader(spec):
with suppress(AttributeError):
return readers.ZipReader(spec.loader, spec.name)
def _namespace_reader(spec):
with suppress(AttributeError, ValueError):
return readers.NamespaceReader(spec.submodule_search_locations)
def _available_reader(spec):
with suppress(AttributeError):
return spec.loader.get_resource_reader(spec.name)
def _native_reader(spec):
reader = _available_reader(spec)
return reader if hasattr(reader, 'files') else None
def _file_reader(spec):
try:
path = pathlib.Path(self.path)
except TypeError:
return None
if path.exists():
return readers.FileReader(self)
return (
# native reader if it supplies 'files'
_native_reader(self.spec)
or
# local ZipReader if a zip module
_zip_reader(self.spec)
or
# local NamespaceReader if a namespace module
_namespace_reader(self.spec)
or
# local FileReader
_file_reader(self.spec)
# fallback - adapt the spec ResourceReader to TraversableReader
or _adapters.CompatibilityFiles(self.spec)
)
def wrap_spec(package):
"""
Construct a package spec with traversable compatibility
on the spec/loader/reader.
Supersedes _adapters.wrap_spec to use TraversableResourcesLoader
from above for older Python compatibility (<3.10).
"""
from . import _adapters
return _adapters.SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)
+35
View File
@@ -0,0 +1,35 @@
from itertools import filterfalse
from typing import (
Callable,
Iterable,
Iterator,
Optional,
Set,
TypeVar,
Union,
)
# Type and type variable definitions
_T = TypeVar('_T')
_U = TypeVar('_U')
def unique_everseen(
iterable: Iterable[_T], key: Optional[Callable[[_T], _U]] = None
) -> Iterator[_T]:
"List unique elements, preserving order. Remember all elements ever seen."
# unique_everseen('AAAABBBCCDAABBB') --> A B C D
# unique_everseen('ABBCcAD', str.lower) --> A B C D
seen: Set[Union[_T, _U]] = set()
seen_add = seen.add
if key is None:
for element in filterfalse(seen.__contains__, iterable):
seen_add(element)
yield element
else:
for element in iterable:
k = key(element)
if k not in seen:
seen_add(k)
yield element
+121
View File
@@ -0,0 +1,121 @@
import functools
import os
import pathlib
import types
import warnings
from typing import Union, Iterable, ContextManager, BinaryIO, TextIO, Any
from . import _common
Package = Union[types.ModuleType, str]
Resource = str
def deprecated(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
warnings.warn(
f"{func.__name__} is deprecated. Use files() instead. "
"Refer to https://importlib-resources.readthedocs.io"
"/en/latest/using.html#migrating-from-legacy for migration advice.",
DeprecationWarning,
stacklevel=2,
)
return func(*args, **kwargs)
return wrapper
def normalize_path(path):
# type: (Any) -> str
"""Normalize a path by ensuring it is a string.
If the resulting string contains path separators, an exception is raised.
"""
str_path = str(path)
parent, file_name = os.path.split(str_path)
if parent:
raise ValueError(f'{path!r} must be only a file name')
return file_name
@deprecated
def open_binary(package: Package, resource: Resource) -> BinaryIO:
"""Return a file-like object opened for binary reading of the resource."""
return (_common.files(package) / normalize_path(resource)).open('rb')
@deprecated
def read_binary(package: Package, resource: Resource) -> bytes:
"""Return the binary contents of the resource."""
return (_common.files(package) / normalize_path(resource)).read_bytes()
@deprecated
def open_text(
package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict',
) -> TextIO:
"""Return a file-like object opened for text reading of the resource."""
return (_common.files(package) / normalize_path(resource)).open(
'r', encoding=encoding, errors=errors
)
@deprecated
def read_text(
package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict',
) -> str:
"""Return the decoded string of the resource.
The decoding-related arguments have the same semantics as those of
bytes.decode().
"""
with open_text(package, resource, encoding, errors) as fp:
return fp.read()
@deprecated
def contents(package: Package) -> Iterable[str]:
"""Return an iterable of entries in `package`.
Note that not all entries are resources. Specifically, directories are
not considered resources. Use `is_resource()` on each entry returned here
to check if it is a resource or not.
"""
return [path.name for path in _common.files(package).iterdir()]
@deprecated
def is_resource(package: Package, name: str) -> bool:
"""True if `name` is a resource inside `package`.
Directories are *not* resources.
"""
resource = normalize_path(name)
return any(
traversable.name == resource and traversable.is_file()
for traversable in _common.files(package).iterdir()
)
@deprecated
def path(
package: Package,
resource: Resource,
) -> ContextManager[pathlib.Path]:
"""A context manager providing a file path object to the resource.
If the resource does not already exist on its own on the file system,
a temporary file will be created. If the file was created, the file
will be deleted upon exiting the context manager (no exception is
raised if the file was deleted prior to the context manager
exiting).
"""
return _common.as_file(_common.files(package) / normalize_path(resource))
+137
View File
@@ -0,0 +1,137 @@
import abc
from typing import BinaryIO, Iterable, Text
from ._compat import runtime_checkable, Protocol
class ResourceReader(metaclass=abc.ABCMeta):
"""Abstract base class for loaders to provide resource reading support."""
@abc.abstractmethod
def open_resource(self, resource: Text) -> BinaryIO:
"""Return an opened, file-like object for binary reading.
The 'resource' argument is expected to represent only a file name.
If the resource cannot be found, FileNotFoundError is raised.
"""
# This deliberately raises FileNotFoundError instead of
# NotImplementedError so that if this method is accidentally called,
# it'll still do the right thing.
raise FileNotFoundError
@abc.abstractmethod
def resource_path(self, resource: Text) -> Text:
"""Return the file system path to the specified resource.
The 'resource' argument is expected to represent only a file name.
If the resource does not exist on the file system, raise
FileNotFoundError.
"""
# This deliberately raises FileNotFoundError instead of
# NotImplementedError so that if this method is accidentally called,
# it'll still do the right thing.
raise FileNotFoundError
@abc.abstractmethod
def is_resource(self, path: Text) -> bool:
"""Return True if the named 'path' is a resource.
Files are resources, directories are not.
"""
raise FileNotFoundError
@abc.abstractmethod
def contents(self) -> Iterable[str]:
"""Return an iterable of entries in `package`."""
raise FileNotFoundError
@runtime_checkable
class Traversable(Protocol):
"""
An object with a subset of pathlib.Path methods suitable for
traversing directories and opening files.
"""
@abc.abstractmethod
def iterdir(self):
"""
Yield Traversable objects in self
"""
def read_bytes(self):
"""
Read contents of self as bytes
"""
with self.open('rb') as strm:
return strm.read()
def read_text(self, encoding=None):
"""
Read contents of self as text
"""
with self.open(encoding=encoding) as strm:
return strm.read()
@abc.abstractmethod
def is_dir(self) -> bool:
"""
Return True if self is a directory
"""
@abc.abstractmethod
def is_file(self) -> bool:
"""
Return True if self is a file
"""
@abc.abstractmethod
def joinpath(self, child):
"""
Return Traversable child in self
"""
def __truediv__(self, child):
"""
Return Traversable child in self
"""
return self.joinpath(child)
@abc.abstractmethod
def open(self, mode='r', *args, **kwargs):
"""
mode may be 'r' or 'rb' to open as text or binary. Return a handle
suitable for reading (same as pathlib.Path.open).
When opening as text, accepts encoding parameters such as those
accepted by io.TextIOWrapper.
"""
@abc.abstractproperty
def name(self) -> str:
"""
The base name of this object without any parent references.
"""
class TraversableResources(ResourceReader):
"""
The required interface for providing traversable
resources.
"""
@abc.abstractmethod
def files(self):
"""Return a Traversable object for the loaded package."""
def open_resource(self, resource):
return self.files().joinpath(resource).open('rb')
def resource_path(self, resource):
raise FileNotFoundError(resource)
def is_resource(self, path):
return self.files().joinpath(path).is_file()
def contents(self):
return (item.name for item in self.files().iterdir())
+122
View File
@@ -0,0 +1,122 @@
import collections
import pathlib
import operator
from . import abc
from ._itertools import unique_everseen
from ._compat import ZipPath
def remove_duplicates(items):
return iter(collections.OrderedDict.fromkeys(items))
class FileReader(abc.TraversableResources):
def __init__(self, loader):
self.path = pathlib.Path(loader.path).parent
def resource_path(self, resource):
"""
Return the file system path to prevent
`resources.path()` from creating a temporary
copy.
"""
return str(self.path.joinpath(resource))
def files(self):
return self.path
class ZipReader(abc.TraversableResources):
def __init__(self, loader, module):
_, _, name = module.rpartition('.')
self.prefix = loader.prefix.replace('\\', '/') + name + '/'
self.archive = loader.archive
def open_resource(self, resource):
try:
return super().open_resource(resource)
except KeyError as exc:
raise FileNotFoundError(exc.args[0])
def is_resource(self, path):
# workaround for `zipfile.Path.is_file` returning true
# for non-existent paths.
target = self.files().joinpath(path)
return target.is_file() and target.exists()
def files(self):
return ZipPath(self.archive, self.prefix)
class MultiplexedPath(abc.Traversable):
"""
Given a series of Traversable objects, implement a merged
version of the interface across all objects. Useful for
namespace packages which may be multihomed at a single
name.
"""
def __init__(self, *paths):
self._paths = list(map(pathlib.Path, remove_duplicates(paths)))
if not self._paths:
message = 'MultiplexedPath must contain at least one path'
raise FileNotFoundError(message)
if not all(path.is_dir() for path in self._paths):
raise NotADirectoryError('MultiplexedPath only supports directories')
def iterdir(self):
files = (file for path in self._paths for file in path.iterdir())
return unique_everseen(files, key=operator.attrgetter('name'))
def read_bytes(self):
raise FileNotFoundError(f'{self} is not a file')
def read_text(self, *args, **kwargs):
raise FileNotFoundError(f'{self} is not a file')
def is_dir(self):
return True
def is_file(self):
return False
def joinpath(self, child):
# first try to find child in current paths
for file in self.iterdir():
if file.name == child:
return file
# if it does not exist, construct it with the first path
return self._paths[0] / child
__truediv__ = joinpath
def open(self, *args, **kwargs):
raise FileNotFoundError(f'{self} is not a file')
@property
def name(self):
return self._paths[0].name
def __repr__(self):
paths = ', '.join(f"'{path}'" for path in self._paths)
return f'MultiplexedPath({paths})'
class NamespaceReader(abc.TraversableResources):
def __init__(self, namespace_path):
if 'NamespacePath' not in str(namespace_path):
raise ValueError('Invalid path')
self.path = MultiplexedPath(*list(namespace_path))
def resource_path(self, resource):
"""
Return the file system path to prevent
`resources.path()` from creating a temporary
copy.
"""
return str(self.path.joinpath(resource))
def files(self):
return self.path
+116
View File
@@ -0,0 +1,116 @@
"""
Interface adapters for low-level readers.
"""
import abc
import io
import itertools
from typing import BinaryIO, List
from .abc import Traversable, TraversableResources
class SimpleReader(abc.ABC):
"""
The minimum, low-level interface required from a resource
provider.
"""
@abc.abstractproperty
def package(self):
# type: () -> str
"""
The name of the package for which this reader loads resources.
"""
@abc.abstractmethod
def children(self):
# type: () -> List['SimpleReader']
"""
Obtain an iterable of SimpleReader for available
child containers (e.g. directories).
"""
@abc.abstractmethod
def resources(self):
# type: () -> List[str]
"""
Obtain available named resources for this virtual package.
"""
@abc.abstractmethod
def open_binary(self, resource):
# type: (str) -> BinaryIO
"""
Obtain a File-like for a named resource.
"""
@property
def name(self):
return self.package.split('.')[-1]
class ResourceHandle(Traversable):
"""
Handle to a named resource in a ResourceReader.
"""
def __init__(self, parent, name):
# type: (ResourceContainer, str) -> None
self.parent = parent
self.name = name # type: ignore
def is_file(self):
return True
def is_dir(self):
return False
def open(self, mode='r', *args, **kwargs):
stream = self.parent.reader.open_binary(self.name)
if 'b' not in mode:
stream = io.TextIOWrapper(*args, **kwargs)
return stream
def joinpath(self, name):
raise RuntimeError("Cannot traverse into a resource")
class ResourceContainer(Traversable):
"""
Traversable container for a package's resources via its reader.
"""
def __init__(self, reader):
# type: (SimpleReader) -> None
self.reader = reader
def is_dir(self):
return True
def is_file(self):
return False
def iterdir(self):
files = (ResourceHandle(self, name) for name in self.reader.resources)
dirs = map(ResourceContainer, self.reader.children())
return itertools.chain(files, dirs)
def open(self, *args, **kwargs):
raise IsADirectoryError()
def joinpath(self, name):
return next(
traversable for traversable in self.iterdir() if traversable.name == name
)
class TraversableReader(TraversableResources, SimpleReader):
"""
A TraversableResources based on SimpleReader. Resource providers
may derive from this class to provide the TraversableResources
interface by supplying the SimpleReader interface.
"""
def files(self):
return ResourceContainer(self)
-127
View File
@@ -1,127 +0,0 @@
# Copyright (c) 2009 Raymond Hettinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
from UserDict import DictMixin
class OrderedDict(dict, DictMixin):
def __init__(self, *args, **kwds):
if len(args) > 1:
raise TypeError('expected at most 1 arguments, got %d' % len(args))
try:
self.__end
except AttributeError:
self.clear()
self.update(*args, **kwds)
def clear(self):
self.__end = end = []
end += [None, end, end] # sentinel node for doubly linked list
self.__map = {} # key --> [key, prev, next]
dict.clear(self)
def __setitem__(self, key, value):
if key not in self:
end = self.__end
curr = end[1]
curr[2] = end[1] = self.__map[key] = [key, curr, end]
dict.__setitem__(self, key, value)
def __delitem__(self, key):
dict.__delitem__(self, key)
key, prev, next = self.__map.pop(key)
prev[2] = next
next[1] = prev
def __iter__(self):
end = self.__end
curr = end[2]
while curr is not end:
yield curr[0]
curr = curr[2]
def __reversed__(self):
end = self.__end
curr = end[1]
while curr is not end:
yield curr[0]
curr = curr[1]
def popitem(self, last=True):
if not self:
raise KeyError('dictionary is empty')
if last:
key = next(reversed(self))
else:
key = next(iter(self))
value = self.pop(key)
return key, value
def __reduce__(self):
items = [[k, self[k]] for k in self]
tmp = self.__map, self.__end
del self.__map, self.__end
inst_dict = vars(self).copy()
self.__map, self.__end = tmp
if inst_dict:
return (self.__class__, (items,), inst_dict)
return self.__class__, (items,)
def keys(self):
return list(self)
setdefault = DictMixin.setdefault
update = DictMixin.update
pop = DictMixin.pop
values = DictMixin.values
items = DictMixin.items
iterkeys = DictMixin.iterkeys
itervalues = DictMixin.itervalues
iteritems = DictMixin.iteritems
def __repr__(self):
if not self:
return '%s()' % (self.__class__.__name__,)
return '%s(%r)' % (self.__class__.__name__, list(self.items()))
def copy(self):
return self.__class__(self)
@classmethod
def fromkeys(cls, iterable, value=None):
d = cls()
for key in iterable:
d[key] = value
return d
def __eq__(self, other):
if isinstance(other, OrderedDict):
if len(self) != len(other):
return False
for p, q in zip(list(self.items()), list(other.items())):
if p != q:
return False
return True
return dict.__eq__(self, other)
def __ne__(self, other):
return not self == other
@@ -1,6 +0,0 @@
import setuptools
setuptools.setup(
name="my-test-package",
version="1.0",
zip_safe=True,
)
@@ -1,10 +0,0 @@
Metadata-Version: 1.0
Name: my-test-package
Version: 1.0
Summary: UNKNOWN
Home-page: UNKNOWN
Author: UNKNOWN
Author-email: UNKNOWN
License: UNKNOWN
Description: UNKNOWN
Platform: UNKNOWN
@@ -1,7 +0,0 @@
setup.cfg
setup.py
my_test_package.egg-info/PKG-INFO
my_test_package.egg-info/SOURCES.txt
my_test_package.egg-info/dependency_links.txt
my_test_package.egg-info/top_level.txt
my_test_package.egg-info/zip-safe
@@ -1,43 +0,0 @@
import py
import pytest
import pkg_resources
TESTS_DATA_DIR = py.path.local(__file__).dirpath('data')
class TestFindDistributions:
@pytest.fixture
def target_dir(self, tmpdir):
target_dir = tmpdir.mkdir('target')
# place a .egg named directory in the target that is not an egg:
target_dir.mkdir('not.an.egg')
return target_dir
def test_non_egg_dir_named_egg(self, target_dir):
dists = pkg_resources.find_distributions(str(target_dir))
assert not list(dists)
def test_standalone_egg_directory(self, target_dir):
(TESTS_DATA_DIR / 'my-test-package_unpacked-egg').copy(target_dir)
dists = pkg_resources.find_distributions(str(target_dir))
assert [dist.project_name for dist in dists] == ['my-test-package']
dists = pkg_resources.find_distributions(str(target_dir), only=True)
assert not list(dists)
def test_zipped_egg(self, target_dir):
(TESTS_DATA_DIR / 'my-test-package_zipped-egg').copy(target_dir)
dists = pkg_resources.find_distributions(str(target_dir))
assert [dist.project_name for dist in dists] == ['my-test-package']
dists = pkg_resources.find_distributions(str(target_dir), only=True)
assert not list(dists)
def test_zipped_sdist_one_level_removed(self, target_dir):
(TESTS_DATA_DIR / 'my-test-package-zip').copy(target_dir)
dists = pkg_resources.find_distributions(
str(target_dir / "my-test-package.zip"))
assert [dist.project_name for dist in dists] == ['my-test-package']
dists = pkg_resources.find_distributions(
str(target_dir / "my-test-package.zip"), only=True)
assert not list(dists)
-8
View File
@@ -1,8 +0,0 @@
import mock
from pkg_resources import evaluate_marker
@mock.patch('platform.python_version', return_value='2.7.10')
def test_ordering(python_version_mock):
assert evaluate_marker("python_full_version > '2.7.3'") is True
@@ -1,415 +0,0 @@
import sys
import tempfile
import os
import zipfile
import datetime
import time
import subprocess
import stat
import distutils.dist
import distutils.command.install_egg_info
try:
from unittest import mock
except ImportError:
import mock
from pkg_resources import (
DistInfoDistribution, Distribution, EggInfoDistribution,
)
import pytest
import pkg_resources
def timestamp(dt):
"""
Return a timestamp for a local, naive datetime instance.
"""
try:
return dt.timestamp()
except AttributeError:
# Python 3.2 and earlier
return time.mktime(dt.timetuple())
class EggRemover(str):
def __call__(self):
if self in sys.path:
sys.path.remove(self)
if os.path.exists(self):
os.remove(self)
class TestZipProvider:
finalizers = []
ref_time = datetime.datetime(2013, 5, 12, 13, 25, 0)
"A reference time for a file modification"
@classmethod
def setup_class(cls):
"create a zip egg and add it to sys.path"
egg = tempfile.NamedTemporaryFile(suffix='.egg', delete=False)
zip_egg = zipfile.ZipFile(egg, 'w')
zip_info = zipfile.ZipInfo()
zip_info.filename = 'mod.py'
zip_info.date_time = cls.ref_time.timetuple()
zip_egg.writestr(zip_info, 'x = 3\n')
zip_info = zipfile.ZipInfo()
zip_info.filename = 'data.dat'
zip_info.date_time = cls.ref_time.timetuple()
zip_egg.writestr(zip_info, 'hello, world!')
zip_info = zipfile.ZipInfo()
zip_info.filename = 'subdir/mod2.py'
zip_info.date_time = cls.ref_time.timetuple()
zip_egg.writestr(zip_info, 'x = 6\n')
zip_info = zipfile.ZipInfo()
zip_info.filename = 'subdir/data2.dat'
zip_info.date_time = cls.ref_time.timetuple()
zip_egg.writestr(zip_info, 'goodbye, world!')
zip_egg.close()
egg.close()
sys.path.append(egg.name)
subdir = os.path.join(egg.name, 'subdir')
sys.path.append(subdir)
cls.finalizers.append(EggRemover(subdir))
cls.finalizers.append(EggRemover(egg.name))
@classmethod
def teardown_class(cls):
for finalizer in cls.finalizers:
finalizer()
def test_resource_listdir(self):
import mod
zp = pkg_resources.ZipProvider(mod)
expected_root = ['data.dat', 'mod.py', 'subdir']
assert sorted(zp.resource_listdir('')) == expected_root
expected_subdir = ['data2.dat', 'mod2.py']
assert sorted(zp.resource_listdir('subdir')) == expected_subdir
assert sorted(zp.resource_listdir('subdir/')) == expected_subdir
assert zp.resource_listdir('nonexistent') == []
assert zp.resource_listdir('nonexistent/') == []
import mod2
zp2 = pkg_resources.ZipProvider(mod2)
assert sorted(zp2.resource_listdir('')) == expected_subdir
assert zp2.resource_listdir('subdir') == []
assert zp2.resource_listdir('subdir/') == []
def test_resource_filename_rewrites_on_change(self):
"""
If a previous call to get_resource_filename has saved the file, but
the file has been subsequently mutated with different file of the
same size and modification time, it should not be overwritten on a
subsequent call to get_resource_filename.
"""
import mod
manager = pkg_resources.ResourceManager()
zp = pkg_resources.ZipProvider(mod)
filename = zp.get_resource_filename(manager, 'data.dat')
actual = datetime.datetime.fromtimestamp(os.stat(filename).st_mtime)
assert actual == self.ref_time
f = open(filename, 'w')
f.write('hello, world?')
f.close()
ts = timestamp(self.ref_time)
os.utime(filename, (ts, ts))
filename = zp.get_resource_filename(manager, 'data.dat')
with open(filename) as f:
assert f.read() == 'hello, world!'
manager.cleanup_resources()
class TestResourceManager:
def test_get_cache_path(self):
mgr = pkg_resources.ResourceManager()
path = mgr.get_cache_path('foo')
type_ = str(type(path))
message = "Unexpected type from get_cache_path: " + type_
assert isinstance(path, str), message
def test_get_cache_path_race(self, tmpdir):
# Patch to os.path.isdir to create a race condition
def patched_isdir(dirname, unpatched_isdir=pkg_resources.isdir):
patched_isdir.dirnames.append(dirname)
was_dir = unpatched_isdir(dirname)
if not was_dir:
os.makedirs(dirname)
return was_dir
patched_isdir.dirnames = []
# Get a cache path with a "race condition"
mgr = pkg_resources.ResourceManager()
mgr.set_extraction_path(str(tmpdir))
archive_name = os.sep.join(('foo', 'bar', 'baz'))
with mock.patch.object(pkg_resources, 'isdir', new=patched_isdir):
mgr.get_cache_path(archive_name)
# Because this test relies on the implementation details of this
# function, these assertions are a sentinel to ensure that the
# test suite will not fail silently if the implementation changes.
called_dirnames = patched_isdir.dirnames
assert len(called_dirnames) == 2
assert called_dirnames[0].split(os.sep)[-2:] == ['foo', 'bar']
assert called_dirnames[1].split(os.sep)[-1:] == ['foo']
"""
Tests to ensure that pkg_resources runs independently from setuptools.
"""
def test_setuptools_not_imported(self):
"""
In a separate Python environment, import pkg_resources and assert
that action doesn't cause setuptools to be imported.
"""
lines = (
'import pkg_resources',
'import sys',
(
'assert "setuptools" not in sys.modules, '
'"setuptools was imported"'
),
)
cmd = [sys.executable, '-c', '; '.join(lines)]
subprocess.check_call(cmd)
def make_test_distribution(metadata_path, metadata):
"""
Make a test Distribution object, and return it.
:param metadata_path: the path to the metadata file that should be
created. This should be inside a distribution directory that should
also be created. For example, an argument value might end with
"<project>.dist-info/METADATA".
:param metadata: the desired contents of the metadata file, as bytes.
"""
dist_dir = os.path.dirname(metadata_path)
os.mkdir(dist_dir)
with open(metadata_path, 'wb') as f:
f.write(metadata)
dists = list(pkg_resources.distributions_from_metadata(dist_dir))
dist, = dists
return dist
def test_get_metadata__bad_utf8(tmpdir):
"""
Test a metadata file with bytes that can't be decoded as utf-8.
"""
filename = 'METADATA'
# Convert the tmpdir LocalPath object to a string before joining.
metadata_path = os.path.join(str(tmpdir), 'foo.dist-info', filename)
# Encode a non-ascii string with the wrong encoding (not utf-8).
metadata = 'née'.encode('iso-8859-1')
dist = make_test_distribution(metadata_path, metadata=metadata)
with pytest.raises(UnicodeDecodeError) as excinfo:
dist.get_metadata(filename)
exc = excinfo.value
actual = str(exc)
expected = (
# The error message starts with "'utf-8' codec ..." However, the
# spelling of "utf-8" can vary (e.g. "utf8") so we don't include it
"codec can't decode byte 0xe9 in position 1: "
'invalid continuation byte in METADATA file at path: '
)
assert expected in actual, 'actual: {}'.format(actual)
assert actual.endswith(metadata_path), 'actual: {}'.format(actual)
def make_distribution_no_version(tmpdir, basename):
"""
Create a distribution directory with no file containing the version.
"""
dist_dir = tmpdir / basename
dist_dir.ensure_dir()
# Make the directory non-empty so distributions_from_metadata()
# will detect it and yield it.
dist_dir.join('temp.txt').ensure()
if sys.version_info < (3, 6):
dist_dir = str(dist_dir)
dists = list(pkg_resources.distributions_from_metadata(dist_dir))
assert len(dists) == 1
dist, = dists
return dist, dist_dir
@pytest.mark.parametrize(
'suffix, expected_filename, expected_dist_type',
[
('egg-info', 'PKG-INFO', EggInfoDistribution),
('dist-info', 'METADATA', DistInfoDistribution),
],
)
def test_distribution_version_missing(
tmpdir, suffix, expected_filename, expected_dist_type):
"""
Test Distribution.version when the "Version" header is missing.
"""
basename = 'foo.{}'.format(suffix)
dist, dist_dir = make_distribution_no_version(tmpdir, basename)
expected_text = (
"Missing 'Version:' header and/or {} file at path: "
).format(expected_filename)
metadata_path = os.path.join(dist_dir, expected_filename)
# Now check the exception raised when the "version" attribute is accessed.
with pytest.raises(ValueError) as excinfo:
dist.version
err = str(excinfo.value)
# Include a string expression after the assert so the full strings
# will be visible for inspection on failure.
assert expected_text in err, str((expected_text, err))
# Also check the args passed to the ValueError.
msg, dist = excinfo.value.args
assert expected_text in msg
# Check that the message portion contains the path.
assert metadata_path in msg, str((metadata_path, msg))
assert type(dist) == expected_dist_type
def test_distribution_version_missing_undetected_path():
"""
Test Distribution.version when the "Version" header is missing and
the path can't be detected.
"""
# Create a Distribution object with no metadata argument, which results
# in an empty metadata provider.
dist = Distribution('/foo')
with pytest.raises(ValueError) as excinfo:
dist.version
msg, dist = excinfo.value.args
expected = (
"Missing 'Version:' header and/or PKG-INFO file at path: "
'[could not detect]'
)
assert msg == expected
@pytest.mark.parametrize('only', [False, True])
def test_dist_info_is_not_dir(tmp_path, only):
"""Test path containing a file with dist-info extension."""
dist_info = tmp_path / 'foobar.dist-info'
dist_info.touch()
assert not pkg_resources.dist_factory(str(tmp_path), str(dist_info), only)
class TestDeepVersionLookupDistutils:
@pytest.fixture
def env(self, tmpdir):
"""
Create a package environment, similar to a virtualenv,
in which packages are installed.
"""
class Environment(str):
pass
env = Environment(tmpdir)
tmpdir.chmod(stat.S_IRWXU)
subs = 'home', 'lib', 'scripts', 'data', 'egg-base'
env.paths = dict(
(dirname, str(tmpdir / dirname))
for dirname in subs
)
list(map(os.mkdir, env.paths.values()))
return env
def create_foo_pkg(self, env, version):
"""
Create a foo package installed (distutils-style) to env.paths['lib']
as version.
"""
ld = "This package has unicode metadata! ❄"
attrs = dict(name='foo', version=version, long_description=ld)
dist = distutils.dist.Distribution(attrs)
iei_cmd = distutils.command.install_egg_info.install_egg_info(dist)
iei_cmd.initialize_options()
iei_cmd.install_dir = env.paths['lib']
iei_cmd.finalize_options()
iei_cmd.run()
def test_version_resolved_from_egg_info(self, env):
version = '1.11.0.dev0+2329eae'
self.create_foo_pkg(env, version)
# this requirement parsing will raise a VersionConflict unless the
# .egg-info file is parsed (see #419 on BitBucket)
req = pkg_resources.Requirement.parse('foo>=1.9')
dist = pkg_resources.WorkingSet([env.paths['lib']]).find(req)
assert dist.version == version
@pytest.mark.parametrize(
'unnormalized, normalized',
[
('foo', 'foo'),
('foo/', 'foo'),
('foo/bar', 'foo/bar'),
('foo/bar/', 'foo/bar'),
],
)
def test_normalize_path_trailing_sep(self, unnormalized, normalized):
"""Ensure the trailing slash is cleaned for path comparison.
See pypa/setuptools#1519.
"""
result_from_unnormalized = pkg_resources.normalize_path(unnormalized)
result_from_normalized = pkg_resources.normalize_path(normalized)
assert result_from_unnormalized == result_from_normalized
@pytest.mark.skipif(
os.path.normcase('A') != os.path.normcase('a'),
reason='Testing case-insensitive filesystems.',
)
@pytest.mark.parametrize(
'unnormalized, normalized',
[
('MiXeD/CasE', 'mixed/case'),
],
)
def test_normalize_path_normcase(self, unnormalized, normalized):
"""Ensure mixed case is normalized on case-insensitive filesystems.
"""
result_from_unnormalized = pkg_resources.normalize_path(unnormalized)
result_from_normalized = pkg_resources.normalize_path(normalized)
assert result_from_unnormalized == result_from_normalized
@pytest.mark.skipif(
os.path.sep != '\\',
reason='Testing systems using backslashes as path separators.',
)
@pytest.mark.parametrize(
'unnormalized, expected',
[
('forward/slash', 'forward\\slash'),
('forward/slash/', 'forward\\slash'),
('backward\\slash\\', 'backward\\slash'),
],
)
def test_normalize_path_backslash_sep(self, unnormalized, expected):
"""Ensure path seps are cleaned on backslash path sep systems.
"""
result = pkg_resources.normalize_path(unnormalized)
assert result.endswith(expected)
-884
View File
@@ -1,884 +0,0 @@
import os
import sys
import string
import platform
import itertools
import pytest
from pkg_resources.extern import packaging
import pkg_resources
from pkg_resources import (
parse_requirements, VersionConflict, parse_version,
Distribution, EntryPoint, Requirement, safe_version, safe_name,
WorkingSet)
# from Python 3.6 docs.
def pairwise(iterable):
"s -> (s0,s1), (s1,s2), (s2, s3), ..."
a, b = itertools.tee(iterable)
next(b, None)
return zip(a, b)
class Metadata(pkg_resources.EmptyProvider):
"""Mock object to return metadata as if from an on-disk distribution"""
def __init__(self, *pairs):
self.metadata = dict(pairs)
def has_metadata(self, name):
return name in self.metadata
def get_metadata(self, name):
return self.metadata[name]
def get_metadata_lines(self, name):
return pkg_resources.yield_lines(self.get_metadata(name))
dist_from_fn = pkg_resources.Distribution.from_filename
class TestDistro:
def testCollection(self):
# empty path should produce no distributions
ad = pkg_resources.Environment([], platform=None, python=None)
assert list(ad) == []
assert ad['FooPkg'] == []
ad.add(dist_from_fn("FooPkg-1.3_1.egg"))
ad.add(dist_from_fn("FooPkg-1.4-py2.4-win32.egg"))
ad.add(dist_from_fn("FooPkg-1.2-py2.4.egg"))
# Name is in there now
assert ad['FooPkg']
# But only 1 package
assert list(ad) == ['foopkg']
# Distributions sort by version
expected = ['1.4', '1.3-1', '1.2']
assert [dist.version for dist in ad['FooPkg']] == expected
# Removing a distribution leaves sequence alone
ad.remove(ad['FooPkg'][1])
assert [dist.version for dist in ad['FooPkg']] == ['1.4', '1.2']
# And inserting adds them in order
ad.add(dist_from_fn("FooPkg-1.9.egg"))
assert [dist.version for dist in ad['FooPkg']] == ['1.9', '1.4', '1.2']
ws = WorkingSet([])
foo12 = dist_from_fn("FooPkg-1.2-py2.4.egg")
foo14 = dist_from_fn("FooPkg-1.4-py2.4-win32.egg")
req, = parse_requirements("FooPkg>=1.3")
# Nominal case: no distros on path, should yield all applicable
assert ad.best_match(req, ws).version == '1.9'
# If a matching distro is already installed, should return only that
ws.add(foo14)
assert ad.best_match(req, ws).version == '1.4'
# If the first matching distro is unsuitable, it's a version conflict
ws = WorkingSet([])
ws.add(foo12)
ws.add(foo14)
with pytest.raises(VersionConflict):
ad.best_match(req, ws)
# If more than one match on the path, the first one takes precedence
ws = WorkingSet([])
ws.add(foo14)
ws.add(foo12)
ws.add(foo14)
assert ad.best_match(req, ws).version == '1.4'
def checkFooPkg(self, d):
assert d.project_name == "FooPkg"
assert d.key == "foopkg"
assert d.version == "1.3.post1"
assert d.py_version == "2.4"
assert d.platform == "win32"
assert d.parsed_version == parse_version("1.3-1")
def testDistroBasics(self):
d = Distribution(
"/some/path",
project_name="FooPkg",
version="1.3-1",
py_version="2.4",
platform="win32",
)
self.checkFooPkg(d)
d = Distribution("/some/path")
assert d.py_version == '{}.{}'.format(*sys.version_info)
assert d.platform is None
def testDistroParse(self):
d = dist_from_fn("FooPkg-1.3.post1-py2.4-win32.egg")
self.checkFooPkg(d)
d = dist_from_fn("FooPkg-1.3.post1-py2.4-win32.egg-info")
self.checkFooPkg(d)
def testDistroMetadata(self):
d = Distribution(
"/some/path", project_name="FooPkg",
py_version="2.4", platform="win32",
metadata=Metadata(
('PKG-INFO', "Metadata-Version: 1.0\nVersion: 1.3-1\n")
),
)
self.checkFooPkg(d)
def distRequires(self, txt):
return Distribution("/foo", metadata=Metadata(('depends.txt', txt)))
def checkRequires(self, dist, txt, extras=()):
assert list(dist.requires(extras)) == list(parse_requirements(txt))
def testDistroDependsSimple(self):
for v in "Twisted>=1.5", "Twisted>=1.5\nZConfig>=2.0":
self.checkRequires(self.distRequires(v), v)
needs_object_dir = pytest.mark.skipif(
not hasattr(object, '__dir__'),
reason='object.__dir__ necessary for self.__dir__ implementation',
)
def test_distribution_dir(self):
d = pkg_resources.Distribution()
dir(d)
@needs_object_dir
def test_distribution_dir_includes_provider_dir(self):
d = pkg_resources.Distribution()
before = d.__dir__()
assert 'test_attr' not in before
d._provider.test_attr = None
after = d.__dir__()
assert len(after) == len(before) + 1
assert 'test_attr' in after
@needs_object_dir
def test_distribution_dir_ignores_provider_dir_leading_underscore(self):
d = pkg_resources.Distribution()
before = d.__dir__()
assert '_test_attr' not in before
d._provider._test_attr = None
after = d.__dir__()
assert len(after) == len(before)
assert '_test_attr' not in after
def testResolve(self):
ad = pkg_resources.Environment([])
ws = WorkingSet([])
# Resolving no requirements -> nothing to install
assert list(ws.resolve([], ad)) == []
# Request something not in the collection -> DistributionNotFound
with pytest.raises(pkg_resources.DistributionNotFound):
ws.resolve(parse_requirements("Foo"), ad)
Foo = Distribution.from_filename(
"/foo_dir/Foo-1.2.egg",
metadata=Metadata(('depends.txt', "[bar]\nBaz>=2.0"))
)
ad.add(Foo)
ad.add(Distribution.from_filename("Foo-0.9.egg"))
# Request thing(s) that are available -> list to activate
for i in range(3):
targets = list(ws.resolve(parse_requirements("Foo"), ad))
assert targets == [Foo]
list(map(ws.add, targets))
with pytest.raises(VersionConflict):
ws.resolve(parse_requirements("Foo==0.9"), ad)
ws = WorkingSet([]) # reset
# Request an extra that causes an unresolved dependency for "Baz"
with pytest.raises(pkg_resources.DistributionNotFound):
ws.resolve(parse_requirements("Foo[bar]"), ad)
Baz = Distribution.from_filename(
"/foo_dir/Baz-2.1.egg", metadata=Metadata(('depends.txt', "Foo"))
)
ad.add(Baz)
# Activation list now includes resolved dependency
assert (
list(ws.resolve(parse_requirements("Foo[bar]"), ad))
== [Foo, Baz]
)
# Requests for conflicting versions produce VersionConflict
with pytest.raises(VersionConflict) as vc:
ws.resolve(parse_requirements("Foo==1.2\nFoo!=1.2"), ad)
msg = 'Foo 0.9 is installed but Foo==1.2 is required'
assert vc.value.report() == msg
def test_environment_marker_evaluation_negative(self):
"""Environment markers are evaluated at resolution time."""
ad = pkg_resources.Environment([])
ws = WorkingSet([])
res = ws.resolve(parse_requirements("Foo;python_version<'2'"), ad)
assert list(res) == []
def test_environment_marker_evaluation_positive(self):
ad = pkg_resources.Environment([])
ws = WorkingSet([])
Foo = Distribution.from_filename("/foo_dir/Foo-1.2.dist-info")
ad.add(Foo)
res = ws.resolve(parse_requirements("Foo;python_version>='2'"), ad)
assert list(res) == [Foo]
def test_environment_marker_evaluation_called(self):
"""
If one package foo requires bar without any extras,
markers should pass for bar without extras.
"""
parent_req, = parse_requirements("foo")
req, = parse_requirements("bar;python_version>='2'")
req_extras = pkg_resources._ReqExtras({req: parent_req.extras})
assert req_extras.markers_pass(req)
parent_req, = parse_requirements("foo[]")
req, = parse_requirements("bar;python_version>='2'")
req_extras = pkg_resources._ReqExtras({req: parent_req.extras})
assert req_extras.markers_pass(req)
def test_marker_evaluation_with_extras(self):
"""Extras are also evaluated as markers at resolution time."""
ad = pkg_resources.Environment([])
ws = WorkingSet([])
Foo = Distribution.from_filename(
"/foo_dir/Foo-1.2.dist-info",
metadata=Metadata(("METADATA", "Provides-Extra: baz\n"
"Requires-Dist: quux; extra=='baz'"))
)
ad.add(Foo)
assert list(ws.resolve(parse_requirements("Foo"), ad)) == [Foo]
quux = Distribution.from_filename("/foo_dir/quux-1.0.dist-info")
ad.add(quux)
res = list(ws.resolve(parse_requirements("Foo[baz]"), ad))
assert res == [Foo, quux]
def test_marker_evaluation_with_extras_normlized(self):
"""Extras are also evaluated as markers at resolution time."""
ad = pkg_resources.Environment([])
ws = WorkingSet([])
Foo = Distribution.from_filename(
"/foo_dir/Foo-1.2.dist-info",
metadata=Metadata(("METADATA", "Provides-Extra: baz-lightyear\n"
"Requires-Dist: quux; extra=='baz-lightyear'"))
)
ad.add(Foo)
assert list(ws.resolve(parse_requirements("Foo"), ad)) == [Foo]
quux = Distribution.from_filename("/foo_dir/quux-1.0.dist-info")
ad.add(quux)
res = list(ws.resolve(parse_requirements("Foo[baz-lightyear]"), ad))
assert res == [Foo, quux]
def test_marker_evaluation_with_multiple_extras(self):
ad = pkg_resources.Environment([])
ws = WorkingSet([])
Foo = Distribution.from_filename(
"/foo_dir/Foo-1.2.dist-info",
metadata=Metadata(("METADATA", "Provides-Extra: baz\n"
"Requires-Dist: quux; extra=='baz'\n"
"Provides-Extra: bar\n"
"Requires-Dist: fred; extra=='bar'\n"))
)
ad.add(Foo)
quux = Distribution.from_filename("/foo_dir/quux-1.0.dist-info")
ad.add(quux)
fred = Distribution.from_filename("/foo_dir/fred-0.1.dist-info")
ad.add(fred)
res = list(ws.resolve(parse_requirements("Foo[baz,bar]"), ad))
assert sorted(res) == [fred, quux, Foo]
def test_marker_evaluation_with_extras_loop(self):
ad = pkg_resources.Environment([])
ws = WorkingSet([])
a = Distribution.from_filename(
"/foo_dir/a-0.2.dist-info",
metadata=Metadata(("METADATA", "Requires-Dist: c[a]"))
)
b = Distribution.from_filename(
"/foo_dir/b-0.3.dist-info",
metadata=Metadata(("METADATA", "Requires-Dist: c[b]"))
)
c = Distribution.from_filename(
"/foo_dir/c-1.0.dist-info",
metadata=Metadata(("METADATA", "Provides-Extra: a\n"
"Requires-Dist: b;extra=='a'\n"
"Provides-Extra: b\n"
"Requires-Dist: foo;extra=='b'"))
)
foo = Distribution.from_filename("/foo_dir/foo-0.1.dist-info")
for dist in (a, b, c, foo):
ad.add(dist)
res = list(ws.resolve(parse_requirements("a"), ad))
assert res == [a, c, b, foo]
def testDistroDependsOptions(self):
d = self.distRequires("""
Twisted>=1.5
[docgen]
ZConfig>=2.0
docutils>=0.3
[fastcgi]
fcgiapp>=0.1""")
self.checkRequires(d, "Twisted>=1.5")
self.checkRequires(
d, "Twisted>=1.5 ZConfig>=2.0 docutils>=0.3".split(), ["docgen"]
)
self.checkRequires(
d, "Twisted>=1.5 fcgiapp>=0.1".split(), ["fastcgi"]
)
self.checkRequires(
d, "Twisted>=1.5 ZConfig>=2.0 docutils>=0.3 fcgiapp>=0.1".split(),
["docgen", "fastcgi"]
)
self.checkRequires(
d, "Twisted>=1.5 fcgiapp>=0.1 ZConfig>=2.0 docutils>=0.3".split(),
["fastcgi", "docgen"]
)
with pytest.raises(pkg_resources.UnknownExtra):
d.requires(["foo"])
class TestWorkingSet:
def test_find_conflicting(self):
ws = WorkingSet([])
Foo = Distribution.from_filename("/foo_dir/Foo-1.2.egg")
ws.add(Foo)
# create a requirement that conflicts with Foo 1.2
req = next(parse_requirements("Foo<1.2"))
with pytest.raises(VersionConflict) as vc:
ws.find(req)
msg = 'Foo 1.2 is installed but Foo<1.2 is required'
assert vc.value.report() == msg
def test_resolve_conflicts_with_prior(self):
"""
A ContextualVersionConflict should be raised when a requirement
conflicts with a prior requirement for a different package.
"""
# Create installation where Foo depends on Baz 1.0 and Bar depends on
# Baz 2.0.
ws = WorkingSet([])
md = Metadata(('depends.txt', "Baz==1.0"))
Foo = Distribution.from_filename("/foo_dir/Foo-1.0.egg", metadata=md)
ws.add(Foo)
md = Metadata(('depends.txt', "Baz==2.0"))
Bar = Distribution.from_filename("/foo_dir/Bar-1.0.egg", metadata=md)
ws.add(Bar)
Baz = Distribution.from_filename("/foo_dir/Baz-1.0.egg")
ws.add(Baz)
Baz = Distribution.from_filename("/foo_dir/Baz-2.0.egg")
ws.add(Baz)
with pytest.raises(VersionConflict) as vc:
ws.resolve(parse_requirements("Foo\nBar\n"))
msg = "Baz 1.0 is installed but Baz==2.0 is required by "
msg += repr(set(['Bar']))
assert vc.value.report() == msg
class TestEntryPoints:
def assertfields(self, ep):
assert ep.name == "foo"
assert ep.module_name == "pkg_resources.tests.test_resources"
assert ep.attrs == ("TestEntryPoints",)
assert ep.extras == ("x",)
assert ep.load() is TestEntryPoints
expect = "foo = pkg_resources.tests.test_resources:TestEntryPoints [x]"
assert str(ep) == expect
def setup_method(self, method):
self.dist = Distribution.from_filename(
"FooPkg-1.2-py2.4.egg", metadata=Metadata(('requires.txt', '[x]')))
def testBasics(self):
ep = EntryPoint(
"foo", "pkg_resources.tests.test_resources", ["TestEntryPoints"],
["x"], self.dist
)
self.assertfields(ep)
def testParse(self):
s = "foo = pkg_resources.tests.test_resources:TestEntryPoints [x]"
ep = EntryPoint.parse(s, self.dist)
self.assertfields(ep)
ep = EntryPoint.parse("bar baz= spammity[PING]")
assert ep.name == "bar baz"
assert ep.module_name == "spammity"
assert ep.attrs == ()
assert ep.extras == ("ping",)
ep = EntryPoint.parse(" fizzly = wocka:foo")
assert ep.name == "fizzly"
assert ep.module_name == "wocka"
assert ep.attrs == ("foo",)
assert ep.extras == ()
# plus in the name
spec = "html+mako = mako.ext.pygmentplugin:MakoHtmlLexer"
ep = EntryPoint.parse(spec)
assert ep.name == 'html+mako'
reject_specs = "foo", "x=a:b:c", "q=x/na", "fez=pish:tush-z", "x=f[a]>2"
@pytest.mark.parametrize("reject_spec", reject_specs)
def test_reject_spec(self, reject_spec):
with pytest.raises(ValueError):
EntryPoint.parse(reject_spec)
def test_printable_name(self):
"""
Allow any printable character in the name.
"""
# Create a name with all printable characters; strip the whitespace.
name = string.printable.strip()
spec = "{name} = module:attr".format(**locals())
ep = EntryPoint.parse(spec)
assert ep.name == name
def checkSubMap(self, m):
assert len(m) == len(self.submap_expect)
for key, ep in self.submap_expect.items():
assert m.get(key).name == ep.name
assert m.get(key).module_name == ep.module_name
assert sorted(m.get(key).attrs) == sorted(ep.attrs)
assert sorted(m.get(key).extras) == sorted(ep.extras)
submap_expect = dict(
feature1=EntryPoint('feature1', 'somemodule', ['somefunction']),
feature2=EntryPoint(
'feature2', 'another.module', ['SomeClass'], ['extra1', 'extra2']),
feature3=EntryPoint('feature3', 'this.module', extras=['something'])
)
submap_str = """
# define features for blah blah
feature1 = somemodule:somefunction
feature2 = another.module:SomeClass [extra1,extra2]
feature3 = this.module [something]
"""
def testParseList(self):
self.checkSubMap(EntryPoint.parse_group("xyz", self.submap_str))
with pytest.raises(ValueError):
EntryPoint.parse_group("x a", "foo=bar")
with pytest.raises(ValueError):
EntryPoint.parse_group("x", ["foo=baz", "foo=bar"])
def testParseMap(self):
m = EntryPoint.parse_map({'xyz': self.submap_str})
self.checkSubMap(m['xyz'])
assert list(m.keys()) == ['xyz']
m = EntryPoint.parse_map("[xyz]\n" + self.submap_str)
self.checkSubMap(m['xyz'])
assert list(m.keys()) == ['xyz']
with pytest.raises(ValueError):
EntryPoint.parse_map(["[xyz]", "[xyz]"])
with pytest.raises(ValueError):
EntryPoint.parse_map(self.submap_str)
def testDeprecationWarnings(self):
ep = EntryPoint(
"foo", "pkg_resources.tests.test_resources", ["TestEntryPoints"],
["x"]
)
with pytest.warns(pkg_resources.PkgResourcesDeprecationWarning):
ep.load(require=False)
class TestRequirements:
def testBasics(self):
r = Requirement.parse("Twisted>=1.2")
assert str(r) == "Twisted>=1.2"
assert repr(r) == "Requirement.parse('Twisted>=1.2')"
assert r == Requirement("Twisted>=1.2")
assert r == Requirement("twisTed>=1.2")
assert r != Requirement("Twisted>=2.0")
assert r != Requirement("Zope>=1.2")
assert r != Requirement("Zope>=3.0")
assert r != Requirement("Twisted[extras]>=1.2")
def testOrdering(self):
r1 = Requirement("Twisted==1.2c1,>=1.2")
r2 = Requirement("Twisted>=1.2,==1.2c1")
assert r1 == r2
assert str(r1) == str(r2)
assert str(r2) == "Twisted==1.2c1,>=1.2"
assert (
Requirement("Twisted")
!=
Requirement("Twisted @ https://localhost/twisted.zip")
)
def testBasicContains(self):
r = Requirement("Twisted>=1.2")
foo_dist = Distribution.from_filename("FooPkg-1.3_1.egg")
twist11 = Distribution.from_filename("Twisted-1.1.egg")
twist12 = Distribution.from_filename("Twisted-1.2.egg")
assert parse_version('1.2') in r
assert parse_version('1.1') not in r
assert '1.2' in r
assert '1.1' not in r
assert foo_dist not in r
assert twist11 not in r
assert twist12 in r
def testOptionsAndHashing(self):
r1 = Requirement.parse("Twisted[foo,bar]>=1.2")
r2 = Requirement.parse("Twisted[bar,FOO]>=1.2")
assert r1 == r2
assert set(r1.extras) == set(("foo", "bar"))
assert set(r2.extras) == set(("foo", "bar"))
assert hash(r1) == hash(r2)
assert (
hash(r1)
==
hash((
"twisted",
None,
packaging.specifiers.SpecifierSet(">=1.2"),
frozenset(["foo", "bar"]),
None
))
)
assert (
hash(Requirement.parse("Twisted @ https://localhost/twisted.zip"))
==
hash((
"twisted",
"https://localhost/twisted.zip",
packaging.specifiers.SpecifierSet(),
frozenset(),
None
))
)
def testVersionEquality(self):
r1 = Requirement.parse("foo==0.3a2")
r2 = Requirement.parse("foo!=0.3a4")
d = Distribution.from_filename
assert d("foo-0.3a4.egg") not in r1
assert d("foo-0.3a1.egg") not in r1
assert d("foo-0.3a4.egg") not in r2
assert d("foo-0.3a2.egg") in r1
assert d("foo-0.3a2.egg") in r2
assert d("foo-0.3a3.egg") in r2
assert d("foo-0.3a5.egg") in r2
def testSetuptoolsProjectName(self):
"""
The setuptools project should implement the setuptools package.
"""
assert (
Requirement.parse('setuptools').project_name == 'setuptools')
# setuptools 0.7 and higher means setuptools.
assert (
Requirement.parse('setuptools == 0.7').project_name
== 'setuptools'
)
assert (
Requirement.parse('setuptools == 0.7a1').project_name
== 'setuptools'
)
assert (
Requirement.parse('setuptools >= 0.7').project_name
== 'setuptools'
)
class TestParsing:
def testEmptyParse(self):
assert list(parse_requirements('')) == []
def testYielding(self):
for inp, out in [
([], []), ('x', ['x']), ([[]], []), (' x\n y', ['x', 'y']),
(['x\n\n', 'y'], ['x', 'y']),
]:
assert list(pkg_resources.yield_lines(inp)) == out
def testSplitting(self):
sample = """
x
[Y]
z
a
[b ]
# foo
c
[ d]
[q]
v
"""
assert (
list(pkg_resources.split_sections(sample))
==
[
(None, ["x"]),
("Y", ["z", "a"]),
("b", ["c"]),
("d", []),
("q", ["v"]),
]
)
with pytest.raises(ValueError):
list(pkg_resources.split_sections("[foo"))
def testSafeName(self):
assert safe_name("adns-python") == "adns-python"
assert safe_name("WSGI Utils") == "WSGI-Utils"
assert safe_name("WSGI Utils") == "WSGI-Utils"
assert safe_name("Money$$$Maker") == "Money-Maker"
assert safe_name("peak.web") != "peak-web"
def testSafeVersion(self):
assert safe_version("1.2-1") == "1.2.post1"
assert safe_version("1.2 alpha") == "1.2.alpha"
assert safe_version("2.3.4 20050521") == "2.3.4.20050521"
assert safe_version("Money$$$Maker") == "Money-Maker"
assert safe_version("peak.web") == "peak.web"
def testSimpleRequirements(self):
assert (
list(parse_requirements('Twis-Ted>=1.2-1'))
==
[Requirement('Twis-Ted>=1.2-1')]
)
assert (
list(parse_requirements('Twisted >=1.2, \\ # more\n<2.0'))
==
[Requirement('Twisted>=1.2,<2.0')]
)
assert (
Requirement.parse("FooBar==1.99a3")
==
Requirement("FooBar==1.99a3")
)
with pytest.raises(ValueError):
Requirement.parse(">=2.3")
with pytest.raises(ValueError):
Requirement.parse("x\\")
with pytest.raises(ValueError):
Requirement.parse("x==2 q")
with pytest.raises(ValueError):
Requirement.parse("X==1\nY==2")
with pytest.raises(ValueError):
Requirement.parse("#")
def test_requirements_with_markers(self):
assert (
Requirement.parse("foobar;os_name=='a'")
==
Requirement.parse("foobar;os_name=='a'")
)
assert (
Requirement.parse("name==1.1;python_version=='2.7'")
!=
Requirement.parse("name==1.1;python_version=='3.6'")
)
assert (
Requirement.parse("name==1.0;python_version=='2.7'")
!=
Requirement.parse("name==1.2;python_version=='2.7'")
)
assert (
Requirement.parse("name[foo]==1.0;python_version=='3.6'")
!=
Requirement.parse("name[foo,bar]==1.0;python_version=='3.6'")
)
def test_local_version(self):
req, = parse_requirements('foo==1.0+org1')
def test_spaces_between_multiple_versions(self):
req, = parse_requirements('foo>=1.0, <3')
req, = parse_requirements('foo >= 1.0, < 3')
@pytest.mark.parametrize(
['lower', 'upper'],
[
('1.2-rc1', '1.2rc1'),
('0.4', '0.4.0'),
('0.4.0.0', '0.4.0'),
('0.4.0-0', '0.4-0'),
('0post1', '0.0post1'),
('0pre1', '0.0c1'),
('0.0.0preview1', '0c1'),
('0.0c1', '0-rc1'),
('1.2a1', '1.2.a.1'),
('1.2.a', '1.2a'),
],
)
def testVersionEquality(self, lower, upper):
assert parse_version(lower) == parse_version(upper)
torture = """
0.80.1-3 0.80.1-2 0.80.1-1 0.79.9999+0.80.0pre4-1
0.79.9999+0.80.0pre2-3 0.79.9999+0.80.0pre2-2
0.77.2-1 0.77.1-1 0.77.0-1
"""
@pytest.mark.parametrize(
['lower', 'upper'],
[
('2.1', '2.1.1'),
('2a1', '2b0'),
('2a1', '2.1'),
('2.3a1', '2.3'),
('2.1-1', '2.1-2'),
('2.1-1', '2.1.1'),
('2.1', '2.1post4'),
('2.1a0-20040501', '2.1'),
('1.1', '02.1'),
('3.2', '3.2.post0'),
('3.2post1', '3.2post2'),
('0.4', '4.0'),
('0.0.4', '0.4.0'),
('0post1', '0.4post1'),
('2.1.0-rc1', '2.1.0'),
('2.1dev', '2.1a0'),
] + list(pairwise(reversed(torture.split()))),
)
def testVersionOrdering(self, lower, upper):
assert parse_version(lower) < parse_version(upper)
def testVersionHashable(self):
"""
Ensure that our versions stay hashable even though we've subclassed
them and added some shim code to them.
"""
assert (
hash(parse_version("1.0"))
==
hash(parse_version("1.0"))
)
class TestNamespaces:
ns_str = "__import__('pkg_resources').declare_namespace(__name__)\n"
@pytest.fixture
def symlinked_tmpdir(self, tmpdir):
"""
Where available, return the tempdir as a symlink,
which as revealed in #231 is more fragile than
a natural tempdir.
"""
if not hasattr(os, 'symlink'):
yield str(tmpdir)
return
link_name = str(tmpdir) + '-linked'
os.symlink(str(tmpdir), link_name)
try:
yield type(tmpdir)(link_name)
finally:
os.unlink(link_name)
@pytest.fixture(autouse=True)
def patched_path(self, tmpdir):
"""
Patch sys.path to include the 'site-pkgs' dir. Also
restore pkg_resources._namespace_packages to its
former state.
"""
saved_ns_pkgs = pkg_resources._namespace_packages.copy()
saved_sys_path = sys.path[:]
site_pkgs = tmpdir.mkdir('site-pkgs')
sys.path.append(str(site_pkgs))
try:
yield
finally:
pkg_resources._namespace_packages = saved_ns_pkgs
sys.path = saved_sys_path
issue591 = pytest.mark.xfail(platform.system() == 'Windows', reason="#591")
@issue591
def test_two_levels_deep(self, symlinked_tmpdir):
"""
Test nested namespace packages
Create namespace packages in the following tree :
site-packages-1/pkg1/pkg2
site-packages-2/pkg1/pkg2
Check both are in the _namespace_packages dict and that their __path__
is correct
"""
real_tmpdir = symlinked_tmpdir.realpath()
tmpdir = symlinked_tmpdir
sys.path.append(str(tmpdir / 'site-pkgs2'))
site_dirs = tmpdir / 'site-pkgs', tmpdir / 'site-pkgs2'
for site in site_dirs:
pkg1 = site / 'pkg1'
pkg2 = pkg1 / 'pkg2'
pkg2.ensure_dir()
(pkg1 / '__init__.py').write_text(self.ns_str, encoding='utf-8')
(pkg2 / '__init__.py').write_text(self.ns_str, encoding='utf-8')
import pkg1
assert "pkg1" in pkg_resources._namespace_packages
# attempt to import pkg2 from site-pkgs2
import pkg1.pkg2
# check the _namespace_packages dict
assert "pkg1.pkg2" in pkg_resources._namespace_packages
assert pkg_resources._namespace_packages["pkg1"] == ["pkg1.pkg2"]
# check the __path__ attribute contains both paths
expected = [
str(real_tmpdir / "site-pkgs" / "pkg1" / "pkg2"),
str(real_tmpdir / "site-pkgs2" / "pkg1" / "pkg2"),
]
assert pkg1.pkg2.__path__ == expected
@issue591
def test_path_order(self, symlinked_tmpdir):
"""
Test that if multiple versions of the same namespace package subpackage
are on different sys.path entries, that only the one earliest on
sys.path is imported, and that the namespace package's __path__ is in
the correct order.
Regression test for https://github.com/pypa/setuptools/issues/207
"""
tmpdir = symlinked_tmpdir
site_dirs = (
tmpdir / "site-pkgs",
tmpdir / "site-pkgs2",
tmpdir / "site-pkgs3",
)
vers_str = "__version__ = %r"
for number, site in enumerate(site_dirs, 1):
if number > 1:
sys.path.append(str(site))
nspkg = site / 'nspkg'
subpkg = nspkg / 'subpkg'
subpkg.ensure_dir()
(nspkg / '__init__.py').write_text(self.ns_str, encoding='utf-8')
(subpkg / '__init__.py').write_text(
vers_str % number, encoding='utf-8')
import nspkg.subpkg
import nspkg
expected = [
str(site.realpath() / 'nspkg')
for site in site_dirs
]
assert nspkg.__path__ == expected
assert nspkg.subpkg.__version__ == 1
-482
View File
@@ -1,482 +0,0 @@
import inspect
import re
import textwrap
import functools
import pytest
import pkg_resources
from .test_resources import Metadata
def strip_comments(s):
return '\n'.join(
line for line in s.split('\n')
if line.strip() and not line.strip().startswith('#')
)
def parse_distributions(s):
'''
Parse a series of distribution specs of the form:
{project_name}-{version}
[optional, indented requirements specification]
Example:
foo-0.2
bar-1.0
foo>=3.0
[feature]
baz
yield 2 distributions:
- project_name=foo, version=0.2
- project_name=bar, version=1.0,
requires=['foo>=3.0', 'baz; extra=="feature"']
'''
s = s.strip()
for spec in re.split(r'\n(?=[^\s])', s):
if not spec:
continue
fields = spec.split('\n', 1)
assert 1 <= len(fields) <= 2
name, version = fields.pop(0).split('-')
if fields:
requires = textwrap.dedent(fields.pop(0))
metadata = Metadata(('requires.txt', requires))
else:
metadata = None
dist = pkg_resources.Distribution(project_name=name,
version=version,
metadata=metadata)
yield dist
class FakeInstaller:
def __init__(self, installable_dists):
self._installable_dists = installable_dists
def __call__(self, req):
return next(iter(filter(lambda dist: dist in req,
self._installable_dists)), None)
def parametrize_test_working_set_resolve(*test_list):
idlist = []
argvalues = []
for test in test_list:
(
name,
installed_dists,
installable_dists,
requirements,
expected1, expected2
) = [
strip_comments(s.lstrip()) for s in
textwrap.dedent(test).lstrip().split('\n\n', 5)
]
installed_dists = list(parse_distributions(installed_dists))
installable_dists = list(parse_distributions(installable_dists))
requirements = list(pkg_resources.parse_requirements(requirements))
for id_, replace_conflicting, expected in (
(name, False, expected1),
(name + '_replace_conflicting', True, expected2),
):
idlist.append(id_)
expected = strip_comments(expected.strip())
if re.match(r'\w+$', expected):
expected = getattr(pkg_resources, expected)
assert issubclass(expected, Exception)
else:
expected = list(parse_distributions(expected))
argvalues.append(pytest.param(installed_dists, installable_dists,
requirements, replace_conflicting,
expected))
return pytest.mark.parametrize('installed_dists,installable_dists,'
'requirements,replace_conflicting,'
'resolved_dists_or_exception',
argvalues, ids=idlist)
@parametrize_test_working_set_resolve(
'''
# id
noop
# installed
# installable
# wanted
# resolved
# resolved [replace conflicting]
''',
'''
# id
already_installed
# installed
foo-3.0
# installable
# wanted
foo>=2.1,!=3.1,<4
# resolved
foo-3.0
# resolved [replace conflicting]
foo-3.0
''',
'''
# id
installable_not_installed
# installed
# installable
foo-3.0
foo-4.0
# wanted
foo>=2.1,!=3.1,<4
# resolved
foo-3.0
# resolved [replace conflicting]
foo-3.0
''',
'''
# id
not_installable
# installed
# installable
# wanted
foo>=2.1,!=3.1,<4
# resolved
DistributionNotFound
# resolved [replace conflicting]
DistributionNotFound
''',
'''
# id
no_matching_version
# installed
# installable
foo-3.1
# wanted
foo>=2.1,!=3.1,<4
# resolved
DistributionNotFound
# resolved [replace conflicting]
DistributionNotFound
''',
'''
# id
installable_with_installed_conflict
# installed
foo-3.1
# installable
foo-3.5
# wanted
foo>=2.1,!=3.1,<4
# resolved
VersionConflict
# resolved [replace conflicting]
foo-3.5
''',
'''
# id
not_installable_with_installed_conflict
# installed
foo-3.1
# installable
# wanted
foo>=2.1,!=3.1,<4
# resolved
VersionConflict
# resolved [replace conflicting]
DistributionNotFound
''',
'''
# id
installed_with_installed_require
# installed
foo-3.9
baz-0.1
foo>=2.1,!=3.1,<4
# installable
# wanted
baz
# resolved
foo-3.9
baz-0.1
# resolved [replace conflicting]
foo-3.9
baz-0.1
''',
'''
# id
installed_with_conflicting_installed_require
# installed
foo-5
baz-0.1
foo>=2.1,!=3.1,<4
# installable
# wanted
baz
# resolved
VersionConflict
# resolved [replace conflicting]
DistributionNotFound
''',
'''
# id
installed_with_installable_conflicting_require
# installed
foo-5
baz-0.1
foo>=2.1,!=3.1,<4
# installable
foo-2.9
# wanted
baz
# resolved
VersionConflict
# resolved [replace conflicting]
baz-0.1
foo-2.9
''',
'''
# id
installed_with_installable_require
# installed
baz-0.1
foo>=2.1,!=3.1,<4
# installable
foo-3.9
# wanted
baz
# resolved
foo-3.9
baz-0.1
# resolved [replace conflicting]
foo-3.9
baz-0.1
''',
'''
# id
installable_with_installed_require
# installed
foo-3.9
# installable
baz-0.1
foo>=2.1,!=3.1,<4
# wanted
baz
# resolved
foo-3.9
baz-0.1
# resolved [replace conflicting]
foo-3.9
baz-0.1
''',
'''
# id
installable_with_installable_require
# installed
# installable
foo-3.9
baz-0.1
foo>=2.1,!=3.1,<4
# wanted
baz
# resolved
foo-3.9
baz-0.1
# resolved [replace conflicting]
foo-3.9
baz-0.1
''',
'''
# id
installable_with_conflicting_installable_require
# installed
foo-5
# installable
foo-2.9
baz-0.1
foo>=2.1,!=3.1,<4
# wanted
baz
# resolved
VersionConflict
# resolved [replace conflicting]
baz-0.1
foo-2.9
''',
'''
# id
conflicting_installables
# installed
# installable
foo-2.9
foo-5.0
# wanted
foo>=2.1,!=3.1,<4
foo>=4
# resolved
VersionConflict
# resolved [replace conflicting]
VersionConflict
''',
'''
# id
installables_with_conflicting_requires
# installed
# installable
foo-2.9
dep==1.0
baz-5.0
dep==2.0
dep-1.0
dep-2.0
# wanted
foo
baz
# resolved
VersionConflict
# resolved [replace conflicting]
VersionConflict
''',
'''
# id
installables_with_conflicting_nested_requires
# installed
# installable
foo-2.9
dep1
dep1-1.0
subdep<1.0
baz-5.0
dep2
dep2-1.0
subdep>1.0
subdep-0.9
subdep-1.1
# wanted
foo
baz
# resolved
VersionConflict
# resolved [replace conflicting]
VersionConflict
''',
)
def test_working_set_resolve(installed_dists, installable_dists, requirements,
replace_conflicting, resolved_dists_or_exception):
ws = pkg_resources.WorkingSet([])
list(map(ws.add, installed_dists))
resolve_call = functools.partial(
ws.resolve,
requirements, installer=FakeInstaller(installable_dists),
replace_conflicting=replace_conflicting,
)
if inspect.isclass(resolved_dists_or_exception):
with pytest.raises(resolved_dists_or_exception):
resolve_call()
else:
assert sorted(resolve_call()) == sorted(resolved_dists_or_exception)
-149
View File
@@ -1,149 +0,0 @@
import time
import random
import datetime
from unittest import mock
import pytest
import pytz
import freezegun
from tempora import schedule
do_nothing = type(None)
def test_delayed_command_order():
"""
delayed commands should be sorted by delay time
"""
delays = [random.randint(0, 99) for x in range(5)]
cmds = sorted(
[schedule.DelayedCommand.after(delay, do_nothing) for delay in delays]
)
assert [c.delay.seconds for c in cmds] == sorted(delays)
def test_periodic_command_delay():
"A PeriodicCommand must have a positive, non-zero delay."
with pytest.raises(ValueError) as exc_info:
schedule.PeriodicCommand.after(0, None)
assert str(exc_info.value) == test_periodic_command_delay.__doc__
def test_periodic_command_fixed_delay():
"""
Test that we can construct a periodic command with a fixed initial
delay.
"""
fd = schedule.PeriodicCommandFixedDelay.at_time(
at=schedule.now(), delay=datetime.timedelta(seconds=2), target=lambda: None
)
assert fd.due() is True
assert fd.next().due() is False
class TestCommands:
def test_delayed_command_from_timestamp(self):
"""
Ensure a delayed command can be constructed from a timestamp.
"""
t = time.time()
schedule.DelayedCommand.at_time(t, do_nothing)
def test_command_at_noon(self):
"""
Create a periodic command that's run at noon every day.
"""
when = datetime.time(12, 0, tzinfo=pytz.utc)
cmd = schedule.PeriodicCommandFixedDelay.daily_at(when, target=None)
assert cmd.due() is False
next_cmd = cmd.next()
daily = datetime.timedelta(days=1)
day_from_now = schedule.now() + daily
two_days_from_now = day_from_now + daily
assert day_from_now < next_cmd < two_days_from_now
@pytest.mark.parametrize("hour", range(10, 14))
@pytest.mark.parametrize("tz_offset", (14, -14))
def test_command_at_noon_distant_local(self, hour, tz_offset):
"""
Run test_command_at_noon, but with the local timezone
more than 12 hours away from UTC.
"""
with freezegun.freeze_time(f"2020-01-10 {hour:02}:01", tz_offset=tz_offset):
self.test_command_at_noon()
class TestTimezones:
def test_alternate_timezone_west(self):
target_tz = pytz.timezone('US/Pacific')
target = schedule.now().astimezone(target_tz)
cmd = schedule.DelayedCommand.at_time(target, target=None)
assert cmd.due()
def test_alternate_timezone_east(self):
target_tz = pytz.timezone('Europe/Amsterdam')
target = schedule.now().astimezone(target_tz)
cmd = schedule.DelayedCommand.at_time(target, target=None)
assert cmd.due()
def test_daylight_savings(self):
"""
A command at 9am should always be 9am regardless of
a DST boundary.
"""
with freezegun.freeze_time('2018-03-10 08:00:00'):
target_tz = pytz.timezone('US/Eastern')
target_time = datetime.time(9, tzinfo=target_tz)
cmd = schedule.PeriodicCommandFixedDelay.daily_at(
target_time, target=lambda: None
)
def naive(dt):
return dt.replace(tzinfo=None)
assert naive(cmd) == datetime.datetime(2018, 3, 10, 9, 0, 0)
next_ = cmd.next()
assert naive(next_) == datetime.datetime(2018, 3, 11, 9, 0, 0)
assert next_ - cmd == datetime.timedelta(hours=23)
class TestScheduler:
def test_invoke_scheduler(self):
sched = schedule.InvokeScheduler()
target = mock.MagicMock()
cmd = schedule.DelayedCommand.after(0, target)
sched.add(cmd)
sched.run_pending()
target.assert_called_once()
assert not sched.queue
def test_callback_scheduler(self):
callback = mock.MagicMock()
sched = schedule.CallbackScheduler(callback)
target = mock.MagicMock()
cmd = schedule.DelayedCommand.after(0, target)
sched.add(cmd)
sched.run_pending()
callback.assert_called_once_with(target)
def test_periodic_command(self):
sched = schedule.InvokeScheduler()
target = mock.MagicMock()
before = datetime.datetime.utcnow()
cmd = schedule.PeriodicCommand.after(10, target)
sched.add(cmd)
sched.run_pending()
target.assert_not_called()
with freezegun.freeze_time(before + datetime.timedelta(seconds=15)):
sched.run_pending()
assert sched.queue
target.assert_called_once()
with freezegun.freeze_time(before + datetime.timedelta(seconds=25)):
sched.run_pending()
assert target.call_count == 2
-50
View File
@@ -1,50 +0,0 @@
import datetime
import time
import contextlib
import os
from unittest import mock
import pytest
from tempora import timing
def test_IntervalGovernor():
"""
IntervalGovernor should prevent a function from being called more than
once per interval.
"""
func_under_test = mock.MagicMock()
# to look like a function, it needs a __name__ attribute
func_under_test.__name__ = 'func_under_test'
interval = datetime.timedelta(seconds=1)
governed = timing.IntervalGovernor(interval)(func_under_test)
governed('a')
governed('b')
governed(3, 'sir')
func_under_test.assert_called_once_with('a')
@pytest.fixture
def alt_tz(monkeypatch):
hasattr(time, 'tzset') or pytest.skip("tzset not available")
@contextlib.contextmanager
def change():
val = 'AEST-10AEDT-11,M10.5.0,M3.5.0'
with monkeypatch.context() as ctx:
ctx.setitem(os.environ, 'TZ', val)
time.tzset()
yield
time.tzset()
return change()
def test_Stopwatch_timezone_change(alt_tz):
"""
The stopwatch should provide a consistent duration even
if the timezone changes.
"""
watch = timing.Stopwatch()
with alt_tz:
assert abs(watch.split().total_seconds()) < 0.1
+329
View File
@@ -0,0 +1,329 @@
import io
import posixpath
import zipfile
import itertools
import contextlib
import sys
import pathlib
if sys.version_info < (3, 7):
from collections import OrderedDict
else:
OrderedDict = dict
__all__ = ['Path']
def _parents(path):
"""
Given a path with elements separated by
posixpath.sep, generate all parents of that path.
>>> list(_parents('b/d'))
['b']
>>> list(_parents('/b/d/'))
['/b']
>>> list(_parents('b/d/f/'))
['b/d', 'b']
>>> list(_parents('b'))
[]
>>> list(_parents(''))
[]
"""
return itertools.islice(_ancestry(path), 1, None)
def _ancestry(path):
"""
Given a path with elements separated by
posixpath.sep, generate all elements of that path
>>> list(_ancestry('b/d'))
['b/d', 'b']
>>> list(_ancestry('/b/d/'))
['/b/d', '/b']
>>> list(_ancestry('b/d/f/'))
['b/d/f', 'b/d', 'b']
>>> list(_ancestry('b'))
['b']
>>> list(_ancestry(''))
[]
"""
path = path.rstrip(posixpath.sep)
while path and path != posixpath.sep:
yield path
path, tail = posixpath.split(path)
_dedupe = OrderedDict.fromkeys
"""Deduplicate an iterable in original order"""
def _difference(minuend, subtrahend):
"""
Return items in minuend not in subtrahend, retaining order
with O(1) lookup.
"""
return itertools.filterfalse(set(subtrahend).__contains__, minuend)
class CompleteDirs(zipfile.ZipFile):
"""
A ZipFile subclass that ensures that implied directories
are always included in the namelist.
"""
@staticmethod
def _implied_dirs(names):
parents = itertools.chain.from_iterable(map(_parents, names))
as_dirs = (p + posixpath.sep for p in parents)
return _dedupe(_difference(as_dirs, names))
def namelist(self):
names = super(CompleteDirs, self).namelist()
return names + list(self._implied_dirs(names))
def _name_set(self):
return set(self.namelist())
def resolve_dir(self, name):
"""
If the name represents a directory, return that name
as a directory (with the trailing slash).
"""
names = self._name_set()
dirname = name + '/'
dir_match = name not in names and dirname in names
return dirname if dir_match else name
@classmethod
def make(cls, source):
"""
Given a source (filename or zipfile), return an
appropriate CompleteDirs subclass.
"""
if isinstance(source, CompleteDirs):
return source
if not isinstance(source, zipfile.ZipFile):
return cls(_pathlib_compat(source))
# Only allow for FastLookup when supplied zipfile is read-only
if 'r' not in source.mode:
cls = CompleteDirs
source.__class__ = cls
return source
class FastLookup(CompleteDirs):
"""
ZipFile subclass to ensure implicit
dirs exist and are resolved rapidly.
"""
def namelist(self):
with contextlib.suppress(AttributeError):
return self.__names
self.__names = super(FastLookup, self).namelist()
return self.__names
def _name_set(self):
with contextlib.suppress(AttributeError):
return self.__lookup
self.__lookup = super(FastLookup, self)._name_set()
return self.__lookup
def _pathlib_compat(path):
"""
For path-like objects, convert to a filename for compatibility
on Python 3.6.1 and earlier.
"""
try:
return path.__fspath__()
except AttributeError:
return str(path)
class Path:
"""
A pathlib-compatible interface for zip files.
Consider a zip file with this structure::
.
a.txt
b
c.txt
d
e.txt
>>> data = io.BytesIO()
>>> zf = zipfile.ZipFile(data, 'w')
>>> zf.writestr('a.txt', 'content of a')
>>> zf.writestr('b/c.txt', 'content of c')
>>> zf.writestr('b/d/e.txt', 'content of e')
>>> zf.filename = 'mem/abcde.zip'
Path accepts the zipfile object itself or a filename
>>> root = Path(zf)
From there, several path operations are available.
Directory iteration (including the zip file itself):
>>> a, b = root.iterdir()
>>> a
Path('mem/abcde.zip', 'a.txt')
>>> b
Path('mem/abcde.zip', 'b/')
name property:
>>> b.name
'b'
join with divide operator:
>>> c = b / 'c.txt'
>>> c
Path('mem/abcde.zip', 'b/c.txt')
>>> c.name
'c.txt'
Read text:
>>> c.read_text()
'content of c'
existence:
>>> c.exists()
True
>>> (b / 'missing.txt').exists()
False
Coercion to string:
>>> import os
>>> str(c).replace(os.sep, posixpath.sep)
'mem/abcde.zip/b/c.txt'
At the root, ``name``, ``filename``, and ``parent``
resolve to the zipfile. Note these attributes are not
valid and will raise a ``ValueError`` if the zipfile
has no filename.
>>> root.name
'abcde.zip'
>>> str(root.filename).replace(os.sep, posixpath.sep)
'mem/abcde.zip'
>>> str(root.parent)
'mem'
"""
__repr = "{self.__class__.__name__}({self.root.filename!r}, {self.at!r})"
def __init__(self, root, at=""):
"""
Construct a Path from a ZipFile or filename.
Note: When the source is an existing ZipFile object,
its type (__class__) will be mutated to a
specialized type. If the caller wishes to retain the
original type, the caller should either create a
separate ZipFile object or pass a filename.
"""
self.root = FastLookup.make(root)
self.at = at
def open(self, mode='r', *args, pwd=None, **kwargs):
"""
Open this entry as text or binary following the semantics
of ``pathlib.Path.open()`` by passing arguments through
to io.TextIOWrapper().
"""
if self.is_dir():
raise IsADirectoryError(self)
zip_mode = mode[0]
if not self.exists() and zip_mode == 'r':
raise FileNotFoundError(self)
stream = self.root.open(self.at, zip_mode, pwd=pwd)
if 'b' in mode:
if args or kwargs:
raise ValueError("encoding args invalid for binary operation")
return stream
return io.TextIOWrapper(stream, *args, **kwargs)
@property
def name(self):
return pathlib.Path(self.at).name or self.filename.name
@property
def suffix(self):
return pathlib.Path(self.at).suffix or self.filename.suffix
@property
def suffixes(self):
return pathlib.Path(self.at).suffixes or self.filename.suffixes
@property
def stem(self):
return pathlib.Path(self.at).stem or self.filename.stem
@property
def filename(self):
return pathlib.Path(self.root.filename).joinpath(self.at)
def read_text(self, *args, **kwargs):
with self.open('r', *args, **kwargs) as strm:
return strm.read()
def read_bytes(self):
with self.open('rb') as strm:
return strm.read()
def _is_child(self, path):
return posixpath.dirname(path.at.rstrip("/")) == self.at.rstrip("/")
def _next(self, at):
return self.__class__(self.root, at)
def is_dir(self):
return not self.at or self.at.endswith("/")
def is_file(self):
return self.exists() and not self.is_dir()
def exists(self):
return self.at in self.root._name_set()
def iterdir(self):
if not self.is_dir():
raise ValueError("Can't listdir a file")
subs = map(self._next, self.root.namelist())
return filter(self._is_child, subs)
def __str__(self):
return posixpath.join(self.root.filename, self.at)
def __repr__(self):
return self.__repr.format(self=self)
def joinpath(self, *other):
next = posixpath.join(self.at, *map(_pathlib_compat, other))
return self._next(self.root.resolve_dir(next))
__truediv__ = joinpath
@property
def parent(self):
if not self.at:
return self.filename.parent
parent_at = posixpath.dirname(self.at.rstrip('/'))
if parent_at:
parent_at += '/'
return self._next(parent_at)