commit: f1f43b235cec228fa0e478f389225729ae2ea3f8
Author: Sam James <sam <AT> gentoo <DOT> org>
AuthorDate: Mon Aug 7 00:05:27 2023 +0000
Commit: Sam James <sam <AT> gentoo <DOT> org>
CommitDate: Mon Aug 7 00:06:39 2023 +0000
URL: https://gitweb.gentoo.org/proj/mirrorselect.git/commit/?id=f1f43b23
Reformat with `black`
Signed-off-by: Sam James <sam <AT> gentoo.org>
.github/workflows/black.yml | 10 +
bin/mirrorselect | 29 +-
mirrorselect/configs.py | 301 ++++++------
mirrorselect/extractor.py | 184 ++++----
mirrorselect/main.py | 761 ++++++++++++++++--------------
mirrorselect/mirrorparser3.py | 117 ++---
mirrorselect/output.py | 187 ++++----
mirrorselect/selectors.py | 1021 ++++++++++++++++++++++-------------------
mirrorselect/version.py | 1 -
setup.py | 163 ++++---
tests/test_write_make_conf.py | 92 ++--
11 files changed, 1532 insertions(+), 1334 deletions(-)
diff --git a/.github/workflows/black.yml b/.github/workflows/black.yml
new file mode 100644
index 0000000..b04fb15
--- /dev/null
+++ b/.github/workflows/black.yml
@@ -0,0 +1,10 @@
+name: Lint
+
+on: [push, pull_request]
+
+jobs:
+ lint:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v2
+ - uses: psf/black@stable
diff --git a/bin/mirrorselect b/bin/mirrorselect
index 7f019e3..3440935 100755
--- a/bin/mirrorselect
+++ b/bin/mirrorselect
@@ -32,29 +32,28 @@ import sys
# This block ensures that ^C interrupts are handled quietly.
try:
- import signal
+ import signal
- def exithandler(signum,frame):
- signal.signal(signal.SIGINT, signal.SIG_IGN)
- signal.signal(signal.SIGTERM, signal.SIG_IGN)
- print('Caught signal %s. Exiting' % signum)
- sys.exit(1)
+ def exithandler(signum, frame):
+ signal.signal(signal.SIGINT, signal.SIG_IGN)
+ signal.signal(signal.SIGTERM, signal.SIG_IGN)
+ print("Caught signal %s. Exiting" % signum)
+ sys.exit(1)
- signal.signal(signal.SIGINT, exithandler)
- signal.signal(signal.SIGTERM, exithandler)
- signal.signal(signal.SIGPIPE, signal.SIG_DFL)
+ signal.signal(signal.SIGINT, exithandler)
+ signal.signal(signal.SIGTERM, exithandler)
+ signal.signal(signal.SIGPIPE, signal.SIG_DFL)
except KeyboardInterrupt:
- print()
- sys.exit(1)
+ print()
+ sys.exit(1)
from mirrorselect.main import MirrorSelect
try:
- MirrorSelect().main(sys.argv)
+ MirrorSelect().main(sys.argv)
except KeyboardInterrupt:
- print("Aborted.")
- sys.exit(130)
+ print("Aborted.")
+ sys.exit(130)
sys.exit(0)
-
diff --git a/mirrorselect/configs.py b/mirrorselect/configs.py
index 303fc1d..39dcdab 100644
--- a/mirrorselect/configs.py
+++ b/mirrorselect/configs.py
@@ -38,159 +38,166 @@ letters = string.ascii_letters
def get_make_conf_path(EPREFIX):
- # try the newer make.conf location
- config_path = EPREFIX + '/etc/portage/make.conf'
- if not os.access(config_path, os.F_OK):
- # check if the old location is what is used
- if os.access(EPREFIX + '/etc/make.conf', os.F_OK):
- config_path = EPREFIX + '/etc/make.conf'
- return config_path
+ # try the newer make.conf location
+ config_path = EPREFIX + "/etc/portage/make.conf"
+ if not os.access(config_path, os.F_OK):
+ # check if the old location is what is used
+ if os.access(EPREFIX + "/etc/make.conf", os.F_OK):
+ config_path = EPREFIX + "/etc/make.conf"
+ return config_path
def write_make_conf(output, config_path, var, mirror_string):
- """Write the make.conf target changes
-
- @param output: file, or output to print messages to
- @param mirror_string: "var='hosts'" string to write
- @param config_path; string
- """
- output.write('\n')
- output.print_info('Modifying %s with new mirrors...\n' % config_path)
- try:
- config = open(config_path)
- output.write('\tReading make.conf\n')
- lines = config.readlines()
- config.close()
- output.write('\tMoving to %s.backup\n' % config_path)
- shutil.move(config_path, config_path + '.backup')
- except OSError:
- lines = []
-
- with open(config_path + '.backup') as f:
- lex = shlex.shlex(f, posix=True)
- lex.wordchars = string.digits + letters + r"~!@#$%*_\:;?,./-+{}"
- lex.quotes = "\"'"
- while True:
- key = lex.get_token()
- if key is None:
- break
-
- if key == var:
- begin_line = lex.lineno
- equ = lex.get_token()
- if equ is None:
- break
- if equ != '=':
- continue
-
- val = lex.get_token()
- if val is None:
- break
- end_line = lex.lineno
-
- new_lines = []
- for index, line in enumerate(lines):
- if index < begin_line - 1 or index >=
end_line - 1:
- new_lines.append(line)
- lines = new_lines
- break
-
- lines.append(mirror_string)
-
- output.write('\tWriting new %s\n' % config_path)
-
- config = open(config_path, 'w')
-
- for line in lines:
- config.write(line)
- config.write('\n')
- config.close()
-
- output.print_info('Done.\n')
+ """Write the make.conf target changes
+
+ @param output: file, or output to print messages to
+ @param mirror_string: "var='hosts'" string to write
+ @param config_path; string
+ """
+ output.write("\n")
+ output.print_info("Modifying %s with new mirrors...\n" % config_path)
+ try:
+ config = open(config_path)
+ output.write("\tReading make.conf\n")
+ lines = config.readlines()
+ config.close()
+ output.write("\tMoving to %s.backup\n" % config_path)
+ shutil.move(config_path, config_path + ".backup")
+ except OSError:
+ lines = []
+
+ with open(config_path + ".backup") as f:
+ lex = shlex.shlex(f, posix=True)
+ lex.wordchars = string.digits + letters + r"~!@#$%*_\:;?,./-+{}"
+ lex.quotes = "\"'"
+ while True:
+ key = lex.get_token()
+ if key is None:
+ break
+
+ if key == var:
+ begin_line = lex.lineno
+ equ = lex.get_token()
+ if equ is None:
+ break
+ if equ != "=":
+ continue
+
+ val = lex.get_token()
+ if val is None:
+ break
+ end_line = lex.lineno
+
+ new_lines = []
+ for index, line in enumerate(lines):
+ if index < begin_line - 1 or index >= end_line - 1:
+ new_lines.append(line)
+ lines = new_lines
+ break
+
+ lines.append(mirror_string)
+
+ output.write("\tWriting new %s\n" % config_path)
+
+ config = open(config_path, "w")
+
+ for line in lines:
+ config.write(line)
+ config.write("\n")
+ config.close()
+
+ output.print_info("Done.\n")
def write_repos_conf(output, config_path, var, value):
- """Saves the new var value to a ConfigParser style file
-
- @param output: file, or output to print messages to
- @param config_path; string
- @param var: string; the variable to save teh value to.
- @param value: string, the value to set var to
- """
- try:
- from configparser import ConfigParser
- except ImportError:
- from ConfigParser import ConfigParser
- config = ConfigParser()
- config.read(config_path)
- if config.has_option('gentoo', var):
- config.set('gentoo', var, value)
- with open(config_path, 'w') as configfile:
- config.write(configfile)
- else:
- output.print_err("write_repos_conf(): Failed to find section
'gentoo',"
- " variable: %s\nChanges NOT SAVED" %var)
+ """Saves the new var value to a ConfigParser style file
+
+ @param output: file, or output to print messages to
+ @param config_path; string
+ @param var: string; the variable to save teh value to.
+ @param value: string, the value to set var to
+ """
+ try:
+ from configparser import ConfigParser
+ except ImportError:
+ from ConfigParser import ConfigParser
+ config = ConfigParser()
+ config.read(config_path)
+ if config.has_option("gentoo", var):
+ config.set("gentoo", var, value)
+ with open(config_path, "w") as configfile:
+ config.write(configfile)
+ else:
+ output.print_err(
+ "write_repos_conf(): Failed to find section 'gentoo',"
+ " variable: %s\nChanges NOT SAVED" % var
+ )
def get_filesystem_mirrors(output, config_path):
- """Read the current mirrors and retain mounted filesystems mirrors
-
- @param config_path: string
- @rtype list
- """
-
- def get_token(lex):
- '''internal function for getting shlex tokens
- '''
- try:
- val = lex.get_token()
- except ValueError:
- val = None
- return val
-
- fsmirrors = []
-
- var = 'GENTOO_MIRRORS'
-
- output.write('get_filesystem_mirrors(): config_path = %s\n' %
config_path, 2)
- try:
- f = open(config_path)
- except OSError:
- return fsmirrors
-
- """ Search for 'var' in make.conf and extract value """
- lex = shlex.shlex(f, posix=True)
- lex.wordchars = string.digits + letters + r"~!@#$%*_\:;?,./-+{}"
- lex.quotes = "\"'"
- p = re.compile('rsync://|http://|https://|ftp://', re.IGNORECASE)
- while 1:
- key = get_token(lex)
- #output.write('get_filesystem_mirrors(): processing key = %s\n'
% key, 2)
-
- if key == var:
- equ = get_token(lex)
- if (equ != '='):
- break
-
- val = get_token(lex)
- if val is None:
- break
-
- """ Look for mounted filesystem in value """
- mirrorlist = val.rsplit()
- output.write('get_filesystem_mirrors(): mirrorlist =
%s\n' % mirrorlist, 2)
- for mirror in mirrorlist:
- if (p.match(mirror) == None):
- if os.access(mirror, os.F_OK):
-
output.write('get_filesystem_mirrors(): found file system mirror = %s\n' %
mirror, 2)
- fsmirrors.append(mirror)
- else:
-
output.write('get_filesystem_mirrors(): ignoring non-accessible mirror = %s\n'
% mirror, 2)
- break
- elif key is None:
- break
-
- output.write('get_filesystem_mirrors(): fsmirrors = %s\n' % fsmirrors,
2)
- return fsmirrors
-
-
+ """Read the current mirrors and retain mounted filesystems mirrors
+
+ @param config_path: string
+ @rtype list
+ """
+
+ def get_token(lex):
+ """internal function for getting shlex tokens"""
+ try:
+ val = lex.get_token()
+ except ValueError:
+ val = None
+ return val
+
+ fsmirrors = []
+
+ var = "GENTOO_MIRRORS"
+
+ output.write("get_filesystem_mirrors(): config_path = %s\n" % config_path,
2)
+ try:
+ f = open(config_path)
+ except OSError:
+ return fsmirrors
+
+ """ Search for 'var' in make.conf and extract value """
+ lex = shlex.shlex(f, posix=True)
+ lex.wordchars = string.digits + letters + r"~!@#$%*_\:;?,./-+{}"
+ lex.quotes = "\"'"
+ p = re.compile("rsync://|http://|https://|ftp://";, re.IGNORECASE)
+ while 1:
+ key = get_token(lex)
+ # output.write('get_filesystem_mirrors(): processing key = %s\n' %
key, 2)
+
+ if key == var:
+ equ = get_token(lex)
+ if equ != "=":
+ break
+
+ val = get_token(lex)
+ if val is None:
+ break
+
+ """ Look for mounted filesystem in value """
+ mirrorlist = val.rsplit()
+ output.write("get_filesystem_mirrors(): mirrorlist = %s\n" %
mirrorlist, 2)
+ for mirror in mirrorlist:
+ if p.match(mirror) == None:
+ if os.access(mirror, os.F_OK):
+ output.write(
+ "get_filesystem_mirrors(): found file system
mirror = %s\n"
+ % mirror,
+ 2,
+ )
+ fsmirrors.append(mirror)
+ else:
+ output.write(
+ "get_filesystem_mirrors(): ignoring non-accessible
mirror = %s\n"
+ % mirror,
+ 2,
+ )
+ break
+ elif key is None:
+ break
+
+ output.write("get_filesystem_mirrors(): fsmirrors = %s\n" % fsmirrors, 2)
+ return fsmirrors
diff --git a/mirrorselect/extractor.py b/mirrorselect/extractor.py
index f9e4076..8aa495c 100644
--- a/mirrorselect/extractor.py
+++ b/mirrorselect/extractor.py
@@ -33,95 +33,97 @@ from mirrorselect.version import version
USERAGENT = "Mirrorselect-" + version
-class Extractor:
- """The Extractor employs a MirrorParser3 object to get a list of valid
- mirrors, and then filters them. Only the mirrors that should be tested,
- based on user input are saved. They will be in the hosts attribute."""
-
- def __init__(self, list_url, options, output):
- self.output = output
- self.output.print_info('Using url: %s\n' % list_url)
- filters = {}
- for opt in ["country", "region"]:
- value = getattr(options, opt)
- if value is not None:
- filters[opt] = value
- self.output.print_info('Limiting test to
"%s=%s" hosts. \n'
- %(opt, value))
- for opt in ["ftp", "http", "https"]:
- if getattr(options, opt):
- filters["proto"] = opt
- self.output.print_info('Limiting test to %s
hosts. \n' % opt )
-
- self.proxies = {}
-
- for proxy in ['http_proxy', 'https_proxy']:
- prox = proxy.split('_')[0]
- if options.proxy and prox + ":" in options.proxy:
- self.proxies[prox] = options.proxy
- elif os.getenv(proxy):
- self.proxies[prox] = os.getenv(proxy)
-
- parser = MirrorParser3()
- self.hosts = []
-
- self.unfiltered_hosts = self.getlist(parser, list_url)
-
- self.hosts = self.filter_hosts(filters, self.unfiltered_hosts)
-
- self.output.write('Extractor(): fetched mirrors,'
- ' %s hosts after filtering\n' %
len(self.hosts), 2)
-
-
- @staticmethod
- def filter_hosts(filters, hosts):
- """Filter the hosts to the criteria passed in
- Return the filtered list
- """
- if not len(filters):
- return hosts
- filtered = []
- for uri, data in hosts:
- good = True
- for f in filters:
- if data[f] != filters[f]:
- good = False
- continue
- if good:
- filtered.append((uri, data))
- return filtered
-
-
- def getlist(self, parser, url):
- """
- Uses the supplied parser to get a list of urls.
- Takes a parser object, url, and filering options.
- """
-
- self.output.write('getlist(): fetching ' + url + '\n', 2)
-
- self.output.print_info('Downloading a list of mirrors...\n')
-
- # setup the ssl-fetch ouptut map
- connector_output = {
- 'info':self.output.write,
- 'debug': self.output.write,
- 'error': self.output.print_err,
- 'kwargs-info': {'level': 2},
- 'kwargs-debug': {'level':2},
- 'kwargs-error': {'level':0},
- }
-
- fetcher = Connector(connector_output, self.proxies, USERAGENT)
- success, mirrorlist, timestamp = fetcher.fetch_content(url,
climit=60)
- parser.parse(mirrorlist)
-
- if (not mirrorlist) or len(parser.tuples()) == 0:
- self.output.print_err('Could not get mirror list. '
- 'Check your internet connection.')
-
- self.output.write(' Got %d mirrors.\n' % len(parser.tuples()))
-
- return parser.tuples()
-
+class Extractor:
+ """The Extractor employs a MirrorParser3 object to get a list of valid
+ mirrors, and then filters them. Only the mirrors that should be tested,
+ based on user input are saved. They will be in the hosts attribute."""
+
+ def __init__(self, list_url, options, output):
+ self.output = output
+ self.output.print_info("Using url: %s\n" % list_url)
+ filters = {}
+ for opt in ["country", "region"]:
+ value = getattr(options, opt)
+ if value is not None:
+ filters[opt] = value
+ self.output.print_info(
+ 'Limiting test to "%s=%s" hosts. \n' % (opt, value)
+ )
+ for opt in ["ftp", "http", "https"]:
+ if getattr(options, opt):
+ filters["proto"] = opt
+ self.output.print_info("Limiting test to %s hosts. \n" % opt)
+
+ self.proxies = {}
+
+ for proxy in ["http_proxy", "https_proxy"]:
+ prox = proxy.split("_")[0]
+ if options.proxy and prox + ":" in options.proxy:
+ self.proxies[prox] = options.proxy
+ elif os.getenv(proxy):
+ self.proxies[prox] = os.getenv(proxy)
+
+ parser = MirrorParser3()
+ self.hosts = []
+
+ self.unfiltered_hosts = self.getlist(parser, list_url)
+
+ self.hosts = self.filter_hosts(filters, self.unfiltered_hosts)
+
+ self.output.write(
+ "Extractor(): fetched mirrors,"
+ " %s hosts after filtering\n" % len(self.hosts),
+ 2,
+ )
+
+ @staticmethod
+ def filter_hosts(filters, hosts):
+ """Filter the hosts to the criteria passed in
+ Return the filtered list
+ """
+ if not len(filters):
+ return hosts
+ filtered = []
+ for uri, data in hosts:
+ good = True
+ for f in filters:
+ if data[f] != filters[f]:
+ good = False
+ continue
+ if good:
+ filtered.append((uri, data))
+ return filtered
+
+ def getlist(self, parser, url):
+ """
+ Uses the supplied parser to get a list of urls.
+ Takes a parser object, url, and filering options.
+ """
+
+ self.output.write("getlist(): fetching " + url + "\n", 2)
+
+ self.output.print_info("Downloading a list of mirrors...\n")
+
+ # setup the ssl-fetch ouptut map
+ connector_output = {
+ "info": self.output.write,
+ "debug": self.output.write,
+ "error": self.output.print_err,
+ "kwargs-info": {"level": 2},
+ "kwargs-debug": {"level": 2},
+ "kwargs-error": {"level": 0},
+ }
+
+ fetcher = Connector(connector_output, self.proxies, USERAGENT)
+ success, mirrorlist, timestamp = fetcher.fetch_content(url, climit=60)
+ parser.parse(mirrorlist)
+
+ if (not mirrorlist) or len(parser.tuples()) == 0:
+ self.output.print_err(
+ "Could not get mirror list. " "Check your internet connection."
+ )
+
+ self.output.write(" Got %d mirrors.\n" % len(parser.tuples()))
+
+ return parser.tuples()
diff --git a/mirrorselect/main.py b/mirrorselect/main.py
index f003b8c..8180143 100755
--- a/mirrorselect/main.py
+++ b/mirrorselect/main.py
@@ -36,15 +36,19 @@ from mirrorselect.mirrorparser3 import MIRRORS_3_XML,
MIRRORS_RSYNC_DATA
from mirrorselect.output import Output, ColoredFormatter
from mirrorselect.selectors import Deep, Shallow, Interactive
from mirrorselect.extractor import Extractor
-from mirrorselect.configs import (get_make_conf_path, write_make_conf,
- write_repos_conf, get_filesystem_mirrors)
+from mirrorselect.configs import (
+ get_make_conf_path,
+ write_make_conf,
+ write_repos_conf,
+ get_filesystem_mirrors,
+)
from mirrorselect.version import version
# eprefix compatibility
try:
- from portage.const import rootuid
+ from portage.const import rootuid
except ImportError:
- rootuid = 0
+ rootuid = 0
# establish the eprefix, initially set so eprefixify can
@@ -53,341 +57,420 @@ EPREFIX = "@GENTOO_PORTAGE_EPREFIX@"
# check and set it if it wasn't
if "GENTOO_PORTAGE_EPREFIX" in EPREFIX:
- EPREFIX = ''
+ EPREFIX = ""
class MirrorSelect:
- '''Main operational class'''
-
- def __init__(self, output=None):
- '''MirrorSelect class init
-
- @param output: mirrorselect.output.Ouptut() class instance
- or None for the default instance
- '''
- self.output = output or Output()
-
-
- @staticmethod
- def _have_bin(name):
- """Determines whether a particular binary is available
- on the host system. It searches in the PATH environment
- variable paths.
-
- @param name: string, binary name to search for
- @rtype: string or None
- """
- for path_dir in os.environ.get("PATH", "").split(":"):
- if not path_dir:
- continue
- file_path = os.path.join(path_dir, name)
- if os.path.isfile(file_path) and os.access(file_path,
os.X_OK):
- return file_path
- return None
-
-
- def change_config(self, hosts, out, config_path, sync=False):
- """Writes the config changes to the given file, or to stdout.
-
- @param hosts: list of host urls to write
- @param out: boolean, used to redirect output to stdout
- @param config_path; string
- @param sync: boolean, used to switch between sync-uri
repos.conf target,
- and GENTOO_MIRRORS make.conf variable target
- """
- if sync:
- var = "sync-uri"
- else:
- var = 'GENTOO_MIRRORS'
-
- for i in range(0, len(hosts)):
- if isinstance(hosts[i], bytes):
- hosts[i] = hosts[i].decode('utf-8')
-
- if var == "sync-uri" and out:
- mirror_string = '{} = {}'.format(var, ' '.join(hosts))
- else:
- mirror_string = '{}="{}"'.format(var, ' \\\n
'.join(hosts))
-
- if out:
- self.write_to_output(mirror_string)
- elif var == "sync-uri":
- write_repos_conf(self.output, config_path, var, '
'.join(hosts))
- else:
- write_make_conf(self.output, config_path, var,
mirror_string)
-
-
- @staticmethod
- def write_to_output(mirror_string):
- print()
- print(mirror_string)
- sys.exit(0)
-
-
- def _parse_args(self, argv, config_path):
- """
- Does argument parsing and some sanity checks.
- Returns an optparse Options object.
-
- The descriptions, grouping, and possibly the amount sanity
checking
- need some finishing touches.
- """
- desc = "\n".join((
- self.output.white("examples:"),
- "",
- self.output.white(" automatic:"),
- " # mirrorselect -s5",
- " # mirrorselect -s3 -b10 -o >>
/mnt/gentoo/etc/portage/make.conf",
- " # mirrorselect -D -s4",
- "",
- self.output.white(" interactive:"),
- " # mirrorselect -i -r",
- ))
-
- def set_servers(option, opt_str, value, parser):
- set_servers.user_configured = True
- setattr(parser.values, option.dest, value)
-
- parser = OptionParser(
- formatter=ColoredFormatter(self.output),
description=desc,
- version='Mirrorselect version: %s' % version)
-
- group = parser.add_option_group("Main modes")
- group.add_option(
- "-a", "--all_mirrors", action="store_true",
default=False,
- help="This will present a list of all filtered search
results "
- "to make it possible to select mirrors you wish to use.
"
- " For the -r, --rsync option, it will select the
rotation server "
- "only. As multiple rsync URL's are not supported.")
- group.add_option(
- "-D", "--deep", action="store_true", default=False,
- help="Deep mode. This is used to give a more accurate "
- "speed test. It will download a 100k file from "
- "each server. Because of this you should only use "
- "this option if you have a good connection.")
- group.add_option(
- "-i", "--interactive", action="store_true",
default=False,
- help="Interactive Mode, this will present a list "
- "to make it possible to select mirrors you wish to
use.")
-
- group = parser.add_option_group(
- "Server type selection (choose at most one)")
- group.add_option(
- "-c", "--country", action="store", default=None,
- help="only use mirrors from the specified country "
- "NOTE: Names with a space must be quoted "
- "eg.: -c 'South Korea'")
- group.add_option(
- "-F", "--ftp", action="store_true", default=False,
- help="ftp only mode. Will not consider hosts of other "
- "types.")
- group.add_option(
- "-H", "--http", action="store_true", default=False,
- help="http only mode. Will not consider hosts of other
types")
- group.add_option(
- "-S", "--https", action="store_true", default=False,
- help="https only mode. Will not consider hosts of other
types")
- group.add_option(
- "-r", "--rsync", action="store_true", default=False,
- help="rsync mode. Allows you to interactively select
your"
- " rsync mirror. Requires -i or -a to be used.")
- group.add_option(
- "-R", "--region", action="store", default=None,
- help="only use mirrors from the specified region "
- "NOTE: Names with a space must be quoted "
- "eg.: -R 'North America'")
- group.add_option(
- "-4", "--ipv4", action="store_true", default=False,
- help="only use IPv4")
- group.add_option(
- "-6", "--ipv6", action="store_true", default=False,
- help="only use IPv6")
-
- group = parser.add_option_group("Other options")
- group.add_option(
- "-b", "--blocksize", action="store", type="int",
- help="This is to be used in automatic mode "
- "and will split the hosts into blocks of BLOCKSIZE for "
- "use with netselect. This is required for certain "
- "routers which block 40+ requests at any given time. "
- "Recommended parameters to pass are: -s3 -b10")
- group.add_option(
- "-d", "--debug", action="store", type="int",
dest="verbosity",
- default=1, help="debug mode, pass in the debug level
[1-9]")
- group.add_option(
- "-f", "--file", action="store",
default='mirrorselect-test',
- help="An alternate file to download for deep testing. "
- "Please choose the file carefully as to not
abuse the system "
- "by selecting an overly large size file. You
must also "
- " use the -m, --md5 option.")
- group.add_option(
- "-m", "--md5", action="store",
- default='bdf077b2e683c506bf9e8f2494eeb044',
- help="An alternate file md5sum value used to compare
the downloaded "
- "file against for deep testing.")
- group.add_option(
- "-o", "--output", action="store_true", default=False,
- help="Output Only Mode, this is especially useful "
- "when being used during installation, to redirect "
- "output to a file other than %s" % config_path)
- group.add_option(
- "-P", "--proxy", action="store",
- default=None,
- help="Proxy server to use if not the default proxy "
- "in the environment")
- group.add_option(
- "-q", "--quiet", action="store_const", const=0,
dest="verbosity",
- help="Quiet mode")
- group.add_option(
- "-s", "--servers", action="callback",
callback=set_servers,
- type="int", default=1, help="Specify Number of servers
for Automatic Mode "
- "to select. this is only valid for download mirrors. "
- "If this is not specified, a default of 1 is used.")
- group.add_option(
- "-t", "--timeout", action="store", type="int",
- default="10", help="Timeout for deep mode. Defaults to
10 seconds.")
- group.add_option(
- "-e", "--exclude", action="append", dest="exclude",
- default=None, help="Exclude host from mirrors list.")
-
-
-
- if len(argv) == 1:
- parser.print_help()
- sys.exit(1)
-
- options, args = parser.parse_args(argv[1:])
-
- # sanity checks
-
- # hack: check if more than one of these is set
- if options.http + options.https + options.ftp + options.rsync >
1:
- self.output.print_err('Choose at most one of -H, -S, -f
and -r')
-
- if options.ipv4 and options.ipv6:
- self.output.print_err('Choose at most one of --ipv4 and
--ipv6')
-
- if (options.ipv6 and not socket.has_ipv6) and not
options.interactive:
- options.ipv6 = False
- self.output.print_err('The --ipv6 option requires
python ipv6 support')
-
- if options.rsync and not (options.interactive or
options.all_mirrors):
- self.output.print_err('rsync servers can only be
selected with -i or -a')
-
- if options.all_mirrors and hasattr(set_servers,
'user_configured'):
- self.output.print_err('Choose at most one of -s or -a')
-
- if options.interactive and (
- options.deep or
- options.blocksize or
- options.servers > 1):
- self.output.print_err('Invalid option combination with
-i')
-
- if (not options.deep) and (not self._have_bin('netselect') ):
- self.output.print_err(
- 'You do not appear to have netselect on your
system. '
- 'You must use the -D flag')
-
- if (os.getuid() != rootuid) and not options.output:
- self.output.print_err('Must be root to write to %s!\n'
% config_path)
-
- if args:
- self.output.print_err('Unexpected arguments passed.')
-
- # return results
- return options
-
-
- def get_available_hosts(self, options):
- '''Returns a list of hosts suitable for consideration by a user
- based on user input
-
- @param options: parser.parse_args() options instance
- @rtype: list
- '''
- if options.rsync:
- self.output.write("using url: %s\n" %
MIRRORS_RSYNC_DATA, 2)
- hosts = Extractor(MIRRORS_RSYNC_DATA, options,
self.output).hosts
- else:
- self.output.write("using url: %s\n" % MIRRORS_3_XML, 2)
- hosts = Extractor(MIRRORS_3_XML, options,
self.output).hosts
-
- if options.exclude:
- hosts = [x for x in hosts if x[0] not in
options.exclude]
-
- return hosts
-
-
- def select_urls(self, hosts, options):
- '''Returns the list of selected host urls using
- the options passed in to run one of the three selector types.
- 1) Interactive ncurses dialog
- 2) Deep mode mirror selection.
- 3) (Shallow) Rapid server selection via netselect
-
- @param hosts: list of hosts to choose from
- @param options: parser.parse_args() options instance
- @rtype: list
- '''
- if options.interactive:
- selector = Interactive(hosts, options, self.output)
- elif options.deep:
- selector = Deep(hosts, options, self.output)
- else:
- selector = Shallow(hosts, options, self.output)
- return selector.urls
-
-
- def get_conf_path(self, rsync=False):
- '''Checks for the existance of repos.conf or make.conf in
/etc/portage/
- Failing that it checks for it in /etc/
- Failing in /etc/ it defaults to /etc/portage/make.conf
-
- @rtype: string
- '''
- if rsync:
- # repos.conf
- config_path = EPREFIX +
'/etc/portage/repos.conf/gentoo.conf'
- if not os.access(config_path, os.F_OK):
- self.output.write("Failed access to
gentoo.conf: "
- "%s\n" % os.access(config_path,
os.F_OK), 2)
- config_path = None
- return config_path
- return get_make_conf_path(EPREFIX)
-
-
- def main(self, argv):
- """Lets Rock!
-
- @param argv: list of command line arguments to parse
- """
- config_path = self.get_conf_path()
- options = self._parse_args(argv, config_path)
- self.output.verbosity = options.verbosity
- self.output.write("main(); config_path = %s\n" % config_path, 2)
-
- # reset config_path to find repos.conf/gentoo.conf
- if options.rsync:
- config_path = self.get_conf_path(options.rsync)
- self.output.write("main(); reset config_path = %s\n" %
config_path, 2)
- if not config_path:
- self.output.print_err("main(); Exiting due to missing
repos.conf/gentoo.conf file\n")
-
- fsmirrors = get_filesystem_mirrors(self.output,
- config_path)
-
- hosts = self.get_available_hosts(options)
-
- if options.all_mirrors:
- urls = sorted([url for url, args in list(hosts)])
- if options.rsync:
- urls = [urls[0]]
- else:
- urls = self.select_urls(hosts, options)
-
- if len(urls):
- self.change_config(fsmirrors + urls, options.output,
- config_path, options.rsync)
- else:
- self.output.write("No search results found. "
- "Check your filter settings and re-run
mirrorselect\n")
+ """Main operational class"""
+
+ def __init__(self, output=None):
+ """MirrorSelect class init
+
+ @param output: mirrorselect.output.Ouptut() class instance
+ or None for the default instance
+ """
+ self.output = output or Output()
+
+ @staticmethod
+ def _have_bin(name):
+ """Determines whether a particular binary is available
+ on the host system. It searches in the PATH environment
+ variable paths.
+
+ @param name: string, binary name to search for
+ @rtype: string or None
+ """
+ for path_dir in os.environ.get("PATH", "").split(":"):
+ if not path_dir:
+ continue
+ file_path = os.path.join(path_dir, name)
+ if os.path.isfile(file_path) and os.access(file_path, os.X_OK):
+ return file_path
+ return None
+
+ def change_config(self, hosts, out, config_path, sync=False):
+ """Writes the config changes to the given file, or to stdout.
+
+ @param hosts: list of host urls to write
+ @param out: boolean, used to redirect output to stdout
+ @param config_path; string
+ @param sync: boolean, used to switch between sync-uri repos.conf
target,
+ and GENTOO_MIRRORS make.conf variable target
+ """
+ if sync:
+ var = "sync-uri"
+ else:
+ var = "GENTOO_MIRRORS"
+
+ for i in range(0, len(hosts)):
+ if isinstance(hosts[i], bytes):
+ hosts[i] = hosts[i].decode("utf-8")
+
+ if var == "sync-uri" and out:
+ mirror_string = "{} = {}".format(var, " ".join(hosts))
+ else:
+ mirror_string = '{}="{}"'.format(var, " \\\n ".join(hosts))
+
+ if out:
+ self.write_to_output(mirror_string)
+ elif var == "sync-uri":
+ write_repos_conf(self.output, config_path, var, " ".join(hosts))
+ else:
+ write_make_conf(self.output, config_path, var, mirror_string)
+
+ @staticmethod
+ def write_to_output(mirror_string):
+ print()
+ print(mirror_string)
+ sys.exit(0)
+
+ def _parse_args(self, argv, config_path):
+ """
+ Does argument parsing and some sanity checks.
+ Returns an optparse Options object.
+
+ The descriptions, grouping, and possibly the amount sanity checking
+ need some finishing touches.
+ """
+ desc = "\n".join(
+ (
+ self.output.white("examples:"),
+ "",
+ self.output.white(" automatic:"),
+ " # mirrorselect -s5",
+ " # mirrorselect -s3 -b10 -o >>
/mnt/gentoo/etc/portage/make.conf",
+ " # mirrorselect -D -s4",
+ "",
+ self.output.white(" interactive:"),
+ " # mirrorselect -i -r",
+ )
+ )
+
+ def set_servers(option, opt_str, value, parser):
+ set_servers.user_configured = True
+ setattr(parser.values, option.dest, value)
+
+ parser = OptionParser(
+ formatter=ColoredFormatter(self.output),
+ description=desc,
+ version="Mirrorselect version: %s" % version,
+ )
+
+ group = parser.add_option_group("Main modes")
+ group.add_option(
+ "-a",
+ "--all_mirrors",
+ action="store_true",
+ default=False,
+ help="This will present a list of all filtered search results "
+ "to make it possible to select mirrors you wish to use. "
+ " For the -r, --rsync option, it will select the rotation server "
+ "only. As multiple rsync URL's are not supported.",
+ )
+ group.add_option(
+ "-D",
+ "--deep",
+ action="store_true",
+ default=False,
+ help="Deep mode. This is used to give a more accurate "
+ "speed test. It will download a 100k file from "
+ "each server. Because of this you should only use "
+ "this option if you have a good connection.",
+ )
+ group.add_option(
+ "-i",
+ "--interactive",
+ action="store_true",
+ default=False,
+ help="Interactive Mode, this will present a list "
+ "to make it possible to select mirrors you wish to use.",
+ )
+
+ group = parser.add_option_group("Server type selection (choose at most
one)")
+ group.add_option(
+ "-c",
+ "--country",
+ action="store",
+ default=None,
+ help="only use mirrors from the specified country "
+ "NOTE: Names with a space must be quoted "
+ "eg.: -c 'South Korea'",
+ )
+ group.add_option(
+ "-F",
+ "--ftp",
+ action="store_true",
+ default=False,
+ help="ftp only mode. Will not consider hosts of other " "types.",
+ )
+ group.add_option(
+ "-H",
+ "--http",
+ action="store_true",
+ default=False,
+ help="http only mode. Will not consider hosts of other types",
+ )
+ group.add_option(
+ "-S",
+ "--https",
+ action="store_true",
+ default=False,
+ help="https only mode. Will not consider hosts of other types",
+ )
+ group.add_option(
+ "-r",
+ "--rsync",
+ action="store_true",
+ default=False,
+ help="rsync mode. Allows you to interactively select your"
+ " rsync mirror. Requires -i or -a to be used.",
+ )
+ group.add_option(
+ "-R",
+ "--region",
+ action="store",
+ default=None,
+ help="only use mirrors from the specified region "
+ "NOTE: Names with a space must be quoted "
+ "eg.: -R 'North America'",
+ )
+ group.add_option(
+ "-4", "--ipv4", action="store_true", default=False, help="only use
IPv4"
+ )
+ group.add_option(
+ "-6", "--ipv6", action="store_true", default=False, help="only use
IPv6"
+ )
+
+ group = parser.add_option_group("Other options")
+ group.add_option(
+ "-b",
+ "--blocksize",
+ action="store",
+ type="int",
+ help="This is to be used in automatic mode "
+ "and will split the hosts into blocks of BLOCKSIZE for "
+ "use with netselect. This is required for certain "
+ "routers which block 40+ requests at any given time. "
+ "Recommended parameters to pass are: -s3 -b10",
+ )
+ group.add_option(
+ "-d",
+ "--debug",
+ action="store",
+ type="int",
+ dest="verbosity",
+ default=1,
+ help="debug mode, pass in the debug level [1-9]",
+ )
+ group.add_option(
+ "-f",
+ "--file",
+ action="store",
+ default="mirrorselect-test",
+ help="An alternate file to download for deep testing. "
+ "Please choose the file carefully as to not abuse the system "
+ "by selecting an overly large size file. You must also "
+ " use the -m, --md5 option.",
+ )
+ group.add_option(
+ "-m",
+ "--md5",
+ action="store",
+ default="bdf077b2e683c506bf9e8f2494eeb044",
+ help="An alternate file md5sum value used to compare the
downloaded "
+ "file against for deep testing.",
+ )
+ group.add_option(
+ "-o",
+ "--output",
+ action="store_true",
+ default=False,
+ help="Output Only Mode, this is especially useful "
+ "when being used during installation, to redirect "
+ "output to a file other than %s" % config_path,
+ )
+ group.add_option(
+ "-P",
+ "--proxy",
+ action="store",
+ default=None,
+ help="Proxy server to use if not the default proxy " "in the
environment",
+ )
+ group.add_option(
+ "-q",
+ "--quiet",
+ action="store_const",
+ const=0,
+ dest="verbosity",
+ help="Quiet mode",
+ )
+ group.add_option(
+ "-s",
+ "--servers",
+ action="callback",
+ callback=set_servers,
+ type="int",
+ default=1,
+ help="Specify Number of servers for Automatic Mode "
+ "to select. this is only valid for download mirrors. "
+ "If this is not specified, a default of 1 is used.",
+ )
+ group.add_option(
+ "-t",
+ "--timeout",
+ action="store",
+ type="int",
+ default="10",
+ help="Timeout for deep mode. Defaults to 10 seconds.",
+ )
+ group.add_option(
+ "-e",
+ "--exclude",
+ action="append",
+ dest="exclude",
+ default=None,
+ help="Exclude host from mirrors list.",
+ )
+
+ if len(argv) == 1:
+ parser.print_help()
+ sys.exit(1)
+
+ options, args = parser.parse_args(argv[1:])
+
+ # sanity checks
+
+ # hack: check if more than one of these is set
+ if options.http + options.https + options.ftp + options.rsync > 1:
+ self.output.print_err("Choose at most one of -H, -S, -f and -r")
+
+ if options.ipv4 and options.ipv6:
+ self.output.print_err("Choose at most one of --ipv4 and --ipv6")
+
+ if (options.ipv6 and not socket.has_ipv6) and not options.interactive:
+ options.ipv6 = False
+ self.output.print_err("The --ipv6 option requires python ipv6
support")
+
+ if options.rsync and not (options.interactive or options.all_mirrors):
+ self.output.print_err("rsync servers can only be selected with -i
or -a")
+
+ if options.all_mirrors and hasattr(set_servers, "user_configured"):
+ self.output.print_err("Choose at most one of -s or -a")
+
+ if options.interactive and (
+ options.deep or options.blocksize or options.servers > 1
+ ):
+ self.output.print_err("Invalid option combination with -i")
+
+ if (not options.deep) and (not self._have_bin("netselect")):
+ self.output.print_err(
+ "You do not appear to have netselect on your system. "
+ "You must use the -D flag"
+ )
+
+ if (os.getuid() != rootuid) and not options.output:
+ self.output.print_err("Must be root to write to %s!\n" %
config_path)
+
+ if args:
+ self.output.print_err("Unexpected arguments passed.")
+
+ # return results
+ return options
+
+ def get_available_hosts(self, options):
+ """Returns a list of hosts suitable for consideration by a user
+ based on user input
+
+ @param options: parser.parse_args() options instance
+ @rtype: list
+ """
+ if options.rsync:
+ self.output.write("using url: %s\n" % MIRRORS_RSYNC_DATA, 2)
+ hosts = Extractor(MIRRORS_RSYNC_DATA, options, self.output).hosts
+ else:
+ self.output.write("using url: %s\n" % MIRRORS_3_XML, 2)
+ hosts = Extractor(MIRRORS_3_XML, options, self.output).hosts
+
+ if options.exclude:
+ hosts = [x for x in hosts if x[0] not in options.exclude]
+
+ return hosts
+
+ def select_urls(self, hosts, options):
+ """Returns the list of selected host urls using
+ the options passed in to run one of the three selector types.
+ 1) Interactive ncurses dialog
+ 2) Deep mode mirror selection.
+ 3) (Shallow) Rapid server selection via netselect
+
+ @param hosts: list of hosts to choose from
+ @param options: parser.parse_args() options instance
+ @rtype: list
+ """
+ if options.interactive:
+ selector = Interactive(hosts, options, self.output)
+ elif options.deep:
+ selector = Deep(hosts, options, self.output)
+ else:
+ selector = Shallow(hosts, options, self.output)
+ return selector.urls
+
+ def get_conf_path(self, rsync=False):
+ """Checks for the existance of repos.conf or make.conf in /etc/portage/
+ Failing that it checks for it in /etc/
+ Failing in /etc/ it defaults to /etc/portage/make.conf
+
+ @rtype: string
+ """
+ if rsync:
+ # repos.conf
+ config_path = EPREFIX + "/etc/portage/repos.conf/gentoo.conf"
+ if not os.access(config_path, os.F_OK):
+ self.output.write(
+ "Failed access to gentoo.conf: "
+ "%s\n" % os.access(config_path, os.F_OK),
+ 2,
+ )
+ config_path = None
+ return config_path
+ return get_make_conf_path(EPREFIX)
+
+ def main(self, argv):
+ """Lets Rock!
+
+ @param argv: list of command line arguments to parse
+ """
+ config_path = self.get_conf_path()
+ options = self._parse_args(argv, config_path)
+ self.output.verbosity = options.verbosity
+ self.output.write("main(); config_path = %s\n" % config_path, 2)
+
+ # reset config_path to find repos.conf/gentoo.conf
+ if options.rsync:
+ config_path = self.get_conf_path(options.rsync)
+ self.output.write("main(); reset config_path = %s\n" %
config_path, 2)
+ if not config_path:
+ self.output.print_err(
+ "main(); Exiting due to missing repos.conf/gentoo.conf file\n"
+ )
+
+ fsmirrors = get_filesystem_mirrors(self.output, config_path)
+
+ hosts = self.get_available_hosts(options)
+
+ if options.all_mirrors:
+ urls = sorted([url for url, args in list(hosts)])
+ if options.rsync:
+ urls = [urls[0]]
+ else:
+ urls = self.select_urls(hosts, options)
+
+ if len(urls):
+ self.change_config(
+ fsmirrors + urls, options.output, config_path, options.rsync
+ )
+ else:
+ self.output.write(
+ "No search results found. "
+ "Check your filter settings and re-run mirrorselect\n"
+ )
diff --git a/mirrorselect/mirrorparser3.py b/mirrorselect/mirrorparser3.py
index 4023973..37245ff 100644
--- a/mirrorselect/mirrorparser3.py
+++ b/mirrorselect/mirrorparser3.py
@@ -28,58 +28,69 @@ Distributed under the terms of the GNU General Public
License v2
from xml.etree import ElementTree as ET
-MIRRORS_3_XML = 'https://api.gentoo.org/mirrors/distfiles.xml'
-MIRRORS_RSYNC_DATA = 'https://api.gentoo.org/mirrors/rsync.xml'
+MIRRORS_3_XML = "https://api.gentoo.org/mirrors/distfiles.xml";
+MIRRORS_RSYNC_DATA = "https://api.gentoo.org/mirrors/rsync.xml";
+
class MirrorParser3:
- def __init__(self, options=None):
- self._reset()
-
- def _reset(self):
- self._dict = {}
-
- def _get_proto(self, uri=None):
- if not uri: # Don't parse if empty
- return None;
- try:
- from urllib.parse import urlparse
- return urlparse(uri).scheme
- except Exception as e: # Add general exception to catch errors
- from mirrorselect.output import Output
- Output.write(('_get_proto(): Exception while parsing
the protocol '
- 'for URI %s: %s\n')% (uri, e), 2)
-
- def parse(self, text):
- self._reset()
- for mirrorgroup in ET.XML(text):
- for mirror in mirrorgroup:
- name = ''
- for e in mirror:
- if e.tag == 'name':
- name = e.text
- if e.tag == 'uri':
- uri = e.text
- self._dict[uri] = {
- "name": name,
- "country":
mirrorgroup.get("countryname"),
- "region":
mirrorgroup.get("region"),
- "ipv4": e.get("ipv4"),
- "ipv6": e.get("ipv6"),
- "proto":
e.get("protocol") or self._get_proto(uri),
- }
-
- def tuples(self):
- return [(url, args) for url, args in list(self._dict.items())]
-
- def uris(self):
- return [url for url, args in list(self._dict.items())]
-
-if __name__ == '__main__':
- import sys
- import urllib.request, urllib.parse, urllib.error
- parser = MirrorParser3()
- parser.parse(urllib.request.urlopen(MIRRORS_3_XML).read())
- print('===== tuples')
- print(parser.tuples())
- print('===== uris')
- print(parser.uris())
+ def __init__(self, options=None):
+ self._reset()
+
+ def _reset(self):
+ self._dict = {}
+
+ def _get_proto(self, uri=None):
+ if not uri: # Don't parse if empty
+ return None
+ try:
+ from urllib.parse import urlparse
+
+ return urlparse(uri).scheme
+ except Exception as e: # Add general exception to catch errors
+ from mirrorselect.output import Output
+
+ Output.write(
+ (
+ "_get_proto(): Exception while parsing the protocol "
+ "for URI %s: %s\n"
+ )
+ % (uri, e),
+ 2,
+ )
+
+ def parse(self, text):
+ self._reset()
+ for mirrorgroup in ET.XML(text):
+ for mirror in mirrorgroup:
+ name = ""
+ for e in mirror:
+ if e.tag == "name":
+ name = e.text
+ if e.tag == "uri":
+ uri = e.text
+ self._dict[uri] = {
+ "name": name,
+ "country": mirrorgroup.get("countryname"),
+ "region": mirrorgroup.get("region"),
+ "ipv4": e.get("ipv4"),
+ "ipv6": e.get("ipv6"),
+ "proto": e.get("protocol") or self._get_proto(uri),
+ }
+
+ def tuples(self):
+ return [(url, args) for url, args in list(self._dict.items())]
+
+ def uris(self):
+ return [url for url, args in list(self._dict.items())]
+
+
+if __name__ == "__main__":
+ import sys
+ import urllib.request, urllib.parse, urllib.error
+
+ parser = MirrorParser3()
+ parser.parse(urllib.request.urlopen(MIRRORS_3_XML).read())
+ print("===== tuples")
+ print(parser.tuples())
+ print("===== uris")
+ print(parser.uris())
diff --git a/mirrorselect/output.py b/mirrorselect/output.py
index aa679cb..8b33cff 100644
--- a/mirrorselect/output.py
+++ b/mirrorselect/output.py
@@ -37,13 +37,13 @@ from optparse import IndentedHelpFormatter
def encoder(text, _encoding_):
- return codecs.encode(text, _encoding_, 'replace')
+ return codecs.encode(text, _encoding_, "replace")
def decode_selection(selection):
- '''utility function to decode a list of strings
+ """utility function to decode a list of strings
accoring to the filesystem encoding
- '''
+ """
# fix None passed in, return an empty list
selection = selection or []
enc = sys.getfilesystemencoding()
@@ -53,8 +53,7 @@ def decode_selection(selection):
def get_encoding(output):
- if hasattr(output, 'encoding') \
- and output.encoding != None:
+ if hasattr(output, "encoding") and output.encoding != None:
return output.encoding
else:
encoding = locale.getpreferredencoding()
@@ -65,111 +64,115 @@ def get_encoding(output):
codecs.lookup(encoding)
except LookupError:
# Python does not know the encoding, so use utf-8.
- encoding = 'utf_8'
+ encoding = "utf_8"
return encoding
class Output:
- """Handles text output. Only prints messages with level <= verbosity.
- Therefore, verbosity=2 is everything (debug), and verbosity=0 is urgent
- messages only (quiet)."""
+ """Handles text output. Only prints messages with level <= verbosity.
+ Therefore, verbosity=2 is everything (debug), and verbosity=0 is urgent
+ messages only (quiet)."""
- def __init__(self, verbosity=1, out=sys.stderr):
- esc_seq = "\x1b["
- codes = {}
+ def __init__(self, verbosity=1, out=sys.stderr):
+ esc_seq = "\x1b["
+ codes = {}
- codes["reset"] = esc_seq + "39;49;00m"
- codes["bold"] = esc_seq + "01m"
- codes["blue"] = esc_seq + "34;01m"
- codes["green"] = esc_seq + "32;01m"
- codes["yellow"] = esc_seq + "33;01m"
- codes["red"] = esc_seq + "31;01m"
+ codes["reset"] = esc_seq + "39;49;00m"
+ codes["bold"] = esc_seq + "01m"
+ codes["blue"] = esc_seq + "34;01m"
+ codes["green"] = esc_seq + "32;01m"
+ codes["yellow"] = esc_seq + "33;01m"
+ codes["red"] = esc_seq + "31;01m"
- self.codes = codes
- del codes
+ self.codes = codes
+ del codes
- self.verbosity = verbosity
- self.file = out
+ self.verbosity = verbosity
+ self.file = out
- def red(self, text):
- return self.codes["red"]+text+self.codes["reset"]
+ def red(self, text):
+ return self.codes["red"] + text + self.codes["reset"]
- def green(self, text):
- return self.codes["green"]+text+self.codes["reset"]
+ def green(self, text):
+ return self.codes["green"] + text + self.codes["reset"]
- def white(self, text):
- return self.codes["bold"]+text+self.codes["reset"]
+ def white(self, text):
+ return self.codes["bold"] + text + self.codes["reset"]
- def blue(self, text):
- return self.codes["blue"]+text+self.codes["reset"]
+ def blue(self, text):
+ return self.codes["blue"] + text + self.codes["reset"]
- def yellow(self, text):
- return self.codes["yellow"]+text+self.codes["reset"]
+ def yellow(self, text):
+ return self.codes["yellow"] + text + self.codes["reset"]
- def print_info(self, message, level=1):
- """Prints an info message with a green star, like einfo."""
- if level <= self.verbosity:
- self.file.write('\r' + self.green('* ') + message)
- self.file.flush()
+ def print_info(self, message, level=1):
+ """Prints an info message with a green star, like einfo."""
+ if level <= self.verbosity:
+ self.file.write("\r" + self.green("* ") + message)
+ self.file.flush()
- def print_warn(self, message, level=1):
- """Prints a warning."""
- if level <= self.verbosity:
- self.file.write(self.yellow('Warning: ') + message)
- self.file.flush()
+ def print_warn(self, message, level=1):
+ """Prints a warning."""
+ if level <= self.verbosity:
+ self.file.write(self.yellow("Warning: ") + message)
+ self.file.flush()
- def print_err(self, message, level=0):
- """Prints an error message with a big red ERROR."""
- if level <= self.verbosity:
- self.file.write(self.red('\nERROR: ') + message + '\n')
- self.file.flush()
- sys.exit(1)
+ def print_err(self, message, level=0):
+ """Prints an error message with a big red ERROR."""
+ if level <= self.verbosity:
+ self.file.write(self.red("\nERROR: ") + message + "\n")
+ self.file.flush()
+ sys.exit(1)
- def write(self, message, level=1):
- """A wrapper around stderr.write, to enforce verbosity
settings."""
- if level <= self.verbosity:
- self.file.write(message)
- self.file.flush()
+ def write(self, message, level=1):
+ """A wrapper around stderr.write, to enforce verbosity settings."""
+ if level <= self.verbosity:
+ self.file.write(message)
+ self.file.flush()
class ColoredFormatter(IndentedHelpFormatter):
- """HelpFormatter with colorful output.
-
- Extends format_option.
- Overrides format_heading.
- """
-
- def __init__(self, output):
- IndentedHelpFormatter.__init__(self)
- self.output = output
-
- def format_heading(self, heading):
- """Return a colorful heading."""
- return "%*s%s:\n" % (self.current_indent, "",
self.output.white(heading))
-
- def format_option(self, option):
- """Return colorful formatted help for an option."""
- option = IndentedHelpFormatter.format_option(self, option)
- # long options with args
- option = re.sub(
- r"--([a-zA-Z]*)=([a-zA-Z]*)",
- lambda m: "-{} {}".format(self.output.green(m.group(1)),
- self.output.blue(m.group(2))),
- option)
- # short options with args
- option = re.sub(
- r"-([a-zA-Z]) ?([0-9A-Z]+)",
- lambda m: " -" + self.output.green(m.group(1)) + ' ' + \
- self.output.blue(m.group(2)),
- option)
- # options without args
- option = re.sub(
- r"-([a-zA-Z\d]+)", lambda m: "-" +
self.output.green(m.group(1)),
- option)
- return option
-
- def format_description(self, description):
- """Do not wrap."""
- return description + '\n'
-
+ """HelpFormatter with colorful output.
+
+ Extends format_option.
+ Overrides format_heading.
+ """
+
+ def __init__(self, output):
+ IndentedHelpFormatter.__init__(self)
+ self.output = output
+
+ def format_heading(self, heading):
+ """Return a colorful heading."""
+ return "%*s%s:\n" % (self.current_indent, "",
self.output.white(heading))
+
+ def format_option(self, option):
+ """Return colorful formatted help for an option."""
+ option = IndentedHelpFormatter.format_option(self, option)
+ # long options with args
+ option = re.sub(
+ r"--([a-zA-Z]*)=([a-zA-Z]*)",
+ lambda m: "-{} {}".format(
+ self.output.green(m.group(1)), self.output.blue(m.group(2))
+ ),
+ option,
+ )
+ # short options with args
+ option = re.sub(
+ r"-([a-zA-Z]) ?([0-9A-Z]+)",
+ lambda m: " -"
+ + self.output.green(m.group(1))
+ + " "
+ + self.output.blue(m.group(2)),
+ option,
+ )
+ # options without args
+ option = re.sub(
+ r"-([a-zA-Z\d]+)", lambda m: "-" + self.output.green(m.group(1)),
option
+ )
+ return option
+
+ def format_description(self, description):
+ """Do not wrap."""
+ return description + "\n"
diff --git a/mirrorselect/selectors.py b/mirrorselect/selectors.py
index 9647e56..4ec4474 100644
--- a/mirrorselect/selectors.py
+++ b/mirrorselect/selectors.py
@@ -37,6 +37,7 @@ import time
import hashlib
import urllib.request, urllib.parse, urllib.error
+
url_parse = urllib.parse.urlparse
url_unparse = urllib.parse.urlunparse
url_open = urllib.request.urlopen
@@ -53,515 +54,595 @@ NETSELECT_SUPPORTS_IPV4_IPV6 = True
class Shallow:
- """handles rapid server selection via netselect"""
-
- def __init__(self, hosts, options, output):
- self._options = options
- self.output = output
- self.urls = []
-
- if options.blocksize is not None:
- self.netselect_split(hosts, options.servers,
- options.blocksize)
- else:
- self.netselect(hosts, options.servers)
-
- if len(self.urls) == 0:
- self.output.print_err('Netselect failed to return any
mirrors.'
- ' Try again using block mode.')
-
+ """handles rapid server selection via netselect"""
- def netselect(self, hosts, number, quiet=False):
- """
- Uses Netselect to choose the closest hosts, _very_ quickly
- """
- if not quiet:
- hosts = [host[0] for host in hosts]
- top_host_dict = {}
- top_hosts = []
+ def __init__(self, hosts, options, output):
+ self._options = options
+ self.output = output
+ self.urls = []
- if not quiet:
- self.output.print_info('Using netselect to choose the
top '
- '%d mirrors...' % number)
+ if options.blocksize is not None:
+ self.netselect_split(hosts, options.servers, options.blocksize)
+ else:
+ self.netselect(hosts, options.servers)
+
+ if len(self.urls) == 0:
+ self.output.print_err(
+ "Netselect failed to return any mirrors." " Try again using
block mode."
+ )
- host_string = ' '.join(hosts)
+ def netselect(self, hosts, number, quiet=False):
+ """
+ Uses Netselect to choose the closest hosts, _very_ quickly
+ """
+ if not quiet:
+ hosts = [host[0] for host in hosts]
+ top_host_dict = {}
+ top_hosts = []
- cmd = ['netselect', '-s%d' % (number,)]
+ if not quiet:
+ self.output.print_info(
+ "Using netselect to choose the top " "%d mirrors..." % number
+ )
- if NETSELECT_SUPPORTS_IPV4_IPV6:
- if self._options.ipv4:
- cmd.append('-4')
- elif self._options.ipv6:
- cmd.append('-6')
+ host_string = " ".join(hosts)
- cmd.extend(hosts)
+ cmd = ["netselect", "-s%d" % (number,)]
- self.output.write('\nnetselect(): running "%s"\n'
- % ' '.join(cmd), 2)
+ if NETSELECT_SUPPORTS_IPV4_IPV6:
+ if self._options.ipv4:
+ cmd.append("-4")
+ elif self._options.ipv6:
+ cmd.append("-6")
+
+ cmd.extend(hosts)
+
+ self.output.write('\nnetselect(): running "%s"\n' % " ".join(cmd), 2)
+
+ proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
+
+ out, err = proc.communicate()
- proc = subprocess.Popen(cmd,
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ if err:
+ self.output.write("netselect(): netselect stderr: %s\n" % err, 2)
- out, err = proc.communicate()
+ for line in out.splitlines():
+ line = line.split()
+ if len(line) < 2:
+ continue
+ top_hosts.append(line[1])
+ top_host_dict[line[0]] = line[1]
- if err:
- self.output.write('netselect(): netselect stderr: %s\n'
% err, 2)
+ if not quiet:
+ self.output.write("Done.\n")
- for line in out.splitlines():
- line = line.split()
- if len(line) < 2:
- continue
- top_hosts.append(line[1])
- top_host_dict[line[0]] = line[1]
+ self.output.write(
+ "\nnetselect(): returning {} and {}\n".format(top_hosts,
top_host_dict), 2
+ )
- if not quiet:
- self.output.write('Done.\n')
+ if quiet:
+ return top_hosts, top_host_dict
+ else:
+ self.urls = top_hosts
- self.output.write('\nnetselect(): returning {} and
{}\n'.format(top_hosts,
- top_host_dict), 2)
+ def netselect_split(self, hosts, number, block_size):
+ """
+ This uses netselect to test mirrors in chunks,
+ each at most block_size in length.
+ This is done in a tournament style.
+ """
+ hosts = [host[0] for host in hosts]
- if quiet:
- return top_hosts, top_host_dict
- else:
- self.urls = top_hosts
+ self.output.write("netselect_split() got %s hosts.\n" % len(hosts), 2)
+ host_blocks = self.host_blocks(hosts, block_size)
- def netselect_split(self, hosts, number, block_size):
- """
- This uses netselect to test mirrors in chunks,
- each at most block_size in length.
- This is done in a tournament style.
- """
- hosts = [host[0] for host in hosts]
+ self.output.write(" split into %s blocks\n" % len(host_blocks), 2)
- self.output.write('netselect_split() got %s hosts.\n' %
len(hosts), 2)
+ top_hosts = []
+ ret_hosts = {}
- host_blocks = self.host_blocks(hosts, block_size)
+ block_index = 0
+ for block in host_blocks:
+ self.output.print_info(
+ "Using netselect to choose the top "
+ "%d hosts, in blocks of %s. %s of %s blocks complete."
+ % (number, block_size, block_index, len(host_blocks))
+ )
- self.output.write(' split into %s blocks\n' % len(host_blocks),
2)
+ host_dict = self.netselect(block, len(block), quiet=True)[1]
- top_hosts = []
- ret_hosts = {}
+ self.output.write(
+ "ran netselect(%s, %s), and got %s\n" % (block, len(block),
host_dict),
+ 2,
+ )
- block_index = 0
- for block in host_blocks:
- self.output.print_info('Using netselect to choose the
top '
- '%d hosts, in blocks of %s. %s of %s blocks complete.'
- % (number, block_size, block_index, len(host_blocks)))
+ for key in list(host_dict.keys()):
+ ret_hosts[key] = host_dict[key]
+ block_index += 1
- host_dict = self.netselect(block, len(block),
quiet=True)[1]
+ sys.stderr.write(
+ "\rUsing netselect to choose the top"
+ "%d hosts, in blocks of %s. %s of %s blocks complete.\n"
+ % (number, block_size, block_index, len(host_blocks))
+ )
- self.output.write('ran netselect(%s, %s), and got %s\n'
- % (block, len(block), host_dict), 2)
+ host_ranking_keys = list(ret_hosts.keys())
+ host_ranking_keys.sort()
- for key in list(host_dict.keys()):
- ret_hosts[key] = host_dict[key]
- block_index += 1
+ for rank in host_ranking_keys[:number]:
+ top_hosts.append(ret_hosts[rank])
- sys.stderr.write('\rUsing netselect to choose the top'
- '%d hosts, in blocks of %s. %s of %s blocks complete.\n'
- % (number, block_size, block_index, len(host_blocks)))
+ self.output.write("netselect_split(): returns %s\n" % top_hosts, 2)
- host_ranking_keys = list(ret_hosts.keys())
- host_ranking_keys.sort()
+ self.urls = top_hosts
- for rank in host_ranking_keys[:number]:
- top_hosts.append(ret_hosts[rank])
+ def host_blocks(self, hosts, block_size):
+ """
+ Takes a list of hosts and a block size,
+ and returns an list of lists of URLs.
+ Each of the sublists is at most block_size in length.
+ """
+ host_array = []
+ mylist = []
- self.output.write('netselect_split(): returns %s\n' %
top_hosts, 2)
+ while len(hosts) > block_size:
+ while len(mylist) < block_size:
+ mylist.append(hosts.pop())
+ host_array.append(mylist)
+ mylist = []
+ host_array.append(hosts)
- self.urls = top_hosts
+ self.output.write(
+ "\n_host_blocks(): returns "
+ "%s blocks, each about %s in size\n"
+ % (len(host_array), len(host_array[0])),
+ 2,
+ )
-
- def host_blocks(self, hosts, block_size):
- """
- Takes a list of hosts and a block size,
- and returns an list of lists of URLs.
- Each of the sublists is at most block_size in length.
- """
- host_array = []
- mylist = []
-
- while len(hosts) > block_size:
- while (len(mylist) < block_size):
- mylist.append(hosts.pop())
- host_array.append(mylist)
- mylist = []
- host_array.append(hosts)
-
- self.output.write('\n_host_blocks(): returns '
- '%s blocks, each about %s in size\n'
- % (len(host_array), len(host_array[0])), 2)
-
- return host_array
+ return host_array
class TimeoutException(Exception):
- pass
+ pass
def timeout_handler(signum, frame):
- raise TimeoutException()
+ raise TimeoutException()
class Deep:
- """handles deep mode mirror selection."""
-
- def __init__(self, hosts, options, output):
- self.output = output
- self.urls = []
- self._hosts = hosts
- self._number = options.servers
- self._dns_timeout = options.timeout
- self._connect_timeout = options.timeout
- self._download_timeout = options.timeout
- self.test_file = options.file
- self.test_md5 = options.md5
-
- addr_families = []
- if options.ipv4:
- addr_families.append(socket.AF_INET)
- elif options.ipv6:
- addr_families.append(socket.AF_INET6)
- else:
- addr_families.append(socket.AF_UNSPEC)
-
- self._addr_families = addr_families
-
- self.deeptest()
-
- def deeptest(self):
- """
- Takes a list of hosts and returns the fastest, using _deeptime()
- Doesn't waste time finnishing a test that has already taken
longer than
- the slowest mirror weve already got.
- """
- top_hosts = {}
- prog = 0
- maxtime = self._download_timeout
- hosts = [host[0] for host in self._hosts]
- num_hosts = len(hosts)
- self.dl_failures = 0
-
- for host in hosts:
-
- prog += 1
- if self.test_file != 'mirrorselect-test':
- self.output.print_info(
- 'Downloading %s files from each
mirror... [%s of %s]'
- % (self.test_file, prog, num_hosts) )
- else:
- self.output.print_info(
- 'Downloading 100k files from each
mirror... [%s of %s]'
- % (prog, num_hosts) )
-
- mytime, ignore = self.deeptime(host, maxtime)
-
- if not ignore and mytime < maxtime:
- maxtime, top_hosts = self._list_add((mytime,
host), \
- maxtime, top_hosts,
self._number)
- else:
- continue
-
- self.output.write('deeptest(): got %s hosts, and returned %s\n'
- % (num_hosts, str(list(top_hosts.values()))), 2)
-
- self.output.write('\n') #this just makes output nicer
-
- #can't just return the dict.values,
- #because we want the fastest mirror first...
- keys = list(top_hosts.keys())
- keys.sort()
-
- rethosts = []
- for key in keys:
- #self.output.write('deeptest(): adding rethost '
- #'%s, %s' % (key, top_hosts[key]), 2)
- rethosts.append(top_hosts[key])
-
- self.output.write('deeptest(): final rethost %s\n' %
(rethosts), 2)
- self.output.write('deeptest(): final md5 failures %s of %s\n'
- % (self.dl_failures, num_hosts), 2)
- self.urls = rethosts
-
-
- def deeptime(self, url, maxtime):
- """
- Takes a single url and fetch command, and downloads the test
file.
- Can be given an optional timeout, for use with a clever
algorithm.
- Like mine.
- """
- self.output.write('\n_deeptime(): maxtime is %s\n' % maxtime, 2)
-
- if url.endswith('/'): #append the path to the testfile to the
URL
- url = url + 'distfiles/' + self.test_file
- else:
- url = url + '/distfiles/' + self.test_file
-
- url_parts = url_parse(url)
-
- signal.signal(signal.SIGALRM, timeout_handler)
-
- ips = []
- for addr_family in self._addr_families:
- try:
- try:
- signal.alarm(self._dns_timeout)
- for result in socket.getaddrinfo(
- url_parts.hostname, None,
addr_family,
- socket.SOCK_STREAM, 0,
socket.AI_ADDRCONFIG):
- family, _, __, ___, sockaddr =
result
- ip = sockaddr[0]
- if family == socket.AF_INET6:
- ip = "[%s]" % ip
- ips.append(ip)
- finally:
- signal.alarm(0)
- except OSError as e:
- self.output.write('deeptime(): dns error for
host %s: %s\n'
- % (url_parts.hostname, e), 2)
- except TimeoutException:
- self.output.write('deeptime(): dns timeout for
host %s\n'
- % url_parts.hostname, 2)
-
- if not ips:
- self.output.write('deeptime(): unable to resolve ip for
host %s\n'
- % url_parts.hostname, 2)
- return (None, True)
-
- self.output.write("deeptime(): ip's for host %s: %s\n"
- % (url_parts.hostname, str(ips)), 2)
- delta = 0
- f = None
-
- for ip in ips:
- test_parts = url_parts._replace(netloc=ip)
- test_url = url_unparse(test_parts)
- self.output.write('deeptime(): testing url: %s\n' %
test_url, 2)
-
- f, test_url, early_out =
self._test_connection(test_url, url_parts,
- ip, ips[ips.index(ip):])
- if early_out:
- break
-
- if f is None:
- self.output.write('deeptime(): unable to ' + \
- 'connect to host %s\n' % \
- (url_parts.hostname,), 2)
- return (None, True)
-
- try:
- # Close the initial "wake up" connection.
- try:
- signal.alarm(self._connect_timeout)
- f.close()
- finally:
- signal.alarm(0)
- except OSError as e:
- self.output.write(('deeptime(): closing connection to
host %s '
- 'failed for ip %s: %s\n') %
(url_parts.hostname, ip, e), 2)
- except TimeoutException:
- self.output.write(('deeptime(): closing connection to
host %s '
- 'timed out for ip %s\n') % (url_parts.hostname,
ip), 2)
-
- self.output.write('deeptime(): timing url: %s\n' % test_url, 2)
- try:
- # The first connection serves to "wake up" the route
between
- # the local and remote machines. A second connection is
used
- # for the timed run.
- try:
- signal.alarm(int(math.ceil(maxtime)))
- stime = time.time()
- r = url_request(test_url)
- r.host = url_parts.netloc
- f = url_open(r)
-
- md5 = hashlib.md5(f.read()).hexdigest()
-
- delta = time.time() - stime
- f.close()
- if md5 != self.test_md5:
- self.output.write(
- "\ndeeptime(): md5sum error for
file: %s\n"
- % self.test_file +
- " expected: %s\n" %
self.test_md5 +
- " got.....: %s\n" % md5
+
- " host....: %s, %s\n"
- % (url_parts.hostname, ip))
- self.dl_failures += 1
- return (None, True)
-
- finally:
- signal.alarm(0)
-
- except (OSError, ssl.CertificateError) as e:
- self.output.write(('\ndeeptime(): download from host %s
'
- 'failed for ip %s: %s\n') %
(url_parts.hostname, ip, e), 2)
- return (None, True)
- except TimeoutException:
- self.output.write(('\ndeeptime(): download from host %s
'
- 'timed out for ip %s\n') % (url_parts.hostname,
ip), 2)
- return (None, True)
- except IncompleteRead as e:
- self.output.write(('\ndeeptime(): download from host %s
'
- 'failed for ip %s: %s\n') %
(url_parts.hostname, ip, e), 2)
- return (None, True)
-
- signal.signal(signal.SIGALRM, signal.SIG_DFL)
-
- self.output.write('deeptime(): download completed.\n', 2)
- self.output.write('deeptime(): %s seconds for host %s\n'
- % (delta, url), 2)
- return (delta, False)
-
-
- def _test_connection(self, test_url, url_parts, ip, ips):
- """Tests the url for a connection, will recurse using
- the original url instead of the ip if an HTTPError occurs
- Returns f, test_url, early_out
- """
- early_out = False
- f = None
- try:
- try:
- signal.alarm(self._connect_timeout)
- r = url_request(test_url)
- r.host = url_parts.netloc
- f = url_open(r)
- early_out = True
- finally:
- signal.alarm(0)
- except HTTPError as e:
- self.output.write('deeptime(): connection to host %s\n'
- ' returned HTTPError: %s for ip %s\n'
- ' Switching back to original url\n'
- % (url_parts.hostname, e, ip), 2)
- if len(ips) == 1:
- test_url = url_unparse(url_parts)
- return self._test_connection(test_url,
url_parts, ip, [])
- except (OSError, ssl.CertificateError) as e:
- self.output.write('deeptime(): connection to host %s '
- 'failed for ip %s:\n %s\n'
- % (url_parts.hostname, ip, e), 2)
- except TimeoutException:
- self.output.write(('deeptime(): connection to host %s '
- 'timed out for ip %s\n') % (url_parts.hostname,
ip), 2)
- except Exception as e: # Add general exception to catch any
other errors
- self.output.print_warn(('deeptime(): connection to host
%s '
- 'errored for ip %s\n %s\n'
- ' Please file a bug for this error at
bugs.gentoo.org')
- % (url_parts.hostname, ip, e), 0)
- return f, test_url, early_out
-
-
- def _list_add(self, time_host, maxtime, host_dict, maxlen):
- """
- Takes argumets ((time, host), maxtime, host_dict, maxlen)
- Adds a new time:host pair to the dictionary of top hosts.
- If the dictionary is full, the slowest host is removed to make
space.
- Returns the new maxtime, be it the specified timeout,
- or the slowest host.
- """
- if len(host_dict) < maxlen: #still have room, and host is
fast. add it.
-
- self.output.write('_list_add(): added host %s. with a
time of %s\n'
- % (time_host[1], time_host[0]), 2)
-
- host_dict.update(dict([time_host]))
- times = list(host_dict.keys())
- times.sort()
-
- else: #We need to make room in the dict before we add. Kill the
slowest.
- self.output.write('_list_add(): Adding host %s with a
time of %s\n'
- % (time_host[1], time_host[0]), 2)
- times = list(host_dict.keys())
- times.sort()
- self.output.write('_list_add(): removing %s\n'
- % host_dict[times[-1]], 2)
- del host_dict[times[-1]]
- host_dict.update(dict([time_host]))
- #done adding. now return the appropriate time
- times = list(host_dict.keys())
- times.sort()
-
- if len(host_dict) < maxlen: #check again to choose new
timeout
- self.output.write('_list_add(): host_dict is not full
yet.'
- ' reusing timeout of %s sec.\n' %
maxtime, 2)
- retval = maxtime
- else:
- self.output.write('_list_add(): host_dict is full. '
- 'Selecting the best timeout\n', 2)
- if times[-1] < maxtime:
- retval = times[-1]
- else:
- retval = maxtime
-
- self.output.write('_list_add(): new max time is %s seconds,'
- ' and now len(host_dict)= %s\n' % (retval,
len(host_dict)), 2)
-
- return retval, host_dict
+ """handles deep mode mirror selection."""
+
+ def __init__(self, hosts, options, output):
+ self.output = output
+ self.urls = []
+ self._hosts = hosts
+ self._number = options.servers
+ self._dns_timeout = options.timeout
+ self._connect_timeout = options.timeout
+ self._download_timeout = options.timeout
+ self.test_file = options.file
+ self.test_md5 = options.md5
+
+ addr_families = []
+ if options.ipv4:
+ addr_families.append(socket.AF_INET)
+ elif options.ipv6:
+ addr_families.append(socket.AF_INET6)
+ else:
+ addr_families.append(socket.AF_UNSPEC)
+
+ self._addr_families = addr_families
+
+ self.deeptest()
+
+ def deeptest(self):
+ """
+ Takes a list of hosts and returns the fastest, using _deeptime()
+ Doesn't waste time finnishing a test that has already taken longer than
+ the slowest mirror weve already got.
+ """
+ top_hosts = {}
+ prog = 0
+ maxtime = self._download_timeout
+ hosts = [host[0] for host in self._hosts]
+ num_hosts = len(hosts)
+ self.dl_failures = 0
+
+ for host in hosts:
+
+ prog += 1
+ if self.test_file != "mirrorselect-test":
+ self.output.print_info(
+ "Downloading %s files from each mirror... [%s of %s]"
+ % (self.test_file, prog, num_hosts)
+ )
+ else:
+ self.output.print_info(
+ "Downloading 100k files from each mirror... [%s of %s]"
+ % (prog, num_hosts)
+ )
+
+ mytime, ignore = self.deeptime(host, maxtime)
+
+ if not ignore and mytime < maxtime:
+ maxtime, top_hosts = self._list_add(
+ (mytime, host), maxtime, top_hosts, self._number
+ )
+ else:
+ continue
+
+ self.output.write(
+ "deeptest(): got %s hosts, and returned %s\n"
+ % (num_hosts, str(list(top_hosts.values()))),
+ 2,
+ )
+
+ self.output.write("\n") # this just makes output nicer
+
+ # can't just return the dict.values,
+ # because we want the fastest mirror first...
+ keys = list(top_hosts.keys())
+ keys.sort()
+
+ rethosts = []
+ for key in keys:
+ # self.output.write('deeptest(): adding rethost '
+ #'%s, %s' % (key, top_hosts[key]), 2)
+ rethosts.append(top_hosts[key])
+
+ self.output.write("deeptest(): final rethost %s\n" % (rethosts), 2)
+ self.output.write(
+ "deeptest(): final md5 failures %s of %s\n" % (self.dl_failures,
num_hosts),
+ 2,
+ )
+ self.urls = rethosts
+
+ def deeptime(self, url, maxtime):
+ """
+ Takes a single url and fetch command, and downloads the test file.
+ Can be given an optional timeout, for use with a clever algorithm.
+ Like mine.
+ """
+ self.output.write("\n_deeptime(): maxtime is %s\n" % maxtime, 2)
+
+ if url.endswith("/"): # append the path to the testfile to the URL
+ url = url + "distfiles/" + self.test_file
+ else:
+ url = url + "/distfiles/" + self.test_file
+
+ url_parts = url_parse(url)
+
+ signal.signal(signal.SIGALRM, timeout_handler)
+
+ ips = []
+ for addr_family in self._addr_families:
+ try:
+ try:
+ signal.alarm(self._dns_timeout)
+ for result in socket.getaddrinfo(
+ url_parts.hostname,
+ None,
+ addr_family,
+ socket.SOCK_STREAM,
+ 0,
+ socket.AI_ADDRCONFIG,
+ ):
+ family, _, __, ___, sockaddr = result
+ ip = sockaddr[0]
+ if family == socket.AF_INET6:
+ ip = "[%s]" % ip
+ ips.append(ip)
+ finally:
+ signal.alarm(0)
+ except OSError as e:
+ self.output.write(
+ "deeptime(): dns error for host %s: %s\n" %
(url_parts.hostname, e),
+ 2,
+ )
+ except TimeoutException:
+ self.output.write(
+ "deeptime(): dns timeout for host %s\n" %
url_parts.hostname, 2
+ )
+
+ if not ips:
+ self.output.write(
+ "deeptime(): unable to resolve ip for host %s\n" %
url_parts.hostname, 2
+ )
+ return (None, True)
+
+ self.output.write(
+ "deeptime(): ip's for host %s: %s\n" % (url_parts.hostname,
str(ips)), 2
+ )
+ delta = 0
+ f = None
+
+ for ip in ips:
+ test_parts = url_parts._replace(netloc=ip)
+ test_url = url_unparse(test_parts)
+ self.output.write("deeptime(): testing url: %s\n" % test_url, 2)
+
+ f, test_url, early_out = self._test_connection(
+ test_url, url_parts, ip, ips[ips.index(ip) :]
+ )
+ if early_out:
+ break
+
+ if f is None:
+ self.output.write(
+ "deeptime(): unable to "
+ + "connect to host %s\n" % (url_parts.hostname,),
+ 2,
+ )
+ return (None, True)
+
+ try:
+ # Close the initial "wake up" connection.
+ try:
+ signal.alarm(self._connect_timeout)
+ f.close()
+ finally:
+ signal.alarm(0)
+ except OSError as e:
+ self.output.write(
+ ("deeptime(): closing connection to host %s " "failed for ip
%s: %s\n")
+ % (url_parts.hostname, ip, e),
+ 2,
+ )
+ except TimeoutException:
+ self.output.write(
+ ("deeptime(): closing connection to host %s " "timed out for
ip %s\n")
+ % (url_parts.hostname, ip),
+ 2,
+ )
+
+ self.output.write("deeptime(): timing url: %s\n" % test_url, 2)
+ try:
+ # The first connection serves to "wake up" the route between
+ # the local and remote machines. A second connection is used
+ # for the timed run.
+ try:
+ signal.alarm(int(math.ceil(maxtime)))
+ stime = time.time()
+ r = url_request(test_url)
+ r.host = url_parts.netloc
+ f = url_open(r)
+
+ md5 = hashlib.md5(f.read()).hexdigest()
+
+ delta = time.time() - stime
+ f.close()
+ if md5 != self.test_md5:
+ self.output.write(
+ "\ndeeptime(): md5sum error for file: %s\n" %
self.test_file
+ + " expected: %s\n" % self.test_md5
+ + " got.....: %s\n" % md5
+ + " host....: %s, %s\n" % (url_parts.hostname,
ip)
+ )
+ self.dl_failures += 1
+ return (None, True)
+
+ finally:
+ signal.alarm(0)
+
+ except (OSError, ssl.CertificateError) as e:
+ self.output.write(
+ ("\ndeeptime(): download from host %s " "failed for ip %s:
%s\n")
+ % (url_parts.hostname, ip, e),
+ 2,
+ )
+ return (None, True)
+ except TimeoutException:
+ self.output.write(
+ ("\ndeeptime(): download from host %s " "timed out for ip
%s\n")
+ % (url_parts.hostname, ip),
+ 2,
+ )
+ return (None, True)
+ except IncompleteRead as e:
+ self.output.write(
+ ("\ndeeptime(): download from host %s " "failed for ip %s:
%s\n")
+ % (url_parts.hostname, ip, e),
+ 2,
+ )
+ return (None, True)
+
+ signal.signal(signal.SIGALRM, signal.SIG_DFL)
+
+ self.output.write("deeptime(): download completed.\n", 2)
+ self.output.write("deeptime(): %s seconds for host %s\n" % (delta,
url), 2)
+ return (delta, False)
+
+ def _test_connection(self, test_url, url_parts, ip, ips):
+ """Tests the url for a connection, will recurse using
+ the original url instead of the ip if an HTTPError occurs
+ Returns f, test_url, early_out
+ """
+ early_out = False
+ f = None
+ try:
+ try:
+ signal.alarm(self._connect_timeout)
+ r = url_request(test_url)
+ r.host = url_parts.netloc
+ f = url_open(r)
+ early_out = True
+ finally:
+ signal.alarm(0)
+ except HTTPError as e:
+ self.output.write(
+ "deeptime(): connection to host %s\n"
+ " returned HTTPError: %s for ip %s\n"
+ " Switching back to original url\n"
+ % (url_parts.hostname, e, ip),
+ 2,
+ )
+ if len(ips) == 1:
+ test_url = url_unparse(url_parts)
+ return self._test_connection(test_url, url_parts, ip, [])
+ except (OSError, ssl.CertificateError) as e:
+ self.output.write(
+ "deeptime(): connection to host %s "
+ "failed for ip %s:\n %s\n" % (url_parts.hostname,
ip, e),
+ 2,
+ )
+ except TimeoutException:
+ self.output.write(
+ ("deeptime(): connection to host %s " "timed out for ip %s\n")
+ % (url_parts.hostname, ip),
+ 2,
+ )
+ except Exception as e: # Add general exception to catch any other
errors
+ self.output.print_warn(
+ (
+ "deeptime(): connection to host %s "
+ "errored for ip %s\n %s\n"
+ " Please file a bug for this error at
bugs.gentoo.org"
+ )
+ % (url_parts.hostname, ip, e),
+ 0,
+ )
+ return f, test_url, early_out
+
+ def _list_add(self, time_host, maxtime, host_dict, maxlen):
+ """
+ Takes argumets ((time, host), maxtime, host_dict, maxlen)
+ Adds a new time:host pair to the dictionary of top hosts.
+ If the dictionary is full, the slowest host is removed to make space.
+ Returns the new maxtime, be it the specified timeout,
+ or the slowest host.
+ """
+ if len(host_dict) < maxlen: # still have room, and host is fast. add
it.
+
+ self.output.write(
+ "_list_add(): added host %s. with a time of %s\n"
+ % (time_host[1], time_host[0]),
+ 2,
+ )
+
+ host_dict.update(dict([time_host]))
+ times = list(host_dict.keys())
+ times.sort()
+
+ else: # We need to make room in the dict before we add. Kill the
slowest.
+ self.output.write(
+ "_list_add(): Adding host %s with a time of %s\n"
+ % (time_host[1], time_host[0]),
+ 2,
+ )
+ times = list(host_dict.keys())
+ times.sort()
+ self.output.write("_list_add(): removing %s\n" %
host_dict[times[-1]], 2)
+ del host_dict[times[-1]]
+ host_dict.update(dict([time_host]))
+ # done adding. now return the appropriate time
+ times = list(host_dict.keys())
+ times.sort()
+
+ if len(host_dict) < maxlen: # check again to choose new timeout
+ self.output.write(
+ "_list_add(): host_dict is not full yet."
+ " reusing timeout of %s sec.\n" % maxtime,
+ 2,
+ )
+ retval = maxtime
+ else:
+ self.output.write(
+ "_list_add(): host_dict is full. " "Selecting the best
timeout\n", 2
+ )
+ if times[-1] < maxtime:
+ retval = times[-1]
+ else:
+ retval = maxtime
+
+ self.output.write(
+ "_list_add(): new max time is %s seconds,"
+ " and now len(host_dict)= %s\n" % (retval, len(host_dict)),
+ 2,
+ )
+
+ return retval, host_dict
class Interactive:
- """Handles interactive host selection."""
-
- def __init__(self, hosts, options, output):
- self.output = output
- self.urls = []
-
- self.interactive(hosts, options)
- self.output.write('Interactive.interactive(): self.urls = %s\n'
- % self.urls, 2)
-
- if not self.urls or len(self.urls[0]) == 0:
- sys.exit(1)
-
-
- def interactive(self, hosts, options):
- """
- Some sort of interactive menu thingy.
- """
- if options.rsync:
- dialog = ['dialog', '--stdout', '--title', '"Gentoo
RSYNC Mirrors"',
- '--radiolist', '"Please select your desired
mirror:"',
- '20', '110', '14']
- else:
- dialog = ['dialog', '--separate-output', '--stdout',
'--title',
- '"Gentoo Download Mirrors"', '--checklist',
- '"Please select your desired mirrors:']
- if not options.ipv4 and not options.ipv6:
- dialog[-1] += '\n* = supports ipv6'
-
- dialog.extend(['20', '110', '14'])
-
- for (url, args) in sorted(hosts, key = lambda x:
- (x[1]['country'].lower(), x[1]['name'].lower())
):
- marker = ""
- if options.rsync and not
url.endswith("/gentoo-portage"):
- url+="/gentoo-portage"
- if (not options.ipv6 and not options.ipv4) and
args['ipv6'] == 'y':
- marker = "* "
- if options.ipv6 and ( args['ipv6'] == 'n' ): continue
- if options.ipv4 and ( args['ipv4'] == 'n' ): continue
-
- #dialog.append('"%s" "%s%s: %s" "OFF"'
- #% ( url, marker, args['country'],
args['name']))
- dialog.extend(["%s" %url,
- "%s%s: %s" %(marker, args['country'],
args['name']),
- "OFF"])
- dialog = [encoder(x, get_encoding(sys.stdout)) for x in dialog]
- proc = subprocess.Popen( dialog,
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
- out, err = proc.communicate()
-
- self.urls = out.splitlines()
-
- sys.stderr.write("\x1b[2J\x1b[H")
- if self.urls:
- if hasattr(self.urls[0], 'decode'):
- self.urls = decode_selection(
- [x.decode('utf-8').rstrip() for x in
self.urls])
- else:
- self.urls = decode_selection([x.rstrip() for x
in self.urls])
-
+ """Handles interactive host selection."""
+
+ def __init__(self, hosts, options, output):
+ self.output = output
+ self.urls = []
+
+ self.interactive(hosts, options)
+ self.output.write("Interactive.interactive(): self.urls = %s\n" %
self.urls, 2)
+
+ if not self.urls or len(self.urls[0]) == 0:
+ sys.exit(1)
+
+ def interactive(self, hosts, options):
+ """
+ Some sort of interactive menu thingy.
+ """
+ if options.rsync:
+ dialog = [
+ "dialog",
+ "--stdout",
+ "--title",
+ '"Gentoo RSYNC Mirrors"',
+ "--radiolist",
+ '"Please select your desired mirror:"',
+ "20",
+ "110",
+ "14",
+ ]
+ else:
+ dialog = [
+ "dialog",
+ "--separate-output",
+ "--stdout",
+ "--title",
+ '"Gentoo Download Mirrors"',
+ "--checklist",
+ '"Please select your desired mirrors:',
+ ]
+ if not options.ipv4 and not options.ipv6:
+ dialog[-1] += "\n* = supports ipv6"
+
+ dialog.extend(["20", "110", "14"])
+
+ for (url, args) in sorted(
+ hosts, key=lambda x: (x[1]["country"].lower(),
x[1]["name"].lower())
+ ):
+ marker = ""
+ if options.rsync and not url.endswith("/gentoo-portage"):
+ url += "/gentoo-portage"
+ if (not options.ipv6 and not options.ipv4) and args["ipv6"] == "y":
+ marker = "* "
+ if options.ipv6 and (args["ipv6"] == "n"):
+ continue
+ if options.ipv4 and (args["ipv4"] == "n"):
+ continue
+
+ # dialog.append('"%s" "%s%s: %s" "OFF"'
+ #% ( url, marker, args['country'], args['name']))
+ dialog.extend(
+ [
+ "%s" % url,
+ "%s%s: %s" % (marker, args["country"], args["name"]),
+ "OFF",
+ ]
+ )
+ dialog = [encoder(x, get_encoding(sys.stdout)) for x in dialog]
+ proc = subprocess.Popen(dialog, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
+
+ out, err = proc.communicate()
+
+ self.urls = out.splitlines()
+
+ sys.stderr.write("\x1b[2J\x1b[H")
+ if self.urls:
+ if hasattr(self.urls[0], "decode"):
+ self.urls = decode_selection(
+ [x.decode("utf-8").rstrip() for x in self.urls]
+ )
+ else:
+ self.urls = decode_selection([x.rstrip() for x in self.urls])
diff --git a/mirrorselect/version.py b/mirrorselect/version.py
index dadd00b..9e67578 100644
--- a/mirrorselect/version.py
+++ b/mirrorselect/version.py
@@ -24,4 +24,3 @@ Distributed under the terms of the GNU General Public License
v2
"""
version = "2.3.0-git"
-
diff --git a/setup.py b/setup.py
index 770d93d..96f11d1 100755
--- a/setup.py
+++ b/setup.py
@@ -12,7 +12,7 @@ import io
import unittest
-__version__ = os.getenv('VERSION', default=os.getenv('PVR', default='9999'))
+__version__ = os.getenv("VERSION", default=os.getenv("PVR", default="9999"))
cwd = os.getcwd()
@@ -22,108 +22,105 @@ EPREFIX = "@GENTOO_PORTAGE_EPREFIX@"
# check and set it if it wasn't
if "GENTOO_PORTAGE_EPREFIX" in EPREFIX:
- EPREFIX = ''
+ EPREFIX = ""
# Python files that need `version = ""` subbed, relative to this dir:
-python_scripts = [os.path.join(cwd, path) for path in (
- 'mirrorselect/version.py',
-)]
+python_scripts = [os.path.join(cwd, path) for path in
("mirrorselect/version.py",)]
-manpage = [os.path.join(cwd, path) for path in (
- 'mirrorselect.8',
-)]
+manpage = [os.path.join(cwd, path) for path in ("mirrorselect.8",)]
class set_version(core.Command):
- """Set python version to our __version__."""
- description = "hardcode scripts' version using VERSION from environment"
- user_options = [] # [(long_name, short_name, desc),]
-
- def initialize_options (self):
- pass
-
- def finalize_options (self):
- pass
-
- def run(self):
- ver = 'git' if __version__ == '9999' else __version__
- print("Setting version to %s" % ver)
- def sub(files, pattern):
- for f in files:
- updated_file = []
- with open(f, 'r', 1, 'utf_8') as s:
- for line in s:
- newline = re.sub(pattern,
'"%s"' % ver, line, 1)
- if newline != line:
- logging.info("{}:
{}".format(f, newline))
- updated_file.append(newline)
- with open(f, 'w', 1, 'utf_8') as s:
- s.writelines(updated_file)
- quote = r'[\'"]{1}'
- python_re = r'(?<=^version = )' + quote + '[^\'"]*' + quote
- sub(python_scripts, python_re)
- man_re = r'(?<=^.TH "mirrorselect" "8" )' + quote + '[^\'"]*' +
quote
- sub(manpage, man_re)
+ """Set python version to our __version__."""
+
+ description = "hardcode scripts' version using VERSION from environment"
+ user_options = [] # [(long_name, short_name, desc),]
+
+ def initialize_options(self):
+ pass
+
+ def finalize_options(self):
+ pass
+
+ def run(self):
+ ver = "git" if __version__ == "9999" else __version__
+ print("Setting version to %s" % ver)
+
+ def sub(files, pattern):
+ for f in files:
+ updated_file = []
+ with open(f, "r", 1, "utf_8") as s:
+ for line in s:
+ newline = re.sub(pattern, '"%s"' % ver, line, 1)
+ if newline != line:
+ logging.info("{}: {}".format(f, newline))
+ updated_file.append(newline)
+ with open(f, "w", 1, "utf_8") as s:
+ s.writelines(updated_file)
+
+ quote = r'[\'"]{1}'
+ python_re = r"(?<=^version = )" + quote + "[^'\"]*" + quote
+ sub(python_scripts, python_re)
+ man_re = r'(?<=^.TH "mirrorselect" "8" )' + quote + "[^'\"]*" + quote
+ sub(manpage, man_re)
class x_sdist(sdist):
- """sdist defaulting to archive files owned by root."""
+ """sdist defaulting to archive files owned by root."""
- def finalize_options(self):
- if self.owner is None:
- self.owner = 'root'
- if self.group is None:
- self.group = 'root'
+ def finalize_options(self):
+ if self.owner is None:
+ self.owner = "root"
+ if self.group is None:
+ self.group = "root"
- sdist.finalize_options(self)
+ sdist.finalize_options(self)
class TestCommand(Command):
- user_options = []
+ user_options = []
- def initialize_options(self):
- pass
+ def initialize_options(self):
+ pass
- def finalize_options(self):
- pass
+ def finalize_options(self):
+ pass
- def run(self):
- suite = unittest.TestSuite()
- tests = unittest.defaultTestLoader.discover('tests')
- suite.addTests(tests)
- result = unittest.TextTestRunner(verbosity=2).run(suite)
- if result.errors or result.failures:
- raise SystemExit(1)
+ def run(self):
+ suite = unittest.TestSuite()
+ tests = unittest.defaultTestLoader.discover("tests")
+ suite.addTests(tests)
+ result = unittest.TextTestRunner(verbosity=2).run(suite)
+ if result.errors or result.failures:
+ raise SystemExit(1)
-test_data = {
- 'mirrorselect': [
- ]
-}
+test_data = {"mirrorselect": []}
core.setup(
- name='mirrorselect',
- version=__version__,
- description='Tool for selecting Gentoo source and rsync mirrors.',
- author='',
- author_email='',
- maintainer='Gentoo Portage Tools Team',
- maintainer_email='tools-port...@gentoo.org',
- url='http://www.gentoo.org/proj/en/portage/tools/index.xml',
-
download_url='http://distfiles.gentoo.org/distfiles/mirrorselect-%s.tar.gz'\
- % __version__,
- packages=['mirrorselect'],
- #package_data = test_data,
- scripts=(['bin/mirrorselect']),
- data_files=(
- (os.path.join(os.sep, EPREFIX.lstrip(os.sep),
'usr/share/man/man8'),
- ['mirrorselect.8']),
- ),
- cmdclass={
- 'test': TestCommand,
- 'sdist': x_sdist,
- 'set_version': set_version,
- },
+ name="mirrorselect",
+ version=__version__,
+ description="Tool for selecting Gentoo source and rsync mirrors.",
+ author="",
+ author_email="",
+ maintainer="Gentoo Portage Tools Team",
+ maintainer_email="tools-port...@gentoo.org",
+ url="http://www.gentoo.org/proj/en/portage/tools/index.xml";,
+ download_url="http://distfiles.gentoo.org/distfiles/mirrorselect-%s.tar.gz";
+ % __version__,
+ packages=["mirrorselect"],
+ # package_data = test_data,
+ scripts=(["bin/mirrorselect"]),
+ data_files=(
+ (
+ os.path.join(os.sep, EPREFIX.lstrip(os.sep), "usr/share/man/man8"),
+ ["mirrorselect.8"],
+ ),
+ ),
+ cmdclass={
+ "test": TestCommand,
+ "sdist": x_sdist,
+ "set_version": set_version,
+ },
)
-
diff --git a/tests/test_write_make_conf.py b/tests/test_write_make_conf.py
index 100c5ed..0deee69 100644
--- a/tests/test_write_make_conf.py
+++ b/tests/test_write_make_conf.py
@@ -10,46 +10,52 @@ from mirrorselect.output import Output
class WriteMakeConfTestCase(unittest.TestCase):
- def test_write_make_conf(self):
-
- def __do_it(var, mirror_string, make_conf, expected_result):
- tempdir = tempfile.mkdtemp()
- status_output = open(os.devnull, 'w')
- #print("------make_conf--------", make_conf,
"----------------------")
- #print("*****expect*****\n", expected_result,
"***********")
- try:
- config_path = os.path.join(tempdir,
'make.conf')
- with open(config_path, 'w') as f:
- f.write(make_conf)
-
write_make_conf(Output(out=status_output), config_path, var, mirror_string)
- with open(config_path) as f:
- result = f.read()
- #print("!!!result!!!\n",
result, "!!!!!!!!!!\n")
- self.assertEqual(result,
"{}".format(expected_result).format(mirror_string))
- finally:
- shutil.rmtree(tempdir)
- status_output.close()
-
- var = 'GENTOO_MIRRORS'
- mirrors = (
- ('{}="a"'.format(var)),
- ('{}="a b"'.format(var)),
- ('{}="a b c"'.format(var)),
- )
-
- cases = (
- ('{}="foo\nbar"\n'.format(var), '{}\n'),
- ('\n{}="foo\nbar"\n'.format(var), '\n{}\n'),
- ('\n{}="foo bar"\n'.format(var), '\n{}\n'),
- ('\n{}="foo bar"\n\n'.format(var), '\n\n{}\n'),
- ('\n{}="foo \\\nbar"\n'.format(var), '\n{}\n'),
- ('\n\n{}="foo \\\nbar"\n'.format(var), '\n\n{}\n'),
- ('\n\n{}="foo \\\nbar"\na="b"\n'.format(var),
'\n\na="b"\n{}\n'),
- ('\n\n{}="foo \\\n bar"\na="b"\n'.format(var),
'\n\na="b"\n{}\n'),
- ('\n\n{}="foo \\\n bar\\\n
baz"\na="b"\n'.format(var), '\n\na="b"\n{}\n'),
- ('', '{}\n'),
- )
-
- for mirror in mirrors:
- for make_conf, expected_result in cases:
- __do_it(var, mirror, make_conf, expected_result)
+ def test_write_make_conf(self):
+ def __do_it(var, mirror_string, make_conf, expected_result):
+ tempdir = tempfile.mkdtemp()
+ status_output = open(os.devnull, "w")
+ # print("------make_conf--------", make_conf,
"----------------------")
+ # print("*****expect*****\n", expected_result, "***********")
+ try:
+ config_path = os.path.join(tempdir, "make.conf")
+ with open(config_path, "w") as f:
+ f.write(make_conf)
+ write_make_conf(
+ Output(out=status_output), config_path, var, mirror_string
+ )
+ with open(config_path) as f:
+ result = f.read()
+ # print("!!!result!!!\n", result, "!!!!!!!!!!\n")
+ self.assertEqual(
+ result, "{}".format(expected_result).format(mirror_string)
+ )
+ finally:
+ shutil.rmtree(tempdir)
+ status_output.close()
+
+ var = "GENTOO_MIRRORS"
+ mirrors = (
+ ('{}="a"'.format(var)),
+ ('{}="a b"'.format(var)),
+ ('{}="a b c"'.format(var)),
+ )
+
+ cases = (
+ ('{}="foo\nbar"\n'.format(var), "{}\n"),
+ ('\n{}="foo\nbar"\n'.format(var), "\n{}\n"),
+ ('\n{}="foo bar"\n'.format(var), "\n{}\n"),
+ ('\n{}="foo bar"\n\n'.format(var), "\n\n{}\n"),
+ ('\n{}="foo \\\nbar"\n'.format(var), "\n{}\n"),
+ ('\n\n{}="foo \\\nbar"\n'.format(var), "\n\n{}\n"),
+ ('\n\n{}="foo \\\nbar"\na="b"\n'.format(var), '\n\na="b"\n{}\n'),
+ ('\n\n{}="foo \\\n bar"\na="b"\n'.format(var),
'\n\na="b"\n{}\n'),
+ (
+ '\n\n{}="foo \\\n bar\\\n baz"\na="b"\n'.format(var),
+ '\n\na="b"\n{}\n',
+ ),
+ ("", "{}\n"),
+ )
+
+ for mirror in mirrors:
+ for make_conf, expected_result in cases:
+ __do_it(var, mirror, make_conf, expected_result)
