from itertools import product

except ImportError:

    def product(*args, **kwds):

        # product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy

        # product(range(2), repeat=3) --> 000 001 010 011 100 101 110 111

        pools = map(tuple, args) * kwds.get('repeat', 1)

        result = [[]]

        for pool in pools:

            result = [x + [y] for x in result for y in pool]

        for prod in result:

            yield tuple(prod)

#==============================================================================

# BytesIO

#==============================================================================

try:

    from io import BytesIO

except ImportError:

    from cStringIO import StringIO as BytesIO

#==============================================================================

# deque

#==============================================================================

if __py_version__ >= (2, 6):

    from collections import deque

else:

    #need to implement our own deque with maxlen

    class deque(object):

            if not hasattr(self, 'data'):

                self.left = self.right = 0

                self.data = {}

            self.extend(iterable)

        def append(self, x):

            self.data[self.right] = x

            self.right += 1

            if self.maxlen != -1 and len(self) > self.maxlen:

                self.popleft()

        def appendleft(self, x):

            self.left -= 1

            self.data[self.left] = x

            if self.maxlen != -1 and len(self) > self.maxlen:

                self.pop()

        def pop(self):

            if self.left == self.right:

                raise IndexError('cannot pop from empty deque')

            self.right -= 1

            elem = self.data[self.right]

            del self.data[self.right]

            return elem

        def popleft(self):

            if self.left == self.right:

            return (tuple(self),)

        def __setstate__(self, s):

            self.__init__(s[0])

        def __hash__(self):

            raise TypeError

        def __copy__(self):

            return self.__class__(self)

        def __deepcopy__(self, memo={}):

            from copy import deepcopy

            result = self.__class__()

            memo[id(self)] = result

            result.__init__(deepcopy(tuple(self), memo))

            return result

#==============================================================================

# threading.Event

#==============================================================================

if __py_version__ >= (2, 6):

else:

    def Event(*args, **kwargs):

        return _Event(*args, **kwargs)

    class _Event(_Verbose):

        # After Tim Peters' event class (without is_posted())

        def __init__(self, verbose=None):

            _Verbose.__init__(self, verbose)

            self.__cond = Condition(Lock())

            self.__flag = False

        def isSet(self):

            return self.__flag

        is_set = isSet

        def set(self):

            self.__cond.acquire()

            try:

                self.__flag = True

                self.__cond.notify_all()

            finally:

input, output, error streams into a meaningfull, non-blocking, concurrent

stream processor exposing the output data as an iterator fitting to be a

return value passed by a WSGI applicaiton to a WSGI server per PEP 3333.

Copyright (c) 2011  Daniel Dotsenko <dotsa@hotmail.com>

This file is part of git_http_backend.py Project.

git_http_backend.py Project is free software: you can redistribute it and/or

modify it under the terms of the GNU Lesser General Public License as

published by the Free Software Foundation, either version 2.1 of the License,

or (at your option) any later version.

git_http_backend.py Project is distributed in the hope that it will be useful,

but WITHOUT ANY WARRANTY; without even the implied warranty of

MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License

along with git_http_backend.py Project.

If not, see <http://www.gnu.org/licenses/>.

'''

import os

import subprocess

    """

    Normal writing into pipe-like is blocking once the buffer is filled.

    This thread allows a thread to seep data from a file-like into a pipe

    without blocking the main thread.

    We close inpipe once the end of the source stream is reached.

    """

    def __init__(self, source):

        super(StreamFeeder, self).__init__()

        self.daemon = True

        filelike = False

        else:  # can be either file pointer or file-like

            if type(source) in (int, long):  # file pointer it is

                ## converting file descriptor (int) stdin into file-like

                try:

                    source = os.fdopen(source, 'rb', 16384)

                except Exception:

                    pass

            # let's see if source is file-like by now

            try:

                filelike = source.read

            except Exception:

                pass

        if not filelike and not self.bytes:

            raise TypeError("StreamFeeder's source object must be a readable "

                            "file-like, a file descriptor, or a string-like.")

        self.source = source

        self.readiface, self.writeiface = os.pipe()

    def run(self):

        t = self.writeiface

        if self.bytes:

            os.write(t, self.bytes)

        else:

            s = self.source

            b = s.read(4096)

            while b:

                os.write(t, b)

                b = s.read(4096)

        os.close(t)

    @property

    def output(self):

        return self.readiface

    def __init__(self, source, target, buffer_size, chunk_size):

        super(InputStreamChunker, self).__init__()

        self.daemon = True  # die die die.

        self.source = source

        self.target = target

        self.chunk_count_max = int(buffer_size / chunk_size) + 1

        self.chunk_size = chunk_size

        self.data_added = Event()

        self.data_added.clear()

        self.keep_reading = Event()

        self.keep_reading.set()

        self.EOF = Event()

        self.EOF.clear()

        self.go = Event()

        self.go.set()

    def stop(self):

        self.go.clear()

        self.EOF.set()

        try:

            # this is not proper, but is done to force the reader thread let

            # go of the input because, if successful, .close() will send EOF

            # down the pipe.

            self.source.close()

        except:

            pass

    def run(self):

        s = self.source

        t = self.target

        cs = self.chunk_size

        ccm = self.chunk_count_max

        kr = self.keep_reading

        da = self.data_added

        go = self.go

        b = s.read(cs)

        while b and go.is_set():

            if len(t) > ccm:

                kr.clear()

                kr.wait(2)

#                # this only works on 2.7.x and up

#                if not kr.wait(10):

#                    raise Exception("Timed out while waiting for input to be read.")

                # instead we'll use this

                if len(t) > ccm + 3:

                    raise IOError("Timed out while waiting for input from subprocess.")

            t.append(b)

            da.set()

            b = s.read(cs)

        self.EOF.set()

        da.set()  # for cases when done but there was no input.

class BufferedGenerator():

    '''

    Class behaves as a non-blocking, buffered pipe reader.

    Reads chunks of data (through a thread)

    from a blocking pipe, and attaches these to an array (Deque) of chunks.

    Reading is halted in the thread when max chunks is internally buffered.

    The .next() may operate in blocking or non-blocking fashion by yielding

        else:

            maxlen = None

        self.data = deque(starting_values, maxlen)

        self.worker = InputStreamChunker(source, self.data, buffer_size,

                                         chunk_size)

        if starting_values:

            self.worker.data_added.set()

        self.worker.start()

    ####################

    # Generator's methods

    ####################

    def __iter__(self):

        return self

    def next(self):

        while not len(self.data) and not self.worker.EOF.is_set():

            self.worker.data_added.clear()

            self.worker.data_added.wait(0.2)

        if len(self.data):

            self.worker.keep_reading.set()

        elif self.worker.EOF.is_set():

            raise StopIteration

    def throw(self, type, value=None, traceback=None):

        if not self.worker.EOF.is_set():

            raise type(value)

    def start(self):

        self.worker.start()

    def stop(self):

        self.worker.stop()

    def close(self):

        try:

            self.worker.stop()

            self.throw(GeneratorExit)

        except (GeneratorExit, StopIteration):

            pass

    def __del__(self):

        self.close()

    ####################

# -*- coding: utf-8 -*-

"""

    rhodecode.tests.test_libs

    ~~~~~~~~~~~~~~~~~~~~~~~~~

    Package for testing various lib/helper functions in rhodecode

    :created_on: Jun 9, 2011

    :copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>

    :license: GPLv3, see COPYING for more details.

"""

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import unittest

import datetime

import hashlib

import mock

from rhodecode.tests import *

proto = 'http'

TEST_URLS = [

    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

     '%s://127.0.0.1:8080' % proto),

    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],

     '%s://domain.org' % proto),

    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',

                                                '8080'],

     '%s://domain.org:8080' % proto),

]

proto = 'https'