Compare commits

..
17 Commits
Author SHA1 Message Date
rembo10 a78f38c174 Update changelog for v0.6.0 2022-11-13 09:39:20 +05:30
rembo10 14f2a6d22c Merge branch 'develop' 2022-11-13 09:34:39 +05:30
rembo10 2e4299efa7 Merge branch 'lastfm-api-key' into develop 2022-11-13 09:31:24 +05:30
rembo10 0610c2fa93 Fix formatting of Last.fm in log messages 2022-11-12 16:45:43 +05:30
rembo10 9add571886 Re-enable Last.fm similar tag cloud 2022-11-12 16:44:01 +05:30
rembo10 fcf59a9b38 add an option for a personal last.fm api key 2022-11-12 15:48:31 +05:30
AdeHub 74f9e91afc Various
xld - use native plistlib instead of biplist

rutracker - use new Result object

seacher - small tweaks

transmission - str no longer required
2022-03-05 19:08:12 +13:00
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
36 changed files with 693 additions and 3114 deletions
+8
View File
@@ -1,5 +1,13 @@
# Changelog
## v0.6.0
Released 13 November 2022
Highlights:
* Updated to python 3
The full list of commits can be found [here](https://github.com/rembo10/headphones/compare/v0.5.20...v0.6.0).
## v0.5.20
Released 15 October 2021
+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
+16 -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">
@@ -1651,6 +1654,16 @@
</div>
</fieldset>
<fieldset>
<legend>Last.fm</legend>
<div id="lastfmoptions">
<div class="row">
<label>API Key</label>
<input type="text" name="lastfm_apikey" value="${config['lastfm_apikey']}" size="40" />
</div>
</div>
</fieldset>
<fieldset>
<legend>Songkick</legend>
<div class="row checkbox">
+2
View File
@@ -155,6 +155,7 @@ _CONFIG_DEFINITIONS = {
'KEEP_TORRENT_FILES': (int, 'General', 0),
'KEEP_TORRENT_FILES_DIR': (path, 'General', ''),
'LASTFM_USERNAME': (str, 'General', ''),
'LASTFM_APIKEY': (str, 'General', ''),
'LAUNCH_BROWSER': (int, 'General', 1),
'LIBRARYSCAN': (int, 'General', 1),
'LIBRARYSCAN_INTERVAL': (int, 'General', 24),
@@ -240,6 +241,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),
+4 -3
View File
@@ -1,6 +1,6 @@
import os.path
import biplist
import plistlib
from headphones import logger
@@ -14,8 +14,9 @@ def getXldProfile(xldProfile):
# Get xld preferences plist
try:
preferences = biplist.readPlist(expanded)
except (biplist.InvalidPlistException, biplist.NotBinaryPlistException) as e:
with open(expanded, 'rb') as _f:
preferences = plistlib.load(_f)
except Exception as e:
logger.error("Error reading xld preferences plist: %s", e)
return (xldProfileNotFound, None, None)
+1 -7
View File
@@ -102,13 +102,7 @@ def artistlist_to_mbids(artistlist, forced=False):
myDB.action('DELETE from newartists WHERE ArtistName=?', [artist])
# Update the similar artist tag cloud:
# 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)
lastfm.getSimilar()
def addArtistIDListToDB(artistidlist):
+24 -17
View File
@@ -23,7 +23,7 @@ from headphones import db, logger, request
TIMEOUT = 60.0 # seconds
REQUEST_LIMIT = 1.0 / 5 # seconds
ENTRY_POINT = "https://ws.audioscrobbler.com/2.0/"
API_KEY = "395e6ec6bb557382fc41fde867bce66f"
APP_API_KEY = "395e6ec6bb557382fc41fde867bce66f"
# Required for API request limit
lastfm_lock = headphones.lock.TimedLock(REQUEST_LIMIT)
@@ -31,7 +31,7 @@ lastfm_lock = headphones.lock.TimedLock(REQUEST_LIMIT)
def request_lastfm(method, **kwargs):
"""
Call a Last.FM API method. Automatically sets the method and API key. Method
Call a Last.fm API method. Automatically sets the method and API key. Method
will return the result if no error occured.
By default, this method will request the JSON format, since it is more
@@ -40,32 +40,39 @@ def request_lastfm(method, **kwargs):
# Prepare request
kwargs["method"] = method
kwargs.setdefault("api_key", API_KEY)
kwargs.setdefault("api_key", headphones.CONFIG.LASTFM_APIKEY or APP_API_KEY)
kwargs.setdefault("format", "json")
# Send request
logger.debug("Calling Last.FM method: %s", method)
logger.debug("Last.FM call parameters: %s", kwargs)
logger.debug("Calling Last.fm method: %s", method)
logger.debug("Last.fm call parameters: %s", kwargs)
data = request.request_json(ENTRY_POINT, timeout=TIMEOUT, params=kwargs, lock=lastfm_lock)
# Parse response and check for errors.
if not data:
logger.error("Error calling Last.FM method: %s", method)
logger.error("Error calling Last.fm method: %s", method)
return
if "error" in data:
logger.debug("Last.FM returned an error: %s", data["message"])
logger.debug("Last.fm returned an error: %s", data["message"])
return
return data
def getSimilar():
if not headphones.CONFIG.LASTFM_APIKEY:
logger.info(
'To update the Similar Artists cloud tag, create a Last.fm application api key '
'and add it under the Advanced config tab'
)
return
myDB = db.DBConnection()
results = myDB.select("SELECT ArtistID from artists ORDER BY HaveTracks DESC LIMIT 10")
logger.info("Fetching similar artists from Last.FM for tag cloud")
logger.info("Fetching similar artists from Last.fm for tag cloud")
artistlist = []
for result in results:
@@ -85,7 +92,7 @@ def getSimilar():
artistlist.append((artist_name, artist_mbid))
# Add new artists to tag cloud
logger.debug("Fetched %d artists from Last.FM", len(artistlist))
logger.debug("Fetched %d artists from Last.fm", len(artistlist))
count = defaultdict(int)
for artist, mbid in artistlist:
@@ -103,7 +110,7 @@ def getSimilar():
myDB.action("INSERT INTO lastfmcloud VALUES( ?, ?, ?)", [artist_name, artist_mbid, count])
logger.debug("Inserted %d artists into Last.FM tag cloud", len(top_list))
logger.debug("Inserted %d artists into Last.fm tag cloud", len(top_list))
def getArtists():
@@ -111,16 +118,16 @@ def getArtists():
results = myDB.select("SELECT ArtistID from artists")
if not headphones.CONFIG.LASTFM_USERNAME:
logger.warn("Last.FM username not set, not importing artists.")
logger.warn("Last.fm username not set, not importing artists.")
return
logger.info("Fetching artists from Last.FM for username: %s", headphones.CONFIG.LASTFM_USERNAME)
logger.info("Fetching artists from Last.fm for username: %s", headphones.CONFIG.LASTFM_USERNAME)
data = request_lastfm("library.getartists", limit=1000, user=headphones.CONFIG.LASTFM_USERNAME)
if data and "artists" in data:
artistlist = []
artists = data["artists"]["artist"]
logger.debug("Fetched %d artists from Last.FM", len(artists))
logger.debug("Fetched %d artists from Last.fm", len(artists))
for artist in artists:
artist_mbid = artist["mbid"]
@@ -133,20 +140,20 @@ def getArtists():
for artistid in artistlist:
importer.addArtisttoDB(artistid)
logger.info("Imported %d new artists from Last.FM", len(artistlist))
logger.info("Imported %d new artists from Last.fm", len(artistlist))
def getTagTopArtists(tag, limit=50):
myDB = db.DBConnection()
results = myDB.select("SELECT ArtistID from artists")
logger.info("Fetching top artists from Last.FM for tag: %s", tag)
logger.info("Fetching top artists from Last.fm for tag: %s", tag)
data = request_lastfm("tag.gettopartists", limit=limit, tag=tag)
if data and "topartists" in data:
artistlist = []
artists = data["topartists"]["artist"]
logger.debug("Fetched %d artists from Last.FM", len(artists))
logger.debug("Fetched %d artists from Last.fm", len(artists))
for artist in artists:
try:
@@ -162,4 +169,4 @@ def getTagTopArtists(tag, limit=50):
for artistid in artistlist:
importer.addArtisttoDB(artistid)
logger.debug("Added %d new artists from Last.FM", len(artistlist))
logger.debug("Added %d new artists from Last.fm", len(artistlist))
+2 -3
View File
@@ -442,9 +442,8 @@ def libraryScan(dir=None, append=False, ArtistID=None, ArtistName=None,
# if not append:
# update_album_status()
# TODO: Fix last.fm api calls
#if not append and not artistScan:
#lastfm.getSimilar()
if not append and not artistScan:
lastfm.getSimilar()
if ArtistName:
logger.info('Scanning complete for artist: %s', ArtistName)
+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
+2 -1
View File
@@ -11,6 +11,7 @@ from bs4 import BeautifulSoup
import headphones
from headphones import logger
from headphones.types import Result
class Rutracker(object):
@@ -160,7 +161,7 @@ class Rutracker(object):
torrent_id = dict([part.split('=') for part in urlparse(url)[4].split('&')])[
't']
topicurl = 'https://rutracker.org/forum/viewtopic.php?t=' + torrent_id
rulist.append((title, size, topicurl, 'rutracker.org', 'torrent', True))
rulist.append(Result(title, size, url, 'rutracker.org', 'torrent', True))
else:
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)))
+260 -185
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:
@@ -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,49 @@ 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])
myDB.action(
"INSERT INTO snatched VALUES (?, ?, ?, ?, DATETIME('NOW', 'localtime'), "
"?, ?, ?, ?)", [
album['AlbumID'],
result.title,
result.size,
result.url,
"Snatched",
folder_name,
kind,
torrentid
]
)
# Store the torrent id so we can check later if it's finished seeding and can be removed
# Additional record for post processing or scheduled job to remove the torrent when finished seeding
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])
"INSERT INTO snatched VALUES (?, ?, ?, ?, DATETIME('NOW', 'localtime'), "
"?, ?, ?, ?)", [
album['AlbumID'],
result.title,
result.size,
result.url,
"Seed_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 +1252,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 +1402,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 +1476,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 +1641,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 +1747,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 +1830,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 +1884,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 +1892,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,28 +1906,51 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
def preprocess(resultlist):
for result in resultlist:
if result[4] == 'torrent':
headers = {}
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':
# rutracker always needs the torrent data
if result[3] == 'rutracker.org':
return ruobj.get_torrent_data(result[2]), result
if result.provider == 'rutracker.org':
return ruobj.get_torrent_data(result.url), 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 result.provider.startswith('Jackett_') or 'torznab' in result.provider.lower():
r = request.request_response(url=result.url, headers=headers, allow_redirects=False)
if r:
jackett_content = r.content
link = r.headers.get('Location')
if link and link != result[2]:
if link and link != result.url:
if link.startswith('magnet:'):
result = (result[0], result[1], link, result[3], "magnet", result[5])
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[0], result[1], link, result[3], result[4], result[5])
result = Result(
result.url,
result.size,
link,
result.provider,
result.kind,
result.matches
)
return True, result
else:
return r.content, result
# 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
@@ -1874,31 +1958,22 @@ def preprocess(resultlist):
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
# 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
+3 -4
View File
@@ -15,7 +15,7 @@
import time
import json
import base64
from base64 import b64encode
import urllib.parse
import os
@@ -36,10 +36,10 @@ def addTorrent(link, data=None):
if link.endswith('.torrent') and not link.startswith(('http', 'magnet')) or data:
if data:
metainfo = str(base64.b64encode(data))
metainfo = b64encode(data).decode("utf-8")
else:
with open(link, 'rb') as f:
metainfo = str(base64.b64encode(f.read()))
metainfo = b64encode(f.read()).decode("utf-8")
arguments = {'metainfo': metainfo, 'download-dir': headphones.CONFIG.DOWNLOAD_TORRENT_DIR}
else:
arguments = {'filename': link, 'download-dir': headphones.CONFIG.DOWNLOAD_TORRENT_DIR}
@@ -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
+13 -22
View File
@@ -24,6 +24,7 @@ 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
@@ -53,6 +54,7 @@ from headphones.helpers import (
replace_illegal_chars,
today,
)
from headphones.types import Result
def serve_template(templatename, **kwargs):
@@ -450,21 +452,8 @@ 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()
@@ -473,17 +462,17 @@ class WebInterface(object):
if 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'}
@@ -1279,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),
@@ -1395,6 +1385,7 @@ class WebInterface(object):
"custompass": headphones.CONFIG.CUSTOMPASS,
"hpuser": headphones.CONFIG.HPUSER,
"hppass": headphones.CONFIG.HPPASS,
"lastfm_apikey": headphones.CONFIG.LASTFM_APIKEY,
"songkick_enabled": checked(headphones.CONFIG.SONGKICK_ENABLED),
"songkick_apikey": headphones.CONFIG.SONGKICK_APIKEY,
"songkick_location": headphones.CONFIG.SONGKICK_LOCATION,
@@ -1471,8 +1462,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",
-803
View File
@@ -1,803 +0,0 @@
"""biplist -- a library for reading and writing binary property list files.
Binary Property List (plist) files provide a faster and smaller serialization
format for property lists on OS X. This is a library for generating binary
plists which can be read by OS X, iOS, or other clients.
The API models the plistlib API, and will call through to plistlib when
XML serialization or deserialization is required.
To generate plists with UID values, wrap the values with the Uid object. The
value must be an int.
To generate plists with NSData/CFData values, wrap the values with the
Data object. The value must be a string.
Date values can only be datetime.datetime objects.
The exceptions InvalidPlistException and NotBinaryPlistException may be
thrown to indicate that the data cannot be serialized or deserialized as
a binary plist.
Plist generation example:
from biplist import *
from datetime import datetime
plist = {'aKey':'aValue',
'0':1.322,
'now':datetime.now(),
'list':[1,2,3],
'tuple':('a','b','c')
}
try:
writePlist(plist, "example.plist")
except (InvalidPlistException, NotBinaryPlistException), e:
print "Something bad happened:", e
Plist parsing example:
from biplist import *
try:
plist = readPlist("example.plist")
print plist
except (InvalidPlistException, NotBinaryPlistException), e:
print "Not a plist:", e
"""
import sys
from collections import namedtuple
import datetime
import io
import math
import plistlib
from struct import pack, unpack
from struct import error as struct_error
import sys
import time
try:
str
unicodeEmpty = r''
except NameError:
str = str
unicodeEmpty = ''
try:
int
except NameError:
long = int
try:
{}.iteritems
iteritems = lambda x: iter(x.items())
except AttributeError:
iteritems = lambda x: list(x.items())
__all__ = [
'Uid', 'Data', 'readPlist', 'writePlist', 'readPlistFromString',
'writePlistToString', 'InvalidPlistException', 'NotBinaryPlistException'
]
# Apple uses Jan 1, 2001 as a base for all plist date/times.
apple_reference_date = datetime.datetime.utcfromtimestamp(978307200)
class Uid(int):
"""Wrapper around integers for representing UID values. This
is used in keyed archiving."""
def __repr__(self):
return "Uid(%d)" % self
class Data(bytes):
"""Wrapper around str types for representing Data values."""
pass
class InvalidPlistException(Exception):
"""Raised when the plist is incorrectly formatted."""
pass
class NotBinaryPlistException(Exception):
"""Raised when a binary plist was expected but not encountered."""
pass
def readPlist(pathOrFile):
"""Raises NotBinaryPlistException, InvalidPlistException"""
didOpen = False
result = None
if isinstance(pathOrFile, (bytes, str)):
pathOrFile = open(pathOrFile, 'rb')
didOpen = True
try:
reader = PlistReader(pathOrFile)
result = reader.parse()
except NotBinaryPlistException as e:
try:
pathOrFile.seek(0)
result = None
if hasattr(plistlib, 'loads'):
contents = None
if isinstance(pathOrFile, (bytes, str)):
with open(pathOrFile, 'rb') as f:
contents = f.read()
else:
contents = pathOrFile.read()
result = plistlib.loads(contents)
else:
result = plistlib.readPlist(pathOrFile)
result = wrapDataObject(result, for_binary=True)
except Exception as e:
raise InvalidPlistException(e)
finally:
if didOpen:
pathOrFile.close()
return result
def wrapDataObject(o, for_binary=False):
if isinstance(o, Data) and not for_binary:
v = sys.version_info
if not (v[0] >= 3 and v[1] >= 4):
o = plistlib.Data(o)
elif isinstance(o, (bytes, plistlib.Data)) and for_binary:
if hasattr(o, 'data'):
o = Data(o.data)
elif isinstance(o, tuple):
o = wrapDataObject(list(o), for_binary)
o = tuple(o)
elif isinstance(o, list):
for i in range(len(o)):
o[i] = wrapDataObject(o[i], for_binary)
elif isinstance(o, dict):
for k in o:
o[k] = wrapDataObject(o[k], for_binary)
return o
def writePlist(rootObject, pathOrFile, binary=True):
if not binary:
rootObject = wrapDataObject(rootObject, binary)
if hasattr(plistlib, "dump"):
if isinstance(pathOrFile, (bytes, str)):
with open(pathOrFile, 'wb') as f:
return plistlib.dump(rootObject, f)
else:
return plistlib.dump(rootObject, pathOrFile)
else:
return plistlib.writePlist(rootObject, pathOrFile)
else:
didOpen = False
if isinstance(pathOrFile, (bytes, str)):
pathOrFile = open(pathOrFile, 'wb')
didOpen = True
writer = PlistWriter(pathOrFile)
result = writer.writeRoot(rootObject)
if didOpen:
pathOrFile.close()
return result
def readPlistFromString(data):
return readPlist(io.BytesIO(data))
def writePlistToString(rootObject, binary=True):
if not binary:
rootObject = wrapDataObject(rootObject, binary)
if hasattr(plistlib, "dumps"):
return plistlib.dumps(rootObject)
elif hasattr(plistlib, "writePlistToBytes"):
return plistlib.writePlistToBytes(rootObject)
else:
return plistlib.writePlistToString(rootObject)
else:
ioObject = io.BytesIO()
writer = PlistWriter(ioObject)
writer.writeRoot(rootObject)
return ioObject.getvalue()
def is_stream_binary_plist(stream):
stream.seek(0)
header = stream.read(7)
if header == b'bplist0':
return True
else:
return False
PlistTrailer = namedtuple('PlistTrailer', 'offsetSize, objectRefSize, offsetCount, topLevelObjectNumber, offsetTableOffset')
PlistByteCounts = namedtuple('PlistByteCounts', 'nullBytes, boolBytes, intBytes, realBytes, dateBytes, dataBytes, stringBytes, uidBytes, arrayBytes, setBytes, dictBytes')
class PlistReader(object):
file = None
contents = ''
offsets = None
trailer = None
currentOffset = 0
def __init__(self, fileOrStream):
"""Raises NotBinaryPlistException."""
self.reset()
self.file = fileOrStream
def parse(self):
return self.readRoot()
def reset(self):
self.trailer = None
self.contents = ''
self.offsets = []
self.currentOffset = 0
def readRoot(self):
result = None
self.reset()
# Get the header, make sure it's a valid file.
if not is_stream_binary_plist(self.file):
raise NotBinaryPlistException()
self.file.seek(0)
self.contents = self.file.read()
if len(self.contents) < 32:
raise InvalidPlistException("File is too short.")
trailerContents = self.contents[-32:]
try:
self.trailer = PlistTrailer._make(unpack("!xxxxxxBBQQQ", trailerContents))
offset_size = self.trailer.offsetSize * self.trailer.offsetCount
offset = self.trailer.offsetTableOffset
offset_contents = self.contents[offset:offset+offset_size]
offset_i = 0
while offset_i < self.trailer.offsetCount:
begin = self.trailer.offsetSize*offset_i
tmp_contents = offset_contents[begin:begin+self.trailer.offsetSize]
tmp_sized = self.getSizedInteger(tmp_contents, self.trailer.offsetSize)
self.offsets.append(tmp_sized)
offset_i += 1
self.setCurrentOffsetToObjectNumber(self.trailer.topLevelObjectNumber)
result = self.readObject()
except TypeError as e:
raise InvalidPlistException(e)
return result
def setCurrentOffsetToObjectNumber(self, objectNumber):
self.currentOffset = self.offsets[objectNumber]
def readObject(self):
result = None
tmp_byte = self.contents[self.currentOffset:self.currentOffset+1]
marker_byte = unpack("!B", tmp_byte)[0]
format = (marker_byte >> 4) & 0x0f
extra = marker_byte & 0x0f
self.currentOffset += 1
def proc_extra(extra):
if extra == 0b1111:
#self.currentOffset += 1
extra = self.readObject()
return extra
# bool, null, or fill byte
if format == 0b0000:
if extra == 0b0000:
result = None
elif extra == 0b1000:
result = False
elif extra == 0b1001:
result = True
elif extra == 0b1111:
pass # fill byte
else:
raise InvalidPlistException("Invalid object found at offset: %d" % (self.currentOffset - 1))
# int
elif format == 0b0001:
extra = proc_extra(extra)
result = self.readInteger(pow(2, extra))
# real
elif format == 0b0010:
extra = proc_extra(extra)
result = self.readReal(extra)
# date
elif format == 0b0011 and extra == 0b0011:
result = self.readDate()
# data
elif format == 0b0100:
extra = proc_extra(extra)
result = self.readData(extra)
# ascii string
elif format == 0b0101:
extra = proc_extra(extra)
result = self.readAsciiString(extra)
# Unicode string
elif format == 0b0110:
extra = proc_extra(extra)
result = self.readUnicode(extra)
# uid
elif format == 0b1000:
result = self.readUid(extra)
# array
elif format == 0b1010:
extra = proc_extra(extra)
result = self.readArray(extra)
# set
elif format == 0b1100:
extra = proc_extra(extra)
result = set(self.readArray(extra))
# dict
elif format == 0b1101:
extra = proc_extra(extra)
result = self.readDict(extra)
else:
raise InvalidPlistException("Invalid object found: {format: %s, extra: %s}" % (bin(format), bin(extra)))
return result
def readInteger(self, byteSize):
result = 0
original_offset = self.currentOffset
data = self.contents[self.currentOffset:self.currentOffset + byteSize]
result = self.getSizedInteger(data, byteSize, as_number=True)
self.currentOffset = original_offset + byteSize
return result
def readReal(self, length):
result = 0.0
to_read = pow(2, length)
data = self.contents[self.currentOffset:self.currentOffset+to_read]
if length == 2: # 4 bytes
result = unpack('>f', data)[0]
elif length == 3: # 8 bytes
result = unpack('>d', data)[0]
else:
raise InvalidPlistException("Unknown real of length %d bytes" % to_read)
return result
def readRefs(self, count):
refs = []
i = 0
while i < count:
fragment = self.contents[self.currentOffset:self.currentOffset+self.trailer.objectRefSize]
ref = self.getSizedInteger(fragment, len(fragment))
refs.append(ref)
self.currentOffset += self.trailer.objectRefSize
i += 1
return refs
def readArray(self, count):
result = []
values = self.readRefs(count)
i = 0
while i < len(values):
self.setCurrentOffsetToObjectNumber(values[i])
value = self.readObject()
result.append(value)
i += 1
return result
def readDict(self, count):
result = {}
keys = self.readRefs(count)
values = self.readRefs(count)
i = 0
while i < len(keys):
self.setCurrentOffsetToObjectNumber(keys[i])
key = self.readObject()
self.setCurrentOffsetToObjectNumber(values[i])
value = self.readObject()
result[key] = value
i += 1
return result
def readAsciiString(self, length):
result = unpack("!%ds" % length, self.contents[self.currentOffset:self.currentOffset+length])[0]
self.currentOffset += length
return result
def readUnicode(self, length):
actual_length = length*2
data = self.contents[self.currentOffset:self.currentOffset+actual_length]
# unpack not needed?!! data = unpack(">%ds" % (actual_length), data)[0]
self.currentOffset += actual_length
return data.decode('utf_16_be')
def readDate(self):
result = unpack(">d", self.contents[self.currentOffset:self.currentOffset+8])[0]
# Use timedelta to workaround time_t size limitation on 32-bit python.
result = datetime.timedelta(seconds=result) + apple_reference_date
self.currentOffset += 8
return result
def readData(self, length):
result = self.contents[self.currentOffset:self.currentOffset+length]
self.currentOffset += length
return Data(result)
def readUid(self, length):
return Uid(self.readInteger(length+1))
def getSizedInteger(self, data, byteSize, as_number=False):
"""Numbers of 8 bytes are signed integers when they refer to numbers, but unsigned otherwise."""
result = 0
# 1, 2, and 4 byte integers are unsigned
if byteSize == 1:
result = unpack('>B', data)[0]
elif byteSize == 2:
result = unpack('>H', data)[0]
elif byteSize == 4:
result = unpack('>L', data)[0]
elif byteSize == 8:
if as_number:
result = unpack('>q', data)[0]
else:
result = unpack('>Q', data)[0]
elif byteSize <= 16:
# Handle odd-sized or integers larger than 8 bytes
# Don't naively go over 16 bytes, in order to prevent infinite loops.
result = 0
if hasattr(int, 'from_bytes'):
result = int.from_bytes(data, 'big')
else:
for byte in data:
result = (result << 8) | unpack('>B', byte)[0]
else:
raise InvalidPlistException("Encountered integer longer than 16 bytes.")
return result
class HashableWrapper(object):
def __init__(self, value):
self.value = value
def __repr__(self):
return "<HashableWrapper: %s>" % [self.value]
class BoolWrapper(object):
def __init__(self, value):
self.value = value
def __repr__(self):
return "<BoolWrapper: %s>" % self.value
class FloatWrapper(object):
_instances = {}
def __new__(klass, value):
# Ensure FloatWrapper(x) for a given float x is always the same object
wrapper = klass._instances.get(value)
if wrapper is None:
wrapper = object.__new__(klass)
wrapper.value = value
klass._instances[value] = wrapper
return wrapper
def __repr__(self):
return "<FloatWrapper: %s>" % self.value
class PlistWriter(object):
header = b'bplist00bybiplist1.0'
file = None
byteCounts = None
trailer = None
computedUniques = None
writtenReferences = None
referencePositions = None
wrappedTrue = None
wrappedFalse = None
def __init__(self, file):
self.reset()
self.file = file
self.wrappedTrue = BoolWrapper(True)
self.wrappedFalse = BoolWrapper(False)
def reset(self):
self.byteCounts = PlistByteCounts(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
self.trailer = PlistTrailer(0, 0, 0, 0, 0)
# A set of all the uniques which have been computed.
self.computedUniques = set()
# A list of all the uniques which have been written.
self.writtenReferences = {}
# A dict of the positions of the written uniques.
self.referencePositions = {}
def positionOfObjectReference(self, obj):
"""If the given object has been written already, return its
position in the offset table. Otherwise, return None."""
return self.writtenReferences.get(obj)
def writeRoot(self, root):
"""
Strategy is:
- write header
- wrap root object so everything is hashable
- compute size of objects which will be written
- need to do this in order to know how large the object refs
will be in the list/dict/set reference lists
- write objects
- keep objects in writtenReferences
- keep positions of object references in referencePositions
- write object references with the length computed previously
- computer object reference length
- write object reference positions
- write trailer
"""
output = self.header
wrapped_root = self.wrapRoot(root)
should_reference_root = True#not isinstance(wrapped_root, HashableWrapper)
self.computeOffsets(wrapped_root, asReference=should_reference_root, isRoot=True)
self.trailer = self.trailer._replace(**{'objectRefSize':self.intSize(len(self.computedUniques))})
(_, output) = self.writeObjectReference(wrapped_root, output)
output = self.writeObject(wrapped_root, output, setReferencePosition=True)
# output size at this point is an upper bound on how big the
# object reference offsets need to be.
self.trailer = self.trailer._replace(**{
'offsetSize':self.intSize(len(output)),
'offsetCount':len(self.computedUniques),
'offsetTableOffset':len(output),
'topLevelObjectNumber':0
})
output = self.writeOffsetTable(output)
output += pack('!xxxxxxBBQQQ', *self.trailer)
self.file.write(output)
def wrapRoot(self, root):
if isinstance(root, bool):
if root is True:
return self.wrappedTrue
else:
return self.wrappedFalse
elif isinstance(root, float):
return FloatWrapper(root)
elif isinstance(root, set):
n = set()
for value in root:
n.add(self.wrapRoot(value))
return HashableWrapper(n)
elif isinstance(root, dict):
n = {}
for key, value in iteritems(root):
n[self.wrapRoot(key)] = self.wrapRoot(value)
return HashableWrapper(n)
elif isinstance(root, list):
n = []
for value in root:
n.append(self.wrapRoot(value))
return HashableWrapper(n)
elif isinstance(root, tuple):
n = tuple([self.wrapRoot(value) for value in root])
return HashableWrapper(n)
else:
return root
def incrementByteCount(self, field, incr=1):
self.byteCounts = self.byteCounts._replace(**{field:self.byteCounts.__getattribute__(field) + incr})
def computeOffsets(self, obj, asReference=False, isRoot=False):
def check_key(key):
if key is None:
raise InvalidPlistException('Dictionary keys cannot be null in plists.')
elif isinstance(key, Data):
raise InvalidPlistException('Data cannot be dictionary keys in plists.')
elif not isinstance(key, (bytes, str)):
raise InvalidPlistException('Keys must be strings.')
def proc_size(size):
if size > 0b1110:
size += self.intSize(size)
return size
# If this should be a reference, then we keep a record of it in the
# uniques table.
if asReference:
if obj in self.computedUniques:
return
else:
self.computedUniques.add(obj)
if obj is None:
self.incrementByteCount('nullBytes')
elif isinstance(obj, BoolWrapper):
self.incrementByteCount('boolBytes')
elif isinstance(obj, Uid):
size = self.intSize(obj)
self.incrementByteCount('uidBytes', incr=1+size)
elif isinstance(obj, int):
size = self.intSize(obj)
self.incrementByteCount('intBytes', incr=1+size)
elif isinstance(obj, FloatWrapper):
size = self.realSize(obj)
self.incrementByteCount('realBytes', incr=1+size)
elif isinstance(obj, datetime.datetime):
self.incrementByteCount('dateBytes', incr=2)
elif isinstance(obj, Data):
size = proc_size(len(obj))
self.incrementByteCount('dataBytes', incr=1+size)
elif isinstance(obj, (str, bytes)):
size = proc_size(len(obj))
self.incrementByteCount('stringBytes', incr=1+size)
elif isinstance(obj, HashableWrapper):
obj = obj.value
if isinstance(obj, set):
size = proc_size(len(obj))
self.incrementByteCount('setBytes', incr=1+size)
for value in obj:
self.computeOffsets(value, asReference=True)
elif isinstance(obj, (list, tuple)):
size = proc_size(len(obj))
self.incrementByteCount('arrayBytes', incr=1+size)
for value in obj:
asRef = True
self.computeOffsets(value, asReference=True)
elif isinstance(obj, dict):
size = proc_size(len(obj))
self.incrementByteCount('dictBytes', incr=1+size)
for key, value in iteritems(obj):
check_key(key)
self.computeOffsets(key, asReference=True)
self.computeOffsets(value, asReference=True)
else:
raise InvalidPlistException("Unknown object type.")
def writeObjectReference(self, obj, output):
"""Tries to write an object reference, adding it to the references
table. Does not write the actual object bytes or set the reference
position. Returns a tuple of whether the object was a new reference
(True if it was, False if it already was in the reference table)
and the new output.
"""
position = self.positionOfObjectReference(obj)
if position is None:
self.writtenReferences[obj] = len(self.writtenReferences)
output += self.binaryInt(len(self.writtenReferences) - 1, byteSize=self.trailer.objectRefSize)
return (True, output)
else:
output += self.binaryInt(position, byteSize=self.trailer.objectRefSize)
return (False, output)
def writeObject(self, obj, output, setReferencePosition=False):
"""Serializes the given object to the output. Returns output.
If setReferencePosition is True, will set the position the
object was written.
"""
def proc_variable_length(format, length):
result = b''
if length > 0b1110:
result += pack('!B', (format << 4) | 0b1111)
result = self.writeObject(length, result)
else:
result += pack('!B', (format << 4) | length)
return result
if isinstance(obj, str) and obj == unicodeEmpty:
# The Apple Plist decoder can't decode a zero length Unicode string.
obj = b''
if setReferencePosition:
self.referencePositions[obj] = len(output)
if obj is None:
output += pack('!B', 0b00000000)
elif isinstance(obj, BoolWrapper):
if obj.value is False:
output += pack('!B', 0b00001000)
else:
output += pack('!B', 0b00001001)
elif isinstance(obj, Uid):
size = self.intSize(obj)
output += pack('!B', (0b1000 << 4) | size - 1)
output += self.binaryInt(obj)
elif isinstance(obj, int):
byteSize = self.intSize(obj)
root = math.log(byteSize, 2)
output += pack('!B', (0b0001 << 4) | int(root))
output += self.binaryInt(obj, as_number=True)
elif isinstance(obj, FloatWrapper):
# just use doubles
output += pack('!B', (0b0010 << 4) | 3)
output += self.binaryReal(obj)
elif isinstance(obj, datetime.datetime):
timestamp = (obj - apple_reference_date).total_seconds()
output += pack('!B', 0b00110011)
output += pack('!d', float(timestamp))
elif isinstance(obj, Data):
output += proc_variable_length(0b0100, len(obj))
output += obj
elif isinstance(obj, str):
byteData = obj.encode('utf_16_be')
output += proc_variable_length(0b0110, len(byteData)//2)
output += byteData
elif isinstance(obj, bytes):
output += proc_variable_length(0b0101, len(obj))
output += obj
elif isinstance(obj, HashableWrapper):
obj = obj.value
if isinstance(obj, (set, list, tuple)):
if isinstance(obj, set):
output += proc_variable_length(0b1100, len(obj))
else:
output += proc_variable_length(0b1010, len(obj))
objectsToWrite = []
for objRef in obj:
(isNew, output) = self.writeObjectReference(objRef, output)
if isNew:
objectsToWrite.append(objRef)
for objRef in objectsToWrite:
output = self.writeObject(objRef, output, setReferencePosition=True)
elif isinstance(obj, dict):
output += proc_variable_length(0b1101, len(obj))
keys = []
values = []
objectsToWrite = []
for key, value in iteritems(obj):
keys.append(key)
values.append(value)
for key in keys:
(isNew, output) = self.writeObjectReference(key, output)
if isNew:
objectsToWrite.append(key)
for value in values:
(isNew, output) = self.writeObjectReference(value, output)
if isNew:
objectsToWrite.append(value)
for objRef in objectsToWrite:
output = self.writeObject(objRef, output, setReferencePosition=True)
return output
def writeOffsetTable(self, output):
"""Writes all of the object reference offsets."""
all_positions = []
writtenReferences = list(self.writtenReferences.items())
writtenReferences.sort(key=lambda x: x[1])
for obj,order in writtenReferences:
# Porting note: Elsewhere we deliberately replace empty unicdoe strings
# with empty binary strings, but the empty unicode string
# goes into writtenReferences. This isn't an issue in Py2
# because u'' and b'' have the same hash; but it is in
# Py3, where they don't.
if bytes != str and obj == unicodeEmpty:
obj = b''
position = self.referencePositions.get(obj)
if position is None:
raise InvalidPlistException("Error while writing offsets table. Object not found. %s" % obj)
output += self.binaryInt(position, self.trailer.offsetSize)
all_positions.append(position)
return output
def binaryReal(self, obj):
# just use doubles
result = pack('>d', obj.value)
return result
def binaryInt(self, obj, byteSize=None, as_number=False):
result = b''
if byteSize is None:
byteSize = self.intSize(obj)
if byteSize == 1:
result += pack('>B', obj)
elif byteSize == 2:
result += pack('>H', obj)
elif byteSize == 4:
result += pack('>L', obj)
elif byteSize == 8:
if as_number:
result += pack('>q', obj)
else:
result += pack('>Q', obj)
elif byteSize <= 16:
try:
result = pack('>Q', 0) + pack('>Q', obj)
except struct_error as e:
raise InvalidPlistException("Unable to pack integer %d: %s" % (obj, e))
else:
raise InvalidPlistException("Core Foundation can't handle integers with size greater than 16 bytes.")
return result
def intSize(self, obj):
"""Returns the number of bytes necessary to store the given integer."""
# SIGNED
if obj < 0: # Signed integer, always 8 bytes
return 8
# UNSIGNED
elif obj <= 0xFF: # 1 byte
return 1
elif obj <= 0xFFFF: # 2 bytes
return 2
elif obj <= 0xFFFFFFFF: # 4 bytes
return 4
# SIGNED
# 0x7FFFFFFFFFFFFFFF is the max.
elif obj <= 0x7FFFFFFFFFFFFFFF: # 8 bytes signed
return 8
elif obj <= 0xffffffffffffffff: # 8 bytes unsigned
return 16
else:
raise InvalidPlistException("Core Foundation can't handle integers with size greater than 8 bytes.")
def realSize(self, obj):
return 8
View File
@@ -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)