kallithea Changeset - 7172f3b0042b

Changeset - 7172f3b0042b

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Mads Kiilerich - 6 years ago 2019-12-28 15:05:53
mads@kiilerich.com

Grafted from: d2190faa42a2

tests: minor updates for how py3 strings/bytes are different

5 files changed with 6 insertions and 7 deletions:

kallithea/tests/functional/test_admin_settings.py

kallithea/tests/functional/test_files.py

kallithea/tests/functional/test_login.py

kallithea/tests/models/test_diff_parsers.py

kallithea/tests/other/test_vcs_operations.py

0 comments (0 inline, 0 general)

kallithea/tests/functional/test_admin_settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.model.db import Setting, Ui
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestAdminSettingsController(base.TestController):
     def test_index_main(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings'))
     def test_index_mapping(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_mapping'))
     def test_index_global(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_global'))
     def test_index_visual(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_visual'))
     def test_index_email(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_email'))
     def test_index_hooks(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_hooks'))
     def test_create_custom_hook(self):
         self.log_user()
         response = self.app.post(base.url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_1',
                                             new_hook_ui_value='cd %s' % base.TESTS_TMP_PATH,
                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Added new hook')
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('cd %s' % base.TESTS_TMP_PATH)
     def test_edit_custom_hook(self):
         self.log_user()
         response = self.app.post(base.url('admin_settings_hooks'),
                                 params=dict(hook_ui_key='test_hooks_1',
                                             hook_ui_value='old_value_of_hook_1',
                                             hook_ui_value_new='new_value_of_hook_1',
                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('new_value_of_hook_1')
     def test_add_existing_custom_hook(self):
         self.log_user()
         response = self.app.post(base.url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_1',
                                             new_hook_ui_value='attempted_new_value',
                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Hook already exists')
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('new_value_of_hook_1')
     def test_create_custom_hook_delete(self):
         self.log_user()
         response = self.app.post(base.url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='test_hooks_2',
                                             new_hook_ui_value='cd %s2' % base.TESTS_TMP_PATH,
                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Added new hook')
         response = response.follow()
         response.mustcontain('test_hooks_2')
         response.mustcontain('cd %s2' % base.TESTS_TMP_PATH)
         hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id
         ## delete
         self.app.post(base.url('admin_settings_hooks'),
                         params=dict(hook_id=hook_id, _session_csrf_secret_token=self.session_csrf_secret_token()))
         response = self.app.get(base.url('admin_settings_hooks'))
         response.mustcontain(no=['test_hooks_2'])
         response.mustcontain(no=['cd %s2' % base.TESTS_TMP_PATH])
     def test_add_existing_builtin_hook(self):
         self.log_user()
         response = self.app.post(base.url('admin_settings_hooks'),
                                 params=dict(new_hook_ui_key='changegroup.update',
                                             new_hook_ui_value='attempted_new_value',
                                             _session_csrf_secret_token=self.session_csrf_secret_token()))
         self.checkSessionFlash(response, 'Builtin hooks are read-only')
         response = response.follow()
         response.mustcontain('changegroup.update')
         response.mustcontain('hg update &gt;&amp;2')
     def test_index_search(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_search'))
     def test_index_system(self):
         self.log_user()
         response = self.app.get(base.url('admin_settings_system'))
     def test_ga_code_active(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = 'ga-test-123456789'
         response = self.app.post(base.url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['ga_code'] == new_ga_code
         response = response.follow()
         response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
     def test_ga_code_inactive(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(base.url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['ga_code'] == new_ga_code
         response = response.follow()
         response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
     def test_captcha_activate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(base.url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='1234567890',
                                  captcha_public_key='1234567890',
                                  _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['captcha_private_key'] == '1234567890'
         response = self.app.get(base.url('register'))
         response.mustcontain('captcha')
     def test_captcha_deactivate(self):
         self.log_user()
         old_title = 'Kallithea'
         old_realm = 'Kallithea authentication'
         new_ga_code = ''
         response = self.app.post(base.url('admin_settings_global'),
                         params=dict(title=old_title,
                                  realm=old_realm,
                                  ga_code=new_ga_code,
                                  captcha_private_key='',
                                  captcha_public_key='1234567890',
                                  _session_csrf_secret_token=self.session_csrf_secret_token(),
                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         assert Setting.get_app_settings()['captcha_private_key'] == ''
         response = self.app.get(base.url('register'))
         response.mustcontain(no=['captcha'])
     def test_title_change(self):
         self.log_user()
         old_title = 'Kallithea'
         new_title = old_title + '_changed'
         old_realm = 'Kallithea authentication'
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.app.post(base.url('admin_settings_global'),
                         params=dict(title=new_title,
                                  realm=old_realm,
                                  ga_code='',
                                  captcha_private_key='',
                                  captcha_public_key='',
                                  _session_csrf_secret_token=self.session_csrf_secret_token(),
                                 ))
             self.checkSessionFlash(response, 'Updated application settings')
-            assert Setting.get_app_settings()['title'] == new_title.decode('utf-8')
             assert Setting.get_app_settings()['title'] == new_title
             response = response.follow()
             response.mustcontain("""<span class="branding">%s</span>""" % new_title)

kallithea/tests/functional/test_files.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import json
 import mimetypes
 import posixpath
 from kallithea.model.db import Repository
 from kallithea.model.meta import Session
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 ARCHIVE_SPECS = {
     '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),
     '.tar.gz': ('application/x-gzip', 'tgz', ''),
     '.zip': ('application/zip', 'zip', ''),
+}
 HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')
 GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')
 def _set_downloads(repo_name, set_to):
     repo = Repository.get_by_repo_name(repo_name)
     repo.enable_downloads = set_to
     Session().commit()
 class TestFilesController(base.TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=base.HG_REPO,
                                     revision='tip',
                                     f_path='/'))
         # Test response...
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/vcs"><i class="icon-folder-open"></i><span>vcs</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.gitignore"><i class="icon-doc"></i><span>.gitignore</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgignore"><i class="icon-doc"></i><span>.hgignore</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgtags"><i class="icon-doc"></i><span>.hgtags</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.travis.yml"><i class="icon-doc"></i><span>.travis.yml</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/MANIFEST.in"><i class="icon-doc"></i><span>MANIFEST.in</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/run_test_and_report.sh"><i class="icon-doc"></i><span>run_test_and_report.sh</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.cfg"><i class="icon-doc"></i><span>setup.cfg</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.py"><i class="icon-doc"></i><span>setup.py</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/test_and_report.sh"><i class="icon-doc"></i><span>test_and_report.sh</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/tox.ini"><i class="icon-doc"></i><span>tox.ini</span></a>' % base.HG_REPO)
     def test_index_revision(self):
         self.log_user()
         response = self.app.get(
             base.url(controller='files', action='index',
                 repo_name=base.HG_REPO,
                 revision='7ba66bec8d6dbba14a2155be32408c435c5f4492',
                 f_path='/')
+        )
         # Test response...
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/tests"><i class="icon-folder-open"></i><span>tests</span></a>' % base.HG_REPO)
         response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % base.HG_REPO)
         response.mustcontain('1.1 KiB')
     def test_index_different_branch(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=base.HG_REPO,
                                     revision='97e8b885c04894463c51898e14387d80c30ed1ee',
                                     f_path='/'))
         response.mustcontain("""<option selected="selected" value="97e8b885c04894463c51898e14387d80c30ed1ee">git at 97e8b885c048</option>""")
     def test_index_paging(self):
         self.log_user()
         for r in [(73, 'a066b25d5df7016b45a41b7e2a78c33b57adc235'),
                   (92, 'cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e'),
                   (109, '75feb4c33e81186c87eac740cee2447330288412'),
                   (1, '3d8f361e72ab303da48d799ff1ac40d5ac37c67e'),
                   (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:
             response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=base.HG_REPO,
                                     revision=r[1],
                                     f_path='/'))
             response.mustcontain("""@ r%s:%s""" % (r[0], r[1][:12]))
     def test_file_source(self):
         # Force the global cache to be populated now when we know the right .ini has been loaded.
         # (Without this, the test would fail.)
         import kallithea.lib.helpers
         kallithea.lib.helpers._urlify_issues_f = None
         self.log_user()
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=base.HG_REPO,
                                     revision='8911406ad776fdd3d0b9932a2e89677e57405a48',
                                     f_path='vcs/nodes.py'))
         response.mustcontain("""<div class="formatted-fixed">Partially implemented <a class="issue-tracker-link" href="https://issues.example.com/vcs_test_hg/issue/16">#16</a>. filecontent/commit message/author/node name are safe_unicode now.<br/>"""
 """In addition some other __str__ are unicode as well<br/>"""
 """Added test for unicode<br/>"""
 """Improved test to clone into uniq repository.<br/>"""
 """removed extra unicode conversion in diff.</div>
 """)
         response.mustcontain("""<option selected="selected" value="8911406ad776fdd3d0b9932a2e89677e57405a48">default at 8911406ad776</option>""")
     def test_file_source_history(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='history',
                                     repo_name=base.HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py'),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
         assert json.loads(response.body) == json.loads(HG_NODE_HISTORY)
     def test_file_source_history_git(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='history',
                                     repo_name=base.GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py'),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
         assert json.loads(response.body) == json.loads(GIT_NODE_HISTORY)
     def test_file_annotation(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=base.HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate='1'))
         response.mustcontain("""r356:25213a5fbb04""")
     def test_file_annotation_git(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=base.GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate='1'))
         response.mustcontain("""r345:c994f0de03b2""")
     def test_file_annotation_history(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='history',
                                     repo_name=base.HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate='1'),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert json.loads(response.body) == json.loads(HG_NODE_HISTORY)
     def test_file_annotation_history_git(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='history',
                                     repo_name=base.GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate=True),
                                 extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert json.loads(response.body) == json.loads(GIT_NODE_HISTORY)
     def test_file_authors(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='authors',
                                     repo_name=base.HG_REPO,
                                     revision='tip',
                                     f_path='vcs/nodes.py',
                                     annotate='1'))
         response.mustcontain('Marcin Kuzminski')
         response.mustcontain('Lukasz Balcerzak')
     def test_file_authors_git(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='authors',
                                     repo_name=base.GIT_REPO,
                                     revision='master',
                                     f_path='vcs/nodes.py',
                                     annotate='1'))
         response.mustcontain('Marcin Kuzminski')
         response.mustcontain('Lukasz Balcerzak')
     def test_archival(self):
         self.log_user()
         _set_downloads(base.HG_REPO, set_to=True)
         for arch_ext, info in ARCHIVE_SPECS.items():
             short = '27cd5cce30c9%s' % arch_ext
             fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
             filename = '%s-%s' % (base.HG_REPO, short)
             response = self.app.get(base.url(controller='files',
                                         action='archivefile',
                                         repo_name=base.HG_REPO,
                                         fname=fname))
             assert response.status == '200 OK'
             heads = [
                 ('Pragma', 'no-cache'),
                 ('Cache-Control', 'no-cache'),
                 ('Content-Disposition', 'attachment; filename=%s' % filename),
                 ('Content-Type', info[0]),
+            ]
             assert sorted(response.response._headers.items()) == sorted(heads)
     def test_archival_wrong_ext(self):
         self.log_user()
         _set_downloads(base.HG_REPO, set_to=True)
         for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:
             fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
             response = self.app.get(base.url(controller='files',
                                         action='archivefile',
                                         repo_name=base.HG_REPO,
                                         fname=fname))
             response.mustcontain('Unknown archive type')
     def test_archival_wrong_revision(self):
         self.log_user()
         _set_downloads(base.HG_REPO, set_to=True)
         for rev in ['00x000000', 'tar', 'wrong', '@##$@$42413232', '232dffcd']:
             fname = '%s.zip' % rev
             response = self.app.get(base.url(controller='files',
                                         action='archivefile',
                                         repo_name=base.HG_REPO,
                                         fname=fname))
             response.mustcontain('Unknown revision')
     #==========================================================================
     # RAW FILE
     #==========================================================================
     def test_raw_file_ok(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='rawfile',
                                     repo_name=base.HG_REPO,
                                     revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
                                     f_path='vcs/nodes.py'))
         assert response.content_disposition == "attachment; filename=nodes.py"
         assert response.content_type == mimetypes.guess_type("nodes.py")[0]
     def test_raw_file_wrong_cs(self):
         self.log_user()
         rev = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
         response = self.app.get(base.url(controller='files', action='rawfile',
                                     repo_name=base.HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = """Such revision does not exist for this repository"""
         response.mustcontain(msg)
     def test_raw_file_wrong_f_path(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(base.url(controller='files', action='rawfile',
                                     repo_name=base.HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = "There is no file nor directory at the given path: &apos;%s&apos; at revision %s" % (f_path, rev[:12])
         response.mustcontain(msg)
     #==========================================================================
     # RAW RESPONSE - PLAIN
     #==========================================================================
     def test_raw_ok(self):
         self.log_user()
         response = self.app.get(base.url(controller='files', action='raw',
                                     repo_name=base.HG_REPO,
                                     revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
                                     f_path='vcs/nodes.py'))
         assert response.content_type == "text/plain"
     def test_raw_wrong_cs(self):
         self.log_user()
         rev = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
         response = self.app.get(base.url(controller='files', action='raw',
                                     repo_name=base.HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = """Such revision does not exist for this repository"""
         response.mustcontain(msg)
     def test_raw_wrong_f_path(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(base.url(controller='files', action='raw',
                                     repo_name=base.HG_REPO,
                                     revision=rev,
                                     f_path=f_path), status=404)
         msg = "There is no file nor directory at the given path: &apos;%s&apos; at revision %s" % (f_path, rev[:12])
         response.mustcontain(msg)
     def test_ajaxed_files_list(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         response = self.app.get(
             base.url('files_nodelist_home', repo_name=base.HG_REPO, f_path='/',
                 revision=rev),
             extra_environ={'HTTP_X_PARTIAL_XHR': '1'},
+        )
         response.mustcontain("vcs/web/simplevcs/views/repository.py")
     # Hg - ADD FILE
     def test_add_file_view_hg(self):
         self.log_user()
         response = self.app.get(base.url('files_add_home',
                                       repo_name=base.HG_REPO,
                                       revision='tip', f_path='/'))
     def test_add_file_into_hg_missing_content(self):
         self.log_user()
         response = self.app.post(base.url('files_add_home',
                                       repo_name=base.HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': '',
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No content')
     def test_add_file_into_hg_missing_filename(self):
         self.log_user()
         response = self.app.post(base.url('files_add_home',
                                       repo_name=base.HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @base.parametrize('location,filename', [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_hg_bad_filenames(self, location, filename):
         self.log_user()
         response = self.app.post(base.url('files_add_home',
                                       repo_name=base.HG_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @base.parametrize('cnt,location,filename', [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_hg(self, cnt, location, filename):
         self.log_user()
         repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='hg')
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - add file
     def test_add_file_view_git(self):
         self.log_user()
         response = self.app.get(base.url('files_add_home',
                                       repo_name=base.GIT_REPO,
                                       revision='tip', f_path='/'))
     def test_add_file_into_git_missing_content(self):
         self.log_user()
         response = self.app.post(base.url('files_add_home',
                                       repo_name=base.GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                      'content': '',
                                      '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No content')
     def test_add_file_into_git_missing_filename(self):
         self.log_user()
         response = self.app.post(base.url('files_add_home',
                                       repo_name=base.GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'No filename')
     @base.parametrize('location,filename', [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_git_bad_filenames(self, location, filename):
         self.log_user()
         response = self.app.post(base.url('files_add_home',
                                       repo_name=base.GIT_REPO,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
     @base.parametrize('cnt,location,filename', [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_git(self, cnt, location, filename):
         self.log_user()
         repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='git')
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "foo",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Hg - EDIT
     def test_edit_file_view_hg(self):
         self.log_user()
         response = self.app.get(base.url('files_edit_home',
                                       repo_name=base.HG_REPO,
                                       revision='tip', f_path='vcs/nodes.py'))
         # Odd error when on tip ...
         self.checkSessionFlash(response, "You can only edit files with revision being a valid branch")
         assert "Commit Message" not in response.body
+        assert b"Commit Message" not in response.body
         # Specify branch head revision to avoid "valid branch" error and get coverage of edit form
         response = self.app.get(base.url('files_edit_home',
                                       repo_name=base.HG_REPO,
                                       revision='96507bd11ecc815ebc6270fdf6db110928c09c1e', f_path='vcs/nodes.py'))
         assert "Commit Message" in response.body
+        assert b"Commit Message" in response.body
     def test_edit_file_view_not_on_branch_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(base.url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only edit files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_edit_file_view_commit_changes_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(base.url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
                                         '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - edit
     def test_edit_file_view_git(self):
         self.log_user()
         response = self.app.get(base.url('files_edit_home',
                                       repo_name=base.GIT_REPO,
                                       revision='tip', f_path='vcs/nodes.py'))
     def test_edit_file_view_not_on_branch_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(base.url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only edit files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_edit_file_view_commit_changes_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(base.url('files_edit_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'content': "def py():\n print 'hello world'\n",
                                         'message': 'i committed',
                                         '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Hg - delete
     def test_delete_file_view_hg(self):
         self.log_user()
         response = self.app.get(base.url('files_delete_home',
                                      repo_name=base.HG_REPO,
                                      revision='tip', f_path='vcs/nodes.py'))
     def test_delete_file_view_not_on_branch_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(base.url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only delete files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_delete_file_view_commit_changes_hg(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(base.url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'message': 'i committed',
                                         '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file %s' % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     # Git - delete
     def test_delete_file_view_git(self):
         self.log_user()
         response = self.app.get(base.url('files_delete_home',
                                      repo_name=base.HG_REPO,
                                      revision='tip', f_path='vcs/nodes.py'))
     def test_delete_file_view_not_on_branch_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip', f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.get(base.url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision='tip', f_path=posixpath.join(location, filename)),
                                     status=302)
             self.checkSessionFlash(response,
                 'You can only delete files with revision being a valid branch')
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_delete_file_view_commit_changes_git(self):
         self.log_user()
         repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
         ## add file
         location = 'vcs'
         filename = 'nodes.py'
         response = self.app.post(base.url('files_add_home',
                                       repo_name=repo.repo_name,
                                       revision='tip',
                                       f_path='/'),
                                  params={
                                     'content': "def py():\n print 'hello'\n",
                                     'filename': filename,
                                     'location': location,
                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response.follow()
         try:
             self.checkSessionFlash(response, 'Successfully committed to %s'
                                    % posixpath.join(location, filename))
             response = self.app.post(base.url('files_delete_home',
                                           repo_name=repo.repo_name,
                                           revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
                                           f_path=posixpath.join(location, filename)),
                                      params={
                                         'message': 'i committed',
                                         '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                      },
                                     status=302)
             self.checkSessionFlash(response,
                                    'Successfully deleted file %s' % posixpath.join(location, filename))
         finally:
             fixture.destroy_repo(repo.repo_name)
     def test_png_diff_no_crash_hg(self):
         self.log_user()
         response = self.app.get(base.url('files_diff_home',
                                     repo_name=base.HG_REPO,
                                     f_path='docs/theme/ADC/static/documentation.png',
                                     diff1='tip', diff2='tip'))
         response.mustcontain("""<pre>Binary file</pre>""")
     def test_png_diff_no_crash_git(self):
         self.log_user()
         response = self.app.get(base.url('files_diff_home',
                                     repo_name=base.GIT_REPO,
                                     f_path='docs/theme/ADC/static/documentation.png',
                                     diff1='master', diff2='master'))
         response.mustcontain("""<pre>Binary file</pre>""")

kallithea/tests/functional/test_login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import re
 import time
 import urllib.parse
 import mock
 from tg.util.webtest import test_context
 import kallithea.lib.celerylib.tasks
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import check_password
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.model import validators
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.db import User
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestLoginController(base.TestController):
     def test_index(self):
         response = self.app.get(base.url(controller='login', action='index'))
         assert response.status == '200 OK'
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_ADMIN_LOGIN,
                                   'password': base.TEST_USER_ADMIN_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, base.TEST_USER_ADMIN_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % base.HG_REPO)
     def test_login_regular_ok(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_REGULAR_LOGIN,
                                   'password': base.TEST_USER_REGULAR_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, base.TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % base.HG_REPO)
     def test_login_regular_email_ok(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_REGULAR_EMAIL,
                                   'password': base.TEST_USER_REGULAR_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, base.TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % base.HG_REPO)
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users'
         response = self.app.post(base.url(controller='login', action='index',
                                      came_from=test_came_from),
                                  {'username': base.TEST_USER_ADMIN_LOGIN,
                                   'password': base.TEST_USER_ADMIN_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('Users Administration')
     def test_login_do_not_remember(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_REGULAR_LOGIN,
                                   'password': base.TEST_USER_REGULAR_PASS,
                                   'remember': False,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie
     def test_login_remember(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_REGULAR_LOGIN,
                                   'password': base.TEST_USER_REGULAR_PASS,
                                   'remember': True,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie
     def test_logout(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_REGULAR_LOGIN,
                                   'password': base.TEST_USER_REGULAR_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         # Verify that a login session has been established.
         response = self.app.get(base.url(controller='login', action='index'))
         response = response.follow()
         assert 'authuser' in response.session
         response.click('Log Out')
         # Verify that the login session has been terminated.
         response = self.app.get(base.url(controller='login', action='index'))
         assert 'authuser' not in response.session
     @base.parametrize('url_came_from', [
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://ftp.example.com',),
           ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),
           ('//evil.example.com/',),
           ('/\r\nX-Header-Injection: boo',),
           ('/invälid_url_bytes',),
           ('non-absolute-path',),
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(base.url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': base.TEST_USER_ADMIN_LOGIN,
                                   'password': base.TEST_USER_ADMIN_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()},
                                  status=400)
     def test_login_short_password(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_ADMIN_LOGIN,
                                   'password': 'as',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '200 OK'
         response.mustcontain('Enter 3 characters or more')
     def test_login_wrong_username_password(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': 'error',
                                   'password': 'test12',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('Invalid username or password')
     def test_login_non_ascii(self):
         response = self.app.post(base.url(controller='login', action='index'),
                                  {'username': base.TEST_USER_REGULAR_LOGIN,
                                   'password': 'blåbærgrød',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('>Invalid username or password<')
     # verify that get arguments are correctly passed along login redirection
     @base.parametrize('args', [
         {'foo':'one', 'bar':'two'},
         {'blue': u'blå', 'green': u'grøn'},
     ])
     def test_redirection_to_login_form_preserves_get_args(self, args):
         with fixture.anon_access(False):
             response = self.app.get(base.url(controller='summary', action='index',
                                         repo_name=base.HG_REPO,
                                         **args))
             assert response.status == '302 Found'
             came_from = urllib.parse.parse_qs(urllib.parse.urlparse(response.location).query)['came_from'][0]
             came_from_qs = urllib.parse.parse_qsl(urllib.parse.urlparse(came_from).query)
-            assert sorted(came_from_qs) == sorted((k, v.encode('utf-8')) for k, v in args.items())
             assert sorted(came_from_qs) == sorted(args.items())
     @base.parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_preserves_get_args(self, args, args_encoded):
         response = self.app.get(base.url(controller='login', action='index',
                                     came_from=base.url('/_admin/users', **args)))
         came_from = urllib.parse.parse_qs(urllib.parse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     @base.parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(base.url(controller='login', action='index',
                                      came_from=base.url('/_admin/users', **args)),
                                  {'username': base.TEST_USER_ADMIN_LOGIN,
                                   'password': base.TEST_USER_ADMIN_PASS,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '302 Found'
         for encoded in args_encoded:
             assert encoded in response.location
     @base.parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(base.url(controller='login', action='index',
                                      came_from=base.url('/_admin/users', **args)),
                                  {'username': 'error',
                                   'password': 'test12',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('Invalid username or password')
         came_from = urllib.parse.parse_qs(urllib.parse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     #==========================================================================
     # REGISTRATIONS
     #==========================================================================
     def test_register(self):
         response = self.app.get(base.url(controller='login', action='register'))
         response.mustcontain('Sign Up')
     def test_register_err_same_username(self):
         uname = base.TEST_USER_ADMIN_LOGIN
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@example.com',
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': uname})
         response.mustcontain(msg)
     def test_register_err_same_email(self):
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': base.TEST_USER_ADMIN_EMAIL,
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': base.TEST_USER_ADMIN_EMAIL.title(),
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_wrong_data(self):
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': 'test',
                                              'password_confirmation': 'test',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
     def test_register_err_username(self):
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': 'error user',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         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 an '
                 'alphanumeric character')
     def test_register_err_case_sensitive(self):
         usr = base.TEST_USER_ADMIN_LOGIN.title()
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('An email address must contain a single @')
         with test_context(self.app):
             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(base.url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test',
                                          '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test',
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         with test_context(self.app):
             msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = 'user4@example.com'
         name = 'testname'
         lastname = 'testlastname'
         response = self.app.post(base.url(controller='login', action='register'),
                                             {'username': username,
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True,
                                              '_session_csrf_secret_token': self.session_csrf_secret_token()})  # This should be overridden
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'You have successfully registered with Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password) == True
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.api_key is not None
         assert ret.admin == False
     #==========================================================================
     # PASSWORD RESET
     #==========================================================================
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username%wrongmail.org'
         response = self.app.post(
                         base.url(controller='login', action='password_reset'),
                             {'email': bad_email,
                              '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('An email address must contain a single @')
     def test_forgot_password(self):
         response = self.app.get(base.url(controller='login',
                                     action='password_reset'))
         assert response.status == '200 OK'
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@example.com'
         name = u'passwd'
         lastname = u'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         token = UserModel().get_reset_password_token(
             User.get_by_username(username), timestamp, self.session_csrf_secret_token())
         collected = []
         def mock_send_email(recipients, subject, body='', html_body='', headers=None, author=None):
             collected.append((recipients, subject, body, html_body))
         with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', mock_send_email), \
                 mock.patch.object(time, 'time', lambda: timestamp):
             response = self.app.post(base.url(controller='login',
                                          action='password_reset'),
                                      {'email': email,
                                       '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         ((recipients, subject, body, html_body),) = collected
         assert recipients == ['username@example.com']
         assert subject == 'Password reset link'
         assert '\n%s\n' % token in body
         (confirmation_url,) = (line for line in body.splitlines() if line.startswith('http://'))
         assert ' href="%s"' % confirmation_url.replace('&', '&amp;').replace('@', '%40') in html_body
         d = urllib.parse.parse_qs(urllib.parse.urlparse(confirmation_url).query)
         assert d['token'] == [token]
         assert d['timestamp'] == [str(timestamp)]
         assert d['email'] == [email]
         response = response.follow()
         # BAD TOKEN
         bad_token = "bad"
         response = self.app.post(base.url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': bad_token,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  })
         assert response.status == '200 OK'
         response.mustcontain('Invalid password reset token')
         # GOOD TOKEN
         response = self.app.get(confirmation_url)
         assert response.status == '200 OK'
         response.mustcontain("You are about to set a new password for the email address %s" % email)
         response.mustcontain('<form action="%s" method="post">' % base.url(controller='login', action='password_reset_confirmation'))
         response.mustcontain('value="%s"' % self.session_csrf_secret_token())
         response.mustcontain('value="%s"' % token)
         response.mustcontain('value="%s"' % timestamp)
         response.mustcontain('value="username@example.com"')
         # fake a submit of that form
         response = self.app.post(base.url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  })
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'Successfully updated password')
         response = response.follow()
     #==========================================================================
     # API
     #==========================================================================
     def _api_key_test(self, api_key, status):
         """Verifies HTTP status code for accessing an auth-requiring page,
         using the given api_key URL parameter as well as using the API key
         with bearer authentication.
         If api_key is None, no api_key is passed at all. If api_key is True,
         a real, working API key is used.
         """
         with fixture.anon_access(False):
             if api_key is None:
                 params = {}
                 headers = {}
             else:
                 if api_key is True:
                     api_key = User.get_first_admin().api_key
                 params = {'api_key': api_key}
                 headers = {'Authorization': 'Bearer ' + str(api_key)}
             self.app.get(base.url(controller='changeset', action='changeset_raw',
                              repo_name=base.HG_REPO, revision='tip', **params),
                          status=status)
             self.app.get(base.url(controller='changeset', action='changeset_raw',
                              repo_name=base.HG_REPO, revision='tip'),
                          headers=headers,
                          status=status)
     @base.parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 403),
         ('fake_number', '123456', 403),
         ('fake_not_alnum', 'a-z', 403),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 403),
         ('proper_api_key', True, 200)
     ])
     def test_access_page_via_api_key(self, test_name, api_key, code):
         self._api_key_test(api_key, code)
     def test_access_page_via_extra_api_key(self):
         new_api_key = ApiKeyModel().create(base.TEST_USER_ADMIN_LOGIN, u'test')
         Session().commit()
         self._api_key_test(new_api_key.api_key, status=200)
     def test_access_page_via_expired_api_key(self):
         new_api_key = ApiKeyModel().create(base.TEST_USER_ADMIN_LOGIN, u'test')
         Session().commit()
         # patch the API key and make it expired
         new_api_key.expires = 0
         Session().commit()
         self._api_key_test(new_api_key.api_key, status=403)

kallithea/tests/models/test_diff_parsers.py

➞

Show inline comments

 from kallithea.lib.diffs import BIN_FILENODE, CHMOD_FILENODE, COPIED_FILENODE, DEL_FILENODE, MOD_FILENODE, NEW_FILENODE, RENAMED_FILENODE, DiffProcessor
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 DIFF_FIXTURES = {
     'hg_diff_add_single_binary_file.diff': [
         ('US Warszawa.jpg', 'added',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {NEW_FILENODE: 'new file 100755',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_mod_single_binary_file.diff': [
         ('US Warszawa.jpg', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {MOD_FILENODE: 'modified file',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_mod_single_file_and_rename_and_chmod.diff': [
         ('README', 'renamed',
          {'added': 3,
           'deleted': 0,
           'binary': False,
           'ops': {RENAMED_FILENODE: 'file renamed from README.rst to README',
                   CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
     ],
     'hg_diff_mod_file_and_rename.diff': [
         ('README.rst', 'renamed',
          {'added': 3,
           'deleted': 0,
           'binary': False,
           'ops': {RENAMED_FILENODE: 'file renamed from README to README.rst'}}),
     ],
     'hg_diff_del_single_binary_file.diff': [
         ('US Warszawa.jpg', 'removed',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {DEL_FILENODE: 'deleted file',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_chmod_and_mod_single_binary_file.diff': [
         ('gravatar.png', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_chmod.diff': [
         ('file', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
     ],
     'hg_diff_rename_file.diff': [
         ('file_renamed', 'renamed',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {RENAMED_FILENODE: 'file renamed from file to file_renamed'}}),
     ],
     'hg_diff_rename_and_chmod_file.diff': [
         ('README', 'renamed',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755',
                   RENAMED_FILENODE: 'file renamed from README.rst to README'}}),
     ],
     'hg_diff_binary_and_normal.diff': [
         ('img/baseline-10px.png', 'added',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {NEW_FILENODE: 'new file 100644',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('img/baseline-20px.png', 'removed',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {DEL_FILENODE: 'deleted file',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('index.html', 'modified',
          {'added': 3,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('js/global.js', 'removed',
          {'added': 0,
           'deleted': 75,
           'binary': False,
           'ops': {DEL_FILENODE: 'deleted file'}}),
         ('js/jquery/hashgrid.js', 'added',
          {'added': 340,
           'deleted': 0,
           'binary': False,
           'ops': {NEW_FILENODE: 'new file 100755'}}),
         ('less/docs.less', 'modified',
          {'added': 34,
           'deleted': 0,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('less/scaffolding.less', 'modified',
          {'added': 1,
           'deleted': 3,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('readme.markdown', 'modified',
          {'added': 1,
           'deleted': 10,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
     ],
     'git_diff_chmod.diff': [
         ('work-horus.xls', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}})
     ],
     'git_diff_rename_file.diff': [
         ('file.xls', 'renamed',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {RENAMED_FILENODE: 'file renamed from work-horus.xls to file.xls'}}),
         ('files/var/www/favicon.ico/DEFAULT',
          'renamed',
          {'added': 0,
           'binary': True,
           'deleted': 0,
           'ops': {4: 'file renamed from files/var/www/favicon.ico to files/var/www/favicon.ico/DEFAULT',
 : 'modified file chmod 100644 => 100755'}})
     ],
     'git_diff_mod_single_binary_file.diff': [
         ('US Warszawa.jpg', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {MOD_FILENODE: 'modified file',
                   BIN_FILENODE: 'binary diff not shown'}})
     ],
     'git_diff_binary_and_normal.diff': [
         ('img/baseline-10px.png', 'added',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {NEW_FILENODE: 'new file 100644',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('img/baseline-20px.png', 'removed',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {DEL_FILENODE: 'deleted file',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('index.html', 'modified',
          {'added': 3,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('js/global.js', 'removed',
          {'added': 0,
           'deleted': 75,
           'binary': False,
           'ops': {DEL_FILENODE: 'deleted file'}}),
         ('js/jquery/hashgrid.js', 'added',
          {'added': 340,
           'deleted': 0,
           'binary': False,
           'ops': {NEW_FILENODE: 'new file 100755'}}),
         ('less/docs.less', 'modified',
          {'added': 34,
           'deleted': 0,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('less/scaffolding.less', 'modified',
          {'added': 1,
           'deleted': 3,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('readme.markdown', 'modified',
          {'added': 1,
           'deleted': 10,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
     ],
     'diff_with_diff_data.diff': [
         ('vcs/backends/base.py', 'modified',
          {'added': 18,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/backends/git/repository.py', 'modified',
          {'added': 46,
           'deleted': 15,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/backends/hg.py', 'modified',
          {'added': 22,
           'deleted': 3,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/tests/test_git.py', 'modified',
          {'added': 5,
           'deleted': 5,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/tests/test_repository.py', 'modified',
          {'added': 174,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
     ],
     'git_diff_modify_binary_file.diff': [
         ('file.name', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {MOD_FILENODE: 'modified file',
                   BIN_FILENODE: 'binary diff not shown'}})
     ],
     'hg_diff_copy_file.diff': [
         ('file2', 'modified',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {COPIED_FILENODE: 'file copied from file1 to file2'}}),
     ],
     'hg_diff_copy_and_modify_file.diff': [
         ('file3', 'modified',
          {'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', 'modified',
          {'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', 'modified',
          {'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', 'renamed',
          {'added': 3,
           'deleted': 2,
           'binary': False,
           'ops': {RENAMED_FILENODE: 'file renamed from oh no to oh yes'}}),
     ],
+}
 class TestDiffLib(base.TestController):
     @base.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]:
             d['line'] = d['line'].encode()  # not needed for py3
             l.append('%(action)-7s %(new_lineno)3s %(old_lineno)3s %(line)r\n' % d)
         s = ''.join(l)
         assert s == r'''
 context ... ... '@@ -51,6 +51,13 @@\n'
 unmod    51  51 '<u>\t</u>begin();\n'
 unmod    52  52 '<u>\t</u>\n'
 add      53     '<u>\t</u>int foo;<u class="cr"></u>\n'
 add      54     '<u>\t</u>int bar; <u class="cr"></u>\n'
 add      55     '<u>\t</u>int baz;<u>\t</u><u class="cr"></u>\n'
 add      56     '<u>\t</u>int space; <i></i>'
 add      57     '<u>\t</u>int tab;<u>\t</u>\n'
 add      58     '<u>\t</u>\n'
 unmod    59  53 ' <i></i>'
 del          54 '<u>\t</u>#define MAX_STEPS (48)\n'
 add      60     '<u>\t</u><u class="cr"></u>\n'
 add      61     '<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'
 unmod    62  55 '\n'
 del          56 '<u>\t</u>#define MIN_STEPS (<del>48</del>)\n'
 add      63     '<u>\t</u>#define MIN_STEPS (<ins>42</ins>)\n'
 '''

kallithea/tests/other/test_vcs_operations.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/>.
 """
 Test suite for vcs push/pull operations.
 The tests need Git > 1.8.1.
 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.
 """
 from __future__ import print_function
 import json
 import os
 import re
 import tempfile
 import time
 import urllib.request
 from subprocess import PIPE, Popen
 from tempfile import _RandomNameSequence
 import pytest
 from kallithea import CONFIG
 from kallithea.lib.utils2 import ascii_bytes
+from kallithea.lib.utils2 import ascii_bytes, safe_str
 from kallithea.model.db import CacheInvalidation, Repository, Ui, User, UserIpMap, UserLog
 from kallithea.model.meta import Session
 from kallithea.model.ssh_key import SshKeyModel
 from kallithea.model.user import UserModel
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 DEBUG = True
 HOST = '127.0.0.1:4999'  # test host
 fixture = Fixture()
 # Parameterize different kinds of VCS testing - both the kind of VCS and the
 # access method (HTTP/SSH)
 # Mixin for using HTTP and SSH URLs
 class HttpVcsTest(object):
     @staticmethod
     def repo_url_param(webserver, repo_name, **kwargs):
         return webserver.repo_url(repo_name, **kwargs)
 class SshVcsTest(object):
     public_keys = {
         base.TEST_USER_REGULAR_LOGIN: u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== kallithea@localhost',
         base.TEST_USER_ADMIN_LOGIN: u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUq== kallithea@localhost',
+    }
     @classmethod
     def repo_url_param(cls, webserver, repo_name, username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS, client_ip=base.IP_ADDR):
         user = User.get_by_username(username)
         if user.ssh_keys:
             ssh_key = user.ssh_keys[0]
         else:
             sshkeymodel = SshKeyModel()
             ssh_key = sshkeymodel.create(user, u'test key', cls.public_keys[user.username])
             Session().commit()
         return cls._ssh_param(repo_name, user, ssh_key, client_ip)
 # Mixins for using Mercurial and Git
 class HgVcsTest(object):
     repo_type = 'hg'
     repo_name = base.HG_REPO
 class GitVcsTest(object):
     repo_type = 'git'
     repo_name = base.GIT_REPO
 # Combine mixins to give the combinations we want to parameterize tests with
 class HgHttpVcsTest(HgVcsTest, HttpVcsTest):
     pass
 class GitHttpVcsTest(GitVcsTest, HttpVcsTest):
     pass
 class HgSshVcsTest(HgVcsTest, SshVcsTest):
     @staticmethod
     def _ssh_param(repo_name, user, ssh_key, client_ip):
         # Specify a custom ssh command on the command line
         return r"""--config ui.ssh="bash -c 'SSH_ORIGINAL_COMMAND=\"\$2\" SSH_CONNECTION=\"%s 1024 127.0.0.1 22\" kallithea-cli ssh-serve -c %s %s %s' --" ssh://someuser@somehost/%s""" % (
             client_ip,
             CONFIG['__file__'],
             user.user_id,
             ssh_key.user_ssh_key_id,
             repo_name)
 class GitSshVcsTest(GitVcsTest, SshVcsTest):
     @staticmethod
     def _ssh_param(repo_name, user, ssh_key, client_ip):
         # Set a custom ssh command in the global environment
         os.environ['GIT_SSH_COMMAND'] = r"""bash -c 'SSH_ORIGINAL_COMMAND="$2" SSH_CONNECTION="%s 1024 127.0.0.1 22" kallithea-cli ssh-serve -c %s %s %s' --""" % (
             client_ip,
             CONFIG['__file__'],
             user.user_id,
             ssh_key.user_ssh_key_id)
         return "ssh://someuser@somehost/%s""" % repo_name
 parametrize_vcs_test = base.parametrize('vt', [
     HgHttpVcsTest,
     GitHttpVcsTest,
     HgSshVcsTest,
     GitSshVcsTest,
 ])
 parametrize_vcs_test_hg = base.parametrize('vt', [
     HgHttpVcsTest,
     HgSshVcsTest,
 ])
 parametrize_vcs_test_http = base.parametrize('vt', [
     HgHttpVcsTest,
     GitHttpVcsTest,
 ])
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, *args, **environ):
         """
         Runs command on the system with given ``args`` using simple space
         join without safe quoting.
         """
         command = ' '.join(args)
         ignoreReturnCode = environ.pop('ignoreReturnCode', False)
         if DEBUG:
             print('*** CMD %s ***' % command)
         testenv = dict(os.environ)
         testenv['LANG'] = 'en_US.UTF-8'
         testenv['LANGUAGE'] = 'en_US:en'
         testenv['HGPLAIN'] = ''
         testenv['HGRCPATH'] = ''
         testenv.update(environ)
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd, env=testenv)
         stdout, stderr = p.communicate()
         if DEBUG:
             if stdout:
                 print('stdout:', stdout)
             if stderr:
                 print('stderr:', stderr)
         if not ignoreReturnCode:
             assert p.returncode == 0
         return stdout, stderr
+        return safe_str(stdout), safe_str(stderr)
 def _get_tmp_dir(prefix='vcs_operations-', suffix=''):
     return tempfile.mkdtemp(dir=base.TESTS_TMP_PATH, prefix=prefix, suffix=suffix)
 def _add_files(vcs, dest_dir, files_no=3):
     """
     Generate some files, add it to dest_dir repo and push back
     vcs is git or hg and defines what VCS we want to make those files for
     :param vcs:
     :param dest_dir:
     """
     added_file = '%ssetup.py' % next(_RandomNameSequence())
     open(os.path.join(dest_dir, added_file), 'a').close()
     Command(dest_dir).execute(vcs, 'add', added_file)
     email = 'me@example.com'
     if os.name == 'nt':
         author_str = 'User <%s>' % email
     else:
         author_str = 'User ǝɯɐᴎ <%s>' % email
     for i in range(files_no):
         cmd = """echo "added_line%s" >> %s""" % (i, added_file)
         Command(dest_dir).execute(cmd)
         if vcs == 'hg':
             cmd = """hg commit -m "committed new %s" -u "%s" "%s" """ % (
                 i, author_str, added_file
+            )
         elif vcs == 'git':
             cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (
                 i, author_str, added_file
+            )
         # git commit needs EMAIL on some machines
         Command(dest_dir).execute(cmd, EMAIL=email)
 def _add_files_and_push(webserver, vt, dest_dir, clone_url, ignoreReturnCode=False, files_no=3):
     _add_files(vt.repo_type, dest_dir, files_no=files_no)
     # PUSH it back
     stdout = stderr = None
     if vt.repo_type == 'hg':
         stdout, stderr = Command(dest_dir).execute('hg push -f --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)
     elif vt.repo_type == 'git':
         stdout, stderr = Command(dest_dir).execute('git push -f --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)
     return stdout, stderr
 def _check_outgoing(vcs, cwd, clone_url):
     if vcs == 'hg':
         # hg removes the password from default URLs, so we have to provide it here via the clone_url
         return Command(cwd).execute('hg -q outgoing', clone_url, ignoreReturnCode=True)
     elif vcs == 'git':
         Command(cwd).execute('git remote update')
         return Command(cwd).execute('git log origin/master..master')
 def set_anonymous_access(enable=True):
     user = User.get_default_user()
     user.active = enable
     Session().commit()
     if enable != User.get_default_user().active:
         raise Exception('Cannot set anonymous access')
 #==============================================================================
 # TESTS
 #==============================================================================
 def _check_proper_git_push(stdout, stderr):
     assert 'fatal' not in stderr
     assert 'rejected' not in stderr
     assert 'Pushing to' in stderr
     assert 'master -> master' in stderr
 @pytest.mark.usefixtures("test_context_fixture")
 class TestVCSOperations(base.TestController):
     @classmethod
     def setup_class(cls):
         # DISABLE ANONYMOUS ACCESS
         set_anonymous_access(False)
     @pytest.fixture()
     def testhook_cleanup(self):
         yield
         # remove hook
         for hook in ['prechangegroup', 'pretxnchangegroup', 'preoutgoing', 'changegroup', 'outgoing', 'incoming']:
             entry = Ui.get_by_key('hooks', '%s.testhook' % hook)
             if entry:
                 Session().delete(entry)
         Session().commit()
     @pytest.fixture(scope="module")
     def testfork(self):
         # create fork so the repo stays untouched
         git_fork_name = u'%s_fork%s' % (base.GIT_REPO, next(_RandomNameSequence()))
         fixture.create_fork(base.GIT_REPO, git_fork_name)
         hg_fork_name = u'%s_fork%s' % (base.HG_REPO, next(_RandomNameSequence()))
         fixture.create_fork(base.HG_REPO, hg_fork_name)
         return {'git': git_fork_name, 'hg': hg_fork_name}
     @parametrize_vcs_test
     def test_clone_repo_by_admin(self, webserver, vt):
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir())
         if vt.repo_type == 'git':
             assert 'Cloning into' in stdout + stderr
             assert stderr == '' or stdout == ''
         elif vt.repo_type == 'hg':
             assert 'requesting all changes' in stdout
             assert 'adding changesets' in stdout
             assert 'adding manifests' in stdout
             assert 'adding file changes' in stdout
             assert stderr == ''
     @parametrize_vcs_test_http
     def test_clone_wrong_credentials(self, webserver, vt):
         clone_url = vt.repo_url_param(webserver, vt.repo_name, password='bad!')
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         if vt.repo_type == 'git':
             assert 'fatal: Authentication failed' in stderr
         elif vt.repo_type == 'hg':
             assert 'abort: authorization failed' in stderr
     def test_clone_git_dir_as_hg(self, webserver):
         clone_url = HgHttpVcsTest.repo_url_param(webserver, base.GIT_REPO)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'HTTP Error 404: Not Found' in stderr or "not a valid repository" in stdout and 'abort:' in stderr
     def test_clone_hg_repo_as_git(self, webserver):
         clone_url = GitHttpVcsTest.repo_url_param(webserver, base.HG_REPO)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'not found' in stderr
     @parametrize_vcs_test
     def test_clone_non_existing_path(self, webserver, vt):
         clone_url = vt.repo_url_param(webserver, 'trololo')
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         if vt.repo_type == 'git':
             assert 'not found' in stderr or 'abort: Access to %r denied' % 'trololo' in stderr
         elif vt.repo_type == 'hg':
             assert 'HTTP Error 404: Not Found' in stderr or 'abort: no suitable response from remote hg' in stderr and 'remote: abort: Access to %r denied' % 'trololo' in stdout
     @parametrize_vcs_test
     def test_push_new_repo(self, webserver, vt):
         # Clear the log so we know what is added
         UserLog.query().delete()
         Session().commit()
         # Create an empty server repo using the API
         repo_name = u'new_%s_%s' % (vt.repo_type, next(_RandomNameSequence()))
         usr = User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         params = {
             "id": 7,
             "api_key": usr.api_key,
             "method": 'create_repo',
             "args": dict(repo_name=repo_name,
                          owner=base.TEST_USER_ADMIN_LOGIN,
                          repo_type=vt.repo_type),
+        }
         req = urllib.request.Request(
             'http://%s:%s/_admin/api' % webserver.server_address,
             data=ascii_bytes(json.dumps(params)),
             headers={'content-type': 'application/json'})
         response = urllib.request.urlopen(req)
         result = json.loads(response.read())
         # Expect something like:
         # {u'result': {u'msg': u'Created new repository `new_XXX`', u'task': None, u'success': True}, u'id': 7, u'error': None}
         assert result[u'result'][u'success']
         # Create local clone of the empty server repo
         local_clone_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, local_clone_dir)
         # Make 3 commits and push to the empty server repo.
         # The server repo doesn't have any other heads than the
         # refs/heads/master we are pushing, but the `git log` in the push hook
         # should still list the 3 commits.
         stdout, stderr = _add_files_and_push(webserver, vt, local_clone_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 'remote: added ' in stdout
         # Verify that we got the right events in UserLog. Expect something like:
         # <UserLog('id:new_git_XXX:started_following_repo')>
         # <UserLog('id:new_git_XXX:user_created_repo')>
         # <UserLog('id:new_git_XXX:pull')>
         # <UserLog('id:new_git_XXX:push:aed9d4c1732a1927da3be42c47eb9afdc200d427,d38b083a07af10a9f44193486959a96a23db78da,4841ff9a2b385bec995f4679ef649adb3f437622')>
         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'started_following_repo', 0),
             (u'user_created_repo', 0),
             (u'pull', 0),
             (u'push', 3)]
             if vt.repo_type == 'git' else [
             (u'started_following_repo', 0),
             (u'user_created_repo', 0),
             # (u'pull', 0), # Mercurial outgoing hook is not called for empty clones
             (u'push', 3)])
     @parametrize_vcs_test
     def test_push_new_file(self, webserver, testfork, vt):
         UserLog.query().delete()
         Session().commit()
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         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(base.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']
         # Test handling of URLs with extra '/' around repo_name
         stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/./%s/' % vt.repo_name), ignoreReturnCode=True)
         if issubclass(vt, HttpVcsTest):
             if vt.repo_type == 'git':
                 # NOTE: when pulling from http://hostname/./vcs_test_git/ , the git client will normalize that and issue an HTTP request to /vcs_test_git/info/refs
                 assert 'Already up to date.' in stdout
             else:
                 assert vt.repo_type == 'hg'
                 assert "abort: HTTP Error 404: Not Found" in stderr
         else:
             assert issubclass(vt, SshVcsTest)
             if vt.repo_type == 'git':
                 assert "abort: Access to './%s' denied" % vt.repo_name in stderr
             else:
                 assert "abort: Access to './%s' denied" % vt.repo_name in stdout
         stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/%s/' % vt.repo_name), ignoreReturnCode=True)
         if vt.repo_type == 'git':
             assert 'Already up to date.' in stdout
         else:
             assert vt.repo_type == 'hg'
             assert "no changes found" in stdout
         assert "denied" not in stderr
         assert "denied" not in stdout
         assert "404" not in stdout
     @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()
         clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url)
         if vt.repo_type == 'git':
             _check_proper_git_push(stdout, stderr)
         post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]
         assert pre_cached_tip != post_cached_tip
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                == testfork[vt.repo_type]).all()
         assert key == []
     @parametrize_vcs_test_http
     def test_push_wrong_credentials(self, webserver, vt):
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         clone_url = webserver.repo_url(vt.repo_name, username='bad', password='name')
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir,
                                              clone_url=clone_url, ignoreReturnCode=True)
         if vt.repo_type == 'git':
             assert 'fatal: Authentication failed' in stderr
         elif vt.repo_type == 'hg':
             assert 'abort: authorization failed' in stderr
     @parametrize_vcs_test
     def test_push_with_readonly_credentials(self, webserver, vt):
         UserLog.query().delete()
         Session().commit()
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name, username=base.TEST_USER_REGULAR_LOGIN, password=base.TEST_USER_REGULAR_PASS)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, ignoreReturnCode=True, clone_url=clone_url)
         if vt.repo_type == 'git':
             assert 'The requested URL returned error: 403' in stderr or 'abort: Push access to %r denied' % str(vt.repo_name) in stderr
         elif vt.repo_type == 'hg':
             assert 'abort: HTTP Error 403: Forbidden' in stderr or 'abort: push failed on remote' in stderr and 'remote: Push access to %r denied' % str(vt.repo_name) 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)]
     @parametrize_vcs_test
     def test_push_back_to_wrong_url(self, webserver, vt):
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(
             webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % (
                 webserver.server_address[0], webserver.server_address[1]),
             ignoreReturnCode=True)
         if vt.repo_type == 'git':
             assert 'not found' in stderr
         elif vt.repo_type == 'hg':
             assert 'HTTP Error 404: Not Found' in stderr
     @parametrize_vcs_test
     def test_ip_restriction(self, webserver, vt):
         user_model = UserModel()
         try:
             # Add IP constraint that excludes the test context:
             user_model.add_extra_ip(base.TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             # IP permissions are cached, need to wait for the cache in the server process to expire
             time.sleep(1.5)
             clone_url = vt.repo_url_param(webserver, vt.repo_name)
             stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
             if vt.repo_type == 'git':
                 # The message apparently changed in Git 1.8.3, so match it loosely.
                 assert re.search(r'\b403\b', stderr) or 'abort: User test_admin from 127.0.0.127 cannot be authorized' in stderr
             elif vt.repo_type == 'hg':
                 assert 'abort: HTTP Error 403: Forbidden' in stderr or 'remote: abort: User test_admin from 127.0.0.127 cannot be authorized' in stdout
         finally:
             # release IP restrictions
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
             # IP permissions are cached, need to wait for the cache in the server process to expire
             time.sleep(1.5)
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir())
         if vt.repo_type == 'git':
             assert 'Cloning into' in stdout + stderr
             assert stderr == '' or stdout == ''
         elif vt.repo_type == 'hg':
             assert 'requesting all changes' in stdout
             assert 'adding changesets' in stdout
             assert 'adding manifests' in stdout
             assert 'adding file changes' in stdout
             assert stderr == ''
     @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks
     def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, vt):
         # set prechangegroup to failing hook (returns True)
         Ui.create_or_update_hook('preoutgoing.testhook', 'python:kallithea.tests.fixture.failing_test_hook')
         Session().commit()
         # clone repo
         clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS)
         dest_dir = _get_tmp_dir()
         stdout, stderr = Command(base.TESTS_TMP_PATH) \
             .execute(vt.repo_type, 'clone', clone_url, dest_dir, ignoreReturnCode=True)
         if vt.repo_type == 'hg':
             assert 'preoutgoing.testhook hook failed' in stdout
         elif vt.repo_type == 'git':
             assert 'error: 406' in stderr
     @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks
     def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, vt):
         # set prechangegroup to failing hook (returns exit code 1)
         Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.failing_test_hook')
         Session().commit()
         # clone repo
         clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS)
         dest_dir = _get_tmp_dir()
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url,
                                              ignoreReturnCode=True)
         assert 'failing_test_hook failed' in stdout + stderr
         assert 'Traceback' not in stdout + stderr
         assert 'prechangegroup.testhook hook failed' in stdout + stderr
         # there are still outgoing changesets
         stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)
         assert stdout != ''
         # set prechangegroup hook to exception throwing method
         Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.exception_test_hook')
         Session().commit()
         # re-try to push
         stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)
         if vt is HgHttpVcsTest:
             # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned
             assert 'HTTP Error 500: INTERNAL SERVER ERROR' in stderr
         elif vt is HgSshVcsTest:
             assert 'remote: Exception: exception_test_hook threw an exception' in stdout
         else: assert False
         # there are still outgoing changesets
         stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)
         assert stdout != ''
         # set prechangegroup hook to method that returns False
         Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.passing_test_hook')
         Session().commit()
         # re-try to push
         stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)
         assert 'passing_test_hook succeeded' in stdout + stderr
         assert 'Traceback' not in stdout + stderr
         assert 'prechangegroup.testhook hook failed' not in stdout + stderr
         # no more outgoing changesets
         stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)
         assert stdout == ''
         assert stderr == ''
     def test_add_submodule_git(self, webserver, testfork):
         dest_dir = _get_tmp_dir()
         clone_url = GitHttpVcsTest.repo_url_param(webserver, base.GIT_REPO)
         fork_url = GitHttpVcsTest.repo_url_param(webserver, testfork['git'])
         # add submodule
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', fork_url, dest_dir)
         stdout, stderr = Command(dest_dir).execute('git submodule add', clone_url, 'testsubmodule')
         stdout, stderr = Command(dest_dir).execute('git commit -am "added testsubmodule pointing to', clone_url, '"', EMAIL=base.TEST_USER_ADMIN_EMAIL)
         stdout, stderr = Command(dest_dir).execute('git push', fork_url, 'master')
         # check for testsubmodule link in files page
         self.log_user()
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=testfork['git'],
                                     revision='tip',
                                     f_path='/'))
         # check _repo_files_url that will be used to reload as AJAX
         response.mustcontain('var _repo_files_url = ("/%s/files/");' % testfork['git'])
         response.mustcontain('<a class="submodule-dir" href="%s" target="_blank"><i class="icon-file-submodule"></i><span>testsubmodule @ ' % clone_url)
         # check that following a submodule link actually works - and redirects
         response = self.app.get(base.url(controller='files', action='index',
                                     repo_name=testfork['git'],
                                     revision='tip',
                                     f_path='/testsubmodule'),
                                 status=302)
         assert response.location == clone_url

0 comments (0 inline, 0 general)