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

# 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/>.

"""

Helpers for fixture generation

"""

import logging

import os

import shutil

import tarfile

from os.path import dirname

import mock

from tg import request

from tg.util.webtest import test_context

from kallithea.model.db import Repository, User, RepoGroup, UserGroup, Gist, ChangesetStatus

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.gist import GistModel

from kallithea.model.scm import ScmModel

from kallithea.model.comment import ChangesetCommentsModel

from kallithea.model.changeset_status import ChangesetStatusModel

from kallithea.model.pull_request import CreatePullRequestAction#, CreatePullRequestIterationAction, PullRequestModel

from kallithea.lib import helpers

from kallithea.lib.auth import AuthUser

from kallithea.lib.db_manage import DbManage

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.tests.base import invalidate_all_caches, GIT_REPO, HG_REPO, \

log = logging.getLogger(__name__)

FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')

def error_function(*args, **kwargs):

    raise Exception('Total Crash !')

class Fixture(object):

    def __init__(self):

        pass

    def anon_access(self, status):

        """

        Context manager for controlling anonymous access.

        Anon access will be set and committed, but restored again when exiting the block.

        Usage:

        fixture = Fixture()

        with fixture.anon_access(False):

            stuff

        """

        class context(object):

            def __enter__(self):

                anon = User.get_default_user()

                self._before = anon.active

                anon.active = status

                Session().commit()

                invalidate_all_caches()

            def __exit__(self, exc_type, exc_val, exc_tb):

                anon = User.get_default_user()

                anon.active = self._before

                Session().commit()

        return context()

    def _get_repo_create_params(self, **custom):

        """Return form values to be validated through RepoForm"""

        defs = dict(

            repo_name=None,

            repo_type='hg',

        return user_group

    def destroy_user_group(self, usergroupid):

        UserGroupModel().delete(user_group=usergroupid, force=True)

        Session().commit()

    def create_gist(self, **kwargs):

        form_data = {

            'description': u'new-gist',

            'owner': TEST_USER_ADMIN_LOGIN,

            'gist_type': Gist.GIST_PUBLIC,

            'lifetime': -1,

            'gist_mapping': {'filename1.txt': {'content': 'hello world'}}

        }

        form_data.update(kwargs)

        gist = GistModel().create(

            description=form_data['description'], owner=form_data['owner'],

            gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],

            lifetime=form_data['lifetime']

        )

        Session().commit()

        return gist

    def destroy_gists(self, gistid=None):

        for g in Gist.query():

            if gistid:

                if gistid == g.gist_access_id:

                    GistModel().delete(g)

            else:

                GistModel().delete(g)

        Session().commit()

    def load_resource(self, resource_name, strip=True):

        with open(os.path.join(FIXTURES, resource_name), 'rb') as f:

            source = f.read()

            if strip:

                source = source.strip()

        return source

    def commit_change(self, repo, filename, content, message, vcs_type,

                      parent=None, newfile=False, author=None):

        repo = Repository.get_by_repo_name(repo)

        _cs = parent

        if parent is None:

            _cs = EmptyChangeset(alias=vcs_type)

        if author is None:

        if newfile:

            nodes = {

                filename: {

                    'content': content

                }

            }

            cs = ScmModel().create_nodes(

                user=TEST_USER_ADMIN_LOGIN, repo=repo,

                message=message,

                nodes=nodes,

                parent_cs=_cs,

                author=author,

            )

        else:

            cs = ScmModel().commit_change(

                repo=repo.scm_instance, repo_name=repo.repo_name,

                cs=parent, user=TEST_USER_ADMIN_LOGIN,

                author=author,

                message=message,

                content=content,

                f_path=filename

            )

        return cs

    def review_changeset(self, repo, revision, status, author=TEST_USER_ADMIN_LOGIN):

        comment = ChangesetCommentsModel().create(u"review comment", repo, author, revision=revision, send_email=False)

        csm = ChangesetStatusModel().set_status(repo, ChangesetStatus.STATUS_APPROVED, author, comment, revision=revision)

        Session().commit()

        return csm

    def create_pullrequest(self, testcontroller, repo_name, pr_src_rev, pr_dst_rev, title=u'title'):

        org_ref = 'branch:stable:%s' % pr_src_rev

        other_ref = 'branch:default:%s' % pr_dst_rev

        with test_context(testcontroller.app): # needed to be able to mock request user

            org_repo = other_repo = Repository.get_by_repo_name(repo_name)

            owner_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

            reviewers = [User.get_by_username(TEST_USER_REGULAR_LOGIN)]

            request.authuser = request.user = AuthUser(dbuser=owner_user)

            # creating a PR sends a message with an absolute URL - without routing that requires mocking

            with mock.patch.object(helpers, 'url', (lambda arg, qualified=False, **kwargs: ('https://localhost' if qualified else '') + '/fake/' + arg)):

                cmd = CreatePullRequestAction(org_repo, other_repo, org_ref, other_ref, title, u'No description', owner_user, reviewers)

                pull_request = cmd.execute()

            Session().commit()

        return pull_request.pull_request_id

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

    def test_remove_raise_node_already_removed(self):

        self.test_add() # Performs first commit

        node = FileNode(self.nodes[0].path)

        self.imc.remove(node)

        self.assertRaises(NodeAlreadyRemovedError, self.imc.remove, node)

    def test_remove_raise_node_already_changed(self):

        self.test_add()  # Performs first commit

        node = FileNode(self.nodes[0].path, content='Bending time')

        self.imc.change(node)

        self.assertRaises(NodeAlreadyChangedError, self.imc.remove, node)

    def test_reset(self):

        self.imc.add(FileNode('foo', content='bar'))

        #self.imc.change(FileNode('baz', content='new'))

        #self.imc.remove(FileNode('qwe'))

        self.imc.reset()

        self.assertTrue(not any((self.imc.added, self.imc.changed,

            self.imc.removed)))

    def test_multiple_commits(self):

        N = 3  # number of commits to perform

        last = None

        for x in xrange(N):

            fname = 'file%s' % str(x).rjust(5, '0')

            content = 'foobar\n' * x

            node = FileNode(fname, content=content)

            self.imc.add(node)

            commit = self.imc.commit(u"Commit no. %s" % (x + 1), author=u'vcs')

            self.assertTrue(last != commit)

            last = commit

        # Check commit number for same repo

        self.assertEqual(len(self.repo.revisions), N)

        # Check commit number for recreated repo

        backend = self.get_backend()

        repo = backend(self.repo_path)

        self.assertEqual(len(repo.revisions), N)

    def test_date_attr(self):

        node = FileNode('foobar.txt', content='Foobared!')

        self.imc.add(node)

        date = datetime.datetime(1985, 1, 30, 1, 45)

        commit = self.imc.commit(u"Committed at time when I was born ;-)",

        self.assertEqual(commit.date, date)

class BackendBaseTestCase(unittest.TestCase):

    """

    Base test class for tests which requires repository.

    """

    backend_alias = 'hg'

    commits = [

        {

            'message': 'Initial commit',

            'author': 'Joe Doe <joe.doe@example.com>',

            'date': datetime.datetime(2010, 1, 1, 20),

            'added': [

                FileNode('foobar', content='Foobar'),

                FileNode('foobar2', content='Foobar II'),

                FileNode('foo/bar/baz', content='baz here!'),

            ],

        },

    ]

    def get_backend(self):

        return vcs.get_backend(self.backend_alias)

    def get_commits(self):

        """

        Returns list of commits which builds repository for each tests.

        """

        if hasattr(self, 'commits'):

            return self.commits

    def get_new_repo_path(self):

        """

        Returns newly created repository's directory.

        """

        backend = self.get_backend()

        key = '%s-%s' % (self.backend_alias, str(time.time()))

        repo_path = get_new_dir(key)

        return repo_path

    def setUp(self):

        Backend = self.get_backend()

        self.backend_class = Backend

        self.repo_path = self.get_new_repo_path()

        self.repo = Backend(self.repo_path, create=True)

        self.imc = self.repo.in_memory_changeset

here = os.path.abspath(os.path.dirname(__file__))

def _get_meta_var(name, data, callback_handler=None):

    import re

    matches = re.compile(r'(?:%s)\s*=\s*(.*)' % name).search(data)

    if matches:

        if not callable(callback_handler):

            callback_handler = lambda v: v

        return callback_handler(eval(matches.groups()[0]))

_meta = open(os.path.join(here, 'kallithea', '__init__.py'), 'rb')

_metadata = _meta.read()

_meta.close()

callback = lambda V: ('.'.join(map(str, V[:3])) + '.'.join(V[3:]))

__version__ = _get_meta_var('VERSION', _metadata, callback)

__license__ = _get_meta_var('__license__', _metadata)

__author__ = _get_meta_var('__author__', _metadata)

__url__ = _get_meta_var('__url__', _metadata)

# defines current platform

__platform__ = platform.system()

is_windows = __platform__ in ['Windows']

requirements = [

    "alembic>=0.8.0,<0.9",

    "GearBox<1",

    "waitress>=0.8.8,<1.0",

    "webob>=1.7,<2",

    "backlash >= 0.1.2, < 1.0.0",

    "TurboGears2 >= 2.3.10, < 3.0.0",

    "tgext.routes >= 0.2.0, < 1.0.0",

    "Beaker>=1.7.0,<2",

    "WebHelpers==1.3",

    "formencode>=1.2.4,<=1.2.6",

    "SQLAlchemy>=1.1,<1.2",

    "Mako>=0.9.0,<=1.0.0",

    "pygments>=1.5",

    "whoosh>=2.5.0,<=2.5.7",

    "celery>=3.1,<3.2",

    "babel>=0.9.6,<2.4",

    "python-dateutil>=1.5.0,<2.0.0",

    "markdown==2.2.1",

    "docutils>=0.8.1",

    "URLObject==2.3.4",

    "Routes==1.13",

    "mercurial>=2.9,<4.5",

    "decorator >= 3.3.2",

    "Paste >= 2.0.3, < 3.0",

]

if sys.version_info < (2, 7):

    requirements.append("importlib==1.0.1")

    requirements.append("argparse")

if not is_windows:

    requirements.append("bcrypt>=3.1.0")

dependency_links = [

]

classifiers = [

    'Development Status :: 4 - Beta',

    'Environment :: Web Environment',

    'Framework :: Pylons',

    'Intended Audience :: Developers',

    'License :: OSI Approved :: GNU General Public License (GPL)',

    'Operating System :: OS Independent',

    'Programming Language :: Python',

    'Programming Language :: Python :: 2.6',

    'Programming Language :: Python :: 2.7',

    'Topic :: Software Development :: Version Control',

]

# additional files from project that goes somewhere in the filesystem

# relative to sys.prefix

data_files = []

description = ('Kallithea is a fast and powerful management tool '

               'for Mercurial and Git with a built in push/pull server, '

               'full text search and code-review.')

keywords = ' '.join([

    'kallithea', 'mercurial', 'git', 'code review',

    'repo groups', 'ldap', 'repository management', 'hgweb replacement',

    'hgwebdir', 'gitweb replacement', 'serving hgweb',

])

# long description

README_FILE = 'README.rst'

try:

    long_description = open(README_FILE).read()

except IOError as err: