Changeset - be1d20bfd2dd
[Not reviewed]
default
0 8 0
Thomas De Schampheleire - 10 years ago 2016-04-26 20:00:39
thomas.de.schampheleire@gmail.com
pytest migration: model: convert all tests to TestControllerPytest

The model tests were based on BaseTestCase which does not exist currently
for pytest-style tests. Nevertheless, there seems to be no advantage of
directly subclassing BaseTestCase over subclassing TestControllerPytest.
Thus, keep things simple and use TestControllerPytest.
8 files changed with 33 insertions and 39 deletions:
0 comments (0 inline, 0 general)
kallithea/tests/models/test_changeset_status.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.model.changeset_status import ChangesetStatusModel
 
from kallithea.model.db import ChangesetStatus as CS
 

	
 
class CSM(object): # ChangesetStatusMock
 

	
 
    def __init__(self, status):
 
        self.status = status
 

	
 
class TestChangesetStatusCalculation(BaseTestCase):
 
class TestChangesetStatusCalculation(TestControllerPytest):
 

	
 
    def setUp(self):
 
    def setup_method(self, method):
 
        self.m = ChangesetStatusModel()
 

	
 
    @parameterized.expand([
 
    @parametrize('name,expected_result,statuses', [
 
        ('empty list', CS.STATUS_UNDER_REVIEW, []),
 
        ('approve', CS.STATUS_APPROVED, [CSM(CS.STATUS_APPROVED)]),
 
        ('approve2', CS.STATUS_APPROVED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_APPROVED)]),
 
        ('approve_reject', CS.STATUS_REJECTED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_REJECTED)]),
 
        ('approve_underreview', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_UNDER_REVIEW)]),
 
        ('approve_notreviewed', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_NOT_REVIEWED)]),
 
        ('underreview', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_UNDER_REVIEW), CSM(CS.STATUS_UNDER_REVIEW)]),
 
        ('reject', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED)]),
 
        ('reject_underreview', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_UNDER_REVIEW)]),
 
        ('reject_notreviewed', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_NOT_REVIEWED)]),
 
        ('notreviewed', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_NOT_REVIEWED)]),
 
        ('approve_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), None]),
 
        ('approve2_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_APPROVED), None]),
 
        ('approve_reject_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_REJECTED), None]),
 
        ('approve_underreview_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('approve_notreviewed_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_NOT_REVIEWED), None]),
 
        ('underreview_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_UNDER_REVIEW), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('reject_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), None]),
 
        ('reject_underreview_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('reject_notreviewed_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_NOT_REVIEWED), None]),
 
        ('notreviewed_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_NOT_REVIEWED), None]),
 
    ])
 
    def test_result(self, name, expected_result, statuses):
 
        result = self.m._calculate_status(statuses)
 
        self.assertEqual(result, expected_result)
kallithea/tests/models/test_diff_parsers.py
Show inline comments
 
@@ -226,90 +226,90 @@ DIFF_FIXTURES = {
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {MOD_FILENODE: 'modified file',
 
                  BIN_FILENODE: 'binary diff not shown'}})
 
    ],
 
    'hg_diff_copy_file.diff': [
 
        ('file2', 'M',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {COPIED_FILENODE: 'file copied from file1 to file2'}}),
 
    ],
 
    'hg_diff_copy_and_modify_file.diff': [
 
        ('file3', 'M',
 
         {'added': 1,
 
          'deleted': 0,
 
          'binary': False,
 
          'ops': {COPIED_FILENODE: 'file copied from file2 to file3',
 
                  MOD_FILENODE: 'modified file'}}),
 
    ],
 
    'hg_diff_copy_and_chmod_file.diff': [
 
        ('file4', 'M',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {COPIED_FILENODE: 'file copied from file3 to file4',
 
                  CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}}),
 
    ],
 
    'hg_diff_copy_chmod_and_edit_file.diff': [
 
        ('file5', 'M',
 
         {'added': 2,
 
          'deleted': 1,
 
          'binary': False,
 
          'ops': {COPIED_FILENODE: 'file copied from file4 to file5',
 
                  CHMOD_FILENODE: 'modified file chmod 100755 => 100644',
 
                  MOD_FILENODE: 'modified file'}}),
 
    ],
 
    'hg_diff_rename_space_cr.diff': [
 
        ('oh yes', 'R',
 
         {'added': 3,
 
          'deleted': 2,
 
          'binary': False,
 
          'ops': {RENAMED_FILENODE: 'file renamed from oh no to oh yes'}}),
 
    ],
 
}
 

	
 

	
 
class DiffLibTest(BaseTestCase):
 
class TestDiffLib(TestControllerPytest):
 

	
 
    @parameterized.expand([(x,) for x in DIFF_FIXTURES])
 
    @parametrize('diff_fixture', DIFF_FIXTURES)
 
    def test_diff(self, diff_fixture):
 
        diff = fixture.load_resource(diff_fixture, strip=False)
 
        vcs = 'hg'
 
        if diff_fixture.startswith('git_'):
 
            vcs = 'git'
 
        diff_proc = DiffProcessor(diff, vcs=vcs)
 
        diff_proc_d = diff_proc.prepare()
 
        data = [(x['filename'], x['operation'], x['stats']) for x in diff_proc_d]
 
        expected_data = DIFF_FIXTURES[diff_fixture]
 
        self.assertListEqual(expected_data, data)
 

	
 
    def test_diff_markup(self):
 
        diff = fixture.load_resource('markuptest.diff', strip=False)
 
        diff_proc = DiffProcessor(diff)
 
        diff_proc_d = diff_proc.prepare()
 
        chunks = diff_proc_d[0]['chunks']
 
        self.assertFalse(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)
 
        print s
 
        self.assertEqual(s, r'''
 
context ... ... u'@@ -51,5 +51,12 @@\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><del>#define MAX_STEPS (48)</del>\n'
 
add      60     u'<u>\t</u><ins><u class="cr"></u></ins>\n'
 
add      61     u'<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'
 
unmod    62  55 u'\n'
 
''')
kallithea/tests/models/test_notifications.py
Show inline comments
 
from kallithea.tests import *
 

	
 
from kallithea.model.db import User, Notification, UserNotification
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.notification import NotificationModel
 

	
 

	
 
class TestNotifications(BaseTestCase):
 
class TestNotifications(TestControllerPytest):
 

	
 
    def __init__(self, methodName='runTest'):
 
    def setup_method(self, method):
 
        Session.remove()
 
        self.u1 = UserModel().create_or_update(username=u'u1',
 
                                        password=u'qweqwe',
 
                                        email=u'u1@example.com',
 
                                        firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 
        self.u1 = self.u1.user_id
 

	
 
        self.u2 = UserModel().create_or_update(username=u'u2',
 
                                        password=u'qweqwe',
 
                                        email=u'u2@example.com',
 
                                        firstname=u'u2', lastname=u'u3')
 
        Session().commit()
 
        self.u2 = self.u2.user_id
 

	
 
        self.u3 = UserModel().create_or_update(username=u'u3',
 
                                        password=u'qweqwe',
 
                                        email=u'u3@example.com',
 
                                        firstname=u'u3', lastname=u'u3')
 
        Session().commit()
 
        self.u3 = self.u3.user_id
 

	
 
        super(TestNotifications, self).__init__(methodName=methodName)
 

	
 
    def setUp(self):
 
        remove_all_notifications()
 
        self.assertEqual([], Notification.query().all())
 
        self.assertEqual([], UserNotification.query().all())
 

	
 
    def test_create_notification(self):
 
        usrs = [self.u1, self.u2]
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'subj', body=u'hi there',
 
                                           recipients=usrs)
 
        Session().commit()
 
        u1 = User.get(self.u1)
 
        u2 = User.get(self.u2)
 
        u3 = User.get(self.u3)
 
        notifications = Notification.query().all()
 
        self.assertEqual(len(notifications), 1)
 

	
 
        self.assertEqual(notifications[0].recipients, [u1, u2])
 
        self.assertEqual(notification.notification_id,
 
                         notifications[0].notification_id)
 

	
 
        unotification = UserNotification.query() \
 
            .filter(UserNotification.notification == notification).all()
 

	
 
        self.assertEqual(len(unotification), len(usrs))
 
        self.assertEqual(set([x.user.user_id for x in unotification]),
 
                         set(usrs))
 

	
 
    def test_user_notifications(self):
 
        notification1 = NotificationModel().create(created_by=self.u1,
 
                                            subject=u'subj', body=u'hi there1',
 
                                            recipients=[self.u3])
 
        Session().commit()
 
        notification2 = NotificationModel().create(created_by=self.u1,
 
                                            subject=u'subj', body=u'hi there2',
 
                                            recipients=[self.u3])
 
        Session().commit()
 
        u3 = Session().query(User).get(self.u3)
 

	
 
        self.assertEqual(sorted([x.notification for x in u3.notifications]),
 
                         sorted([notification2, notification1]))
 

	
 
    def test_delete_notifications(self):
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'title', body=u'hi there3',
 
                                    recipients=[self.u3, self.u1, self.u2])
 
        Session().commit()
 
        notifications = Notification.query().all()
 
        self.assertTrue(notification in notifications)
kallithea/tests/models/test_permissions.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import RepoGroup, User, UserGroupRepoGroupToPerm, \
 
    Permission, UserToPerm
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.lib.auth import AuthUser
 
from kallithea.model.permission import PermissionModel
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestPermissions(BaseTestCase):
 
    def __init__(self, methodName='runTest'):
 
        super(TestPermissions, self).__init__(methodName=methodName)
 
class TestPermissions(TestControllerPytest):
 

	
 
    @classmethod
 
    def setUpClass(cls):
 
    def setup_class(cls):
 
        #recreate default user to get a clean start
 
        PermissionModel().create_default_permissions(user=User.DEFAULT_USER,
 
                                                     force=True)
 
        Session().commit()
 

	
 
    def setUp(self):
 
    def setup_method(self, method):
 
        self.u1 = UserModel().create_or_update(
 
            username=u'u1', password=u'qweqwe',
 
            email=u'u1@example.com', firstname=u'u1', lastname=u'u1'
 
        )
 
        self.u2 = UserModel().create_or_update(
 
            username=u'u2', password=u'qweqwe',
 
            email=u'u2@example.com', firstname=u'u2', lastname=u'u2'
 
        )
 
        self.u3 = UserModel().create_or_update(
 
            username=u'u3', password=u'qweqwe',
 
            email=u'u3@example.com', firstname=u'u3', lastname=u'u3'
 
        )
 
        self.anon = User.get_default_user()
 
        self.a1 = UserModel().create_or_update(
 
            username=u'a1', password=u'qweqwe',
 
            email=u'a1@example.com', firstname=u'a1', lastname=u'a1', admin=True
 
        )
 
        Session().commit()
 

	
 
    def tearDown(self):
 
    def teardown_method(self, method):
 
        if hasattr(self, 'test_repo'):
 
            RepoModel().delete(repo=self.test_repo)
 

	
 
        UserModel().delete(self.u1)
 
        UserModel().delete(self.u2)
 
        UserModel().delete(self.u3)
 
        UserModel().delete(self.a1)
 
        if hasattr(self, 'g1'):
 
            RepoGroupModel().delete(self.g1.group_id)
 
        if hasattr(self, 'g2'):
 
            RepoGroupModel().delete(self.g2.group_id)
 

	
 
        if hasattr(self, 'ug1'):
 
            UserGroupModel().delete(self.ug1, force=True)
 
        if hasattr(self, 'ug2'):
 
            UserGroupModel().delete(self.ug2, force=True)
 

	
 
        Session().commit()
 

	
 
    def test_default_perms_set(self):
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.create.repository', 'repository.read',
 
                           'hg.register.manual_activate']),
 
            'repositories': {HG_REPO: 'repository.read'}
 
        }
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        new_perm = 'repository.write'
 
        RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1,
 
                                          perm=new_perm)
 
        Session().commit()
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         new_perm)
 

	
 
    def test_default_admin_perms_set(self):
 
        a1_auth = AuthUser(user_id=self.a1.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.admin', 'hg.create.write_on_repogroup.true']),
 
            'repositories': {HG_REPO: 'repository.admin'}
 
        }
 
        self.assertEqual(a1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        new_perm = 'repository.write'
 
@@ -671,76 +669,76 @@ class TestPermissions(BaseTestCase):
 
                         'repository.admin')
 

	
 
    def test_owner_permissions_doesnot_get_overwritten_by_others(self):
 
        #create repo as USER,
 
        self.test_repo = fixture.create_repo(name=u'myownrepo',
 
                                             repo_type='hg',
 
                                             cur_user=self.u1)
 

	
 
        #he has permissions of admin as owner
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 
        #set his permission as user, he should still be admin
 
        RepoModel().grant_user_permission(self.test_repo, user=self.u1,
 
                                          perm='repository.none')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 

	
 
    def _test_def_perm_equal(self, user, change_factor=0):
 
        perms = UserToPerm.query() \
 
                .filter(UserToPerm.user == user) \
 
                .all()
 
        self.assertEqual(len(perms),
 
                         len(Permission.DEFAULT_USER_PERMISSIONS,)+change_factor,
 
                         msg=perms)
 

	
 
    def test_set_default_permissions(self):
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 

	
 
    def test_set_default_permissions_after_one_is_missing(self):
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 
        #now we delete one, it should be re-created after another call
 
        perms = UserToPerm.query() \
 
                .filter(UserToPerm.user == self.u1) \
 
                .all()
 
        Session().delete(perms[0])
 
        Session().commit()
 

	
 
        self._test_def_perm_equal(user=self.u1, change_factor=-1)
 

	
 
        #create missing one !
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 

	
 
    @parameterized.expand([
 
    @parametrize('perm,modify_to', [
 
        ('repository.read', 'repository.none'),
 
        ('group.read', 'group.none'),
 
        ('usergroup.read', 'usergroup.none'),
 
        ('hg.create.repository', 'hg.create.none'),
 
        ('hg.fork.repository', 'hg.fork.none'),
 
        ('hg.register.manual_activate', 'hg.register.auto_activate',)
 
    ])
 
    def test_set_default_permissions_after_modification(self, perm, modify_to):
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 

	
 
        old = Permission.get_by_key(perm)
 
        new = Permission.get_by_key(modify_to)
 
        self.assertNotEqual(old, None)
 
        self.assertNotEqual(new, None)
 

	
 
        #now modify permissions
 
        p = UserToPerm.query() \
 
                .filter(UserToPerm.user == self.u1) \
 
                .filter(UserToPerm.permission == old) \
 
                .one()
 
        p.permission = new
 
        Session().add(p)
 
        Session().commit()
 

	
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
kallithea/tests/models/test_repo_groups.py
Show inline comments
 
import os
 
from sqlalchemy.exc import IntegrityError
 

	
 
from kallithea.tests import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import RepoGroup
 
from kallithea.model.meta import Session
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
def _update_group(id_, group_name, desc=u'desc', parent_id=None):
 
    form_data = fixture._get_group_create_params(group_name=group_name,
 
                                                 group_desc=desc,
 
                                                 group_parent_id=parent_id)
 
    gr = RepoGroupModel().update(id_, form_data)
 
    return gr
 

	
 

	
 
def _update_repo(name, **kwargs):
 
    form_data = fixture._get_repo_create_params(**kwargs)
 
    if not 'repo_name' in kwargs:
 
        form_data['repo_name'] = name
 
    if not 'perms_new' in kwargs:
 
        form_data['perms_new'] = []
 
    if not 'perms_updates' in kwargs:
 
        form_data['perms_updates'] = []
 
    r = RepoModel().update(name, **form_data)
 
    return r
 

	
 

	
 
class TestRepoGroups(BaseTestCase):
 
class TestRepoGroups(TestControllerPytest):
 

	
 
    def setUp(self):
 
    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(self):
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def __check_path(self, *path):
 
        """
 
        Checks the path for existance !
 
        """
 
        path = [TESTS_TMP_PATH] + list(path)
 
        path = os.path.join(*path)
 
        return os.path.isdir(path)
 

	
 
    def _check_folders(self):
 
        print os.listdir(TESTS_TMP_PATH)
 

	
 
    def __delete_group(self, id_):
 
        RepoGroupModel().delete(id_)
 

	
 
    def test_create_group(self):
 
        g = fixture.create_repo_group(u'newGroup')
 
        Session().commit()
 
        self.assertEqual(g.full_path, 'newGroup')
 

	
 
        self.assertTrue(self.__check_path('newGroup'))
 

	
 
    def test_create_same_name_group(self):
 
        self.assertRaises(IntegrityError, lambda: fixture.create_repo_group(u'newGroup'))
 
        Session().rollback()
 

	
 
    def test_same_subgroup(self):
 
        sg1 = fixture.create_repo_group(u'sub1', group_parent_id=self.g1.group_id)
 
        self.assertEqual(sg1.parent_group, self.g1)
 
        self.assertEqual(sg1.full_path, 'test1/sub1')
 
        self.assertTrue(self.__check_path('test1', 'sub1'))
 

	
 
        ssg1 = fixture.create_repo_group(u'subsub1', group_parent_id=sg1.group_id)
 
        self.assertEqual(ssg1.parent_group, sg1)
 
        self.assertEqual(ssg1.full_path, 'test1/sub1/subsub1')
 
        self.assertTrue(self.__check_path('test1', 'sub1', 'subsub1'))
 

	
 
    def test_remove_group(self):
 
        sg1 = fixture.create_repo_group(u'deleteme')
 
        self.__delete_group(sg1.group_id)
 

	
 
        self.assertEqual(RepoGroup.get(sg1.group_id), None)
 
        self.assertFalse(self.__check_path('deteteme'))
 

	
 
        sg1 = fixture.create_repo_group(u'deleteme', group_parent_id=self.g1.group_id)
 
        self.__delete_group(sg1.group_id)
 

	
kallithea/tests/models/test_repos.py
Show inline comments
 
from kallithea.tests import *
 

	
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import Repository
 
from kallithea.lib.exceptions import AttachedForksError
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestRepos(BaseTestCase):
 
class TestRepos(TestControllerPytest):
 

	
 
    def setUp(self):
 
        pass
 

	
 
    def tearDown(self):
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_remove_repo(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        RepoModel().delete(repo=repo)
 
        Session().commit()
 

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

	
 
    def test_remove_repo_repo_raises_exc_when_attached_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

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

	
 
        self.assertRaises(AttachedForksError, lambda: RepoModel().delete(repo=repo))
 
        # cleanup
 
        RepoModel().delete(repo=u'test-repo-fork-1')
 
        RepoModel().delete(repo=u'test-repo-1')
 
        Session().commit()
 

	
 
    def test_remove_repo_delete_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

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

	
 
        #fork of fork
 
        fixture.create_fork(fork.repo_name, u'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=u'test-repo-1'))
 
        self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-1'))
 
        self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1'))
 

	
 
    def test_remove_repo_detach_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

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

	
 
        #fork of fork
 
        fixture.create_fork(fork.repo_name, u'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=u'test-repo-1'))
 
            self.assertNotEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-1'))
 
            self.assertNotEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1'))
 
        finally:
 
            RepoModel().delete(repo=u'test-repo-fork-fork-1')
 
            RepoModel().delete(repo=u'test-repo-fork-1')
 
            Session().commit()
kallithea/tests/models/test_user_groups.py
Show inline comments
 
from kallithea.model.db import User
 

	
 
from kallithea.tests import BaseTestCase, parameterized, TEST_USER_REGULAR_LOGIN
 
from kallithea.tests import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.model.meta import Session
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestUserGroups(BaseTestCase):
 
class TestUserGroups(TestControllerPytest):
 

	
 
    def tearDown(self):
 
    def teardown_method(self, method):
 
        # delete all groups
 
        for gr in UserGroupModel.get_all():
 
            fixture.destroy_user_group(gr)
 
        Session().commit()
 

	
 
    @parameterized.expand([
 
    @parametrize('pre_existing,regular_should_be,external_should_be,groups,expected', [
 
        ([], [], [], [], []),
 
        ([], [u'regular'], [], [], [u'regular']),  # no changes of regular
 
        ([u'some_other'], [], [], [u'some_other'], []),   # not added to regular group
 
        ([], [u'regular'], [u'container'], [u'container'], [u'regular', u'container']),
 
        ([], [u'regular'], [], [u'container', u'container2'], [u'regular', u'container', u'container2']),
 
        ([], [u'regular'], [u'other'], [], [u'regular']),  # remove not used
 
        ([u'some_other'], [u'regular'], [u'other', u'container'], [u'container', u'container2'], [u'regular', u'container', u'container2']),
 
    ])
 
    def test_enforce_groups(self, pre_existing, regular_should_be,
 
                            external_should_be, groups, expected):
 
        # delete all groups
 
        for gr in UserGroupModel.get_all():
 
            fixture.destroy_user_group(gr)
 
        Session().commit()
 

	
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        for gr in pre_existing:
 
            gr = fixture.create_user_group(gr)
 
        Session().commit()
 

	
 
        # make sure use is just in those groups
 
        for gr in regular_should_be:
 
            gr = fixture.create_user_group(gr)
 
            Session().commit()
 
            UserGroupModel().add_user_to_group(gr, user)
 
            Session().commit()
 

	
 
        # now special external groups created by auth plugins
 
        for gr in external_should_be:
 
            gr = fixture.create_user_group(gr, user_group_data={'extern_type': 'container'})
 
            Session().commit()
 
            UserGroupModel().add_user_to_group(gr, user)
 
            Session().commit()
 

	
 
        UserGroupModel().enforce_groups(user, groups, 'container')
 
        Session().commit()
 

	
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        in_groups = user.group_member
 
        self.assertEqual(expected, [x.users_group.users_group_name for x in in_groups])
kallithea/tests/models/test_users.py
Show inline comments
 
from kallithea.tests import *
 

	
 
from kallithea.model.db import User, UserGroup, UserGroupMember, UserEmailMap, \
 
    Permission
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestUser(BaseTestCase):
 
    def __init__(self, methodName='runTest'):
 
class TestUser(TestControllerPytest):
 

	
 
    @classmethod
 
    def setup_class(cls):
 
        Session.remove()
 
        super(TestUser, self).__init__(methodName=methodName)
 

	
 
    def tearDown(self):
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_create_and_remove(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                           email=u'u232@example.com',
 
                                           firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 
        self.assertEqual(User.get_by_username(u'test_user'), usr)
 
        self.assertEqual(User.get_by_username(u'test_USER', case_insensitive=True), usr)
 
        self.assertEqual(User.get_by_username(u'test_USER', case_insensitive=False), None)
 

	
 
        # make user group
 
        user_group = fixture.create_user_group(u'some_example_group')
 
        Session().commit()
 

	
 
        UserGroupModel().add_user_to_group(user_group, usr)
 
        Session().commit()
 

	
 
        self.assertEqual(UserGroup.get(user_group.users_group_id), user_group)
 
        self.assertEqual(UserGroupMember.query().count(), 1)
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 
        self.assertEqual(UserGroupMember.query().all(), [])
 

	
 
    def test_additonal_email_as_main(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                     email=u'main_email@example.com',
 
                                     firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 

	
 
        def do():
 
            m = UserEmailMap()
 
            m.email = u'main_email@example.com'
 
            m.user = usr
 
            Session().add(m)
 
            Session().commit()
 
        self.assertRaises(AttributeError, do)
 

	
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 
    def test_extra_email_map(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                     email=u'main_email@example.com',
 
                                     firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 

	
 
        m = UserEmailMap()
 
        m.email = u'main_email2@example.com'
 
        m.user = usr
 
        Session().add(m)
 
        Session().commit()
 

	
 
        u = User.get_by_email(email='MAIN_email@example.com')
 
        self.assertEqual(usr.user_id, u.user_id)
 
        self.assertEqual(usr.username, u.username)
 

	
 
        u = User.get_by_email(email='main_email@example.com')
 
        self.assertEqual(usr.user_id, u.user_id)
 
        self.assertEqual(usr.username, u.username)
 

	
 
        u = User.get_by_email(email='main_email2@example.com')
 
        self.assertEqual(usr.user_id, u.user_id)
 
        self.assertEqual(usr.username, u.username)
 
        u = User.get_by_email(email='main_email3@example.com')
 
        self.assertEqual(None, u)
 

	
 
        u = User.get_by_email(email='main_e%ail@example.com')
 
        self.assertEqual(None, u)
 
        u = User.get_by_email(email='main_emai_@example.com')
 
        self.assertEqual(None, u)
 

	
 

	
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 

	
 
class TestUsers(BaseTestCase):
 
class TestUsers(TestControllerPytest):
 

	
 
    def __init__(self, methodName='runTest'):
 
        super(TestUsers, self).__init__(methodName=methodName)
 

	
 
    def setUp(self):
 
    def setup_method(self, method):
 
        self.u1 = UserModel().create_or_update(username=u'u1',
 
                                        password=u'qweqwe',
 
                                        email=u'u1@example.com',
 
                                        firstname=u'u1', lastname=u'u1')
 

	
 
    def tearDown(self):
 
    def teardown_method(self, method):
 
        perm = Permission.query().all()
 
        for p in perm:
 
            UserModel().revoke_perm(self.u1, p)
 

	
 
        UserModel().delete(self.u1)
 
        Session().commit()
 
        Session.remove()
 

	
 
    def test_add_perm(self):
 
        perm = Permission.query().all()[0]
 
        UserModel().grant_perm(self.u1, perm)
 
        Session().commit()
 
        self.assertEqual(UserModel().has_perm(self.u1, perm), True)
 

	
 
    def test_has_perm(self):
 
        perm = Permission.query().all()
 
        for p in perm:
 
            has_p = UserModel().has_perm(self.u1, p)
 
            self.assertEqual(False, has_p)
 

	
 
    def test_revoke_perm(self):
 
        perm = Permission.query().all()[0]
 
        UserModel().grant_perm(self.u1, perm)
 
        Session().commit()
 
        self.assertEqual(UserModel().has_perm(self.u1, perm), True)
 

	
 
        #revoke
 
        UserModel().revoke_perm(self.u1, perm)
 
        Session().commit()
 
        self.assertEqual(UserModel().has_perm(self.u1, perm), False)
0 comments (0 inline, 0 general)