metadata: Fix typo in DISC_TOTAL

Headphones.py

@@ -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

data/interfaces/default/config.html

@@ -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">

headphones/config.py

@@ -240,6 +240,7 @@ _CONFIG_DEFINITIONS = {
     'QBITTORRENT_PASSWORD': (str, 'QBitTorrent', ''),
     'QBITTORRENT_USERNAME': (str, 'QBitTorrent', ''),
     'RENAME_FILES': (int, 'General', 0),
+    'RENAME_SINGLE_DISC_IGNORE': (int, 'General', 0),
     'RENAME_UNPROCESSED': (bool_int, 'General', 1),
     'RENAME_FROZEN': (bool_int, 'General', 1),
     'REPLACE_EXISTING_FOLDERS': (int, 'General', 0),

headphones/metadata.py

@@ -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,

headphones/music_encoder.py

@@ -30,7 +30,6 @@ from . import getXldProfile
 
 
 def encode(albumPath):
-    print(albumPath)
     use_xld = headphones.CONFIG.ENCODER == 'xld'
 
     # Return if xld details not found

headphones/nzbget.py

@@ -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)

headphones/postprocessor.py

@@ -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

headphones/searcher.py

@@ -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,84 +423,75 @@ 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:
+    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'])
+    logger.info(
+        f"No appropriate matches found for {album['ArtistName']} - "
+        f"{album['AlbumTitle']}"
+    )
     return None
 
-    return finallist
-
 
 def get_year_from_release_date(release_date):
     try:
@@ -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,33 @@ def send_to_downloader(data, bestqual, album):
                 return
 
             # Set Seed Ratio
-            seed_ratio = get_seed_ratio(bestqual[3])
+            # Oh my god why is this repeated again for the 100th time
+            seed_ratio = get_seed_ratio(result.provider)
             if seed_ratio is not None:
                 qbittorrent.setSeedRatio(torrentid, seed_ratio)
 
     myDB = db.DBConnection()
     myDB.action('UPDATE albums SET status = "Snatched" WHERE AlbumID=?', [album['AlbumID']])
-    myDB.action('INSERT INTO snatched VALUES( ?, ?, ?, ?, DATETIME("NOW", "localtime"), ?, ?, ?, ?)',
-                [album['AlbumID'], bestqual[0], bestqual[1], bestqual[2], "Snatched", folder_name,
-                 kind, torrentid])
-
-    # Store the torrent id so we can check later if it's finished seeding and can be removed
-    if seed_ratio is not None and seed_ratio != 0 and torrentid:
     myDB.action(
-            'INSERT INTO snatched VALUES( ?, ?, ?, ?, DATETIME("NOW", "localtime"), ?, ?, ?, ?)',
-            [album['AlbumID'], bestqual[0], bestqual[1], bestqual[2], "Seed_Snatched", folder_name,
-             kind, torrentid])
+        "INSERT INTO snatched VALUES (?, ?, ?, ?, DATETIME('NOW', 'localtime'), "
+        "?, ?, ?, ?)", [
+            album['AlbumID'],
+            result.title,
+            result.size,
+            result.url,
+            "Seed_Snatched" if seed_ratio and torrentid else "Snatched",
+            folder_name,
+            kind,
+            torrentid
+        ]
+    )
 
     # notify
     artist = album[1]
     albumname = album[2]
     rgid = album[6]
     title = artist + ' - ' + albumname
-    provider = bestqual[3]
+    provider = result.provider
     if provider.startswith(("http://", "https://")):
         provider = provider.split("//")[1]
     name = folder_name if folder_name else None
@@ -1209,13 +1236,22 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
     year = get_year_from_release_date(reldate)
 
     # MERGE THIS WITH THE TERM CLEANUP FROM searchNZB
-    dic = {'...': '', ' & ': ' ', ' = ': ' ', '?': '', '$': 's', ' + ': ' ', '"': '', ',': ' ',
-           '*': ''}
+    replacements = {
+        '...': '',
+        ' & ': ' ',
+        ' = ': ' ',
+        '?': '',
+        '$': 's',
+        ' + ': ' ',
+        '"': '',
+        ',': ' ',
+        '*': ''
+    }
 
-    semi_cleanalbum = helpers.replace_all(album['AlbumTitle'], dic)
-    cleanalbum = helpers.latinToAscii(semi_cleanalbum)
-    semi_cleanartist = helpers.replace_all(album['ArtistName'], dic)
-    cleanartist = helpers.latinToAscii(semi_cleanartist)
+    semi_cleanalbum = helpers.replace_all(album['AlbumTitle'], replacements)
+    cleanalbum = unidecode(semi_cleanalbum)
+    semi_cleanartist = helpers.replace_all(album['ArtistName'], replacements)
+    cleanartist = unidecode(semi_cleanartist)
 
     # Use provided term if available, otherwise build our own (this code needs to be cleaned up since a lot
     # of these torrent providers are just using cleanartist/cleanalbum terms
@@ -1350,7 +1386,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
                             if all(word.lower() in title.lower() for word in term.split()):
                                 if size < maxsize and minimumseeders < seeders:
                                     logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
-                                    resultlist.append((title, size, url, provider, 'torrent', True))
+                                    resultlist.append(Result(title, size, url, provider, 'torrent', True))
                                 else:
                                     logger.info(
                                         '%s is larger than the maxsize or has too little seeders for this category, '
@@ -1424,7 +1460,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
                         desc_match = re.search(r"Size: (\d+)<", item.description)
                         size = int(desc_match.group(1))
                         url = item.link
-                        resultlist.append((title, size, url, provider, 'torrent', True))
+                        resultlist.append(Result(title, size, url, provider, 'torrent', True))
                         logger.info('Found %s. Size: %s', title, helpers.bytes_to_mb(size))
                     except Exception as e:
                         logger.error(
@@ -1589,11 +1625,16 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
             for torrent in match_torrents:
                 if not torrent.file_path:
                     torrent.group.update_group_data()  # will load the file_path for the individual torrents
-                resultlist.append((torrent.file_path,
+                resultlist.append(
+                    Result(
+                        torrent.file_path,
                         torrent.size,
                         orpheusobj.generate_torrent_link(torrent.id),
                         provider,
-                                   'torrent', True))
+                        'torrent',
+                        True
+                    )
+                )
 
     # Redacted - Using same logic as What.CD as it's also Gazelle, so should really make this into something reusable
     if headphones.CONFIG.REDACTED:
@@ -1690,11 +1731,16 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
                 if not torrent.file_path:
                     torrent.group.update_group_data()  # will load the file_path for the individual torrents
                 use_token = headphones.CONFIG.REDACTED_USE_FLTOKEN and torrent.can_use_token
-                resultlist.append((torrent.file_path,
+                resultlist.append(
+                    Result(
+                        torrent.file_path,
                         torrent.size,
                         redobj.generate_torrent_link(torrent.id, use_token),
                         provider,
-                                   'torrent', True))
+                        'torrent',
+                        True
+                    )
+                )
 
     # Pirate Bay
     if headphones.CONFIG.PIRATEBAY:
@@ -1768,7 +1814,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
                             logger.info('%s is larger than the maxsize or has too little seeders for this category, '
                                         'skipping. (Size: %i bytes, Seeders: %i)' % (title, size, int(seeds)))
 
-                        resultlist.append((title, size, url, provider, "torrent", match))
+                        resultlist.append(Result(title, size, url, provider, "torrent", match))
                     except Exception as e:
                         logger.error("An unknown error occurred in the Pirate Bay parser: %s" % e)
 
@@ -1822,7 +1868,7 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
                             logger.info('%s is larger than the maxsize or has too little seeders for this category, '
                                         'skipping. (Size: %i bytes, Seeders: %i)' % (title, size, int(seeds)))
 
-                        resultlist.append((title, size, url, provider, "torrent", match))
+                        resultlist.append(Result(title, size, url, provider, "torrent", match))
                     except Exception as e:
                         logger.error(
                             "An unknown error occurred in the Old Pirate Bay parser: %s" % e)
@@ -1830,10 +1876,9 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
     # attempt to verify that this isn't a substring result
     # when looking for "Foo - Foo" we don't want "Foobar"
     # this should be less of an issue when it isn't a self-titled album so we'll only check vs artist
-    results = [result for result in resultlist if verifyresult(result[0], artistterm, term, losslessOnly)]
+    results = [result for result in resultlist if verifyresult(result.title, artistterm, term, losslessOnly)]
 
     # Additional filtering for size etc
-    # if results and not choose_specific_download and result[3] != 'Orpheus.network':
     if results and not choose_specific_download:
         results = more_filtering(results, album, albumlength, new)
 
@@ -1845,60 +1890,74 @@ def searchTorrent(album, new=False, losslessOnly=False, albumlength=None,
 
 def preprocess(resultlist):
     for result in resultlist:
-        if result[4] == 'torrent':
 
-            headers = {}
-
-            # rutracker always needs the torrent data
-            if result[3] == 'rutracker.org':
-                return ruobj.get_torrent_data(result[2]), result
-
-            # Jackett sometimes redirects
-            jackett_content = None
-            if result[3].startswith('Jackett_') or 'torznab' in result[3].lower():
-                r = request.request_response(url=result[2], headers=headers, allow_redirects=False)
-                if r:
-                    jackett_content = r.content
-                    link = r.headers.get('Location')
-                    if link and link != result[2]:
-                        if link.startswith('magnet:'):
-                            result = (result[0], result[1], link, result[3], "magnet", result[5])
-                            return "d10:magnet-uri%d:%se" % (len(link), link), result
+        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:
-                            result = (result[0], result[1], link, result[3], result[4], result[5])
-                            return True, result
+            headers = {'User-Agent': USER_AGENT}
 
+        if result.kind == 'torrent':
             # Get out of here if we're using Transmission or Deluge
             # if not a magnet link still need the .torrent to generate hash... uTorrent support labeling
             if headphones.CONFIG.TORRENT_DOWNLOADER in [1, 3]:
                 return True, result
 
             # Get out of here if it's a magnet link
-            if result[2].lower().startswith("magnet:"):
+            if result.url.lower().startswith("magnet:"):
                 return True, result
 
+            # rutracker always needs the torrent data
+            if result.provider == 'rutracker.org':
+                return ruobj.get_torrent_data(result.url), result
+
+            # Jackett sometimes redirects
+            if result.provider.startswith('Jackett_') or 'torznab' in result.provider.lower():
+                r = request.request_response(url=result.url, headers=headers, allow_redirects=False)
+                if r:
+                    link = r.headers.get('Location')
+                    if link and link != result.url:
+                        if link.startswith('magnet:'):
+                            result = Result(
+                                result.url,
+                                result.size,
+                                link,
+                                result.provider,
+                                "magnet",
+                                result.matches
+                            )
+                            return "d10:magnet-uri%d:%se" % (len(link), link), result
+                        else:
+                            result = Result(
+                                result.url,
+                                result.size,
+                                link,
+                                result.provider,
+                                result.kind,
+                                result.matches
+                            )
+                            return True, result
+                    else:
+                        return r.content, result
+
+
             # Download the torrent file
+            return request.request_content(url=result.url, headers=headers), result
 
-            if result[3] == 'Orpheus.network':
-                headers['User-Agent'] = 'Headphones'
-            elif result[3] == 'Redacted':
-                headers['User-Agent'] = 'Headphones'
-            elif result[3] == "The Pirate Bay" or result[3] == "Old Pirate Bay":
-                headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2243.2 Safari/537.36'
-            elif jackett_content:
-                return jackett_content, result
-
-            return request.request_content(url=result[2], headers=headers), result
-
-        if result[4] == 'magnet':
-            magnet_link = result[2]
+        if result.kind == 'magnet':
+            magnet_link = result.url
             return "d10:magnet-uri%d:%se" % (len(magnet_link), magnet_link), result
 
         else:
-            headers = {'User-Agent': USER_AGENT}
-
-            if result[3] == 'headphones':
-                return request.request_content(url=result[2], headers=headers,
-                                               auth=(headphones.CONFIG.HPUSER, headphones.CONFIG.HPPASS)), result
+            if result.provider == 'headphones':
+                return request.request_content(
+                    url=result.url,
+                    headers=headers,
+                    auth=(headphones.CONFIG.HPUSER, headphones.CONFIG.HPPASS)
+                    ), result
             else:
-                return request.request_content(url=result[2], headers=headers), result
+                return request.request_content(url=result.url, headers=headers), result

headphones/transmission.py

@@ -205,5 +205,4 @@ def torrentAction(method, arguments):
             continue
 
         resp_json = response.json()
-        print(resp_json)
         return resp_json

headphones/types.py

@@ -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

headphones/webserve.py

@@ -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),
@@ -1471,8 +1461,8 @@ class WebInterface(object):
             "use_waffles", "use_rutracker",
             "use_orpheus", "use_redacted", "redacted_use_fltoken", "preferred_bitrate_allow_lossless",
             "detect_bitrate", "ignore_clean_releases", "freeze_db", "cue_split", "move_files",
-            "rename_files", "correct_metadata", "cleanup_files", "keep_nfo", "add_album_art",
-            "embed_album_art", "embed_lyrics",
+            "rename_files", "rename_single_disc_ignore", "correct_metadata", "cleanup_files",
+            "keep_nfo", "add_album_art", "embed_album_art", "embed_lyrics",
             "replace_existing_folders", "keep_original_folder", "file_underscores",
             "include_extras", "official_releases_only",
             "wait_until_release_date", "autowant_upcoming", "autowant_all",

lib/pkg_resources/tests/data/my-test-package-source/setup.py

@@ -1,6 +0,0 @@
-import setuptools
-setuptools.setup(
-    name="my-test-package",
-    version="1.0",
-    zip_safe=True,
-)

lib/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/PKG-INFO

@@ -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

lib/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/SOURCES.txt

@@ -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

lib/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/dependency_links.txt

@@ -1 +0,0 @@
-

lib/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/top_level.txt

@@ -1 +0,0 @@
-

lib/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/zip-safe

@@ -1 +0,0 @@
-

lib/pkg_resources/tests/test_find_distributions.py

@@ -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)

lib/pkg_resources/tests/test_markers.py

@@ -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

lib/pkg_resources/tests/test_pkg_resources.py

@@ -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)

lib/pkg_resources/tests/test_resources.py

@@ -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

lib/pkg_resources/tests/test_working_set.py

@@ -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)

lib/tempora/tests/test_schedule.py

@@ -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

lib/tempora/tests/test_timing.py

@@ -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

lib/zipp.py

@@ -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)