Changeset - 790da8154ef8
[Not reviewed]
default
0 8 0
Mads Kiilerich - 11 years ago 2015-04-28 16:55:05
madski@unity3d.com
tests: minimize the amount of references to rc and the past
8 files changed with 42 insertions and 42 deletions:
0 comments (0 inline, 0 general)
kallithea/tests/fixtures/journal_dump.csv
Show inline comments
 
@@ -1737,99 +1737,99 @@ user_log_id,user_id,username,repository_
 
1799,3,demo,350,misisko-repo,"",user_created_repo,2012-10-29 20:19:56.190032
 
1800,3,demo,350,misisko-repo,195.168.239.153,push:34bd2e2b6cf2abdb0e026182b9342f95d8f6f440,2012-10-29 20:21:33.067763
 
1801,3,demo,350,misisko-repo,"",user_commented_revision:34bd2e2b6cf2abdb0e026182b9342f95d8f6f440,2012-10-29 20:23:32.816207
 
1802,2,admin,350,misisko-repo,"",push_local:89a041cce422153ba0716a2beb2b5711ee4f4280,2012-10-29 21:23:56.649892
 
1806,306,foobar,,fork-foobar,"",started_following_repo,2012-10-30 05:39:47.86196
 
1808,306,foobar,,fork-foobar,"",user_created_fork:fork-foobar,2012-10-30 05:39:47.956653
 
1811,306,foobar,,fork-foobar,"",user_deleted_repo,2012-10-30 05:56:06.059093
 
1803,306,foobar,,foobar,"",started_following_repo,2012-10-30 05:36:49.915108
 
1804,306,foobar,,foobar,"",user_created_repo,2012-10-30 05:36:49.929097
 
1805,306,foobar,,foobar,24.67.50.255,push:3b0ad3c9b89dec8bcac5ec8bc9846f307dd5f291,2012-10-30 05:38:30.960676
 
1810,306,foobar,,foobar,24.67.50.255,push:5066e223f81d3983b9bc4d959c509bbce326474c,2012-10-30 05:53:14.168715
 
1812,306,foobar,,foobar,"",user_deleted_repo,2012-10-30 05:56:35.211293
 
1813,3,demo,354,DibaDaba,"",started_following_repo,2012-10-30 10:55:03.068862
 
1814,3,demo,354,DibaDaba,"",user_created_repo,2012-10-30 10:55:03.088077
 
1815,3,demo,354,DibaDaba,"",push_local:cf41636a602d2a69128cc58fdf441857afb83827,2012-10-30 10:56:16.451197
 
1816,3,demo,354,DibaDaba,"",user_updated_repo,2012-10-30 11:05:50.836108
 
1817,3,demo,354,DibaDaba,"",push_local:15453a2035510dcefbc8b7f3e45b916003ae4a0f,2012-10-30 11:18:42.79919
 
1818,3,demo,355,Test-RohdeCode,"",started_following_repo,2012-10-30 21:07:27.016976
 
1819,3,demo,355,Test-RohdeCode,"",user_created_repo,2012-10-30 21:07:27.040381
 
1820,3,demo,355,Test-RohdeCode,"",push_local:a2c7c944a5bae039181a84bec8bc89a9bb4ac68c,2012-10-31 10:52:21.038578
 
1821,3,demo,355,Test-RohdeCode,"",push_local:303c0051c7a04b5a4f77654d8cfeb09a61322e15,2012-10-31 16:32:16.917289
 
1822,3,demo,356,testdzone,"",started_following_repo,2012-11-02 15:51:18.982652
 
1823,3,demo,356,testdzone,"",user_created_repo,2012-11-02 15:51:18.994144
 
1824,3,demo,357,firs,"",started_following_repo,2012-11-04 00:37:08.689703
 
1825,3,demo,357,firs,"",user_created_repo,2012-11-04 00:37:08.700977
 
1826,316,theo,358,brightandshiny,"",started_following_repo,2012-11-04 14:17:04.270938
 
1827,316,theo,358,brightandshiny,"",user_created_repo,2012-11-04 14:17:04.286049
 
1828,316,theo,358,brightandshiny,87.165.13.203,"push:831fd48093c38f64bb541a4c12bec45d27c13f9d,0177264991ad113c2dbeed00716738931bc0222f",2012-11-04 14:21:33.759458
 
1829,3,demo,45,group/fork-bootstrap,"",started_following_repo,2012-11-04 18:25:45.459487
 
1830,3,demo,45,group/fork-bootstrap,"",stopped_following_repo,2012-11-04 18:25:48.166886
 
1831,3,demo,45,group/fork-bootstrap,"",started_following_repo,2012-11-04 18:25:51.905362
 
1832,3,demo,45,group/fork-bootstrap,"",stopped_following_repo,2012-11-04 18:25:54.406445
 
1833,3,demo,227,fork-aaa-project-2,"",user_commented_revision:bc829c34912709c7d129e5dae9a3e5ff46e77fc7,2012-11-05 18:37:48.772572
 
1834,3,demo,227,fork-aaa-project-2,"",user_commented_revision:bc829c34912709c7d129e5dae9a3e5ff46e77fc7,2012-11-05 18:38:42.521316
 
1835,3,demo,359,MyUberRepo,"",started_following_repo,2012-11-05 22:04:02.939575
 
1836,3,demo,359,MyUberRepo,"",user_created_repo,2012-11-05 22:04:02.95609
 
1837,3,demo,359,MyUberRepo,"",push_local:a67ddbdca6d533e0cd43175068cc12f526b2f95f,2012-11-05 22:07:40.846096
 
1838,3,demo,360,marese_test,"",started_following_repo,2012-11-05 22:10:35.535744
 
1839,3,demo,360,marese_test,"",user_created_repo,2012-11-05 22:10:35.547878
 
1840,3,demo,361,test-23,"",started_following_repo,2012-11-06 04:01:46.385817
 
1841,3,demo,361,test-23,"",user_created_repo,2012-11-06 04:01:46.410423
 
1842,3,demo,361,test-23,"",push_local:7d8547b154a01c7055a09292e871e80f3c6b574b,2012-11-06 04:03:03.320736
 
1843,3,demo,361,test-23,"",user_commented_revision:7d8547b154a01c7055a09292e871e80f3c6b574b,2012-11-06 04:07:20.51212
 
1844,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:fba17a64fa4978bfea19222da5e64a18cfddeecd,2012-11-06 04:12:51.753053
 
1845,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:fba17a64fa4978bfea19222da5e64a18cfddeecd,2012-11-06 04:13:30.727101
 
1846,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:fba17a64fa4978bfea19222da5e64a18cfddeecd,2012-11-06 04:15:29.032768
 
1847,3,demo,68,aaa-project,"",user_commented_revision:0c33fa58efc5a5541d8d3f1a3d3b77367e3d94f5,2012-11-06 04:17:16.091927
 
1848,3,demo,249,abcdefg,"",user_commented_revision:5d80e28538141e322b317168e2367fb03178d58c,2012-11-06 06:10:53.372505
 
1849,3,demo,362,Marcin,"",started_following_repo,2012-11-06 10:45:38.505485
 
1850,3,demo,362,Marcin,"",user_created_repo,2012-11-06 10:45:38.518969
 
1851,3,demo,362,hidden/Marcin,"",user_updated_repo,2012-11-06 10:45:46.281581
 
1849,3,demo,362,Username,"",started_following_repo,2012-11-06 10:45:38.505485
 
1850,3,demo,362,Username,"",user_created_repo,2012-11-06 10:45:38.518969
 
1851,3,demo,362,hidden/Username,"",user_updated_repo,2012-11-06 10:45:46.281581
 
1852,3,demo,177,blah,62.200.22.2,push:b369fb18c8d61fe0d3b14c417466680230cabe46,2012-11-06 10:47:55.655029
 
1853,3,demo,99,another-fork-to-check-code-review,"",user_commented_revision:d5422faf648cc589425cd3b0dbf1f6dbf93036a0,2012-11-06 13:12:05.517155
 
1854,3,demo,38,code-review-test,"",user_commented_revision:6d7db5794e8cad7da042b6ae6238116c6e59a4d2,2012-11-06 16:12:59.38977
 
1855,3,demo,363,utf8-test,"",started_following_repo,2012-11-06 16:52:34.020264
 
1856,3,demo,363,utf8-test,"",user_created_repo,2012-11-06 16:52:34.036171
 
1857,3,demo,363,utf8-test,178.6.80.143,push:e9e69068b8277c8e857984fb44e54d4e057710bd,2012-11-06 16:54:30.357337
 
1858,320,blahblub,364,abc_test,"",started_following_repo,2012-11-06 22:06:45.476725
 
1859,320,blahblub,364,abc_test,"",user_created_repo,2012-11-06 22:06:45.488546
 
1860,320,blahblub,364,abc_test,81.190.220.154,"push:e05d82d631f4a61972c2c76eb5b473c55ee6133f,dccf54ebc0638183872cfd9d35f0f49aa44ad4bc",2012-11-06 22:07:39.657839
 
1861,320,blahblub,364,abc_test,81.190.220.154,push:8ef2f06b68eb25067c4ca3a9c73c89bb681f031b,2012-11-06 22:09:18.572762
 
1862,3,demo,365,testing_do,"",started_following_repo,2012-11-07 00:19:56.15291
 
1863,3,demo,365,testing_do,"",user_created_repo,2012-11-07 00:19:56.165626
 
1864,3,demo,365,testing_do,"",user_updated_repo,2012-11-07 00:21:24.482025
 
1865,3,demo,321,Mecurial-Test,"",user_commented_revision:041e48fd7acad4c27385c27a4789d3d5cc7ccbea,2012-11-07 01:16:37.129361
 
1866,321,ifco89,284,fork-django,"",user_commented_revision:129f1ac8484d63c2e61a44fb2a18dd17246c1c4d,2012-11-07 01:26:40.887017
 
1867,3,demo,284,fork-django,"",user_commented_revision:c5d6f6d6829e730bdddf63c1252304f0c49a9053,2012-11-07 01:27:33.995596
 
1868,3,demo,68,aaa-project,"",push_local:5ecd97456233badb3680873dc5b7d4bfb52ede79,2012-11-07 06:43:22.378406
 
1869,3,demo,319,fork1-code-review-test,"",user_commented_revision:e216971a777e3132487405dd2c8e093f66788765,2012-11-07 08:44:19.419642
 
1870,3,demo,366,test123123,"",started_following_repo,2012-11-07 14:16:13.029479
 
1871,3,demo,366,test123123,"",user_created_repo,2012-11-07 14:16:13.04346
 
1872,323,didi,364,abc_test,"",user_commented_revision:8ef2f06b68eb25067c4ca3a9c73c89bb681f031b,2012-11-07 15:45:10.378961
 
1873,324,tutty,367,fork-bootstrap,"",started_following_repo,2012-11-07 16:00:47.116111
 
1874,324,tutty,36,bootstrap,"",user_forked_repo:fork-bootstrap,2012-11-07 16:00:48.378051
 
1875,324,tutty,367,fork-bootstrap,"",user_created_fork:fork-bootstrap,2012-11-07 16:00:48.393002
 
1876,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-08 18:41:07.016059
 
1877,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-08 18:41:34.311311
 
1878,3,demo,68,aaa-project,"",user_commented_revision:b4bf1f77b1288a24eb395537319b4f3155cffa71,2012-11-08 18:41:38.071834
 
1879,3,demo,368,Repotest,"",started_following_repo,2012-11-09 10:40:40.804085
 
1880,3,demo,368,Repotest,"",user_created_repo,2012-11-09 10:40:40.813114
 
1881,3,demo,369,fork-HelloWorld,"",started_following_repo,2012-11-10 19:57:46.238558
 
1882,3,demo,280,HelloWorld,"",user_forked_repo:fork-HelloWorld,2012-11-10 19:57:46.306001
 
1883,3,demo,369,fork-HelloWorld,"",user_created_fork:fork-HelloWorld,2012-11-10 19:57:46.320675
 
1884,3,demo,370,fork-django-more,"",started_following_repo,2012-11-11 01:18:51.709667
 
1885,3,demo,110,django,"",user_forked_repo:fork-django-more,2012-11-11 01:18:58.716934
 
1886,3,demo,370,fork-django-more,"",user_created_fork:fork-django-more,2012-11-11 01:18:58.731127
 
1887,3,demo,371,hgh,"",started_following_repo,2012-11-12 06:07:06.509759
 
1888,3,demo,371,hgh,"",user_created_repo,2012-11-12 06:07:06.525795
 
1889,3,demo,371,hgh,"",push_local:b43f437976ff61a8fa37709f105091ffd00bbd2c,2012-11-12 06:08:34.009007
 
1890,3,demo,371,hgh,"",push_local:93c5a23d3cbde4c6dbd21d722eb1beff90a1d67c,2012-11-12 06:09:44.749181
 
1891,3,demo,371,hgh,"",push_local:d3e97cf131b0feaa586272d88e71e97a65564806,2012-11-12 06:13:04.240444
 
1892,336,demodemo,373,test-for,"",started_following_repo,2012-11-12 18:29:12.02868
 
1893,336,demodemo,373,test-for,"",user_created_repo,2012-11-12 18:29:12.041121
 
1894,336,demodemo,374,fork-test-for,"",started_following_repo,2012-11-12 18:29:25.869055
 
1895,336,demodemo,373,test-for,"",user_forked_repo:fork-test-for,2012-11-12 18:29:25.922561
 
1896,336,demodemo,374,fork-test-for,"",user_created_fork:fork-test-for,2012-11-12 18:29:25.936869
 
1897,336,demodemo,374,fork-test-for,101.140.96.42,push:a7495b86f0bfc299740cdbfbbca1b9d5ca1f62de,2012-11-12 18:31:51.107494
 
1898,2,admin,376,linux-kernel,"",started_following_repo,2012-11-13 22:14:17.466937
 
1899,342,gauravve,377,myrep,"",started_following_repo,2012-11-14 12:12:46.280751
kallithea/tests/functional/test_login.py
Show inline comments
 
@@ -163,139 +163,139 @@ class TestLoginController(TestController
 

	
 
        response.mustcontain('An email address must contain a single @')
 
        response.mustcontain('Username may only contain '
 
                'alphanumeric characters underscores, '
 
                'periods or dashes and must begin with '
 
                'alphanumeric character')
 

	
 
    def test_register_err_case_sensitive(self):
 
        usr = 'Test_Admin'
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': usr,
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'goodmailm',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        response.mustcontain('An email address must contain a single @')
 
        msg = validators.ValidUsername()._messages['username_exists']
 
        msg = h.html_escape(msg % {'username': usr})
 
        response.mustcontain(msg)
 

	
 
    def test_register_special_chars(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                        {'username': 'xxxaxn',
 
                                         'password': 'ąćźżąśśśś',
 
                                         'password_confirmation': 'ąćźżąśśśś',
 
                                         'email': 'goodmailm@test.plx',
 
                                         'firstname': 'test',
 
                                         'lastname': 'test'})
 

	
 
        msg = validators.ValidPassword()._messages['invalid_password']
 
        response.mustcontain(msg)
 

	
 
    def test_register_password_mismatch(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': 'xs',
 
                                             'password': '123qwe',
 
                                             'password_confirmation': 'qwe123',
 
                                             'email': 'goodmailm@test.plxa',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 
        msg = validators.ValidPasswordsMatch()._messages['password_mismatch']
 
        response.mustcontain(msg)
 

	
 
    def test_register_ok(self):
 
        username = 'test_regular4'
 
        password = 'qweqwe'
 
        email = 'marcin@test.com'
 
        email = 'username@test.com'
 
        name = 'testname'
 
        lastname = 'testlastname'
 

	
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': username,
 
                                             'password': password,
 
                                             'password_confirmation': password,
 
                                             'email': email,
 
                                             'firstname': name,
 
                                             'lastname': lastname,
 
                                             'admin': True})  # This should be overriden
 
        self.assertEqual(response.status, '302 Found')
 
        self.checkSessionFlash(response, 'You have successfully registered into Kallithea')
 

	
 
        ret = Session().query(User).filter(User.username == 'test_regular4').one()
 
        self.assertEqual(ret.username, username)
 
        self.assertEqual(check_password(password, ret.password), True)
 
        self.assertEqual(ret.email, email)
 
        self.assertEqual(ret.name, name)
 
        self.assertEqual(ret.lastname, lastname)
 
        self.assertNotEqual(ret.api_key, None)
 
        self.assertEqual(ret.admin, False)
 

	
 
    def test_forgot_password_wrong_mail(self):
 
        bad_email = 'marcin@wrongmail.org'
 
        bad_email = 'username@wrongmail.org'
 
        response = self.app.post(
 
                        url(controller='login', action='password_reset'),
 
                            {'email': bad_email, }
 
        )
 

	
 
        msg = validators.ValidSystemEmail()._messages['non_existing_email']
 
        msg = h.html_escape(msg % {'email': bad_email})
 
        response.mustcontain()
 

	
 
    def test_forgot_password(self):
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset'))
 
        self.assertEqual(response.status, '200 OK')
 

	
 
        username = 'test_password_reset_1'
 
        password = 'qweqwe'
 
        email = 'marcin@python-works.com'
 
        email = 'username@python-works.com'
 
        name = 'passwd'
 
        lastname = 'reset'
 

	
 
        new = User()
 
        new.username = username
 
        new.password = password
 
        new.email = email
 
        new.name = name
 
        new.lastname = lastname
 
        new.api_key = generate_api_key(username)
 
        Session().add(new)
 
        Session().commit()
 

	
 
        response = self.app.post(url(controller='login',
 
                                     action='password_reset'),
 
                                 {'email': email, })
 

	
 
        self.checkSessionFlash(response, 'Your password reset link was sent')
 

	
 
        response = response.follow()
 

	
 
        # BAD KEY
 

	
 
        key = "bad"
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset_confirmation',
 
                                    key=key))
 
        self.assertEqual(response.status, '302 Found')
 
        self.assertTrue(response.location.endswith(url('reset_password')))
 

	
 
        # GOOD KEY
 

	
 
        key = User.get_by_username(username).api_key
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset_confirmation',
 
                                    key=key))
 
        self.assertEqual(response.status, '302 Found')
 
        self.assertTrue(response.location.endswith(url('login_home')))
 

	
 
        self.checkSessionFlash(response,
 
                               ('Your password reset was successful, '
 
                                'new password has been sent to your email'))
 

	
 
        response = response.follow()
 

	
 
    def _get_api_whitelist(self, values=None):
 
        config = {'api_access_controllers_whitelist': values or []}
 
        return config
kallithea/tests/other/manual_test_vcs_operations.py
Show inline comments
 
@@ -66,97 +66,97 @@ class Command(object):
 
        stdout, stderr = p.communicate()
 
        if DEBUG:
 
            print 'stdout:', repr(stdout)
 
            print 'stderr:', repr(stderr)
 
        return stdout, stderr
 

	
 

	
 
def _get_tmp_dir():
 
    return tempfile.mkdtemp(prefix='rc_integration_test')
 

	
 

	
 
def _construct_url(repo, dest=None, **kwargs):
 
    if dest is None:
 
        #make temp clone
 
        dest = _get_tmp_dir()
 
    params = {
 
        'user': TEST_USER_ADMIN_LOGIN,
 
        'passwd': TEST_USER_ADMIN_PASS,
 
        'host': HOST,
 
        'cloned_repo': repo,
 
        'dest': dest
 
    }
 
    params.update(**kwargs)
 
    if params['user'] and params['passwd']:
 
        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s %(dest)s' % params
 
    else:
 
        _url = 'http://(host)s/%(cloned_repo)s %(dest)s' % params
 
    return _url
 

	
 

	
 
def _add_files_and_push(vcs, DEST, **kwargs):
 
    """
 
    Generate some files, add it to DEST repo and push back
 
    vcs is git or hg and defines what VCS we want to make those files for
 

	
 
    :param vcs:
 
    :param DEST:
 
    """
 
    # commit some stuff into this repo
 
    cwd = path = jn(DEST)
 
    #added_file = jn(path, '%ssetupążźć.py' % _RandomNameSequence().next())
 
    added_file = jn(path, '%ssetup.py' % _RandomNameSequence().next())
 
    Command(cwd).execute('touch %s' % added_file)
 
    Command(cwd).execute('%s add %s' % (vcs, added_file))
 

	
 
    for i in xrange(kwargs.get('files_no', 3)):
 
        cmd = """echo 'added_line%s' >> %s""" % (i, added_file)
 
        Command(cwd).execute(cmd)
 
        author_str = 'Marcin Kuźminski <me@email.com>'
 
        author_str = 'User ǝɯɐᴎ <me@email.com>'
 
        if vcs == 'hg':
 
            cmd = """hg commit -m 'commited new %s' -u '%s' %s """ % (
 
                i, author_str, added_file
 
            )
 
        elif vcs == 'git':
 
            cmd = """EMAIL="me@email.com" git commit -m 'commited new %s' --author '%s' %s """ % (
 
                i, author_str, added_file
 
            )
 
        Command(cwd).execute(cmd)
 

	
 
    # PUSH it back
 
    _REPO = None
 
    if vcs == 'hg':
 
        _REPO = HG_REPO
 
    elif vcs == 'git':
 
        _REPO = GIT_REPO
 

	
 
    kwargs['dest'] = ''
 
    clone_url = _construct_url(_REPO, **kwargs)
 
    if 'clone_url' in kwargs:
 
        clone_url = kwargs['clone_url']
 
    stdout = stderr = None
 
    if vcs == 'hg':
 
        stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url)
 
    elif vcs == 'git':
 
        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url + " master")
 

	
 
    return stdout, stderr
 

	
 

	
 
def set_anonymous_access(enable=True):
 
    user = User.get_by_username(User.DEFAULT_USER)
 
    user.active = enable
 
    Session().add(user)
 
    Session().commit()
 
    print '\tanonymous access is now:', enable
 
    if enable != User.get_by_username(User.DEFAULT_USER).active:
 
        raise Exception('Cannot set anonymous access')
 

	
 

	
 
#==============================================================================
 
# TESTS
 
#==============================================================================
 

	
 

	
 
def _check_proper_git_push(stdout, stderr):
 
    #WTF Git stderr is output ?!
 
    assert 'fatal' not in stderr
kallithea/tests/other/test_libs.py
Show inline comments
 
# -*- 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/>.
 
"""
 
kallithea.tests.test_libs
 
~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Package for testing various lib/helper functions in kallithea
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Jun 9, 2011
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
from __future__ import with_statement
 
import datetime
 
import hashlib
 
import mock
 
from kallithea.tests import *
 
from kallithea.lib.utils2 import AttributeDict
 
from kallithea.model.db import Repository
 

	
 
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://username@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://username: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'
 
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://username@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://username: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),
 
]
 

	
 

	
 
class TestLibs(BaseTestCase):
 

	
 
    @parameterized.expand(TEST_URLS)
 
    def test_uri_filter(self, test_url, expected, expected_creds):
 
        from kallithea.lib.utils2 import uri_filter
 
        self.assertEqual(uri_filter(test_url), expected)
 

	
 
    @parameterized.expand(TEST_URLS)
 
    def test_credentials_filter(self, test_url, expected, expected_creds):
 
        from kallithea.lib.utils2 import credentials_filter
 
        self.assertEqual(credentials_filter(test_url), expected_creds)
 

	
 
    @parameterized.expand([('t', True),
 
                           ('true', True),
 
                           ('y', True),
 
                           ('yes', True),
 
                           ('on', True),
 
                           ('1', True),
 
                           ('Y', True),
 
                           ('yeS', True),
 
                           ('Y', True),
 
                           ('TRUE', True),
 
                           ('T', True),
 
                           ('False', False),
 
                           ('F', False),
 
                           ('FALSE', False),
 
                           ('0', False),
 
                           ('-1', False),
 
                           ('', False)
 
    ])
 
    def test_str2bool(self, str_bool, expected):
 
        from kallithea.lib.utils2 import str2bool
 
        self.assertEqual(str2bool(str_bool), expected)
 

	
 
    def test_mention_extractor(self):
 
        from kallithea.lib.utils2 import extract_mentioned_users
 
        sample = (
 
            "@first hi there @marcink here's my email marcin@email.com "
 
            "@first hi there @world here's my email username@email.com "
 
            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
 
            "@MARCIN    @maRCiN @2one_more22 @john please see this http://org.pl "
 
            "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "
 
            "@marian.user just do it @marco-polo and next extract @marco_polo "
 
            "user.dot  hej ! not-needed maril@domain.org"
 
        )
 

	
 
        s = sorted([
 
            '2one_more22', 'first', 'marcink', 'lukaszb', 'one', 'one_more22', 'MARCIN', 'maRCiN', 'john',
 
            'marian.user', 'marco-polo', 'marco_polo'], key=lambda k: k.lower())
 
            '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',
 
            'marian.user', 'marco-polo', 'marco_polo', 'world'], key=lambda k: k.lower())
 
        self.assertEqual(s, extract_mentioned_users(sample))
 

	
 
    @parameterized.expand([
 
        (dict(), u'just now'),
 
        (dict(seconds= -1), u'1 second ago'),
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),
 
        (dict(hours= -1), u'1 hour ago'),
 
        (dict(hours= -24), u'1 day ago'),
 
        (dict(hours= -24 * 5), u'5 days ago'),
 
        (dict(months= -1), u'1 month ago'),
 
        (dict(months= -1, days= -2), u'1 month and 2 days ago'),
 
        (dict(months= -1, days= -20), u'1 month and 19 days ago'),
 
        (dict(years= -1, months= -1), u'1 year and 1 month ago'),
 
        (dict(years= -1, months= -10), u'1 year and 10 months ago'),
 
        (dict(years= -2, months= -4), u'2 years and 4 months ago'),
 
        (dict(years= -2, months= -11), u'2 years and 11 months ago'),
 
        (dict(years= -3, months= -2), u'3 years and 2 months ago'),
 
    ])
 
    def test_age(self, age_args, expected):
 
        from kallithea.lib.utils2 import age
 
        from dateutil import relativedelta
 
        n = datetime.datetime(year=2012, month=5, day=17)
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
        self.assertEqual(age(n + delt(**age_args), now=n), expected)
 

	
 
    @parameterized.expand([
 
        (dict(), u'just now'),
 
        (dict(seconds= -1), u'1 second ago'),
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),
 
        (dict(hours= -1), u'1 hour ago'),
 
        (dict(hours= -24), u'1 day ago'),
 
        (dict(hours= -24 * 5), u'5 days ago'),
 
        (dict(months= -1), u'1 month ago'),
 
        (dict(months= -1, days= -2), u'1 month ago'),
 
        (dict(months= -1, days= -20), u'1 month ago'),
 
        (dict(years= -1, months= -1), u'13 months ago'),
 
        (dict(years= -1, months= -10), u'22 months ago'),
 
        (dict(years= -2, months= -4), u'2 years ago'),
 
        (dict(years= -2, months= -11), u'3 years ago'),
 
        (dict(years= -3, months= -2), u'3 years ago'),
 
        (dict(years= -4, months= -8), u'5 years ago'),
 
    ])
 
    def test_age_short(self, age_args, expected):
 
        from kallithea.lib.utils2 import age
 
        from dateutil import relativedelta
 
        n = datetime.datetime(year=2012, month=5, day=17)
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
        self.assertEqual(age(n + delt(**age_args), show_short_version=True, now=n), expected)
 
@@ -206,109 +206,109 @@ class TestLibs(BaseTestCase):
 
        class fake_url(object):
 
            @classmethod
 
            def current(cls, *args, **kwargs):
 
                return 'https://server.com'
 

	
 
        #mock pylons.tmpl_context
 
        def fake_tmpl_context(_url):
 
            _c = AttributeDict()
 
            _c.visual = AttributeDict()
 
            _c.visual.use_gravatar = True
 
            _c.visual.gravatar_url = _url
 

	
 
            return _c
 

	
 

	
 
        with mock.patch('pylons.url', fake_url):
 
            fake = fake_tmpl_context(_url='http://test.com/{email}')
 
            with mock.patch('pylons.tmpl_context', fake):
 
                    from pylons import url
 
                    assert url.current() == 'https://server.com'
 
                    grav = gravatar_url(email_address='test@foo.com', size=24)
 
                    assert grav == 'http://test.com/test@foo.com'
 

	
 
            fake = fake_tmpl_context(_url='http://test.com/{email}')
 
            with mock.patch('pylons.tmpl_context', fake):
 
                grav = gravatar_url(email_address='test@foo.com', size=24)
 
                assert grav == 'http://test.com/test@foo.com'
 

	
 
            fake = fake_tmpl_context(_url='http://test.com/{md5email}')
 
            with mock.patch('pylons.tmpl_context', fake):
 
                em = 'test@foo.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'http://test.com/%s' % (_md5(em))
 

	
 
            fake = fake_tmpl_context(_url='http://test.com/{md5email}/{size}')
 
            with mock.patch('pylons.tmpl_context', fake):
 
                em = 'test@foo.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)
 

	
 
            fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')
 
            with mock.patch('pylons.tmpl_context', fake):
 
                em = 'test@foo.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'https://server.com/%s/%s' % (_md5(em), 24)
 

	
 
    @parameterized.expand([
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/rc', 'http://vps1:8000/rc/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/rc', 'http://user@vps1:8000/rc/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '/rc', 'http://marcink@vps1:8000/rc/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/rc/', 'http://user@vps1:8000/rc/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '/rc/', 'http://marcink@vps1:8000/rc/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/_23'),
 
        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/_23'),
 
        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://vps1:8000/_23'),
 
        ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {'user': 'marcink'}, '', 'https://marcink@proxy1.server.com/group/repo1'),
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
 
        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
 
        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),
 
        ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.server.com/group/repo1'),
 
        ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.server.com/group/repo1'),
 
        ('https://proxy1.server.com/{user}/{repo}', 'group/repo1', {'user': 'marcink'}, '', 'https://proxy1.server.com/marcink/group/repo1'),
 
        ('https://proxy1.server.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.server.com/username/group/repo1'),
 
    ])
 
    def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):
 
        from kallithea.lib.utils2 import get_clone_url
 
        clone_url = get_clone_url(uri_tmpl=tmpl, qualifed_home_url='http://vps1:8000'+prefix,
 
                                  repo_name=repo_name, repo_id=23, **overrides)
 
        self.assertEqual(clone_url, expected)
 

	
 
    def _quick_url(self, text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):
 
        """
 
        Changes `some text url[foo]` => `some text <a href="/">foo</a>
 

	
 
        :param text:
 
        """
 
        import re
 
        # quickly change expected url[] into a link
 
        URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')
 

	
 
        def url_func(match_obj):
 
            _url = match_obj.groups()[0]
 
            return tmpl % (url_ or '/some-url', _url)
 
        return URL_PAT.sub(url_func, text)
 

	
 
    @parameterized.expand([
 
      ("",
 
       ""),
 
      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
 
       "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),
 
      ("from rev 000000000000",
 
       "from rev url[000000000000]"),
 
      ("from rev 000000000000123123 also rev 000000000000",
 
       "from rev url[000000000000123123] also rev url[000000000000]"),
 
      ("this should-000 00",
 
       "this should-000 00"),
 
      ("longtextffffffffff rev 123123123123",
 
       "longtextffffffffff rev url[123123123123]"),
 
      ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
 
       "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),
 
      ("ffffffffffff some text traalaa",
 
       "url[ffffffffffff] some text traalaa"),
 
       ("""Multi line
 
       123123123123
 
       some text 123123123123
 
       sometimes !
 
       """,
 
       """Multi line
 
       url[123123123123]
 
       some text url[123123123123]
 
       sometimes !
 
@@ -321,55 +321,55 @@ class TestLibs(BaseTestCase):
 
        expected = self._quick_url(expected)
 

	
 
        with mock.patch('pylons.url', fake_url):
 
            from kallithea.lib.helpers import urlify_changesets
 
            self.assertEqual(urlify_changesets(sample, 'repo_name'), expected)
 

	
 
    @parameterized.expand([
 
      ("",
 
       "",
 
       ""),
 
      ("https://svn.apache.org/repos",
 
       "url[https://svn.apache.org/repos]",
 
       "https://svn.apache.org/repos"),
 
      ("http://svn.apache.org/repos",
 
       "url[http://svn.apache.org/repos]",
 
       "http://svn.apache.org/repos"),
 
      ("from rev a also rev http://google.com",
 
       "from rev a also rev url[http://google.com]",
 
       "http://google.com"),
 
       ("""Multi line
 
       https://foo.bar.com
 
       some text lalala""",
 
       """Multi line
 
       url[https://foo.bar.com]
 
       some text lalala""",
 
       "https://foo.bar.com")
 
    ])
 
    def test_urlify_test(self, sample, expected, url_):
 
        from kallithea.lib.helpers import urlify_text
 
        expected = self._quick_url(expected,
 
                                   tmpl="""<a href="%s">%s</a>""", url_=url_)
 
        self.assertEqual(urlify_text(sample), expected)
 

	
 
    @parameterized.expand([
 
      ("", None),
 
      ("/_2", '2'),
 
      ("_2", '2'),
 
      ("/_2/", '2'),
 
      ("_2/", '2'),
 

	
 
      ("/_21", '21'),
 
      ("_21", '21'),
 
      ("/_21/", '21'),
 
      ("_21/", '21'),
 

	
 
      ("/_21/foobar", '21'),
 
      ("_21/121", '21'),
 
      ("/_21/_12", '21'),
 
      ("_21/rc/foo", '21'),
 
      ("_21/prefix/foo", '21'),
 
    ])
 
    def test_get_repo_by_id(self, test, expected):
 
        from kallithea.lib.utils import _extract_id_from_repo_name
 
        _test = _extract_id_from_repo_name(test)
 
        self.assertEqual(_test, expected, msg='url:%s, got:`%s` expected: `%s`'
 
                                              % (test, _test, expected))
kallithea/tests/scripts/create_rc.sh
Show inline comments
 
#!/bin/sh
 
psql -U postgres -h localhost -c 'drop database if exists kallithea;'
 
psql -U postgres -h localhost -c 'create database kallithea;'
 
paster setup-db rc.ini --force-yes --user=username --password=qweqwe --email=username@example.com --repos=/home/username/repos --no-public-access
 
paster setup-db server.ini --force-yes --user=username --password=qweqwe --email=username@example.com --repos=/home/username/repos --no-public-access
 
API_KEY=`psql -R " " -A -U postgres -h localhost -c "select api_key from users where admin=TRUE" -d kallithea | awk '{print $2}'`
 
echo "run those after running server"
 
paster serve rc.ini --pid-file=rc.pid --daemon
 
paster serve server.ini --pid-file=server.pid --daemon
 
sleep 3
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user username:demo1 password:qweqwe email:demo1@example.com
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user username:demo2 password:qweqwe email:demo2@example.com
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user username:demo3 password:qweqwe email:demo3@example.com
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user_group group_name:demo12
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 add_user_to_user_group usergroupid:demo12 userid:demo1
 
kallithea-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 add_user_to_user_group usergroupid:demo12 userid:demo2
 
echo "killing server"
 
kill `cat rc.pid`
 
rm rc.pid
 
kill `cat server.pid`
 
rm server.pid
kallithea/tests/scripts/test_concurency.py
Show inline comments
 
@@ -4,104 +4,104 @@
 
# 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/>.
 
"""
 
kallithea.tests.test_hg_operations
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Test suite for making push/pull operations
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Dec 30, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 

	
 
"""
 

	
 
import os
 
import sys
 
import shutil
 
import logging
 
from os.path import join as jn
 
from os.path import dirname as dn
 

	
 
from tempfile import _RandomNameSequence
 
from subprocess import Popen, PIPE
 

	
 
from paste.deploy import appconfig
 
from sqlalchemy import engine_from_config
 

	
 
from kallithea.lib.utils import add_cache
 
from kallithea.model import init_model
 
from kallithea.model import meta
 
from kallithea.model.db import User, Repository
 
from kallithea.lib.auth import get_crypt_password
 

	
 
from kallithea.tests import TESTS_TMP_PATH, HG_REPO
 
from kallithea.config.environment import load_environment
 

	
 
rel_path = dn(dn(dn(dn(os.path.abspath(__file__)))))
 
conf = appconfig('config:rc.ini', relative_to=rel_path)
 
conf = appconfig('config:development.ini', relative_to=rel_path)
 
load_environment(conf.global_conf, conf.local_conf)
 

	
 
add_cache(conf)
 

	
 
USER = 'test_admin'
 
PASS = 'test12'
 
HOST = 'rc.local'
 
HOST = 'server.local'
 
METHOD = 'pull'
 
DEBUG = True
 
log = logging.getLogger(__name__)
 

	
 

	
 
class Command(object):
 

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

	
 
    def execute(self, cmd, *args):
 
        """Runs command on the system with given ``args``.
 
        """
 

	
 
        command = cmd + ' ' + ' '.join(args)
 
        log.debug('Executing %s' % command)
 
        if DEBUG:
 
            print command
 
        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)
 
        stdout, stderr = p.communicate()
 
        if DEBUG:
 
            print stdout, stderr
 
        return stdout, stderr
 

	
 

	
 
def get_session():
 
    engine = engine_from_config(conf, 'sqlalchemy.db1.')
 
    init_model(engine)
 
    sa = meta.Session
 
    return sa
 

	
 

	
 
def create_test_user(force=True):
 
    print 'creating test user'
 
    sa = get_session()
 

	
 
    user = sa.query(User).filter(User.username == USER).scalar()
 

	
 
    if force and user is not None:
 
        print 'removing current user'
 
        for repo in sa.query(Repository).filter(Repository.user == user).all():
 
            sa.delete(repo)
 
        sa.delete(user)
 
        sa.commit()
 

	
 
    if user is None or force:
 
        print 'creating new one'
 
        new_usr = User()
kallithea/tests/scripts/test_crawler.py
Show inline comments
 
@@ -15,97 +15,97 @@
 
kallithea.tests.test_crawer
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Test for crawling a project for memory usage
 
This should be runned just as regular script together
 
with a watch script that will show memory usage.
 

	
 
watch -n1 ./kallithea/tests/mem_watch
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Apr 21, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 

	
 
import cookielib
 
import urllib
 
import urllib2
 
import time
 
import os
 
import sys
 
from os.path import join as jn
 
from os.path import dirname as dn
 

	
 
__here__ = os.path.abspath(__file__)
 
__root__ = dn(dn(dn(__here__)))
 
sys.path.append(__root__)
 

	
 
from kallithea.lib import vcs
 
from kallithea.lib.compat import OrderedSet
 
from kallithea.lib.vcs.exceptions import RepositoryError
 

	
 
PASES = 3
 
HOST = 'http://127.0.0.1'
 
PORT = 5000
 
BASE_URI = '%s:%s/' % (HOST, PORT)
 

	
 
if len(sys.argv) == 2:
 
    BASE_URI = sys.argv[1]
 

	
 
if not BASE_URI.endswith('/'):
 
    BASE_URI += '/'
 

	
 
print 'Crawling @ %s' % BASE_URI
 
BASE_URI += '%s'
 
PROJECT_PATH = jn('/', 'home', 'marcink', 'repos')
 
PROJECT_PATH = jn('/', 'home', 'username', 'repos')
 
PROJECTS = [
 
    #'linux-magx-pbranch',
 
    'CPython',
 
    'kallithea',
 
]
 

	
 

	
 
cj = cookielib.FileCookieJar('/tmp/rc_test_cookie.txt')
 
o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
 
o.addheaders = [
 
    ('User-agent', 'kallithea-crawler'),
 
    ('Accept-Language', 'en - us, en;q = 0.5')
 
]
 

	
 
urllib2.install_opener(o)
 

	
 

	
 
def _get_repo(proj):
 
    if isinstance(proj, basestring):
 
        repo = vcs.get_repo(jn(PROJECT_PATH, proj))
 
        proj = proj
 
    else:
 
        repo = proj
 
        proj = repo.name
 

	
 
    return repo, proj
 

	
 

	
 
def test_changelog_walk(proj, pages=100):
 
    repo, proj = _get_repo(proj)
 

	
 
    total_time = 0
 
    for i in range(1, pages):
 

	
 
        page = '/'.join((proj, 'changelog',))
 

	
 
        full_uri = (BASE_URI % page) + '?' + urllib.urlencode({'page': i})
 
        s = time.time()
 
        f = o.open(full_uri)
 

	
 
        assert f.url == full_uri, 'URL:%s does not match %s' % (f.url, full_uri)
 

	
 
        size = len(f.read())
 
        e = time.time() - s
 
        total_time += e
 
        print 'visited %s size:%s req:%s ms' % (full_uri, size, e)
 

	
 
    print 'total_time', total_time
kallithea/tests/vcs/test_utils.py
Show inline comments
 
@@ -140,104 +140,104 @@ class TestParseDatetime(unittest.TestCas
 
    def test_now(self):
 
        self.assertTrue(parse_datetime('now') - datetime.datetime.now() <
 
            datetime.timedelta(seconds=1))
 

	
 
    def test_today(self):
 
        today = datetime.date.today()
 
        self.assertEqual(parse_datetime('today'),
 
            datetime.datetime(*today.timetuple()[:3]))
 

	
 
    def test_yesterday(self):
 
        yesterday = datetime.date.today() - datetime.timedelta(days=1)
 
        self.assertEqual(parse_datetime('yesterday'),
 
            datetime.datetime(*yesterday.timetuple()[:3]))
 

	
 
    def test_tomorrow(self):
 
        tomorrow = datetime.date.today() + datetime.timedelta(days=1)
 
        args = tomorrow.timetuple()[:3] + (23, 59, 59)
 
        self.assertEqual(parse_datetime('tomorrow'), datetime.datetime(*args))
 

	
 
    def test_days(self):
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=3)
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
 
        expected = datetime.datetime(*args)
 
        self.assertEqual(parse_datetime('3d'), expected)
 
        self.assertEqual(parse_datetime('3 d'), expected)
 
        self.assertEqual(parse_datetime('3 day'), expected)
 
        self.assertEqual(parse_datetime('3 days'), expected)
 

	
 
    def test_weeks(self):
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=3 * 7)
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
 
        expected = datetime.datetime(*args)
 
        self.assertEqual(parse_datetime('3w'), expected)
 
        self.assertEqual(parse_datetime('3 w'), expected)
 
        self.assertEqual(parse_datetime('3 week'), expected)
 
        self.assertEqual(parse_datetime('3 weeks'), expected)
 

	
 
    def test_mixed(self):
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=2 * 7 + 3)
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
 
        expected = datetime.datetime(*args)
 
        self.assertEqual(parse_datetime('2w3d'), expected)
 
        self.assertEqual(parse_datetime('2w 3d'), expected)
 
        self.assertEqual(parse_datetime('2w 3 days'), expected)
 
        self.assertEqual(parse_datetime('2 weeks 3 days'), expected)
 

	
 

	
 
class TestAuthorExtractors(unittest.TestCase):
 
    TEST_AUTHORS = [('Marcin Kuzminski <marcin@python-works.com>',
 
                    ('Marcin Kuzminski', 'marcin@python-works.com')),
 
                  ('Marcin Kuzminski Spaces < marcin@python-works.com >',
 
                    ('Marcin Kuzminski Spaces', 'marcin@python-works.com')),
 
                  ('Marcin Kuzminski <marcin.kuzminski@python-works.com>',
 
                    ('Marcin Kuzminski', 'marcin.kuzminski@python-works.com')),
 
                  ('mrf RFC_SPEC <marcin+kuzminski@python-works.com>',
 
                    ('mrf RFC_SPEC', 'marcin+kuzminski@python-works.com')),
 
    TEST_AUTHORS = [("Username Last'o'Name <username@python-works.com>",
 
                    ("Username Last'o'Name", "username@python-works.com")),
 
                  ("Username Last'o'Name Spaces < username@python-works.com >",
 
                    ("Username Last'o'Name Spaces", "username@python-works.com")),
 
                  ("Username Last'o'Name <username.lastname@python-works.com>",
 
                    ("Username Last'o'Name", "username.lastname@python-works.com")),
 
                  ('mrf RFC_SPEC <username+lastname@python-works.com>',
 
                    ('mrf RFC_SPEC', 'username+lastname@python-works.com')),
 
                  ('username <user@email.com>',
 
                    ('username', 'user@email.com')),
 
                  ('username <user@email.com',
 
                   ('username', 'user@email.com')),
 
                  ('broken missing@email.com',
 
                   ('broken', 'missing@email.com')),
 
                  ('<justemail@mail.com>',
 
                   ('', 'justemail@mail.com')),
 
                  ('justname',
 
                   ('justname', '')),
 
                  ('Mr Double Name withemail@email.com ',
 
                   ('Mr Double Name', 'withemail@email.com')),
 
                  ]
 

	
 
    def test_author_email(self):
 

	
 
        for test_str, result in self.TEST_AUTHORS:
 
            self.assertEqual(result[1], author_email(test_str))
 

	
 

	
 
    def test_author_name(self):
 

	
 
        for test_str, result in self.TEST_AUTHORS:
 
            self.assertEqual(result[0], author_name(test_str))
 

	
 

	
 
class TestGetDictForAttrs(unittest.TestCase):
 

	
 
    def test_returned_dict_has_expected_attrs(self):
 
        obj = mock.Mock()
 
        obj.NOT_INCLUDED = 'this key/value should not be included'
 
        obj.CONST = True
 
        obj.foo = 'aaa'
 
        obj.attrs = {'foo': 'bar'}
 
        obj.date = datetime.datetime(2010, 12, 31)
 
        obj.count = 1001
 

	
 
        self.assertEqual(get_dict_for_attrs(obj, ['CONST', 'foo', 'attrs',
 
            'date', 'count']), {
 
            'CONST': True,
 
            'foo': 'aaa',
 
            'attrs': {'foo': 'bar'},
 
            'date': datetime.datetime(2010, 12, 31),
 
            'count': 1001,
 
        })
 

	
 

	
 
class TestGetTotalSeconds(unittest.TestCase):
0 comments (0 inline, 0 general)