"""

Helpers for fixture generation

"""

import os

import unittest

from rhodecode.tests import *

from rhodecode.model.db import Repository, User

from rhodecode.model.meta import Session

from rhodecode.model.repo import RepoModel

class Fixture(object):

    def __init__(self):

        pass

    def create_repo(self, name, **kwargs):

        form_data = _get_repo_create_params(repo_name=name, **kwargs)

        cur_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

        RepoModel().create(form_data, cur_user)

        return Repository.get_by_repo_name(name)

    def create_fork(self, repo_to_fork, fork_name, **kwargs):

        repo_to_fork = Repository.get_by_repo_name(repo_to_fork)

        vcs_type = repo_to_fork.repo_type

        form_data = dict(

            repo_name=fork_name,

            repo_name_full=fork_name,

            repo_group=None,

            repo_type=vcs_type,

            description='',

            private=False,

            copy_permissions=False,

            landing_rev='tip',

            update_after_clone=False,

            fork_parent_id=repo_to_fork,

        )

        cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        RepoModel().create_fork(form_data, cur_user=cur_user)

        Session().commit()

        return Repository.get_by_repo_name(fork_name)

import os

import unittest

from rhodecode.tests import *

from rhodecode.model.meta import Session

from rhodecode.tests.fixture import Fixture

from rhodecode.model.repo import RepoModel

from rhodecode.model.db import Repository

from rhodecode.lib.exceptions import AttachedForksError

fixture = Fixture()

class TestRepos(unittest.TestCase):

    def setUp(self):

        pass

    def tearDown(self):

        Session.remove()

    def test_remove_repo(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        RepoModel().delete(repo=repo)

        Session().commit()

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-1'))

    def test_remove_repo_repo_raises_exc_when_attached_forks(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        fixture.create_fork(repo.repo_name, 'test-repo-fork-1')

        Session().commit()

        self.assertRaises(AttachedForksError, lambda: RepoModel().delete(repo=repo))

    def test_remove_repo_delete_forks(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        fork = fixture.create_fork(repo.repo_name, 'test-repo-fork-1')

        Session().commit()

        #fork of fork

        fixture.create_fork(fork.repo_name, 'test-repo-fork-fork-1')

        Session().commit()

        RepoModel().delete(repo=repo, forks='delete')

        Session().commit()

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-1'))

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-fork-1'))

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-fork-fork-1'))

    def test_remove_repo_detach_forks(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        fork = fixture.create_fork(repo.repo_name, 'test-repo-fork-1')

        Session().commit()

        #fork of fork

        fixture.create_fork(fork.repo_name, 'test-repo-fork-fork-1')

        Session().commit()

        RepoModel().delete(repo=repo, forks='detach')

        Session().commit()

        try:

            self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-1'))

            self.assertNotEqual(None, Repository.get_by_repo_name(repo_name='test-repo-fork-1'))

            self.assertNotEqual(None, Repository.get_by_repo_name(repo_name='test-repo-fork-fork-1'))

        finally:

            RepoModel().delete(repo='test-repo-fork-fork-1')

            RepoModel().delete(repo='test-repo-fork-1')

            Session().commit()