# -*- coding: utf-8 -*-
#
# Copyright (C) 2012-2023 Vinay Sajip.
# Licensed to the Python Software Foundation under a contributor agreement.
# See LICENSE.txt and CONTRIBUTORS.txt.
#
from io import BytesIO
from itertools import islice
import os
import re
import shutil
try:
    import ssl
except ImportError:
    ssl = None
import sys
import tempfile
import textwrap
import time

from compat import unittest
from support import TempdirManager, DistlibTestCase, in_github_workflow

from distlib import DistlibException
from distlib.compat import cache_from_source
from distlib.util import (get_export_entry, ExportEntry, resolve, get_cache_base, path_to_cache_dir, zip_dir,
                          parse_credentials, ensure_slash, split_filename, EventMixin, Sequencer, unarchive, Progress,
                          iglob, RICH_GLOB, parse_requirement, get_extras, Configurator, read_exports, write_exports,
                          FileOperator, is_string_sequence, get_package_data, convert_path)

HERE = os.path.dirname(os.path.abspath(__file__))
IN_GITHUB_WORKFLOW = in_github_workflow()


class TestContainer(object):

    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs


class UtilTestCase(DistlibTestCase):

    def check_entry(self, entry, name, prefix, suffix, flags):
        self.assertEqual(entry.name, name)
        self.assertEqual(entry.prefix, prefix)
        self.assertEqual(entry.suffix, suffix)
        self.assertEqual(entry.flags, flags)

    def test_export_entry(self):
        self.assertIsNone(get_export_entry('foo.py'))
        self.assertIsNone(get_export_entry('foo.py='))
        for spec in ('foo=foo:main', 'foo =foo:main', 'foo= foo:main', 'foo = foo:main'):
            self.check_entry(get_export_entry(spec), 'foo', 'foo', 'main', [])
        self.check_entry(get_export_entry('foo=foo.bar:main'), 'foo', 'foo.bar', 'main', [])
        self.check_entry(get_export_entry('foo=foo.bar:main [a]'), 'foo', 'foo.bar', 'main', ['a'])
        # See issue #127 - allow hyphens
        self.check_entry(get_export_entry('foo=foo.bar:main [with-foo]'), 'foo', 'foo.bar', 'main', ['with-foo'])
        self.check_entry(get_export_entry('foo=foo.bar:main [ a ]'), 'foo', 'foo.bar', 'main', ['a'])
        self.check_entry(get_export_entry('foo=foo.bar:main [a=b, c=d,e, f=g]'), 'foo', 'foo.bar', 'main',
                         ['a=b', 'c=d', 'e', 'f=g'])
        self.check_entry(get_export_entry('foo=foo.bar:main [a=9, 9=8,e, f9=g8]'), 'foo', 'foo.bar', 'main',
                         ['a=9', '9=8', 'e', 'f9=g8'])
        self.check_entry(get_export_entry('foo=foo.bar:main[x]'), 'foo', 'foo.bar', 'main', ['x'])
        self.check_entry(get_export_entry('foo=abc'), 'foo', 'abc', None, [])
        self.check_entry(get_export_entry('smc++ = smcpp.frontend:console'), 'smc++', 'smcpp.frontend', 'console', [])
        # See issue #203 - correct name parsing to allow non-name-like names like ","
        self.check_entry(get_export_entry(', = comma:main'), ',', 'comma', 'main', [])
        self.check_entry(get_export_entry(',comma = comma:main'), ',comma', 'comma', 'main', [])

        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x:y')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x [')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x ]')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x []')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x [\\]')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x [a=]')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x [a,]')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x [a,,b]')
        self.assertRaises(DistlibException, get_export_entry, 'foo=foo.bar:x [a b]')

    def test_resolve(self):
        import logging
        import logging.handlers
        self.assertIs(resolve('logging', None), logging)
        self.assertIs(resolve('logging.handlers', None), logging.handlers)
        self.assertIs(resolve('logging', 'root'), logging.root)
        self.assertEqual(resolve('logging', 'root.debug'), logging.root.debug)

    def test_cache_base(self):
        actual = get_cache_base()
        if os.name == 'nt' and 'LOCALAPPDATA' in os.environ:
            expected = os.path.expandvars('$localappdata')
        else:
            expected = os.path.expanduser('~')
        expected = os.path.join(expected, '.distlib')
        self.assertEqual(expected, actual)
        self.assertTrue(os.path.isdir(expected))

    @unittest.skipIf(os.name != 'posix', 'Test is only valid for POSIX')
    def test_path_to_cache_dir_posix(self):
        self.assertEqual(path_to_cache_dir('/home/user/some-file.zip'), '--home--user--some-file.zip.cache')
        self.assertEqual(path_to_cache_dir('path/to/some-file.zip'),
                         os.path.splitext(path_to_cache_dir(os.path.dirname(HERE)))[0] +
                         '--path--to--some-file.zip.cache')
        self.assertEqual(path_to_cache_dir('path/to/some-file.zip', use_abspath=False),
                         'path--to--some-file.zip.cache')

    @unittest.skipIf(os.name != 'nt', 'Test is only valid for Windows')
    def test_path_to_cache_dir_nt(self):
        self.assertEqual(path_to_cache_dir(r'c:\Users\User\Some-File.zip'), 'c-----Users--User--Some-File.zip.cache')

    def test_parse_credentials(self):
        cases = (
            ('example.com', (None, None, 'example.com')),
            ('user@example.com', ('user', None, 'example.com')),
            ('user:pwd@example.com', ('user', 'pwd', 'example.com')),
            ('user:@example.com', ('user', '', 'example.com')),
            ('user:pass@word@example.com', ('user', 'pass@word', 'example.com')),
            ('user:pass:word@example.com', ('user', 'pass:word', 'example.com')),
            ('user%3Aname:%23%5E%40@example.com', ('user:name', '#^@', 'example.com')),
        )

        for s, expected in cases:
            self.assertEqual(parse_credentials(s), expected)

    def test_ensure_slash(self):
        self.assertEqual(ensure_slash(''), '/')
        self.assertEqual(ensure_slash('/'), '/')
        self.assertEqual(ensure_slash('abc'), 'abc/')
        self.assertEqual(ensure_slash('def/'), 'def/')

    def test_split_filename(self):
        self.assertIsNone(split_filename('abl.jquery'))
        self.assertEqual(split_filename('abl.jquery-1.4.2-2'), ('abl.jquery', '1.4.2-2', None))
        self.assertEqual(split_filename('python-gnupg-0.1'), ('python-gnupg', '0.1', None))
        self.assertEqual(split_filename('baklabel-1.0.3-2729-py3.2'), ('baklabel', '1.0.3-2729', '3.2'))
        self.assertEqual(split_filename('baklabel-1.0.3-2729-py27'), ('baklabel', '1.0.3-2729', '27'))
        self.assertEqual(split_filename('advpy-0.99b'), ('advpy', '0.99b', None))
        self.assertEqual(split_filename('asv_files-dev-20120501-01', 'asv_files'),
                         ('asv_files', 'dev-20120501-01', None))
        self.assertEqual(split_filename('greenlet-0.4.0-py27-win32'), ('greenlet', '0.4.0', '27'))
        self.assertEqual(split_filename('greenlet-0.4.0-py27-linux_x86_64'), ('greenlet', '0.4.0', '27'))
        self.assertEqual(split_filename('django-altuser-v0.6.8'), ('django-altuser', 'v0.6.8', None))
        self.assertEqual(split_filename('youtube_dl_server-alpha.1'), ('youtube_dl_server', 'alpha.1', None))
        self.assertEqual(split_filename('pytest-xdist-dev'), ('pytest-xdist', 'dev', None))
        self.assertEqual(split_filename('pytest_xdist-0.1_myfork', None), ('pytest_xdist', '0.1_myfork', None))
        self.assertEqual(split_filename('pytest_xdist-0.1_myfork', 'pytest-xdist'),
                         ('pytest_xdist', '0.1_myfork', None))
        self.assertEqual(split_filename('pytest_xdist-0.1_myfork', 'pytest_dist'), ('pytest_xdist', '0.1_myfork', None))

    def test_convert_path(self):
        CP = convert_path
        if os.sep == '/':
            d = os.path.dirname(__file__)
            self.assertEqual(CP(d), d)
        else:
            self.assertEqual(CP(''), '')
            self.assertRaises(ValueError, CP, '/foo')
            self.assertRaises(ValueError, CP, 'foo/')

    def test_events(self):
        collected = []

        def handler1(e, *args, **kwargs):
            collected.append((1, e, args, kwargs))

        def handler2(e, *args, **kwargs):
            collected.append((2, e, args, kwargs))

        def handler3(e, *args, **kwargs):
            if not args:
                raise NotImplementedError('surprise!')
            collected.append((3, e, args, kwargs))
            return (args, kwargs)

        e = EventMixin()
        e.add('A', handler1)
        self.assertRaises(ValueError, e.remove, 'B', handler1)

        cases = (
            ((1, 2), {
                'buckle': 'my shoe'
            }),
            ((3, 4), {
                'shut': 'the door'
            }),
        )

        for case in cases:
            e.publish('A', *case[0], **case[1])
            e.publish('B', *case[0], **case[1])

        for actual, source in zip(collected, cases):
            self.assertEqual(actual, (1, 'A') + source[:1] + source[1:])

        collected = []
        e.add('B', handler2)

        self.assertEqual(tuple(e.get_subscribers('A')), (handler1, ))
        self.assertEqual(tuple(e.get_subscribers('B')), (handler2, ))
        self.assertEqual(tuple(e.get_subscribers('C')), ())

        for case in cases:
            e.publish('A', *case[0], **case[1])
            e.publish('B', *case[0], **case[1])

        actuals = islice(collected, 0, None, 2)
        for actual, source in zip(actuals, cases):
            self.assertEqual(actual, (1, 'A') + source[:1] + source[1:])

        actuals = islice(collected, 1, None, 2)
        for actual, source in zip(actuals, cases):
            self.assertEqual(actual, (2, 'B') + source[:1] + source[1:])

        e.remove('B', handler2)

        collected = []

        for case in cases:
            e.publish('A', *case[0], **case[1])
            e.publish('B', *case[0], **case[1])

        for actual, source in zip(collected, cases):
            self.assertEqual(actual, (1, 'A') + source[:1] + source[1:])

        e.add('C', handler3)

        collected = []
        returned = []

        for case in cases:
            returned.extend(e.publish('C', *case[0], **case[1]))
            returned.extend(e.publish('C'))

        for actual, source in zip(collected, cases):
            self.assertEqual(actual, (3, 'C') + source[:1] + source[1:])

        self.assertEqual(tuple(islice(returned, 1, None, 2)), (None, None))
        actuals = islice(returned, 0, None, 2)
        for actual, expected in zip(actuals, cases):
            self.assertEqual(actual, expected)

    def test_sequencer_basic(self):
        seq = Sequencer()

        steps = (('check', 'sdist'), ('check', 'register'), ('check', 'sdist'), ('check', 'register'),
                 ('register', 'upload_sdist'), ('sdist', 'upload_sdist'), ('check', 'build_clibs'),
                 ('build_clibs', 'build_ext'), ('build_ext', 'build_py'), ('build_py', 'build_scripts'),
                 ('build_scripts', 'build'), ('build', 'test'), ('register', 'upload_bdist'), ('build', 'upload_bdist'),
                 ('build', 'install_headers'), ('install_headers', 'install_lib'), ('install_lib', 'install_scripts'),
                 ('install_scripts', 'install_data'), ('install_data', 'install_distinfo'), ('install_distinfo',
                                                                                             'install'))

        for pred, succ in steps:
            seq.add(pred, succ)

        # Note: these tests are sensitive to dictionary ordering
        # but work under Python 2.6, 2.7, 3.2, 3.3, 3.4 and PyPy 2.5
        cases = (
            ('check', ['check']),
            ('register', ['check', 'register']),
            ('sdist', ['check', 'sdist']),
            ('build_clibs', ['check', 'build_clibs']),
            ('build_ext', ['check', 'build_clibs', 'build_ext']),
            ('build_py', ['check', 'build_clibs', 'build_ext', 'build_py']),
            ('build_scripts', ['check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts']),
            ('build', ['check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build']),
            ('test', ['check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'test']),
            ('install_headers',
             ['check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'install_headers']),
            ('install_lib', [
                'check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'install_headers',
                'install_lib'
            ]),
            ('install_scripts', [
                'check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'install_headers',
                'install_lib', 'install_scripts'
            ]),
            ('install_data', [
                'check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'install_headers',
                'install_lib', 'install_scripts', 'install_data'
            ]),
            ('install_distinfo', [
                'check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'install_headers',
                'install_lib', 'install_scripts', 'install_data', 'install_distinfo'
            ]),
            ('install', [
                'check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'install_headers',
                'install_lib', 'install_scripts', 'install_data', 'install_distinfo', 'install'
            ]),
            ('upload_sdist', (['check', 'register', 'sdist',
                               'upload_sdist'], ['check', 'sdist', 'register', 'upload_sdist'])),
            ('upload_bdist',
             (['check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'build', 'register', 'upload_bdist'],
              ['check', 'build_clibs', 'build_ext', 'build_py', 'build_scripts', 'register', 'build', 'upload_bdist'])),
        )

        for final, expected in cases:
            actual = list(seq.get_steps(final))
            if isinstance(expected, tuple):
                self.assertIn(actual, expected)
            else:
                self.assertEqual(actual, expected)

        dot = seq.dot
        expected = '''
        digraph G {
          check -> build_clibs;
          install_lib -> install_scripts;
          register -> upload_bdist;
          build -> upload_bdist;
          build_ext -> build_py;
          install_scripts -> install_data;
          check -> sdist;
          check -> register;
          build -> install_headers;
          install_data -> install_distinfo;
          sdist -> upload_sdist;
          register -> upload_sdist;
          install_distinfo -> install;
          build -> test;
          install_headers -> install_lib;
          build_py -> build_scripts;
          build_clibs -> build_ext;
          build_scripts -> build;
        }
        '''
        expected = textwrap.dedent(expected).strip().splitlines()
        actual = dot.splitlines()
        self.assertEqual(expected[0], actual[0])
        self.assertEqual(expected[-1], actual[-1])
        self.assertEqual(set(expected[1:-1]), set(actual[1:-1]))
        actual = seq.strong_connections
        expected = (
            [('test', ), ('upload_bdist', ), ('install', ), ('install_distinfo', ), ('install_data', ),
             ('install_scripts', ), ('install_lib', ), ('install_headers', ), ('build', ), ('build_scripts', ),
             ('build_py', ), ('build_ext', ), ('build_clibs', ), ('upload_sdist', ), ('sdist', ), ('register', ),
             ('check', )],
            [('install', ), ('install_distinfo', ), ('install_data', ), ('install_scripts', ), ('install_lib', ),
             ('install_headers', ), ('test', ), ('upload_bdist', ), ('build', ), ('build_scripts', ), ('build_py', ),
             ('build_ext', ), ('build_clibs', ), ('upload_sdist', ), ('sdist', ), ('register', ), ('check', )],
            [('upload_sdist', ), ('sdist', ), ('install', ), ('install_distinfo', ), ('install_data', ),
             ('upload_bdist', ), ('register', ), ('install_scripts', ), ('install_lib', ), ('install_headers', ),
             ('test', ), ('build', ), ('build_scripts', ), ('build_py', ), ('build_ext', ), ('build_clibs', ),
             ('check', )],
            # Next case added for PyPy
            [('upload_sdist', ), ('sdist', ), ('upload_bdist', ), ('register', ), ('test', ), ('install', ),
             ('install_distinfo', ), ('install_data', ),
             ('install_scripts', ), ('install_lib', ), ('install_headers', ), ('build', ), ('build_scripts', ),
             ('build_py', ), ('build_ext', ), ('build_clibs', ), ('check', )],
            # Next case added for Python 3.6
            [('upload_sdist', ), ('sdist', ), ('upload_bdist', ), ('register', ), ('install', ), ('install_distinfo', ),
             ('install_data', ), ('install_scripts', ), ('install_lib', ), ('install_headers', ), ('test', ),
             ('build', ), ('build_scripts', ), ('build_py', ), ('build_ext', ), ('build_clibs', ), ('check', )],
            # Next case added for Python 3.11
            [('upload_sdist', ), ('sdist', ), ('install', ), ('install_distinfo', ), ('install_data', ),
             ('install_scripts', ), ('install_lib', ), ('install_headers', ), ('test', ), ('upload_bdist', ),
             ('build', ), ('build_scripts', ), ('build_py', ), ('build_ext', ), ('build_clibs', ), ('register', ),
             ('check', )])
        self.assertIn(actual, expected)

    def test_sequencer_cycle(self):
        seq = Sequencer()
        seq.add('A', 'B')
        seq.add('B', 'C')
        seq.add('C', 'D')
        self.assertEqual(list(seq.get_steps('D')), ['A', 'B', 'C', 'D'])
        seq.add('C', 'A')
        self.assertEqual(list(seq.get_steps('D')), ['C', 'A', 'B', 'D'])
        self.assertFalse(seq.is_step('E'))
        self.assertRaises(ValueError, seq.get_steps, 'E')
        seq.add_node('E')
        self.assertTrue(seq.is_step('E'))
        self.assertEqual(list(seq.get_steps('E')), ['E'])
        seq.remove_node('E')
        self.assertFalse(seq.is_step('E'))
        self.assertRaises(ValueError, seq.get_steps, 'E')
        seq.remove('C', 'A')
        self.assertEqual(list(seq.get_steps('D')), ['A', 'B', 'C', 'D'])

    def test_sequencer_removal(self):
        seq = Sequencer()
        seq.add('A', 'B')
        seq.add('B', 'C')
        seq.add('C', 'D')
        preds = {'B': set(['A']), 'C': set(['B']), 'D': set(['C'])}
        succs = {'A': set(['B']), 'B': set(['C']), 'C': set(['D'])}
        self.assertEqual(seq._preds, preds)
        self.assertEqual(seq._succs, succs)
        seq.remove_node('C')
        self.assertEqual(seq._preds, preds)
        self.assertEqual(seq._succs, succs)
        seq.remove_node('C', True)
        self.assertEqual(seq._preds, {'B': set(['A'])})
        self.assertEqual(seq._succs, {'A': set(['B'])})

    def test_unarchive(self):
        import zipfile, tarfile

        good_archives = (
            ('good.zip', zipfile.ZipFile, 'r', 'namelist'),
            ('good.tar', tarfile.open, 'r', 'getnames'),
            ('good.tar.gz', tarfile.open, 'r:gz', 'getnames'),
            ('good.tar.bz2', tarfile.open, 'r:bz2', 'getnames'),
        )
        bad_archives = ('bad.zip', 'bad.tar', 'bad.tar.gz', 'bad.tar.bz2')

        # Test "evil" tarball on 3.12 *or* on Python with PEP-706 backported
        if sys.version_info > (3, 12) or hasattr(tarfile, 'data_filter'):
            bad_archives += ('evil.tar.gz', )

        for name, cls, mode, lister in good_archives:
            td = tempfile.mkdtemp()
            archive = None
            try:
                name = os.path.join(HERE, name)
                unarchive(name, td)
                archive = cls(name, mode)
                names = getattr(archive, lister)()
                for name in names:
                    p = os.path.join(td, name)
                    self.assertTrue(os.path.exists(p))
            finally:
                shutil.rmtree(td)
                if archive:
                    archive.close()

        for name in bad_archives:
            name = os.path.join(HERE, name)
            td = tempfile.mkdtemp()
            try:
                self.assertRaises(ValueError, unarchive, name, td)
            finally:
                shutil.rmtree(td)

    def test_string_sequence(self):
        self.assertTrue(is_string_sequence(['a']))
        self.assertTrue(is_string_sequence(['a', 'b']))
        self.assertFalse(is_string_sequence(['a', 'b', None]))
        self.assertRaises(AssertionError, is_string_sequence, [])

    @unittest.skipIf('SKIP_ONLINE' in os.environ, 'Skipping online test')
    @unittest.skipUnless(ssl, 'SSL required for this test.')
    def test_package_data(self):
        data = get_package_data(name='config', version='0.3.6')
        self.assertTrue(data)
        self.assertTrue('index-metadata' in data)
        metadata = data['index-metadata']
        self.assertEqual(metadata['name'], 'config')
        self.assertEqual(metadata['version'], '0.3.6')
        data = get_package_data(name='config', version='0.3.5')
        self.assertFalse(data)

    def test_zip_dir(self):
        d = os.path.join(HERE, 'foofoo')
        data = zip_dir(d)
        self.assertIsInstance(data, BytesIO)

    def test_configurator(self):
        d = {
            'a': 1,
            'b': 2.0,
            'c': 'xyz',
            'd': 'inc://' + os.path.join(HERE, 'included.json'),
            'e': 'inc://' + 'included.json',
            'stderr': 'ext://sys.stderr',
            'list_o_stuff': [
                'cfg://stderr',
                'ext://sys.stdout',
                'ext://logging.NOTSET',
            ],
            'dict_o_stuff': {
                'k1': 'cfg://list_o_stuff[1]',
                'k2': 'abc',
                'k3': 'cfg://list_o_stuff',
            },
            'another_dict_o_stuff': {
                'k1': 'cfg://dict_o_stuff[k2]',
                'k2': 'ext://re.I',
                'k3': 'cfg://dict_o_stuff[k3][0]',
            },
            'custom': {
                '()': __name__ + '.TestContainer',
                '[]': [1, 'a', 2.0, ('b', 'c', 'd')],
                '.': {
                    'p1': 'a',
                    'p2': 'b',
                    'p3': {
                        '()': __name__ + '.TestContainer',
                        '[]': [1, 2],
                        '.': {
                            'p1': 'c',
                        },
                    },
                },
                'k1': 'v1',
                'k2': 'v2',
            }
        }

        cfg = Configurator(d, HERE)
        self.assertEqual(cfg['a'], 1)
        self.assertEqual(cfg['b'], 2.0)
        self.assertEqual(cfg['c'], 'xyz')
        self.assertIs(cfg['stderr'], sys.stderr)
        self.assertIs(cfg['list_o_stuff'][0], sys.stderr)
        self.assertIs(cfg['list_o_stuff'][1], sys.stdout)
        self.assertIs(cfg['list_o_stuff'][-1], 0)  # logging.NOTSET == 0
        self.assertIs(cfg['dict_o_stuff']['k1'], sys.stdout)
        self.assertIs(cfg['another_dict_o_stuff']['k1'], 'abc')
        self.assertIs(cfg['another_dict_o_stuff']['k2'], re.I)
        self.assertIs(cfg['another_dict_o_stuff']['k3'], sys.stderr)
        custom = cfg['custom']
        self.assertIsInstance(custom, TestContainer)
        self.assertEqual(custom.args, (1, 'a', 2.0, ('b', 'c', 'd')))
        self.assertEqual(custom.kwargs, {'k1': 'v1', 'k2': 'v2'})
        self.assertEqual(custom.p1, 'a')
        self.assertEqual(custom.p2, 'b')
        self.assertIsInstance(custom.p3, TestContainer)
        self.assertEqual(custom.p3.args, (1, 2))
        self.assertEqual(custom.p3.kwargs, {})
        self.assertEqual(custom.p3.p1, 'c')
        self.assertEqual(cfg['d'], {'foo': 'bar', 'bar': 'baz'})
        self.assertEqual(cfg['e'], {'foo': 'bar', 'bar': 'baz'})


def _speed_range(min_speed, max_speed):
    return tuple(['%d KB/s' % v for v in range(min_speed, max_speed + 1)])


def _eta_range(min_eta, max_eta, prefix='ETA '):
    msg = prefix + ': 00:00:%02d'
    return tuple([msg % v for v in range(min_eta, max_eta + 1)])


class ProgressTestCase(DistlibTestCase):
    # Of late, the speed tests keep failing on AppVeyor and Windows
    @unittest.skipIf(IN_GITHUB_WORKFLOW or (os.name == 'nt' and os.environ.get('APPVEYOR') == 'True'),
                     'Test disabled on some environments due to performance')
    def test_basic(self):

        # These ranges may need tweaking to cater for especially slow
        # machines
        if os.name == 'nt':
            speed1 = _speed_range(18, 20)
            speed2 = _speed_range(20, 22)
        else:
            speed1 = _speed_range(16, 19)
            speed2 = _speed_range(20, 22)
        expected = (
            (' 10 %', _eta_range(4, 7), speed1),
            (' 20 %', _eta_range(4, 7), speed1),
            (' 30 %', _eta_range(3, 4), speed1),
            (' 40 %', _eta_range(3, 3), speed1),
            (' 50 %', _eta_range(2, 2), speed1),
            (' 60 %', _eta_range(2, 2), speed1),
            (' 70 %', _eta_range(1, 1), speed1),
            (' 80 %', _eta_range(1, 1), speed1),
            (' 90 %', _eta_range(0, 0), speed1),
            ('100 %', _eta_range(4, 5, 'Done'), speed2),
        )
        bar = Progress(maxval=100000).start()
        for i, v in enumerate(range(10000, 100000, 10000)):
            time.sleep(0.5)
            bar.update(v)
            p, e, s = expected[i]
            self.assertEqual(bar.percentage, p)
            self.assertIn(bar.ETA, e, p)
            self.assertIn(bar.speed, s)
        bar.stop()
        p, e, s = expected[i + 1]
        self.assertEqual(bar.percentage, p)
        self.assertIn(bar.ETA, e, p)
        self.assertIn(bar.speed, s)

    # Of late, the speed tests keep failing on AppVeyor and Windows
    @unittest.skipIf(IN_GITHUB_WORKFLOW or (os.name == 'nt' and os.environ.get('APPVEYOR') == 'True'),
                     'Test disabled on some environments due to performance')
    def test_unknown(self):
        if os.name == 'nt':
            speed = _speed_range(17, 20)
        else:
            speed = _speed_range(17, 19)
        expected = (
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            (' ?? %', 'ETA : ??:??:??', speed),
            ('100 %', 'Done: 00:00:04', speed),
        )
        bar = Progress(maxval=None).start()
        for i, v in enumerate(range(10000, 100000, 10000)):
            time.sleep(0.5)
            bar.update(v)
            p, e, s = expected[i]
            self.assertEqual(bar.percentage, p)
            self.assertEqual(bar.ETA, e)
            self.assertIn(bar.speed, s)
        bar.stop()
        p, e, s = expected[i + 1]
        self.assertEqual(bar.percentage, p)
        self.assertEqual(bar.ETA, e)
        self.assertIn(bar.speed, s)


class FileOpsTestCase(DistlibTestCase):

    def setUp(self):
        self.fileop = FileOperator()
        self.workdir = tempfile.mkdtemp()

    def tearDown(self):
        if os.path.isdir(self.workdir):
            shutil.rmtree(self.workdir)

    def test_ensure_dir(self):
        td = self.workdir
        os.rmdir(td)
        self.fileop.ensure_dir(td)
        self.assertTrue(os.path.exists(td))
        self.fileop.dry_run = True
        os.rmdir(td)
        self.fileop.ensure_dir(td)
        self.assertFalse(os.path.exists(td))

    def test_ensure_removed(self):
        td = self.workdir
        self.assertTrue(os.path.exists(td))
        self.fileop.dry_run = True
        self.fileop.ensure_removed(td)
        self.assertTrue(os.path.exists(td))
        self.fileop.dry_run = False
        self.fileop.ensure_removed(td)
        self.assertFalse(os.path.exists(td))

    def test_is_writable(self):
        sd = 'subdir'
        ssd = 'subsubdir'
        path = os.path.join(self.workdir, sd, ssd)
        os.makedirs(path)
        path = os.path.join(path, 'test')
        self.assertTrue(self.fileop.is_writable(path))
        if os.name == 'posix':
            self.assertFalse(self.fileop.is_writable('/etc'))

    def test_byte_compile(self):
        path = os.path.join(self.workdir, 'hello.py')
        dpath = cache_from_source(path, True)
        self.fileop.write_text_file(path, 'print("Hello, world!")', 'utf-8')
        self.fileop.byte_compile(path, optimize=False)
        self.assertTrue(os.path.exists(dpath))

    def write_some_files(self):
        path = os.path.join(self.workdir, 'file1')
        written = []
        self.fileop.write_text_file(path, 'test', 'utf-8')
        written.append(path)
        path = os.path.join(self.workdir, 'file2')
        self.fileop.copy_file(written[0], path)
        written.append(path)
        path = os.path.join(self.workdir, 'dir1')
        self.fileop.ensure_dir(path)
        return set(written), set([path])

    def test_copy_check(self):
        srcpath = os.path.join(self.workdir, 'file1')
        self.fileop.write_text_file(srcpath, 'test', 'utf-8')
        dstpath = os.path.join(self.workdir, 'file2')
        os.mkdir(dstpath)
        self.assertRaises(ValueError, self.fileop.copy_file, srcpath, dstpath)
        os.rmdir(dstpath)
        if os.name == 'posix':  # symlinks available
            linkpath = os.path.join(self.workdir, 'file3')
            self.fileop.write_text_file(linkpath, 'linkdest', 'utf-8')
            os.symlink(linkpath, dstpath)
            self.assertRaises(ValueError, self.fileop.copy_file, srcpath, dstpath)

    def test_commit(self):
        # will assert if record isn't set
        self.assertRaises(AssertionError, self.fileop.commit)
        self.fileop.record = True
        expected = self.write_some_files()
        actual = self.fileop.commit()
        self.assertEqual(actual, expected)
        self.assertFalse(self.fileop.record)

    def test_rollback(self):
        # will assert if record isn't set
        self.assertRaises(AssertionError, self.fileop.commit)
        self.fileop.record = True
        self.write_some_files()
        self.fileop.rollback()
        self.assertEqual(os.listdir(self.workdir), [])
        self.assertFalse(self.fileop.record)


class GlobTestCaseBase(TempdirManager, DistlibTestCase):

    def build_files_tree(self, files):
        tempdir = self.mkdtemp()
        for filepath in files:
            is_dir = filepath.endswith('/')
            filepath = os.path.join(tempdir, *filepath.split('/'))
            if is_dir:
                dirname = filepath
            else:
                dirname = os.path.dirname(filepath)
            if dirname and not os.path.exists(dirname):
                os.makedirs(dirname)
            if not is_dir:
                self.write_file(filepath, 'babar')
        return tempdir

    @staticmethod
    def os_dependent_path(path):
        path = path.rstrip('/').split('/')
        return os.path.join(*path)

    def clean_tree(self, spec):
        files = []
        for path, includes in spec.items():
            if includes:
                files.append(self.os_dependent_path(path))
        return sorted(files)


class GlobTestCase(GlobTestCaseBase):

    def assertGlobMatch(self, glob, spec):
        tempdir = self.build_files_tree(spec)
        expected = self.clean_tree(spec)
        os.chdir(tempdir)
        result = sorted(iglob(glob))
        self.assertEqual(expected, result)

    def test_regex_rich_glob(self):
        matches = RICH_GLOB.findall(r"babar aime les {fraises} est les {huitres}")
        self.assertEqual(["fraises", "huitres"], matches)

    def test_simple_glob(self):
        glob = '*.tp?'
        spec = {'coucou.tpl': True, 'coucou.tpj': True, 'Donotwant': False}
        self.assertGlobMatch(glob, spec)

    def test_simple_glob_in_dir(self):
        glob = os.path.join('babar', '*.tp?')
        spec = {'babar/coucou.tpl': True, 'babar/coucou.tpj': True, 'babar/toto.bin': False, 'Donotwant': False}
        self.assertGlobMatch(glob, spec)

    def test_recursive_glob_head(self):
        glob = os.path.join('**', 'tip', '*.t?l')
        spec = {
            'babar/zaza/zuzu/tip/coucou.tpl': True,
            'babar/z/tip/coucou.tpl': True,
            'babar/tip/coucou.tpl': True,
            'babar/zeop/tip/babar/babar.tpl': False,
            'babar/z/tip/coucou.bin': False,
            'babar/toto.bin': False,
            'zozo/zuzu/tip/babar.tpl': True,
            'zozo/tip/babar.tpl': True,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_recursive_glob_tail(self):
        glob = os.path.join('babar', '**')
        spec = {
            'babar/zaza/': True,
            'babar/zaza/zuzu/': True,
            'babar/zaza/zuzu/babar.xml': True,
            'babar/zaza/zuzu/toto.xml': True,
            'babar/zaza/zuzu/toto.csv': True,
            'babar/zaza/coucou.tpl': True,
            'babar/bubu.tpl': True,
            'zozo/zuzu/tip/babar.tpl': False,
            'zozo/tip/babar.tpl': False,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_recursive_glob_middle(self):
        glob = os.path.join('babar', '**', 'tip', '*.t?l')
        spec = {
            'babar/zaza/zuzu/tip/coucou.tpl': True,
            'babar/z/tip/coucou.tpl': True,
            'babar/tip/coucou.tpl': True,
            'babar/zeop/tip/babar/babar.tpl': False,
            'babar/z/tip/coucou.bin': False,
            'babar/toto.bin': False,
            'zozo/zuzu/tip/babar.tpl': False,
            'zozo/tip/babar.tpl': False,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_glob_set_tail(self):
        glob = os.path.join('bin', '*.{bin,sh,exe}')
        spec = {
            'bin/babar.bin': True,
            'bin/zephir.sh': True,
            'bin/celestine.exe': True,
            'bin/cornelius.bat': False,
            'bin/cornelius.xml': False,
            'toto/yurg': False,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_glob_set_middle(self):
        glob = os.path.join('xml', '{babar,toto}.xml')
        spec = {
            'xml/babar.xml': True,
            'xml/toto.xml': True,
            'xml/babar.xslt': False,
            'xml/cornelius.sgml': False,
            'xml/zephir.xml': False,
            'toto/yurg.xml': False,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_glob_set_head(self):
        glob = os.path.join('{xml,xslt}', 'babar.*')
        spec = {
            'xml/babar.xml': True,
            'xml/toto.xml': False,
            'xslt/babar.xslt': True,
            'xslt/toto.xslt': False,
            'toto/yurg.xml': False,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_glob_all(self):
        dirs = '{%s,%s}' % (os.path.join('xml', '*'), os.path.join('xslt', '**'))
        glob = os.path.join(dirs, 'babar.xml')
        spec = {
            'xml/a/babar.xml': True,
            'xml/b/babar.xml': True,
            'xml/a/c/babar.xml': False,
            'xslt/a/babar.xml': True,
            'xslt/b/babar.xml': True,
            'xslt/a/c/babar.xml': True,
            'toto/yurg.xml': False,
            'Donotwant': False
        }
        self.assertGlobMatch(glob, spec)

    def test_invalid_glob_pattern(self):
        invalids = [
            'ppooa**',
            'azzaeaz4**/',
            '/**ddsfs',
            '**##1e"&e',
            'DSFb**c009',
            '{',
            '{aaQSDFa',
            '}',
            'aQSDFSaa}',
            '{**a,',
            ',**a}',
            '{a**,',
            ',b**}',
            '{a**a,babar}',
            '{bob,b**z}',
        ]
        for pattern in invalids:
            self.assertRaises(ValueError, iglob, pattern)

    def test_parse_requirement(self):
        # Empty requirements
        for empty in ('', '#this should be ignored'):
            self.assertIsNone(parse_requirement(empty))

        # Invalid requirements
        for invalid in ('a (', 'a/', 'a$', 'a [', 'a () [],', 'a 1.2'):
            self.assertRaises(SyntaxError, parse_requirement, invalid)

        # Valid requirements
        def validate(r, values):
            self.assertEqual(r.name, values[0])
            self.assertEqual(r.constraints, values[1])
            self.assertEqual(r.extras, values[2])
            self.assertEqual(r.requirement, values[3])
            self.assertEqual(r.url, values[4])

        r = parse_requirement('a')
        validate(r, ('a', None, None, 'a', None))
        r = parse_requirement('a >= 1.2, <2.0,!=1.7')
        validate(r, ('a', [('>=', '1.2'), ('<', '2.0'), ('!=', '1.7')], None, 'a >= 1.2, < 2.0, != 1.7', None))
        r = parse_requirement('a [ab,cd , ef] >= 1.2, <2.0')
        validate(r, ('a', [('>=', '1.2'), ('<', '2.0')], ['ab', 'cd', 'ef'], 'a >= 1.2, < 2.0', None))
        r = parse_requirement('a[]')
        validate(r, ('a', None, None, 'a', None))
        r = parse_requirement('a (== 1.2.*, != 1.2.1.*)')
        validate(r, ('a', [('==', '1.2.*'), ('!=', '1.2.1.*')], None, 'a == 1.2.*, != 1.2.1.*', None))
        r = parse_requirement('a @ http://domain.com/path#abc=def')
        validate(r, ('a', None, None, 'a', 'http://domain.com/path#abc=def'))
        # See issue #148
        r = parse_requirement('a >=3.6')
        validate(r, ('a', [('>=', '3.6')], None, 'a >= 3.6', None))
        r = parse_requirement('a >=3.6,')
        validate(r, ('a', [('>=', '3.6')], None, 'a >= 3.6', None))

        if False:  # TODO re-enable
            for e in ('*', ':*:', ':meta:', '-', '-abc'):
                r = parse_requirement('a [%s]' % e)
                validate(r, ('a', None, [e], 'a', None))

    def test_write_exports(self):
        exports = {
            'foo': {
                'v1': ExportEntry('v1', 'p1', 's1', []),
                'v2': ExportEntry('v2', 'p2', 's2', ['f2=a', 'g2']),
            },
            'bar': {
                'v3': ExportEntry('v3', 'p3', 's3', ['f3', 'g3=h']),
                'v4': ExportEntry('v4', 'p4', 's4', ['f4', 'g4']),
            },
        }

        fd, fn = tempfile.mkstemp()
        try:
            os.close(fd)
            with open(fn, 'wb') as f:
                write_exports(exports, f)
            with open(fn, 'rb') as f:
                actual = read_exports(f)
            self.assertEqual(actual, exports)
        finally:
            os.remove(fn)

    def test_get_extras(self):
        cases = (
            (['*'], ['i18n'], set(['i18n'])),
            (['*', '-bar'], ['foo', 'bar'], set(['foo'])),
        )
        for requested, available, expected in cases:
            actual = get_extras(requested, available)
            self.assertEqual(actual, expected)


if __name__ == '__main__':  # pragma: no cover
    unittest.main()