kallithea Changeset - 082780404e6c

Changeset - 082780404e6c

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

Mads Kiilerich - 6 years ago 2019-12-16 02:29:34
mads@kiilerich.com

Grafted from: 3ebc4af045e2

cleanup: fix incorrect backslash escaping - mainly in regexps

6 files changed with 35 insertions and 35 deletions:

kallithea/tests/api/api_base.py

kallithea/tests/functional/test_changeset_pullrequests_comments.py

kallithea/tests/functional/test_pullrequests.py

kallithea/tests/other/test_libs.py

kallithea/tests/scripts/manual_test_crawler.py

kallithea/tests/vcs/test_repository.py

0 comments (0 inline, 0 general)

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -2320,249 +2320,249 @@ class _BaseTestApi(object): @@
                                   lifetime=10,
                                   description='foobar-gist',
                                   gist_type='public',
                                   files={'foobar': {'content': 'foo'}})
         response = api_call(self, params)
         response_json = response.json
         expected = {
             'gist': {
                 'access_id': response_json['result']['gist']['access_id'],
                 'created_on': response_json['result']['gist']['created_on'],
                 'description': 'foobar-gist',
                 'expires': response_json['result']['gist']['expires'],
                 'gist_id': response_json['result']['gist']['gist_id'],
                 'type': 'public',
                 'url': response_json['result']['gist']['url']
             },
             'msg': 'created new gist'
+        }
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(GistModel, 'create', crash)
     def test_api_create_gist_exception_occurred(self):
         id_, params = _build_data(self.apikey_regular, 'create_gist',
                                   files={})
         response = api_call(self, params)
         expected = 'failed to create gist'
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_gist(self):
         gist_id = fixture.create_gist().gist_access_id
         id_, params = _build_data(self.apikey, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
         self._compare_ok(id_, expected, given=response.body)
     def test_api_delete_gist_regular_user(self):
         gist_id = fixture.create_gist(owner=self.TEST_USER_LOGIN).gist_access_id
         id_, params = _build_data(self.apikey_regular, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
         self._compare_ok(id_, expected, given=response.body)
     def test_api_delete_gist_regular_user_no_permission(self):
         gist_id = fixture.create_gist().gist_access_id
         id_, params = _build_data(self.apikey_regular, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = 'gist `%s` does not exist' % (gist_id,)
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(GistModel, 'delete', crash)
     def test_api_delete_gist_exception_occurred(self):
         gist_id = fixture.create_gist().gist_access_id
         id_, params = _build_data(self.apikey, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = 'failed to delete gist ID:%s' % (gist_id,)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_ip(self):
         id_, params = _build_data(self.apikey, 'get_ip')
         response = api_call(self, params)
         expected = {
             'server_ip_addr': '0.0.0.0',
             'user_ips': []
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_server_info(self):
         id_, params = _build_data(self.apikey, 'get_server_info')
         response = api_call(self, params)
         expected = Setting.get_server_info()
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_changesets(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start=0, end=2)
         response = api_call(self, params)
         result = json.loads(response.body)["result"]
         assert len(result) == 3
         assert 'message' in result[0]
         assert 'added' not in result[0]
     def test_api_get_changesets_with_max_revisions(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start_date="2011-02-24T00:00:00", max_revisions=10)
         response = api_call(self, params)
         result = json.loads(response.body)["result"]
         assert len(result) == 10
         assert 'message' in result[0]
         assert 'added' not in result[0]
     def test_api_get_changesets_with_branch(self):
         if self.REPO == 'vcs_test_hg':
             branch = 'stable'
         else:
             pytest.skip("skipping due to missing branches in git test repo")
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, branch_name=branch, start_date="2011-02-24T00:00:00")
         response = api_call(self, params)
         result = json.loads(response.body)["result"]
         assert len(result) == 5
         assert 'message' in result[0]
         assert 'added' not in result[0]
     def test_api_get_changesets_with_file_list(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start_date="2010-04-07T23:30:30", end_date="2010-04-08T00:31:14", with_file_list=True)
         response = api_call(self, params)
         result = json.loads(response.body)["result"]
         assert len(result) == 3
         assert 'message' in result[0]
         assert 'added' in result[0]
     def test_api_get_changeset(self):
         review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION)
         response = api_call(self, params)
         result = json.loads(response.body)["result"]
         assert result["raw_id"] == self.TEST_REVISION
         assert "reviews" not in result
     def test_api_get_changeset_with_reviews(self):
         reviewobjs = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION,
                                   with_reviews=True)
         response = api_call(self, params)
         result = json.loads(response.body)["result"]
         assert result["raw_id"] == self.TEST_REVISION
         assert "reviews" in result
         assert len(result["reviews"]) == 1
         review = result["reviews"][0]
         expected = {
             'status': 'approved',
             'modified_at': reviewobjs[0].modified_at.replace(microsecond=0).isoformat(),
             'reviewer': 'test_admin',
+        }
         assert review == expected
     def test_api_get_changeset_that_does_not_exist(self):
         """ Fetch changeset status for non-existant changeset.
         revision id is the above git hash used in the test above with the
         last 3 nibbles replaced with 0xf.  Should not exist for git _or_ hg.
         """
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id = '7ab37bc680b4aa72c34d07b230c866c28e9fcfff')
         response = api_call(self, params)
         expected = u'Changeset %s does not exist' % ('7ab37bc680b4aa72c34d07b230c866c28e9fcfff',)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_changeset_without_permission(self):
         review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         RepoModel().revoke_user_permission(repo=self.REPO, user=self.TEST_USER_LOGIN)
         RepoModel().revoke_user_permission(repo=self.REPO, user="default")
         id_, params = _build_data(self.apikey_regular, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION)
         response = api_call(self, params)
         expected = u'Access denied to repo %s' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, u'get test')
         random_id = random.randrange(1, 9999)
         params = json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": 'get_pullrequest',
             "args": {"pullrequest_id": pull_request_id},
         })
         response = api_call(self, params)
         pullrequest = PullRequest().get(pull_request_id)
         expected = {
             "status": "new",
             "pull_request_id": pull_request_id,
             "description": "No description",
             "url": "/%s/pull-request/%s/_/%s" % (self.REPO, pull_request_id, "stable"),
             "reviewers": [{"username": "test_regular"}],
             "org_repo_url": "http://localhost:80/%s" % self.REPO,
             "org_ref_parts": ["branch", "stable", self.TEST_PR_SRC],
             "other_ref_parts": ["branch", "default", self.TEST_PR_DST],
             "comments": [{"username": TEST_USER_ADMIN_LOGIN, "text": "",
                          "comment_id": pullrequest.comments[0].comment_id}],
             "owner": TEST_USER_ADMIN_LOGIN,
             "statuses": [{"status": "under_review", "reviewer": TEST_USER_ADMIN_LOGIN, "modified_at": "2000-01-01T00:00:00"} for i in range(0, len(self.TEST_PR_REVISIONS))],
             "title": "get test",
             "revisions": self.TEST_PR_REVISIONS,
+        }
         self._compare_ok(random_id, expected,
                          given=re.sub("\d\d\d\d\-\d\d\-\d\dT\d\d\:\d\d\:\d\d",
+                         given=re.sub(r"\d\d\d\d\-\d\d\-\d\dT\d\d\:\d\d\:\d\d",
                                       "2000-01-01T00:00:00", response.body))
     def test_api_close_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, u'close test')
         random_id = random.randrange(1, 9999)
         params = json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "close_pr": True},
         })
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == ''
         assert pullrequest.status == PullRequest.STATUS_CLOSED
         assert pullrequest.is_closed() == True
     def test_api_status_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, u"status test")
         random_id = random.randrange(1, 9999)
         params = json.dumps({
             "id": random_id,
             "api_key": User.get_by_username(TEST_USER_REGULAR2_LOGIN).api_key,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "status": ChangesetStatus.STATUS_APPROVED},
         })
         response = api_call(self, params)
         pullrequest = PullRequest().get(pull_request_id)
         self._compare_error(random_id, "No permission to change pull request status. User needs to be admin, owner or reviewer.", given=response.body)
         assert ChangesetStatus.STATUS_UNDER_REVIEW == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]
         params = json.dumps({
             "id": random_id,
             "api_key": User.get_by_username(TEST_USER_REGULAR_LOGIN).api_key,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "status": ChangesetStatus.STATUS_APPROVED},
         })
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = PullRequest().get(pull_request_id)
         assert ChangesetStatus.STATUS_APPROVED == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]
     def test_api_comment_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, u"comment test")
         random_id = random.randrange(1, 9999)
         params = json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "comment_msg": "Looks good to me"},
         })
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == u'Looks good to me'

kallithea/tests/functional/test_changeset_pullrequests_comments.py

➞

Show inline comments

 import re
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.db import ChangesetComment, PullRequest
 from kallithea.model.meta import Session
 from kallithea.tests.base import *
 class TestChangeSetCommentsController(TestController):
     def setup_method(self, method):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
     def test_create(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'general comment on changeset'
         params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 1
     def test_create_inline(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'inline comment on changeset'
         f_path = 'vcs/web/simplevcs/views/repository.py'
         line = 'n1'
         params = {'text': text, 'f_path': f_path, 'line': line, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (1 inline, 0 general)'''
+        )
         response.mustcontain(
             '''<div class="comments-list-chunk" '''
             '''data-f_path="vcs/web/simplevcs/views/repository.py" '''
             '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 1
     def test_create_with_mention(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN
         params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain('<b>@%s</b> check CommentOnRevision' % TEST_USER_REGULAR_LOGIN)
         # test DB
         assert ChangesetComment.query().count() == 1
     def test_create_status_change(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'general comment on changeset'
         params = {'text': text, 'changeset_status': 'rejected',
                 '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 1
         # check status
         status = ChangesetStatusModel().get_status(repo=HG_REPO, revision=rev)
         assert status == 'rejected'
     def test_delete(self):
         self.log_user()
         rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
         text = u'general comment on changeset to be deleted'
         params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='changeset', action='comment',
                                      repo_name=HG_REPO, revision=rev),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         comments = ChangesetComment.query().all()
         assert len(comments) == 1
         comment_id = comments[0].comment_id
         self.app.post(url("changeset_comment_delete",
                                     repo_name=HG_REPO,
                                     comment_id=comment_id),
             params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         comments = ChangesetComment.query().all()
         assert len(comments) == 0
         response = self.app.get(url(controller='changeset', action='index',
                                 repo_name=HG_REPO, revision=rev))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 0 comments (0 inline, 0 general)'''
+        )
         response.mustcontain(no=text)
 class TestPullrequestsCommentsController(TestController):
     def setup_method(self, method):
         for x in ChangesetComment.query().all():
             Session().delete(x)
         Session().commit()
     def _create_pr(self):
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         pr_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
+        pr_id = int(re.search(r'/pull-request/(\d+)/', response.location).group(1))
         return pr_id
     def test_create(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         # PRs currently always have an initial 'Under Review' status change
         # that counts as a general comment, hence '2' in the test below. That
         # could be counted as a misfeature, to be reworked later.
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (0 inline, 2 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 2
     def test_create_inline(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'inline comment on changeset'
         f_path = 'vcs/web/simplevcs/views/repository.py'
         line = 'n1'
         params = {'text': text, 'f_path': f_path, 'line': line, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (1 inline, 1 general)'''
+        )
         response.mustcontain(
             '''<div class="comments-list-chunk" '''
             '''data-f_path="vcs/web/simplevcs/views/repository.py" '''
             '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 2
     def test_create_with_mention(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN
         params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (0 inline, 2 general)'''
+        )
         response.mustcontain('<b>@%s</b> check CommentOnRevision' % TEST_USER_REGULAR_LOGIN)
         # test DB
         assert ChangesetComment.query().count() == 2
     def test_create_status_change(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, 'changeset_status': 'rejected',
                 '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         # PRs currently always have an initial 'Under Review' status change
         # that counts as a general comment, hence '2' in the test below. That
         # could be counted as a misfeature, to be reworked later.
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 2 comments (0 inline, 2 general)'''
+        )
         response.mustcontain(text)
         # test DB
         assert ChangesetComment.query().count() == 2
         # check status
         status = ChangesetStatusModel().get_status(repo=HG_REPO, pull_request=pr_id)
         assert status == 'rejected'
     def test_delete(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on changeset to be deleted'
         params = {'text': text, '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         comments = ChangesetComment.query().all()
         assert len(comments) == 2
         comment_id = comments[-1].comment_id
         self.app.post(url("pullrequest_comment_delete",
                                     repo_name=HG_REPO,
                                     comment_id=comment_id),
             params={'_session_csrf_secret_token': self.session_csrf_secret_token()})
         comments = ChangesetComment.query().all()
         assert len(comments) == 1
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''<div class="comments-number">'''
             ''' 1 comment (0 inline, 1 general)'''
+        )
         response.mustcontain(no=text)
     def test_close_pr(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, 'save_close': 'close',
                 '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''))
         response.mustcontain(
             '''title (Closed)'''
+        )
         response.mustcontain(text)
         # test DB
         assert PullRequest.get(pr_id).status == PullRequest.STATUS_CLOSED
     def test_delete_pr(self):
         self.log_user()
         pr_id = self._create_pr()
         text = u'general comment on pullrequest'
         params = {'text': text, 'save_delete': 'delete',
                 '_session_csrf_secret_token': self.session_csrf_secret_token()}
         response = self.app.post(url(controller='pullrequests', action='comment',
                                      repo_name=HG_REPO, pull_request_id=pr_id),
                                      params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         # Test response...
         assert response.status == '200 OK'
         response = self.app.get(url(controller='pullrequests', action='show',
                                 repo_name=HG_REPO, pull_request_id=pr_id, extra=''), status=404)
         # test DB
         assert PullRequest.get(pr_id) is None
     def test_delete_closed_pr(self):
         self.log_user()
         pr_id = self._create_pr()
         # first close
         text = u'general comment on pullrequest'
         params = {'text': text, 'save_close': 'close',
                 '_session_csrf_secret_token': self.session_csrf_secret_token()}

kallithea/tests/functional/test_pullrequests.py

➞

Show inline comments

 import re
 import pytest
 from kallithea.controllers.pullrequests import PullrequestsController
 from kallithea.model.db import PullRequest, User
 from kallithea.model.meta import Session
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestPullrequestsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='pullrequests', action='index',
                                     repo_name=HG_REPO))
     def test_create_trivial(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('Successfully opened new pull request')
         response.mustcontain('No additional changesets found for iterating on this pull request')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')
     def test_available(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain(no='No additional changesets found for iterating on this pull request')
         response.mustcontain('The following additional changes are available on stable:')
         response.mustcontain('<input id="updaterev_4f7e2131323e0749a740c0a56ab68ae9269c562a" name="updaterev" type="radio" value="4f7e2131323e0749a740c0a56ab68ae9269c562a" />')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"') # as update
     def test_range(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('No additional changesets found for iterating on this pull request')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')
     def test_update_reviewers(self):
         self.log_user()
         regular_user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         regular_user2 = User.get_by_username(TEST_USER_REGULAR2_LOGIN)
         admin_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         # create initial PR
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                  status=302)
         pull_request1_id = re.search('/pull-request/(\d+)/', response.location).group(1)
+        pull_request1_id = re.search(r'/pull-request/(\d+)/', response.location).group(1)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request1_id)
         # create new iteration
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request1_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'review_members': [regular_user.user_id],
                                  },
                                  status=302)
         pull_request2_id = re.search('/pull-request/(\d+)/', response.location).group(1)
+        pull_request2_id = re.search(r'/pull-request/(\d+)/', response.location).group(1)
         assert pull_request2_id != pull_request1_id
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
         # verify reviewer was added
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         # update without creating new iteration
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request2_id),
+                                 {
                                   'pullrequest_title': 'Title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'org_review_members': [admin_user.user_id], # fake - just to get some 'meanwhile' warning ... but it is also added ...
                                   'review_members': [regular_user2.user_id, admin_user.user_id],
                                  },
                                  status=302)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
         # verify reviewers were added / removed
         response.mustcontain('Meanwhile, the following reviewers have been added: test_regular')
         response.mustcontain('Meanwhile, the following reviewers have been removed: test_admin')
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user2.user_id)
         response.mustcontain(no='<input type="hidden" value="%s" name="review_members" />' % admin_user.user_id)
     def test_update_with_invalid_reviewer(self):
         invalid_user_id = 99999
         self.log_user()
         # create a valid pull request
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                 status=302)
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
+        m = re.search(r'/pull-request/(\d+)/', response.location)
         assert m is not None
         pull_request_id = m.group(1)
         # update it
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'review_members': [str(invalid_user_id)],
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
     def test_edit_with_invalid_reviewer(self):
         invalid_user_id = 99999
         self.log_user()
         # create a valid pull request
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                  },
                                 status=302)
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
+        m = re.search(r'/pull-request/(\d+)/', response.location)
         assert m is not None
         pull_request_id = m.group(1)
         # edit it
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token(),
                                   'review_members': [str(invalid_user_id)],
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
     def test_iteration_refs(self):
         # Repo graph excerpt:
         #   o   fb95b340e0d0 webvcs
         #  /:
         # o :   41d2568309a0 default
         # : :
         # : o   5ec21f21aafe webvcs
         # : :
         # : o   9e6119747791 webvcs
         # : :
         # o :   3d1091ee5a53 default
         # :/
         # o     948da46b29c1 default
         self.log_user()
         # create initial PR
         response = self.app.post(
             url(controller='pullrequests', action='create', repo_name=HG_REPO),
+            {
                 'org_repo': HG_REPO,
                 'org_ref': 'rev:9e6119747791:9e6119747791ff886a5abe1193a730b6bf874e1c',
                 'other_repo': HG_REPO,
                 'other_ref': 'branch:default:3d1091ee5a533b1f4577ec7d8a226bb315fb1336',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 '_session_csrf_secret_token': self.session_csrf_secret_token(),
             },
             status=302)
         pr1_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
+        pr1_id = int(re.search(r'/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         assert pr1.org_ref == 'branch:webvcs:9e6119747791ff886a5abe1193a730b6bf874e1c'
         assert pr1.other_ref == 'branch:default:948da46b29c125838a717f6a8496eb409717078d'
         Session().rollback() # invalidate loaded PR objects before issuing next request.
         # create PR 2 (new iteration with same ancestor)
         response = self.app.post(
             url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr1_id),
+            {
                 'updaterev': '5ec21f21aafe95220f1fc4843a4a57c378498b71',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 'owner': TEST_USER_REGULAR_LOGIN,
                 '_session_csrf_secret_token': self.session_csrf_secret_token(),
              },
              status=302)
         pr2_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
+        pr2_id = int(re.search(r'/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         pr2 = PullRequest.get(pr2_id)
         assert pr2_id != pr1_id
         assert pr1.status == PullRequest.STATUS_CLOSED
         assert pr2.org_ref == 'branch:webvcs:5ec21f21aafe95220f1fc4843a4a57c378498b71'
         assert pr2.other_ref == pr1.other_ref
         Session().rollback() # invalidate loaded PR objects before issuing next request.
         # create PR 3 (new iteration with new ancestor)
         response = self.app.post(
             url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr2_id),
+            {
                 'updaterev': 'fb95b340e0d03fa51f33c56c991c08077c99303e',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 'owner': TEST_USER_REGULAR_LOGIN,
                 '_session_csrf_secret_token': self.session_csrf_secret_token(),
              },
              status=302)
         pr3_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
+        pr3_id = int(re.search(r'/pull-request/(\d+)/', response.location).group(1))
         pr2 = PullRequest.get(pr2_id)
         pr3 = PullRequest.get(pr3_id)
         assert pr3_id != pr2_id
         assert pr2.status == PullRequest.STATUS_CLOSED
         assert pr3.org_ref == 'branch:webvcs:fb95b340e0d03fa51f33c56c991c08077c99303e'
         assert pr3.other_ref == 'branch:default:41d2568309a05f422cffb8008e599d385f8af439'
 @pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods
 class TestPullrequestsGetRepoRefs(TestController):
     def setup_method(self, method):
         self.repo_name = u'main'
         repo = fixture.create_repo(self.repo_name, repo_type='hg')
         self.repo_scm_instance = repo.scm_instance
         Session().commit()
         self.c = PullrequestsController()
     def teardown_method(self, method):
         fixture.destroy_repo(u'main')
         Session().commit()
         Session.remove()
     def test_repo_refs_empty_repo(self):
         # empty repo with no commits, no branches, no bookmarks, just one tag
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         assert default == 'tag:null:0000000000000000000000000000000000000000'
     def test_repo_refs_one_commit_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         assert default == 'branch:default:%s' % cs0.raw_id
         assert ([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],
                 'Branches') in refs
     def test_repo_refs_one_commit_rev_hint(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
         expected = 'branch:default:%s' % cs0.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_rev_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
         expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)
         assert default == expected
         assert ([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special') in refs
         assert ([('branch:default:%s' % cs1.raw_id, 'default (current tip)')], 'Branches') in refs
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs1.raw_id)
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_branch_hint(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, branch='default')
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_one_branch_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         # TODO

kallithea/tests/other/test_libs.py

➞

Show inline comments

@@ @@ -293,322 +293,322 @@ class TestLibs(TestController): @@
         assert clone_url == expected
     def _quick_url(self, text, tmpl="""<a class="changeset_hash" href="%s">%s</a>""", url_=None):
         """
         Changes `some text url[foo]` => `some text <a href="/">foo</a>
         :param text:
         """
         import re
         # quickly change expected url[] into a link
         url_pattern = re.compile(r'(?:url\[)(.+?)(?:\])')
         def url_func(match_obj):
             _url = match_obj.groups()[0]
             return tmpl % (url_ or '/repo_name/changeset/%s' % _url, _url)
         return url_pattern.sub(url_func, text)
     @parametrize('sample,expected', [
       ("",
        ""),
       ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
        """git-svn-id: <a href="https://svn.apache.org/repos/asf/libcloud/trunk@1441655">https://svn.apache.org/repos/asf/libcloud/trunk@1441655</a> 13f79535-47bb-0310-9956-ffa450edef68"""),
       ("from rev 000000000000",
        """from rev url[000000000000]"""),
       ("from rev 000000000000123123 also rev 000000000000",
        """from rev url[000000000000123123] also rev url[000000000000]"""),
       ("this should-000 00",
        """this should-000 00"""),
       ("longtextffffffffff rev 123123123123",
        """longtextffffffffff rev url[123123123123]"""),
       ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
        """rev ffffffffffffffffffffffffffffffffffffffffffffffffff"""),
       ("ffffffffffff some text traalaa",
        """url[ffffffffffff] some text traalaa"""),
        ("""Multi line
        123123123123
        some text 123123123123
        sometimes !
        """,
        """Multi line<br/>"""
        """       url[123123123123]<br/>"""
        """       some text url[123123123123]<br/>"""
        """       sometimes !"""),
     ])
     def test_urlify_text(self, sample, expected):
         expected = self._quick_url(expected)
         fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')
         with mock.patch('kallithea.config.routing.url', fake_url):
             from kallithea.lib.helpers import urlify_text
             assert urlify_text(sample, 'repo_name') == expected
     @parametrize('sample,expected,url_', [
       ("",
        "",
        ""),
       ("https://svn.apache.org/repos",
        """url[https://svn.apache.org/repos]""",
        "https://svn.apache.org/repos"),
       ("http://svn.apache.org/repos",
        """url[http://svn.apache.org/repos]""",
        "http://svn.apache.org/repos"),
       ("from rev a also rev http://google.com",
        """from rev a also rev url[http://google.com]""",
        "http://google.com"),
       ("http://imgur.com/foo.gif inline http://imgur.com/foo.gif ending http://imgur.com/foo.gif",
        """url[http://imgur.com/foo.gif] inline url[http://imgur.com/foo.gif] ending url[http://imgur.com/foo.gif]""",
        "http://imgur.com/foo.gif"),
       ("""Multi line
        https://foo.bar.example.com
        some text lalala""",
        """Multi line<br/>"""
        """       url[https://foo.bar.example.com]<br/>"""
        """       some text lalala""",
        "https://foo.bar.example.com"),
       ("@mention @someone",
        """<b>@mention</b> <b>@someone</b>""",
        ""),
       ("deadbeefcafe 123412341234",
        """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a> <a class="changeset_hash" href="/repo_name/changeset/123412341234">123412341234</a>""",
        ""),
       ("We support * markup for *bold* markup of *single or multiple* words, "
        "*a bit @like http://slack.com*. "
        "The first * must come after whitespace and not be followed by whitespace, "
        "contain anything but * and newline until the next *, "
        "which must not come after whitespace "
        "and not be followed by * or alphanumerical *characters*.",
        """We support * markup for <b>*bold*</b> markup of <b>*single or multiple*</b> words, """
        """<b>*a bit <b>@like</b> <a href="http://slack.com">http://slack.com</a>*</b>. """
        """The first * must come after whitespace and not be followed by whitespace, """
        """contain anything but * and newline until the next *, """
        """which must not come after whitespace """
        """and not be followed by * or alphanumerical <b>*characters*</b>.""",
        "-"),
       ("HTML escaping: <abc> 'single' \"double\" &pointer",
        "HTML escaping: &lt;abc&gt; &#39;single&#39; &quot;double&quot; &amp;pointer",
        "-"),
       # tags are covered by test_tag_extractor
     ])
     def test_urlify_test(self, sample, expected, url_):
         expected = self._quick_url(expected,
                                    tmpl="""<a href="%s">%s</a>""", url_=url_)
         fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')
         with mock.patch('kallithea.config.routing.url', fake_url):
             from kallithea.lib.helpers import urlify_text
             assert urlify_text(sample, 'repo_name', stylize=True) == expected
     @parametrize('sample,expected', [
       ("deadbeefcafe @mention, and http://foo.bar/ yo",
        """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a>"""
        """<a class="message-link" href="#the-link"> <b>@mention</b>, and </a>"""
        """<a href="http://foo.bar/">http://foo.bar/</a>"""
        """<a class="message-link" href="#the-link"> yo</a>"""),
     ])
     def test_urlify_link(self, sample, expected):
         fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')
         with mock.patch('kallithea.config.routing.url', fake_url):
             from kallithea.lib.helpers import urlify_text
             assert urlify_text(sample, 'repo_name', link_='#the-link') == expected
     @parametrize('issue_pat,issue_server,issue_sub,sample,expected', [
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue #123 and issue#456',
             """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
         (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue #123 and issue#456',
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
         # to require whitespace before the issue reference, one may be tempted to use \b...
         (r'\bPR(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue PR123 and issuePR456',
             """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issuePR456"""),
         # ... but it turns out that \b does not work well in combination with '#': the expectations
         # are reversed from what is actually happening.
         (r'\b#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue #123 and issue#456',
             """issue #123 and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
         # ... so maybe try to be explicit? Unfortunately the whitespace before the issue
         # reference is not retained, again, because it is part of the pattern.
         (r'(?:^|\s)#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             '#15 and issue #123 and issue#456',
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/15">#15</a> and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue#456"""),
         # ... instead, use lookbehind assertions.
         (r'(?:^|(?<=\s))#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             '#15 and issue #123 and issue#456',
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/15">#15</a> and """
             """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue#456"""),
         (r'(?:pullrequest|pull request|PR|pr) ?#?(\d+)', 'http://foo/{repo}/issue/\\1', 'PR#\\1',
             'fixed with pullrequest #1, pull request#2, PR 3, pr4',
             """fixed with <a class="issue-tracker-link" href="http://foo/repo_name/issue/1">PR#1</a>, """
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/2">PR#2</a>, """
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/3">PR#3</a>, """
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/4">PR#4</a>"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', 'PR\\1',
             'interesting issue #123',
             """interesting issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">PR123</a>"""),
         (r'BUG\d{5}', 'https://bar/{repo}/\\1', '\\1',
             'silly me, I did not parenthesize the id, BUG12345.',
             """silly me, I did not parenthesize the id, <a class="issue-tracker-link" href="https://bar/repo_name/\\1">BUG12345</a>."""),
         (r'BUG(\d{5})', 'https://bar/{repo}/', 'BUG\\1',
             'silly me, the URL does not contain id, BUG12345.',
             """silly me, the URL does not contain id, <a class="issue-tracker-link" href="https://bar/repo_name/">BUG12345</a>."""),
         (r'(PR-\d+)', 'http://foo/{repo}/issue/\\1', '',
             'interesting issue #123, err PR-56',
             """interesting issue #123, err <a class="issue-tracker-link" href="http://foo/repo_name/issue/PR-56">PR-56</a>"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             "some 'standard' text with apostrophes",
             """some &#39;standard&#39; text with apostrophes"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             "some 'standard' issue #123",
             """some &#39;standard&#39; issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'an issue   #123       with extra whitespace',
             """an issue   <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>       with extra whitespace"""),
         (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'an issue   #123       with extra whitespace',
             """an issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>       with extra whitespace"""),
         # invalid issue pattern
         (r'(PR\d+', 'http://foo/{repo}/issue/{id}', '',
             'PR135',
             """PR135"""),
         # other character than #
         (r'(?:^|(?<=\s))\$(\d+)', 'http://foo/{repo}/issue/\\1', '',
             'empty issue_sub $123 and issue$456',
             """empty issue_sub <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">$123</a> and """
             """issue$456"""),
         # named groups
         (r'(PR|pullrequest|pull request) ?(?P<sitecode>BRU|CPH|BER)-(?P<id>\d+)', 'http://foo/\g<sitecode>/pullrequest/\g<id>/', 'PR-\g<sitecode>-\g<id>',
+        (r'(PR|pullrequest|pull request) ?(?P<sitecode>BRU|CPH|BER)-(?P<id>\d+)', r'http://foo/\g<sitecode>/pullrequest/\g<id>/', r'PR-\g<sitecode>-\g<id>',
             'pullrequest CPH-789 is similar to PRBRU-747',
             """<a class="issue-tracker-link" href="http://foo/CPH/pullrequest/789/">PR-CPH-789</a> is similar to """
             """<a class="issue-tracker-link" href="http://foo/BRU/pullrequest/747/">PR-BRU-747</a>"""),
     ])
     def test_urlify_issues(self, issue_pat, issue_server, issue_sub, sample, expected):
         from kallithea.lib.helpers import urlify_text
         config_stub = {
             'sqlalchemy.url': 'foo',
             'issue_pat': issue_pat,
             'issue_server_link': issue_server,
             'issue_sub': issue_sub,
+        }
         # force recreation of lazy function
         with mock.patch('kallithea.lib.helpers._urlify_issues_f', None):
             with mock.patch('kallithea.CONFIG', config_stub):
                 assert urlify_text(sample, 'repo_name') == expected
     @parametrize('sample,expected', [
         ('abc X5', 'abc <a class="issue-tracker-link" href="http://main/repo_name/main/5/">#5</a>'),
         ('abc pullrequest #6 xyz', 'abc <a class="issue-tracker-link" href="http://pr/repo_name/pr/6">PR#6</a> xyz'),
         ('pull request7 #', '<a class="issue-tracker-link" href="http://pr/repo_name/pr/7">PR#7</a> #'),
         ('look PR9 and pr #11', 'look <a class="issue-tracker-link" href="http://pr/repo_name/pr/9">PR#9</a> and <a class="issue-tracker-link" href="http://pr/repo_name/pr/11">PR#11</a>'),
         ('pullrequest#10 solves issue 9', '<a class="issue-tracker-link" href="http://pr/repo_name/pr/10">PR#10</a> solves <a class="issue-tracker-link" href="http://bug/repo_name/bug/9">bug#9</a>'),
         ('issue FAIL67', 'issue <a class="issue-tracker-link" href="http://fail/repo_name/67">FAIL67</a>'),
         ('issue FAILMORE89', 'issue FAILMORE89'), # no match because absent prefix
     ])
     def test_urlify_issues_multiple_issue_patterns(self, sample, expected):
         from kallithea.lib.helpers import urlify_text
         config_stub = {
             'sqlalchemy.url': 'foo',
             'issue_pat': 'X(\d+)',
             'issue_server_link': 'http://main/{repo}/main/\\1/',
             'issue_sub': '#\\1',
             'issue_pat_pr': '(?:pullrequest|pull request|PR|pr) ?#?(\d+)',
             'issue_server_link_pr': 'http://pr/{repo}/pr/\\1',
             'issue_sub_pr': 'PR#\\1',
             'issue_pat_bug': '(?:BUG|bug|issue) ?#?(\d+)',
             'issue_server_link_bug': 'http://bug/{repo}/bug/\\1',
             'issue_sub_bug': 'bug#\\1',
             'issue_pat_empty_prefix': 'FAIL(\d+)',
             'issue_server_link_empty_prefix': 'http://fail/{repo}/\\1',
             'issue_sub_empty_prefix': '',
             'issue_pat_absent_prefix': 'FAILMORE(\d+)',
             'issue_server_link_absent_prefix': 'http://failmore/{repo}/\\1',
             'sqlalchemy.url': r'foo',
             'issue_pat': r'X(\d+)',
             'issue_server_link': r'http://main/{repo}/main/\1/',
             'issue_sub': r'#\1',
             'issue_pat_pr': r'(?:pullrequest|pull request|PR|pr) ?#?(\d+)',
             'issue_server_link_pr': r'http://pr/{repo}/pr/\1',
             'issue_sub_pr': r'PR#\1',
             'issue_pat_bug': r'(?:BUG|bug|issue) ?#?(\d+)',
             'issue_server_link_bug': r'http://bug/{repo}/bug/\1',
             'issue_sub_bug': r'bug#\1',
             'issue_pat_empty_prefix': r'FAIL(\d+)',
             'issue_server_link_empty_prefix': r'http://fail/{repo}/\1',
             'issue_sub_empty_prefix': r'',
             'issue_pat_absent_prefix': r'FAILMORE(\d+)',
             'issue_server_link_absent_prefix': r'http://failmore/{repo}/\1',
+        }
         # force recreation of lazy function
         with mock.patch('kallithea.lib.helpers._urlify_issues_f', None):
             with mock.patch('kallithea.CONFIG', config_stub):
                 assert urlify_text(sample, 'repo_name') == expected
     @parametrize('test,expected', [
       ("", None),
       ("/_2", None),
       ("_2", 2),
       ("_2/", None),
     ])
     def test_get_permanent_id(self, test, expected):
         from kallithea.lib.utils import _get_permanent_id
         extracted = _get_permanent_id(test)
         assert extracted == expected, 'url:%s, got:`%s` expected: `%s`' % (test, _test, expected)
     @parametrize('test,expected', [
       ("", ""),
       ("/", "/"),
       ("/_ID", '/_ID'),
       ("ID", "ID"),
       ("_ID", 'NAME'),
       ("_ID/", 'NAME/'),
       ("_ID/1/2", 'NAME/1/2'),
       ("_IDa", '_IDa'),
     ])
     def test_fix_repo_id_name(self, test, expected):
         repo = Repository.get_by_repo_name(HG_REPO)
         test = test.replace('ID', str(repo.repo_id))
         expected = expected.replace('NAME', repo.repo_name).replace('ID', str(repo.repo_id))
         from kallithea.lib.utils import fix_repo_id_name
         replaced = fix_repo_id_name(test)
         assert replaced == expected, 'url:%s, got:`%s` expected: `%s`' % (test, replaced, expected)
     @parametrize('canonical,test,expected', [
         ('http://www.example.org/', '/abc/xyz', 'http://www.example.org/abc/xyz'),
         ('http://www.example.org', '/abc/xyz', 'http://www.example.org/abc/xyz'),
         ('http://www.example.org', '/abc/xyz/', 'http://www.example.org/abc/xyz/'),
         ('http://www.example.org', 'abc/xyz/', 'http://www.example.org/abc/xyz/'),
         ('http://www.example.org', 'about', 'http://www.example.org/about-page'),
         ('http://www.example.org/repos/', 'abc/xyz/', 'http://www.example.org/repos/abc/xyz/'),
         ('http://www.example.org/kallithea/repos/', 'abc/xyz/', 'http://www.example.org/kallithea/repos/abc/xyz/'),
     ])
     def test_canonical_url(self, canonical, test, expected):
         from kallithea.lib.helpers import canonical_url
         from tg import request
         # setup url(), used by canonical_url
         import routes
         m = routes.Mapper()
         m.connect('about', '/about-page')
         url = routes.URLGenerator(m, {'HTTP_HOST': 'http_host.example.org'})
         config_mock = {
             'canonical_url': canonical,
+        }
         with test_context(self.app):
             request.environ['routes.url'] = url
             with mock.patch('kallithea.CONFIG', config_mock):
                 assert canonical_url(test) == expected
     @parametrize('canonical,expected', [
         ('http://www.example.org', 'www.example.org'),
         ('http://www.example.org/repos/', 'www.example.org'),
         ('http://www.example.org/kallithea/repos/', 'www.example.org'),
     ])
     def test_canonical_hostname(self, canonical, expected):
         from kallithea.lib.helpers import canonical_hostname
         from tg import request
         # setup url(), used by canonical_hostname
         import routes
         m = routes.Mapper()
         url = routes.URLGenerator(m, {'HTTP_HOST': 'http_host.example.org'})
         config_mock = {
             'canonical_url': canonical,
+        }
         with test_context(self.app):
             request.environ['routes.url'] = url
             with mock.patch('kallithea.CONFIG', config_mock):
                 assert canonical_hostname() == expected

kallithea/tests/scripts/manual_test_crawler.py

➞

Show inline comments

 #!/usr/bin/env python2
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.tests.scripts.manual_test_crawler
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Test for crawling a project for memory usage
 This should be runned just as regular script together
 with a watch script that will show memory usage.
 watch -n1 ./kallithea/tests/mem_watch
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import print_function
 import cookielib
 import os
 import sys
 import tempfile
 import time
 import urllib
 import urllib2
 from os.path import dirname
 from kallithea.lib import vcs
 from kallithea.lib.compat import OrderedSet
 from kallithea.lib.vcs.exceptions import RepositoryError
 __here__ = os.path.abspath(__file__)
 __root__ = dirname(dirname(dirname(__here__)))
 sys.path.append(__root__)
 PASES = 3
 HOST = 'http://127.0.0.1'
 PORT = 5000
 BASE_URI = '%s:%s/' % (HOST, PORT)
 if len(sys.argv) == 2:
     BASE_URI = sys.argv[1]
 if not BASE_URI.endswith('/'):
     BASE_URI += '/'
 print('Crawling @ %s' % BASE_URI)
 BASE_URI += '%s'
 PROJECT_PATH = os.path.join('/', 'home', 'username', 'repos')
 PROJECTS = [
     # 'linux-magx-pbranch',
     'CPython',
     'kallithea',
+]
 cj = cookielib.FileCookieJar(os.path.join(tempfile.gettempdir(), 'rc_test_cookie.txt'))
 o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
 o.addheaders = [
     ('User-agent', 'kallithea-crawler'),
     ('Accept-Language', 'en - us, en;q = 0.5')
+]
 urllib2.install_opener(o)
 def _get_repo(proj):
     if isinstance(proj, basestring):
         repo = vcs.get_repo(os.path.join(PROJECT_PATH, proj))
         proj = proj
     else:
         repo = proj
         proj = repo.name
     return repo, proj
 def test_changelog_walk(proj, pages=100):
     repo, proj = _get_repo(proj)
     total_time = 0
     for i in range(1, pages):
         page = '/'.join((proj, 'changelog',))
         full_uri = (BASE_URI % page) + '?' + urllib.urlencode({'page': i})
         s = time.time()
         f = o.open(full_uri)
         assert f.url == full_uri, 'URL:%s does not match %s' % (f.url, full_uri)
         size = len(f.read())
         e = time.time() - s
         total_time += e
         print('visited %s size:%s req:%s ms' % (full_uri, size, e))
     print('total_time', total_time)
     print('average on req', total_time / float(pages))
 def test_changeset_walk(proj, limit=None):
     repo, proj = _get_repo(proj)
     print('processing', os.path.join(PROJECT_PATH, proj))
     total_time = 0
     cnt = 0
     for i in repo:
         cnt += 1
         raw_cs = '/'.join((proj, 'changeset', i.raw_id))
         if limit and limit == cnt:
             break
         full_uri = (BASE_URI % raw_cs)
-        print('%s visiting %s\%s' % (cnt, full_uri, i))
+        print('%s visiting %s/%s' % (cnt, full_uri, i))
         s = time.time()
         f = o.open(full_uri)
         size = len(f.read())
         e = time.time() - s
         total_time += e
-        print('%s visited %s\%s size:%s req:%s ms' % (cnt, full_uri, i, size, e))
+        print('%s visited %s/%s size:%s req:%s ms' % (cnt, full_uri, i, size, e))
     print('total_time', total_time)
     print('average on req', total_time / float(cnt))
 def test_files_walk(proj, limit=100):
     repo, proj = _get_repo(proj)
     print('processing', os.path.join(PROJECT_PATH, proj))
     total_time = 0
     paths_ = OrderedSet([''])
     try:
         tip = repo.get_changeset('tip')
         for topnode, dirs, files in tip.walk('/'):
             for dir in dirs:
                 paths_.add(dir.path)
                 for f in dir:
                     paths_.add(f.path)
             for f in files:
                 paths_.add(f.path)
     except RepositoryError as e:
         pass
     cnt = 0
     for f in paths_:
         cnt += 1
         if limit and limit == cnt:
             break
         file_path = '/'.join((proj, 'files', 'tip', f))
         full_uri = (BASE_URI % file_path)
         print('%s visiting %s' % (cnt, full_uri))
         s = time.time()
         f = o.open(full_uri)
         size = len(f.read())
         e = time.time() - s
         total_time += e
         print('%s visited OK size:%s req:%s ms' % (cnt, size, e))
     print('total_time', total_time)
     print('average on req', total_time / float(cnt))
 if __name__ == '__main__':
     for path in PROJECTS:
         repo = vcs.get_repo(os.path.join(PROJECT_PATH, path))
         for i in range(PASES):
             print('PASS %s/%s' % (i, PASES))
             test_changelog_walk(repo, pages=80)
             test_changeset_walk(repo, limit=100)
             test_files_walk(repo, limit=100)

kallithea/tests/vcs/test_repository.py

➞

Show inline comments

 import datetime
 import pytest
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.tests.vcs import TEST_USER_CONFIG_FILE
 from kallithea.tests.vcs.base import _BackendTestMixin
 class RepositoryBaseTest(_BackendTestMixin):
     recreate_repo_per_test = False
     @classmethod
     def _get_commits(cls):
         return super(RepositoryBaseTest, cls)._get_commits()[:1]
     def test_get_config_value(self):
         assert self.repo.get_config_value('universal', 'foo', TEST_USER_CONFIG_FILE) == 'bar'
     def test_get_config_value_defaults_to_None(self):
         assert self.repo.get_config_value('universal', 'nonexist', TEST_USER_CONFIG_FILE) == None
     def test_get_user_name(self):
         assert self.repo.get_user_name(TEST_USER_CONFIG_FILE) == 'Foo Bar'
     def test_get_user_email(self):
         assert self.repo.get_user_email(TEST_USER_CONFIG_FILE) == 'foo.bar@example.com'
     def test_repo_equality(self):
         assert self.repo == self.repo
     def test_repo_equality_broken_object(self):
         import copy
         _repo = copy.copy(self.repo)
         delattr(_repo, 'path')
         assert self.repo != _repo
     def test_repo_equality_other_object(self):
         class dummy(object):
             path = self.repo.path
         assert self.repo != dummy()
 class TestGitRepositoryBase(RepositoryBaseTest):
     backend_alias = 'git'
 class TestHgRepositoryBase(RepositoryBaseTest):
     backend_alias = 'hg'
 class RepositoryGetDiffTest(_BackendTestMixin):
     @classmethod
     def _get_commits(cls):
         commits = [
+            {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar', content='foobar'),
                     FileNode('foobar2', content='foobar2'),
                 ],
             },
+            {
                 'message': 'Changed foobar, added foobar3',
                 'author': 'Jane Doe <jane.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'added': [
                     FileNode('foobar3', content='foobar3'),
                 ],
                 'changed': [
                     FileNode('foobar', 'FOOBAR'),
                 ],
             },
+            {
                 'message': 'Removed foobar, changed foobar3',
                 'author': 'Jane Doe <jane.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'changed': [
                     FileNode('foobar3', content='FOOBAR\nFOOBAR\nFOOBAR\n'),
                 ],
                 'removed': [FileNode('foobar')],
             },
+            {
                 'message': u'Commit that contains glob pattern in filename',
                 'author': 'Jane Doe <jane.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('README{', content='Strangely-named README file'),
                 ],
             },
+        ]
         return commits
     def test_raise_for_wrong(self):
         with pytest.raises(ChangesetDoesNotExistError):
             self.repo.get_diff('a' * 40, 'b' * 40)
     def test_glob_patterns_in_filename_do_not_raise_exception(self):
         revs = self.repo.revisions
         diff = self.repo.get_diff(revs[2], revs[3], path='README{') # should not raise
 class TestGitRepositoryGetDiff(RepositoryGetDiffTest):
     backend_alias = 'git'
     def test_initial_commit_diff(self):
         initial_rev = self.repo.revisions[0]
         assert self.repo.get_diff(self.repo.EMPTY_CHANGESET, initial_rev) == '''diff --git a/foobar b/foobar
+        assert self.repo.get_diff(self.repo.EMPTY_CHANGESET, initial_rev) == r'''diff --git a/foobar b/foobar
 new file mode 100644
 index 0000000000000000000000000000000000000000..f6ea0495187600e7b2288c8ac19c5886383a4632
 --- /dev/null
 +++ b/foobar
@@ -0,0 +1 @@
 +foobar
 \ No newline at end of file
 diff --git a/foobar2 b/foobar2
 new file mode 100644
 index 0000000000000000000000000000000000000000..e8c9d6b98e3dce993a464935e1a53f50b56a3783
 --- /dev/null
 +++ b/foobar2
@@ -0,0 +1 @@
 +foobar2
 \ No newline at end of file
 '''
     def test_second_changeset_diff(self):
         revs = self.repo.revisions
         assert self.repo.get_diff(revs[0], revs[1]) == '''diff --git a/foobar b/foobar
+        assert self.repo.get_diff(revs[0], revs[1]) == r'''diff --git a/foobar b/foobar
 index f6ea0495187600e7b2288c8ac19c5886383a4632..389865bb681b358c9b102d79abd8d5f941e96551 100644
 --- a/foobar
 +++ b/foobar
@@ -1 +1 @@
 -foobar
 \ No newline at end of file
 +FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 new file mode 100644
 index 0000000000000000000000000000000000000000..c11c37d41d33fb47741cff93fa5f9d798c1535b0
 --- /dev/null
 +++ b/foobar3
@@ -0,0 +1 @@
 +foobar3
 \ No newline at end of file
 '''
     def test_third_changeset_diff(self):
         revs = self.repo.revisions
         assert self.repo.get_diff(revs[1], revs[2]) == '''diff --git a/foobar b/foobar
+        assert self.repo.get_diff(revs[1], revs[2]) == r'''diff --git a/foobar b/foobar
 deleted file mode 100644
 index 389865bb681b358c9b102d79abd8d5f941e96551..0000000000000000000000000000000000000000
 --- a/foobar
 +++ /dev/null
@@ -1 +0,0 @@
 -FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 index c11c37d41d33fb47741cff93fa5f9d798c1535b0..f9324477362684ff692aaf5b9a81e01b9e9a671c 100644
 --- a/foobar3
 +++ b/foobar3
@@ -1 +1,3 @@
 -foobar3
 \ No newline at end of file
 +FOOBAR
 +FOOBAR
 +FOOBAR
 '''
     def test_fourth_changeset_diff(self):
         revs = self.repo.revisions
         assert self.repo.get_diff(revs[2], revs[3]) == '''diff --git a/README{ b/README{
+        assert self.repo.get_diff(revs[2], revs[3]) == r'''diff --git a/README{ b/README{
 new file mode 100644
 index 0000000000000000000000000000000000000000..cdc0c1b5d234feedb37bbac19cd1b6442061102d
 --- /dev/null
 +++ b/README{
@@ -0,0 +1 @@
 +Strangely-named README file
 \ No newline at end of file
 '''
 class TestHgRepositoryGetDiff(RepositoryGetDiffTest):
     backend_alias = 'hg'
     def test_initial_commit_diff(self):
         initial_rev = self.repo.revisions[0]
         assert self.repo.get_diff(self.repo.EMPTY_CHANGESET, initial_rev) == '''diff --git a/foobar b/foobar
+        assert self.repo.get_diff(self.repo.EMPTY_CHANGESET, initial_rev) == r'''diff --git a/foobar b/foobar
 new file mode 100644
 --- /dev/null
 +++ b/foobar
@@ -0,0 +1,1 @@
 +foobar
 \ No newline at end of file
 diff --git a/foobar2 b/foobar2
 new file mode 100644
 --- /dev/null
 +++ b/foobar2
@@ -0,0 +1,1 @@
 +foobar2
 \ No newline at end of file
 '''
     def test_second_changeset_diff(self):
         revs = self.repo.revisions
         assert self.repo.get_diff(revs[0], revs[1]) == '''diff --git a/foobar b/foobar
+        assert self.repo.get_diff(revs[0], revs[1]) == r'''diff --git a/foobar b/foobar
 --- a/foobar
 +++ b/foobar
@@ -1,1 +1,1 @@
 -foobar
 \ No newline at end of file
 +FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 new file mode 100644
 --- /dev/null
 +++ b/foobar3
@@ -0,0 +1,1 @@
 +foobar3
 \ No newline at end of file
 '''
     def test_third_changeset_diff(self):
         revs = self.repo.revisions
         assert self.repo.get_diff(revs[1], revs[2]) == '''diff --git a/foobar b/foobar
+        assert self.repo.get_diff(revs[1], revs[2]) == r'''diff --git a/foobar b/foobar
 deleted file mode 100644
 --- a/foobar
 +++ /dev/null
@@ -1,1 +0,0 @@
 -FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 --- a/foobar3
 +++ b/foobar3
@@ -1,1 +1,3 @@
 -foobar3
 \ No newline at end of file
 +FOOBAR
 +FOOBAR
 +FOOBAR
 '''
     def test_fourth_changeset_diff(self):
         revs = self.repo.revisions
         assert self.repo.get_diff(revs[2], revs[3]) == '''diff --git a/README{ b/README{
+        assert self.repo.get_diff(revs[2], revs[3]) == r'''diff --git a/README{ b/README{
 new file mode 100644
 --- /dev/null
 +++ b/README{
@@ -0,0 +1,1 @@
 +Strangely-named README file
 \ No newline at end of file
 '''

0 comments (0 inline, 0 general)