class TestDiffLib(TestController):

    @parametrize('diff_fixture', DIFF_FIXTURES)

    def test_diff(self, diff_fixture):

        raw_diff = fixture.load_resource(diff_fixture, strip=False)

        vcs = 'hg'

        if diff_fixture.startswith('git_'):

            vcs = 'git'

        diff_processor = DiffProcessor(raw_diff, vcs=vcs)

        data = [(x['filename'], x['operation'], x['stats']) for x in diff_processor.parsed]

        expected_data = DIFF_FIXTURES[diff_fixture]

        assert expected_data == data

    def test_diff_markup(self):

        raw_diff = fixture.load_resource('markuptest.diff', strip=False)

        diff_processor = DiffProcessor(raw_diff)

        chunks = diff_processor.parsed[0]['chunks']

        assert not chunks[0]

        #from pprint import pprint; pprint(chunks[1])

        l = ['\n']

        for d in chunks[1]:

            l.append('%(action)-7s %(new_lineno)3s %(old_lineno)3s %(line)r\n' % d)

        s = ''.join(l)

        assert s == r'''

context ... ... u'@@ -51,6 +51,13 @@\n'

unmod    51  51 u'<u>\t</u>begin();\n'

unmod    52  52 u'<u>\t</u>\n'

add      53     u'<u>\t</u>int foo;<u class="cr"></u>\n'

add      54     u'<u>\t</u>int bar; <u class="cr"></u>\n'

add      55     u'<u>\t</u>int baz;<u>\t</u><u class="cr"></u>\n'

add      56     u'<u>\t</u>int space; <i></i>'

add      57     u'<u>\t</u>int tab;<u>\t</u>\n'

add      58     u'<u>\t</u>\n'

unmod    59  53 u' <i></i>'

del          54 u'<u>\t</u>#define MAX_STEPS (48)\n'

add      60     u'<u>\t</u><u class="cr"></u>\n'

add      61     u'<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'

unmod    62  55 u'\n'

del          56 u'<u>\t</u>#define MIN_STEPS (<del>48</del>)\n'

add      63     u'<u>\t</u>#define MIN_STEPS (<ins>42</ins>)\n'

'''

    if 'perms_updates' not in kwargs:

        kwargs['perms_updates'] = []

    r = RepoModel().update(name, **kwargs)

    return r

class TestRepoGroups(TestController):

    def setup_method(self, method):

        self.g1 = fixture.create_repo_group(u'test1', skip_if_exists=True)

        self.g2 = fixture.create_repo_group(u'test2', skip_if_exists=True)

        self.g3 = fixture.create_repo_group(u'test3', skip_if_exists=True)

    def teardown_method(self, method):

        Session.remove()

    def __check_path(self, *path):

        """

        Checks the path for existence !

        """

        path = [TESTS_TMP_PATH] + list(path)

        path = os.path.join(*path)

        return os.path.isdir(path)

    def __delete_group(self, id_):

        RepoGroupModel().delete(id_)

    def test_create_group(self):

        g = fixture.create_repo_group(u'newGroup')

        Session().commit()

        assert g.full_path == 'newGroup'

        assert self.__check_path('newGroup')

    def test_create_same_name_group(self):

        with pytest.raises(IntegrityError):

            fixture.create_repo_group(u'newGroup')

        Session().rollback()

    def test_same_subgroup(self):

        sg1 = fixture.create_repo_group(u'sub1', parent_group_id=self.g1.group_id)

        assert sg1.parent_group == self.g1

        assert sg1.full_path == 'test1/sub1'

        assert self.__check_path('test1', 'sub1')

        ssg1 = fixture.create_repo_group(u'subsub1', parent_group_id=sg1.group_id)

        assert ssg1.parent_group == sg1

        assert ssg1.full_path == 'test1/sub1/subsub1'

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url=clone_url)

        if vt.repo_type == 'git':

            _check_proper_git_push(stdout, stderr)

        elif vt.repo_type == 'hg':

            assert 'pushing to' in stdout

            assert 'Repository size' in stdout

            assert 'Last revision is now' in stdout

        action_parts = [ul.action.split(':', 1) for ul in UserLog.query().order_by(UserLog.user_log_id)]

        assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == \

            [(u'pull', 0), (u'push', 3)]

    @parametrize_vcs_test

    def test_pull(self, webserver, testfork, vt):

        UserLog.query().delete()

        Session().commit()

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'init', dest_dir)

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url)

        if vt.repo_type == 'git':

            assert 'FETCH_HEAD' in stderr

        elif vt.repo_type == 'hg':

            assert 'new changesets' in stdout

        action_parts = [ul.action for ul in UserLog.query().order_by(UserLog.user_log_id)]

        assert action_parts == [u'pull']

    @parametrize_vcs_test

    def test_push_invalidates_cache(self, webserver, testfork, vt):

        pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               == testfork[vt.repo_type]).scalar()

        if not key:

            key = CacheInvalidation(testfork[vt.repo_type], testfork[vt.repo_type])

            Session().add(key)

        key.cache_active = True

        Session().commit()

        dest_dir = _get_tmp_dir()