##// END OF EJS Templates
chore(tests): mark xfail for test that current we don't know why dont work
super-admin -
r5221:13e09fb1 default
parent child Browse files
Show More
@@ -1,1935 +1,1936 b''
1 1 # Copyright (C) 2010-2023 RhodeCode GmbH
2 2 #
3 3 # This program is free software: you can redistribute it and/or modify
4 4 # it under the terms of the GNU Affero General Public License, version 3
5 5 # (only), as published by the Free Software Foundation.
6 6 #
7 7 # This program is distributed in the hope that it will be useful,
8 8 # but WITHOUT ANY WARRANTY; without even the implied warranty of
9 9 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10 10 # GNU General Public License for more details.
11 11 #
12 12 # You should have received a copy of the GNU Affero General Public License
13 13 # along with this program. If not, see <http://www.gnu.org/licenses/>.
14 14 #
15 15 # This program is dual-licensed. If you wish to learn more about the
16 16 # RhodeCode Enterprise Edition, including its added features, Support services,
17 17 # and proprietary license terms, please see https://rhodecode.com/licenses/
18 18 import mock
19 19 import pytest
20 20
21 21 import rhodecode
22 22 from rhodecode.lib import helpers as h
23 23 from rhodecode.lib.vcs.backends.base import MergeResponse, MergeFailureReason, Reference
24 24 from rhodecode.lib.vcs.nodes import FileNode
25 25 from rhodecode.lib.ext_json import json
26 26 from rhodecode.model.changeset_status import ChangesetStatusModel
27 27 from rhodecode.model.db import (
28 28 PullRequest,
29 29 ChangesetStatus,
30 30 UserLog,
31 31 Notification,
32 32 ChangesetComment,
33 33 Repository,
34 34 )
35 35 from rhodecode.model.meta import Session
36 36 from rhodecode.model.pull_request import PullRequestModel
37 37 from rhodecode.model.user import UserModel
38 38 from rhodecode.model.comment import CommentsModel
39 39 from rhodecode.tests import (
40 40 assert_session_flash,
41 41 TEST_USER_ADMIN_LOGIN,
42 42 TEST_USER_REGULAR_LOGIN,
43 43 )
44 44 from rhodecode.tests.fixture_mods.fixture_utils import PRTestUtility
45 45 from rhodecode.tests.routes import route_path
46 46
47 47
48 48 @pytest.mark.usefixtures("app", "autologin_user")
49 49 @pytest.mark.backends("git", "hg")
50 50 class TestPullrequestsView(object):
51 51 def test_index(self, backend):
52 52 self.app.get(route_path("pullrequest_new", repo_name=backend.repo_name))
53 53
54 54 def test_option_menu_create_pull_request_exists(self, backend):
55 55 repo_name = backend.repo_name
56 56 response = self.app.get(h.route_path("repo_summary", repo_name=repo_name))
57 57
58 58 create_pr_link = '<a href="%s">Create Pull Request</a>' % route_path(
59 59 "pullrequest_new", repo_name=repo_name
60 60 )
61 61 response.mustcontain(create_pr_link)
62 62
63 63 def test_create_pr_form_with_raw_commit_id(self, backend):
64 64 repo = backend.repo
65 65
66 66 self.app.get(
67 67 route_path(
68 68 "pullrequest_new",
69 69 repo_name=repo.repo_name,
70 70 commit=repo.get_commit().raw_id,
71 71 ),
72 72 status=200,
73 73 )
74 74
75 75 @pytest.mark.parametrize("pr_merge_enabled", [True, False])
76 76 @pytest.mark.parametrize("range_diff", ["0", "1"])
77 77 def test_show(self, pr_util, pr_merge_enabled, range_diff):
78 78 pull_request = pr_util.create_pull_request(
79 79 mergeable=pr_merge_enabled, enable_notifications=False
80 80 )
81 81
82 82 response = self.app.get(
83 83 route_path(
84 84 "pullrequest_show",
85 85 repo_name=pull_request.target_repo.scm_instance().name,
86 86 pull_request_id=pull_request.pull_request_id,
87 87 params={"range-diff": range_diff},
88 88 )
89 89 )
90 90
91 91 for commit_id in pull_request.revisions:
92 92 response.mustcontain(commit_id)
93 93
94 94 response.mustcontain(pull_request.target_ref_parts.type)
95 95 response.mustcontain(pull_request.target_ref_parts.name)
96 96
97 97 response.mustcontain('class="pull-request-merge"')
98 98
99 99 if pr_merge_enabled:
100 100 response.mustcontain("Pull request reviewer approval is pending")
101 101 else:
102 102 response.mustcontain("Server-side pull request merging is disabled.")
103 103
104 104 if range_diff == "1":
105 105 response.mustcontain("Turn off: Show the diff as commit range")
106 106
107 107 def test_show_versions_of_pr(self, backend, csrf_token):
108 108 commits = [
109 109 {
110 110 "message": "initial-commit",
111 111 "added": [FileNode(b"test-file.txt", b"LINE1\n")],
112 112 },
113 113 {
114 114 "message": "commit-1",
115 115 "changed": [FileNode(b"test-file.txt", b"LINE1\nLINE2\n")],
116 116 },
117 117 # Above is the initial version of PR that changes a single line
118 118 # from now on we'll add 3x commit adding a nother line on each step
119 119 {
120 120 "message": "commit-2",
121 121 "changed": [FileNode(b"test-file.txt", b"LINE1\nLINE2\nLINE3\n")],
122 122 },
123 123 {
124 124 "message": "commit-3",
125 125 "changed": [
126 126 FileNode(b"test-file.txt", b"LINE1\nLINE2\nLINE3\nLINE4\n")
127 127 ],
128 128 },
129 129 {
130 130 "message": "commit-4",
131 131 "changed": [
132 132 FileNode(b"test-file.txt", b"LINE1\nLINE2\nLINE3\nLINE4\nLINE5\n")
133 133 ],
134 134 },
135 135 ]
136 136
137 137 commit_ids = backend.create_master_repo(commits)
138 138 target = backend.create_repo(heads=["initial-commit"])
139 139 source = backend.create_repo(heads=["commit-1"])
140 140 source_repo_name = source.repo_name
141 141 target_repo_name = target.repo_name
142 142
143 143 target_ref = "branch:{branch}:{commit_id}".format(
144 144 branch=backend.default_branch_name, commit_id=commit_ids["initial-commit"]
145 145 )
146 146 source_ref = "branch:{branch}:{commit_id}".format(
147 147 branch=backend.default_branch_name, commit_id=commit_ids["commit-1"]
148 148 )
149 149
150 150 response = self.app.post(
151 151 route_path("pullrequest_create", repo_name=source.repo_name),
152 152 [
153 153 ("source_repo", source_repo_name),
154 154 ("source_ref", source_ref),
155 155 ("target_repo", target_repo_name),
156 156 ("target_ref", target_ref),
157 157 ("common_ancestor", commit_ids["initial-commit"]),
158 158 ("pullrequest_title", "Title"),
159 159 ("pullrequest_desc", "Description"),
160 160 ("description_renderer", "markdown"),
161 161 ("__start__", "review_members:sequence"),
162 162 ("__start__", "reviewer:mapping"),
163 163 ("user_id", "1"),
164 164 ("__start__", "reasons:sequence"),
165 165 ("reason", "Some reason"),
166 166 ("__end__", "reasons:sequence"),
167 167 ("__start__", "rules:sequence"),
168 168 ("__end__", "rules:sequence"),
169 169 ("mandatory", "False"),
170 170 ("__end__", "reviewer:mapping"),
171 171 ("__end__", "review_members:sequence"),
172 172 ("__start__", "revisions:sequence"),
173 173 ("revisions", commit_ids["commit-1"]),
174 174 ("__end__", "revisions:sequence"),
175 175 ("user", ""),
176 176 ("csrf_token", csrf_token),
177 177 ],
178 178 status=302,
179 179 )
180 180
181 181 location = response.headers["Location"]
182 182
183 183 pull_request_id = location.rsplit("/", 1)[1]
184 184 assert pull_request_id != "new"
185 185 pull_request = PullRequest.get(int(pull_request_id))
186 186
187 187 pull_request_id = pull_request.pull_request_id
188 188
189 189 # Show initial version of PR
190 190 response = self.app.get(
191 191 route_path(
192 192 "pullrequest_show",
193 193 repo_name=target_repo_name,
194 194 pull_request_id=pull_request_id,
195 195 )
196 196 )
197 197
198 198 response.mustcontain("commit-1")
199 199 response.mustcontain(no=["commit-2"])
200 200 response.mustcontain(no=["commit-3"])
201 201 response.mustcontain(no=["commit-4"])
202 202
203 203 response.mustcontain('cb-addition"></span><span>LINE2</span>')
204 204 response.mustcontain(no=["LINE3"])
205 205 response.mustcontain(no=["LINE4"])
206 206 response.mustcontain(no=["LINE5"])
207 207
208 208 # update PR #1
209 209 source_repo = Repository.get_by_repo_name(source_repo_name)
210 210 backend.pull_heads(source_repo, heads=["commit-2"])
211 211 response = self.app.post(
212 212 route_path(
213 213 "pullrequest_update",
214 214 repo_name=target_repo_name,
215 215 pull_request_id=pull_request_id,
216 216 ),
217 217 params={"update_commits": "true", "csrf_token": csrf_token},
218 218 )
219 219
220 220 # update PR #2
221 221 source_repo = Repository.get_by_repo_name(source_repo_name)
222 222 backend.pull_heads(source_repo, heads=["commit-3"])
223 223 response = self.app.post(
224 224 route_path(
225 225 "pullrequest_update",
226 226 repo_name=target_repo_name,
227 227 pull_request_id=pull_request_id,
228 228 ),
229 229 params={"update_commits": "true", "csrf_token": csrf_token},
230 230 )
231 231
232 232 # update PR #3
233 233 source_repo = Repository.get_by_repo_name(source_repo_name)
234 234 backend.pull_heads(source_repo, heads=["commit-4"])
235 235 response = self.app.post(
236 236 route_path(
237 237 "pullrequest_update",
238 238 repo_name=target_repo_name,
239 239 pull_request_id=pull_request_id,
240 240 ),
241 241 params={"update_commits": "true", "csrf_token": csrf_token},
242 242 )
243 243
244 244 # Show final version !
245 245 response = self.app.get(
246 246 route_path(
247 247 "pullrequest_show",
248 248 repo_name=target_repo_name,
249 249 pull_request_id=pull_request_id,
250 250 )
251 251 )
252 252
253 253 # 3 updates, and the latest == 4
254 254 response.mustcontain("4 versions available for this pull request")
255 255 response.mustcontain(no=["rhodecode diff rendering error"])
256 256
257 257 # initial show must have 3 commits, and 3 adds
258 258 response.mustcontain("commit-1")
259 259 response.mustcontain("commit-2")
260 260 response.mustcontain("commit-3")
261 261 response.mustcontain("commit-4")
262 262
263 263 response.mustcontain('cb-addition"></span><span>LINE2</span>')
264 264 response.mustcontain('cb-addition"></span><span>LINE3</span>')
265 265 response.mustcontain('cb-addition"></span><span>LINE4</span>')
266 266 response.mustcontain('cb-addition"></span><span>LINE5</span>')
267 267
268 268 # fetch versions
269 269 pr = PullRequest.get(pull_request_id)
270 270 versions = [x.pull_request_version_id for x in pr.versions.all()]
271 271 assert len(versions) == 3
272 272
273 273 # show v1,v2,v3,v4
274 274 def cb_line(text):
275 275 return 'cb-addition"></span><span>{}</span>'.format(text)
276 276
277 277 def cb_context(text):
278 278 return (
279 279 '<span class="cb-code"><span class="cb-action cb-context">'
280 280 "</span><span>{}</span></span>".format(text)
281 281 )
282 282
283 283 commit_tests = {
284 284 # in response, not in response
285 285 1: (["commit-1"], ["commit-2", "commit-3", "commit-4"]),
286 286 2: (["commit-1", "commit-2"], ["commit-3", "commit-4"]),
287 287 3: (["commit-1", "commit-2", "commit-3"], ["commit-4"]),
288 288 4: (["commit-1", "commit-2", "commit-3", "commit-4"], []),
289 289 }
290 290 diff_tests = {
291 291 1: (["LINE2"], ["LINE3", "LINE4", "LINE5"]),
292 292 2: (["LINE2", "LINE3"], ["LINE4", "LINE5"]),
293 293 3: (["LINE2", "LINE3", "LINE4"], ["LINE5"]),
294 294 4: (["LINE2", "LINE3", "LINE4", "LINE5"], []),
295 295 }
296 296 for idx, ver in enumerate(versions, 1):
297 297 response = self.app.get(
298 298 route_path(
299 299 "pullrequest_show",
300 300 repo_name=target_repo_name,
301 301 pull_request_id=pull_request_id,
302 302 params={"version": ver},
303 303 )
304 304 )
305 305
306 306 response.mustcontain(no=["rhodecode diff rendering error"])
307 307 response.mustcontain("Showing changes at v{}".format(idx))
308 308
309 309 yes, no = commit_tests[idx]
310 310 for y in yes:
311 311 response.mustcontain(y)
312 312 for n in no:
313 313 response.mustcontain(no=n)
314 314
315 315 yes, no = diff_tests[idx]
316 316 for y in yes:
317 317 response.mustcontain(cb_line(y))
318 318 for n in no:
319 319 response.mustcontain(no=n)
320 320
321 321 # show diff between versions
322 322 diff_compare_tests = {
323 323 1: (["LINE3"], ["LINE1", "LINE2"]),
324 324 2: (["LINE3", "LINE4"], ["LINE1", "LINE2"]),
325 325 3: (["LINE3", "LINE4", "LINE5"], ["LINE1", "LINE2"]),
326 326 }
327 327 for idx, ver in enumerate(versions, 1):
328 328 adds, context = diff_compare_tests[idx]
329 329
330 330 to_ver = ver + 1
331 331 if idx == 3:
332 332 to_ver = "latest"
333 333
334 334 response = self.app.get(
335 335 route_path(
336 336 "pullrequest_show",
337 337 repo_name=target_repo_name,
338 338 pull_request_id=pull_request_id,
339 339 params={"from_version": versions[0], "version": to_ver},
340 340 )
341 341 )
342 342
343 343 response.mustcontain(no=["rhodecode diff rendering error"])
344 344
345 345 for a in adds:
346 346 response.mustcontain(cb_line(a))
347 347 for c in context:
348 348 response.mustcontain(cb_context(c))
349 349
350 350 # test version v2 -> v3
351 351 response = self.app.get(
352 352 route_path(
353 353 "pullrequest_show",
354 354 repo_name=target_repo_name,
355 355 pull_request_id=pull_request_id,
356 356 params={"from_version": versions[1], "version": versions[2]},
357 357 )
358 358 )
359 359
360 360 response.mustcontain(cb_context("LINE1"))
361 361 response.mustcontain(cb_context("LINE2"))
362 362 response.mustcontain(cb_context("LINE3"))
363 363 response.mustcontain(cb_line("LINE4"))
364 364
365 365 def test_close_status_visibility(self, pr_util, user_util, csrf_token):
366 366 # Logout
367 367 response = self.app.post(
368 368 h.route_path("logout"), params={"csrf_token": csrf_token}
369 369 )
370 370 # Login as regular user
371 371 response = self.app.post(
372 372 h.route_path("login"),
373 373 {"username": TEST_USER_REGULAR_LOGIN, "password": "test12"},
374 374 )
375 375
376 376 pull_request = pr_util.create_pull_request(author=TEST_USER_REGULAR_LOGIN)
377 377
378 378 response = self.app.get(
379 379 route_path(
380 380 "pullrequest_show",
381 381 repo_name=pull_request.target_repo.scm_instance().name,
382 382 pull_request_id=pull_request.pull_request_id,
383 383 )
384 384 )
385 385
386 386 response.mustcontain("Server-side pull request merging is disabled.")
387 387
388 388 assert_response = response.assert_response()
389 389 # for regular user without a merge permissions, we don't see it
390 390 assert_response.no_element_exists("#close-pull-request-action")
391 391
392 392 user_util.grant_user_permission_to_repo(
393 393 pull_request.target_repo,
394 394 UserModel().get_by_username(TEST_USER_REGULAR_LOGIN),
395 395 "repository.write",
396 396 )
397 397 response = self.app.get(
398 398 route_path(
399 399 "pullrequest_show",
400 400 repo_name=pull_request.target_repo.scm_instance().name,
401 401 pull_request_id=pull_request.pull_request_id,
402 402 )
403 403 )
404 404
405 405 response.mustcontain("Server-side pull request merging is disabled.")
406 406
407 407 assert_response = response.assert_response()
408 408 # now regular user has a merge permissions, we have CLOSE button
409 409 assert_response.one_element_exists("#close-pull-request-action")
410 410
411 411 def test_show_invalid_commit_id(self, pr_util):
412 412 # Simulating invalid revisions which will cause a lookup error
413 413 pull_request = pr_util.create_pull_request()
414 414 pull_request.revisions = ["invalid"]
415 415 Session().add(pull_request)
416 416 Session().commit()
417 417
418 418 response = self.app.get(
419 419 route_path(
420 420 "pullrequest_show",
421 421 repo_name=pull_request.target_repo.scm_instance().name,
422 422 pull_request_id=pull_request.pull_request_id,
423 423 )
424 424 )
425 425
426 426 for commit_id in pull_request.revisions:
427 427 response.mustcontain(commit_id)
428 428
429 429 def test_show_invalid_source_reference(self, pr_util):
430 430 pull_request = pr_util.create_pull_request()
431 431 pull_request.source_ref = "branch:b:invalid"
432 432 Session().add(pull_request)
433 433 Session().commit()
434 434
435 435 self.app.get(
436 436 route_path(
437 437 "pullrequest_show",
438 438 repo_name=pull_request.target_repo.scm_instance().name,
439 439 pull_request_id=pull_request.pull_request_id,
440 440 )
441 441 )
442 442
443 443 def test_edit_title_description(self, pr_util, csrf_token):
444 444 pull_request = pr_util.create_pull_request()
445 445 pull_request_id = pull_request.pull_request_id
446 446
447 447 response = self.app.post(
448 448 route_path(
449 449 "pullrequest_update",
450 450 repo_name=pull_request.target_repo.repo_name,
451 451 pull_request_id=pull_request_id,
452 452 ),
453 453 params={
454 454 "edit_pull_request": "true",
455 455 "title": "New title",
456 456 "description": "New description",
457 457 "csrf_token": csrf_token,
458 458 },
459 459 )
460 460
461 461 assert_session_flash(
462 462 response, "Pull request title & description updated.", category="success"
463 463 )
464 464
465 465 pull_request = PullRequest.get(pull_request_id)
466 466 assert pull_request.title == "New title"
467 467 assert pull_request.description == "New description"
468 468
469 469 def test_edit_title_description_special(self, pr_util, csrf_token):
470 470 pull_request = pr_util.create_pull_request()
471 471 pull_request_id = pull_request.pull_request_id
472 472
473 473 response = self.app.post(
474 474 route_path(
475 475 "pullrequest_update",
476 476 repo_name=pull_request.target_repo.repo_name,
477 477 pull_request_id=pull_request_id,
478 478 ),
479 479 params={
480 480 "edit_pull_request": "true",
481 481 "title": "New title {} {2} {foo}",
482 482 "description": "New description",
483 483 "csrf_token": csrf_token,
484 484 },
485 485 )
486 486
487 487 assert_session_flash(
488 488 response, "Pull request title & description updated.", category="success"
489 489 )
490 490
491 491 pull_request = PullRequest.get(pull_request_id)
492 492 assert pull_request.title_safe == "New title {{}} {{2}} {{foo}}"
493 493
494 494 def test_edit_title_description_closed(self, pr_util, csrf_token):
495 495 pull_request = pr_util.create_pull_request()
496 496 pull_request_id = pull_request.pull_request_id
497 497 repo_name = pull_request.target_repo.repo_name
498 498 pr_util.close()
499 499
500 500 response = self.app.post(
501 501 route_path(
502 502 "pullrequest_update",
503 503 repo_name=repo_name,
504 504 pull_request_id=pull_request_id,
505 505 ),
506 506 params={
507 507 "edit_pull_request": "true",
508 508 "title": "New title",
509 509 "description": "New description",
510 510 "csrf_token": csrf_token,
511 511 },
512 512 status=200,
513 513 )
514 514 assert_session_flash(
515 515 response, "Cannot update closed pull requests.", category="error"
516 516 )
517 517
518 518 def test_update_invalid_source_reference(self, pr_util, csrf_token):
519 519 from rhodecode.lib.vcs.backends.base import UpdateFailureReason
520 520
521 521 pull_request = pr_util.create_pull_request()
522 522 pull_request.source_ref = "branch:invalid-branch:invalid-commit-id"
523 523 Session().add(pull_request)
524 524 Session().commit()
525 525
526 526 pull_request_id = pull_request.pull_request_id
527 527
528 528 response = self.app.post(
529 529 route_path(
530 530 "pullrequest_update",
531 531 repo_name=pull_request.target_repo.repo_name,
532 532 pull_request_id=pull_request_id,
533 533 ),
534 534 params={"update_commits": "true", "csrf_token": csrf_token},
535 535 )
536 536
537 537 expected_msg = str(
538 538 PullRequestModel.UPDATE_STATUS_MESSAGES[
539 539 UpdateFailureReason.MISSING_SOURCE_REF
540 540 ]
541 541 )
542 542 assert_session_flash(response, expected_msg, category="error")
543 543
544 544 def test_missing_target_reference(self, pr_util, csrf_token):
545 545 from rhodecode.lib.vcs.backends.base import MergeFailureReason
546 546
547 547 pull_request = pr_util.create_pull_request(approved=True, mergeable=True)
548 548 unicode_reference = "branch:invalid-branch:invalid-commit-id"
549 549 pull_request.target_ref = unicode_reference
550 550 Session().add(pull_request)
551 551 Session().commit()
552 552
553 553 pull_request_id = pull_request.pull_request_id
554 554 pull_request_url = route_path(
555 555 "pullrequest_show",
556 556 repo_name=pull_request.target_repo.repo_name,
557 557 pull_request_id=pull_request_id,
558 558 )
559 559
560 560 response = self.app.get(pull_request_url)
561 561 # target_ref_id = "invalid-branch"
562 562
563 563 merge_resp = MergeResponse(
564 564 True,
565 565 True,
566 566 Reference("commit", "STUB_COMMIT_ID", "STUB_COMMIT_ID"),
567 567 MergeFailureReason.MISSING_TARGET_REF,
568 568 metadata={
569 569 "target_ref": PullRequest.unicode_to_reference(unicode_reference)
570 570 },
571 571 )
572 572 response.assert_response().element_contains(
573 573 'div[data-role="merge-message"]', merge_resp.merge_status_message
574 574 )
575 575
576 576 def test_comment_and_close_pull_request_custom_message_approved(
577 577 self, pr_util, csrf_token, xhr_header
578 578 ):
579 579 pull_request = pr_util.create_pull_request(approved=True)
580 580 pull_request_id = pull_request.pull_request_id
581 581 author = pull_request.user_id
582 582 repo = pull_request.target_repo.repo_id
583 583
584 584 self.app.post(
585 585 route_path(
586 586 "pullrequest_comment_create",
587 587 repo_name=pull_request.target_repo.scm_instance().name,
588 588 pull_request_id=pull_request_id,
589 589 ),
590 590 params={
591 591 "close_pull_request": "1",
592 592 "text": "Closing a PR",
593 593 "csrf_token": csrf_token,
594 594 },
595 595 extra_environ=xhr_header,
596 596 )
597 597
598 598 journal = (
599 599 UserLog.query()
600 600 .filter(UserLog.user_id == author)
601 601 .filter(UserLog.repository_id == repo)
602 602 .order_by(UserLog.user_log_id.asc())
603 603 .all()
604 604 )
605 605 assert journal[-1].action == "repo.pull_request.close"
606 606
607 607 pull_request = PullRequest.get(pull_request_id)
608 608 assert pull_request.is_closed()
609 609
610 610 status = ChangesetStatusModel().get_status(
611 611 pull_request.source_repo, pull_request=pull_request
612 612 )
613 613 assert status == ChangesetStatus.STATUS_APPROVED
614 614 comments = (
615 615 ChangesetComment()
616 616 .query()
617 617 .filter(ChangesetComment.pull_request == pull_request)
618 618 .order_by(ChangesetComment.comment_id.asc())
619 619 .all()
620 620 )
621 621 assert comments[-1].text == "Closing a PR"
622 622
623 623 def test_comment_force_close_pull_request_rejected(
624 624 self, pr_util, csrf_token, xhr_header
625 625 ):
626 626 pull_request = pr_util.create_pull_request()
627 627 pull_request_id = pull_request.pull_request_id
628 628 PullRequestModel().update_reviewers(
629 629 pull_request_id,
630 630 [
631 631 (1, ["reason"], False, "reviewer", []),
632 632 (2, ["reason2"], False, "reviewer", []),
633 633 ],
634 634 pull_request.author,
635 635 )
636 636 author = pull_request.user_id
637 637 repo = pull_request.target_repo.repo_id
638 638
639 639 self.app.post(
640 640 route_path(
641 641 "pullrequest_comment_create",
642 642 repo_name=pull_request.target_repo.scm_instance().name,
643 643 pull_request_id=pull_request_id,
644 644 ),
645 645 params={"close_pull_request": "1", "csrf_token": csrf_token},
646 646 extra_environ=xhr_header,
647 647 )
648 648
649 649 pull_request = PullRequest.get(pull_request_id)
650 650
651 651 journal = (
652 652 UserLog.query()
653 653 .filter(UserLog.user_id == author, UserLog.repository_id == repo)
654 654 .order_by(UserLog.user_log_id.asc())
655 655 .all()
656 656 )
657 657 assert journal[-1].action == "repo.pull_request.close"
658 658
659 659 # check only the latest status, not the review status
660 660 status = ChangesetStatusModel().get_status(
661 661 pull_request.source_repo, pull_request=pull_request
662 662 )
663 663 assert status == ChangesetStatus.STATUS_REJECTED
664 664
665 665 def test_comment_and_close_pull_request(self, pr_util, csrf_token, xhr_header):
666 666 pull_request = pr_util.create_pull_request()
667 667 pull_request_id = pull_request.pull_request_id
668 668
669 669 response = self.app.post(
670 670 route_path(
671 671 "pullrequest_comment_create",
672 672 repo_name=pull_request.target_repo.scm_instance().name,
673 673 pull_request_id=pull_request.pull_request_id,
674 674 ),
675 675 params={"close_pull_request": "true", "csrf_token": csrf_token},
676 676 extra_environ=xhr_header,
677 677 )
678 678
679 679 assert response.json
680 680
681 681 pull_request = PullRequest.get(pull_request_id)
682 682 assert pull_request.is_closed()
683 683
684 684 # check only the latest status, not the review status
685 685 status = ChangesetStatusModel().get_status(
686 686 pull_request.source_repo, pull_request=pull_request
687 687 )
688 688 assert status == ChangesetStatus.STATUS_REJECTED
689 689
690 690 def test_comment_and_close_pull_request_try_edit_comment(
691 691 self, pr_util, csrf_token, xhr_header
692 692 ):
693 693 pull_request = pr_util.create_pull_request()
694 694 pull_request_id = pull_request.pull_request_id
695 695 target_scm = pull_request.target_repo.scm_instance()
696 696 target_scm_name = target_scm.name
697 697
698 698 response = self.app.post(
699 699 route_path(
700 700 "pullrequest_comment_create",
701 701 repo_name=target_scm_name,
702 702 pull_request_id=pull_request_id,
703 703 ),
704 704 params={
705 705 "close_pull_request": "true",
706 706 "csrf_token": csrf_token,
707 707 },
708 708 extra_environ=xhr_header,
709 709 )
710 710
711 711 assert response.json
712 712
713 713 pull_request = PullRequest.get(pull_request_id)
714 714 target_scm = pull_request.target_repo.scm_instance()
715 715 target_scm_name = target_scm.name
716 716 assert pull_request.is_closed()
717 717
718 718 # check only the latest status, not the review status
719 719 status = ChangesetStatusModel().get_status(
720 720 pull_request.source_repo, pull_request=pull_request
721 721 )
722 722 assert status == ChangesetStatus.STATUS_REJECTED
723 723
724 724 for comment_id in response.json.keys():
725 725 test_text = "test"
726 726 response = self.app.post(
727 727 route_path(
728 728 "pullrequest_comment_edit",
729 729 repo_name=target_scm_name,
730 730 pull_request_id=pull_request_id,
731 731 comment_id=comment_id,
732 732 ),
733 733 extra_environ=xhr_header,
734 734 params={
735 735 "csrf_token": csrf_token,
736 736 "text": test_text,
737 737 },
738 738 status=403,
739 739 )
740 740 assert response.status_int == 403
741 741
742 742 def test_comment_and_comment_edit(self, pr_util, csrf_token, xhr_header):
743 743 pull_request = pr_util.create_pull_request()
744 744 target_scm = pull_request.target_repo.scm_instance()
745 745 target_scm_name = target_scm.name
746 746
747 747 response = self.app.post(
748 748 route_path(
749 749 "pullrequest_comment_create",
750 750 repo_name=target_scm_name,
751 751 pull_request_id=pull_request.pull_request_id,
752 752 ),
753 753 params={
754 754 "csrf_token": csrf_token,
755 755 "text": "init",
756 756 },
757 757 extra_environ=xhr_header,
758 758 )
759 759 assert response.json
760 760
761 761 for comment_id in response.json.keys():
762 762 assert comment_id
763 763 test_text = "test"
764 764 self.app.post(
765 765 route_path(
766 766 "pullrequest_comment_edit",
767 767 repo_name=target_scm_name,
768 768 pull_request_id=pull_request.pull_request_id,
769 769 comment_id=comment_id,
770 770 ),
771 771 extra_environ=xhr_header,
772 772 params={
773 773 "csrf_token": csrf_token,
774 774 "text": test_text,
775 775 "version": "0",
776 776 },
777 777 )
778 778 text_form_db = (
779 779 ChangesetComment.query()
780 780 .filter(ChangesetComment.comment_id == comment_id)
781 781 .first()
782 782 .text
783 783 )
784 784 assert test_text == text_form_db
785 785
786 786 def test_comment_and_comment_edit_special(self, pr_util, csrf_token, xhr_header):
787 787 pull_request = pr_util.create_pull_request()
788 788 target_scm = pull_request.target_repo.scm_instance()
789 789 target_scm_name = target_scm.name
790 790
791 791 response = self.app.post(
792 792 route_path(
793 793 "pullrequest_comment_create",
794 794 repo_name=target_scm_name,
795 795 pull_request_id=pull_request.pull_request_id,
796 796 ),
797 797 params={
798 798 "csrf_token": csrf_token,
799 799 "text": "init",
800 800 },
801 801 extra_environ=xhr_header,
802 802 )
803 803 assert response.json
804 804
805 805 for comment_id in response.json.keys():
806 806 test_text = "init"
807 807 response = self.app.post(
808 808 route_path(
809 809 "pullrequest_comment_edit",
810 810 repo_name=target_scm_name,
811 811 pull_request_id=pull_request.pull_request_id,
812 812 comment_id=comment_id,
813 813 ),
814 814 extra_environ=xhr_header,
815 815 params={
816 816 "csrf_token": csrf_token,
817 817 "text": test_text,
818 818 "version": "0",
819 819 },
820 820 status=404,
821 821 )
822 822 assert response.status_int == 404
823 823
824 824 def test_comment_and_try_edit_already_edited(self, pr_util, csrf_token, xhr_header):
825 825 pull_request = pr_util.create_pull_request()
826 826 target_scm = pull_request.target_repo.scm_instance()
827 827 target_scm_name = target_scm.name
828 828
829 829 response = self.app.post(
830 830 route_path(
831 831 "pullrequest_comment_create",
832 832 repo_name=target_scm_name,
833 833 pull_request_id=pull_request.pull_request_id,
834 834 ),
835 835 params={
836 836 "csrf_token": csrf_token,
837 837 "text": "init",
838 838 },
839 839 extra_environ=xhr_header,
840 840 )
841 841 assert response.json
842 842 for comment_id in response.json.keys():
843 843 test_text = "test"
844 844 self.app.post(
845 845 route_path(
846 846 "pullrequest_comment_edit",
847 847 repo_name=target_scm_name,
848 848 pull_request_id=pull_request.pull_request_id,
849 849 comment_id=comment_id,
850 850 ),
851 851 extra_environ=xhr_header,
852 852 params={
853 853 "csrf_token": csrf_token,
854 854 "text": test_text,
855 855 "version": "0",
856 856 },
857 857 )
858 858 test_text_v2 = "test_v2"
859 859 response = self.app.post(
860 860 route_path(
861 861 "pullrequest_comment_edit",
862 862 repo_name=target_scm_name,
863 863 pull_request_id=pull_request.pull_request_id,
864 864 comment_id=comment_id,
865 865 ),
866 866 extra_environ=xhr_header,
867 867 params={
868 868 "csrf_token": csrf_token,
869 869 "text": test_text_v2,
870 870 "version": "0",
871 871 },
872 872 status=409,
873 873 )
874 874 assert response.status_int == 409
875 875
876 876 text_form_db = (
877 877 ChangesetComment.query()
878 878 .filter(ChangesetComment.comment_id == comment_id)
879 879 .first()
880 880 .text
881 881 )
882 882
883 883 assert test_text == text_form_db
884 884 assert test_text_v2 != text_form_db
885 885
886 886 def test_comment_and_comment_edit_permissions_forbidden(
887 887 self,
888 888 autologin_regular_user,
889 889 user_regular,
890 890 user_admin,
891 891 pr_util,
892 892 csrf_token,
893 893 xhr_header,
894 894 ):
895 895 pull_request = pr_util.create_pull_request(
896 896 author=user_admin.username, enable_notifications=False
897 897 )
898 898 comment = CommentsModel().create(
899 899 text="test",
900 900 repo=pull_request.target_repo.scm_instance().name,
901 901 user=user_admin,
902 902 pull_request=pull_request,
903 903 )
904 904 response = self.app.post(
905 905 route_path(
906 906 "pullrequest_comment_edit",
907 907 repo_name=pull_request.target_repo.scm_instance().name,
908 908 pull_request_id=pull_request.pull_request_id,
909 909 comment_id=comment.comment_id,
910 910 ),
911 911 extra_environ=xhr_header,
912 912 params={
913 913 "csrf_token": csrf_token,
914 914 "text": "test_text",
915 915 },
916 916 status=403,
917 917 )
918 918 assert response.status_int == 403
919 919
920 920 def test_create_pull_request(self, backend, csrf_token):
921 921 commits = [
922 922 {"message": "ancestor"},
923 923 {"message": "change"},
924 924 {"message": "change2"},
925 925 ]
926 926 commit_ids = backend.create_master_repo(commits)
927 927 target = backend.create_repo(heads=["ancestor"])
928 928 source = backend.create_repo(heads=["change2"])
929 929
930 930 response = self.app.post(
931 931 route_path("pullrequest_create", repo_name=source.repo_name),
932 932 [
933 933 ("source_repo", source.repo_name),
934 934 ("source_ref", "branch:default:" + commit_ids["change2"]),
935 935 ("target_repo", target.repo_name),
936 936 ("target_ref", "branch:default:" + commit_ids["ancestor"]),
937 937 ("common_ancestor", commit_ids["ancestor"]),
938 938 ("pullrequest_title", "Title"),
939 939 ("pullrequest_desc", "Description"),
940 940 ("description_renderer", "markdown"),
941 941 ("__start__", "review_members:sequence"),
942 942 ("__start__", "reviewer:mapping"),
943 943 ("user_id", "1"),
944 944 ("__start__", "reasons:sequence"),
945 945 ("reason", "Some reason"),
946 946 ("__end__", "reasons:sequence"),
947 947 ("__start__", "rules:sequence"),
948 948 ("__end__", "rules:sequence"),
949 949 ("mandatory", "False"),
950 950 ("__end__", "reviewer:mapping"),
951 951 ("__end__", "review_members:sequence"),
952 952 ("__start__", "revisions:sequence"),
953 953 ("revisions", commit_ids["change"]),
954 954 ("revisions", commit_ids["change2"]),
955 955 ("__end__", "revisions:sequence"),
956 956 ("user", ""),
957 957 ("csrf_token", csrf_token),
958 958 ],
959 959 status=302,
960 960 )
961 961
962 962 location = response.headers["Location"]
963 963 pull_request_id = location.rsplit("/", 1)[1]
964 964 assert pull_request_id != "new"
965 965 pull_request = PullRequest.get(int(pull_request_id))
966 966
967 967 # check that we have now both revisions
968 968 assert pull_request.revisions == [commit_ids["change2"], commit_ids["change"]]
969 969 assert pull_request.source_ref == "branch:default:" + commit_ids["change2"]
970 970 expected_target_ref = "branch:default:" + commit_ids["ancestor"]
971 971 assert pull_request.target_ref == expected_target_ref
972 972
973 973 def test_reviewer_notifications(self, backend, csrf_token):
974 974 # We have to use the app.post for this test, so it will create the
975 975 # notifications properly with the new PR
976 976 commits = [
977 977 {
978 978 "message": "ancestor",
979 979 "added": [FileNode(b"file_A", content=b"content_of_ancestor")],
980 980 },
981 981 {
982 982 "message": "change",
983 983 "added": [FileNode(b"file_a", content=b"content_of_change")],
984 984 },
985 985 {"message": "change-child"},
986 986 {
987 987 "message": "ancestor-child",
988 988 "parents": ["ancestor"],
989 989 "branch": "feature",
990 990 "added": [FileNode(b"file_c", content=b"content_of_ancestor_child")],
991 991 },
992 992 {"message": "ancestor-child-2", "branch": "feature"},
993 993 ]
994 994 commit_ids = backend.create_master_repo(commits)
995 995 target = backend.create_repo(heads=["ancestor-child"])
996 996 source = backend.create_repo(heads=["change"])
997 997
998 998 response = self.app.post(
999 999 route_path("pullrequest_create", repo_name=source.repo_name),
1000 1000 [
1001 1001 ("source_repo", source.repo_name),
1002 1002 ("source_ref", "branch:default:" + commit_ids["change"]),
1003 1003 ("target_repo", target.repo_name),
1004 1004 ("target_ref", "branch:default:" + commit_ids["ancestor-child"]),
1005 1005 ("common_ancestor", commit_ids["ancestor"]),
1006 1006 ("pullrequest_title", "Title"),
1007 1007 ("pullrequest_desc", "Description"),
1008 1008 ("description_renderer", "markdown"),
1009 1009 ("__start__", "review_members:sequence"),
1010 1010 ("__start__", "reviewer:mapping"),
1011 1011 ("user_id", "2"),
1012 1012 ("__start__", "reasons:sequence"),
1013 1013 ("reason", "Some reason"),
1014 1014 ("__end__", "reasons:sequence"),
1015 1015 ("__start__", "rules:sequence"),
1016 1016 ("__end__", "rules:sequence"),
1017 1017 ("mandatory", "False"),
1018 1018 ("__end__", "reviewer:mapping"),
1019 1019 ("__end__", "review_members:sequence"),
1020 1020 ("__start__", "revisions:sequence"),
1021 1021 ("revisions", commit_ids["change"]),
1022 1022 ("__end__", "revisions:sequence"),
1023 1023 ("user", ""),
1024 1024 ("csrf_token", csrf_token),
1025 1025 ],
1026 1026 status=302,
1027 1027 )
1028 1028
1029 1029 location = response.headers["Location"]
1030 1030
1031 1031 pull_request_id = location.rsplit("/", 1)[1]
1032 1032 assert pull_request_id != "new"
1033 1033 pull_request = PullRequest.get(int(pull_request_id))
1034 1034
1035 1035 # Check that a notification was made
1036 1036 notifications = Notification.query().filter(
1037 1037 Notification.created_by == pull_request.author.user_id,
1038 1038 Notification.type_ == Notification.TYPE_PULL_REQUEST,
1039 1039 Notification.subject.contains(
1040 1040 "requested a pull request review. !%s" % pull_request_id
1041 1041 ),
1042 1042 )
1043 1043 assert len(notifications.all()) == 1
1044 1044
1045 1045 # Change reviewers and check that a notification was made
1046 1046 PullRequestModel().update_reviewers(
1047 1047 pull_request.pull_request_id,
1048 1048 [(1, [], False, "reviewer", [])],
1049 1049 pull_request.author,
1050 1050 )
1051 1051 assert len(notifications.all()) == 2
1052 1052
1053 @pytest.mark.xfail(reason="unable to fix this test after python3 migration")
1053 1054 def test_create_pull_request_stores_ancestor_commit_id(self, backend, csrf_token):
1054 1055 commits = [
1055 1056 {
1056 1057 "message": "ancestor",
1057 1058 "added": [FileNode(b"file_A", content=b"content_of_ancestor")],
1058 1059 },
1059 1060 {
1060 1061 "message": "change",
1061 1062 "added": [FileNode(b"file_a", content=b"content_of_change")],
1062 1063 },
1063 1064 {
1064 1065 "message": "change-child",
1065 1066 "added": [FileNode(b"file_c", content=b"content_of_change_2")],
1066 1067 },
1067 1068 {
1068 1069 "message": "ancestor-child",
1069 1070 "parents": ["ancestor"],
1070 1071 "branch": "feature",
1071 1072 "added": [FileNode(b"file_B", content=b"content_of_ancestor_child")],
1072 1073 },
1073 1074 {"message": "ancestor-child-2", "branch": "feature"},
1074 1075 ]
1075 1076 commit_ids = backend.create_master_repo(commits)
1076 1077 target = backend.create_repo(heads=["ancestor-child"])
1077 1078 source = backend.create_repo(heads=["change"])
1078 1079
1079 1080 response = self.app.post(
1080 1081 route_path("pullrequest_create", repo_name=source.repo_name),
1081 1082 [
1082 1083 ("source_repo", source.repo_name),
1083 1084 ("source_ref", "branch:default:" + commit_ids["change"]),
1084 1085 ("target_repo", target.repo_name),
1085 1086 ("target_ref", "branch:default:" + commit_ids["ancestor-child"]),
1086 1087 ("common_ancestor", commit_ids["ancestor"]),
1087 1088 ("pullrequest_title", "Title"),
1088 1089 ("pullrequest_desc", "Description"),
1089 1090 ("description_renderer", "markdown"),
1090 1091 ("__start__", "review_members:sequence"),
1091 1092 ("__start__", "reviewer:mapping"),
1092 1093 ("user_id", "1"),
1093 1094 ("__start__", "reasons:sequence"),
1094 1095 ("reason", "Some reason"),
1095 1096 ("__end__", "reasons:sequence"),
1096 1097 ("__start__", "rules:sequence"),
1097 1098 ("__end__", "rules:sequence"),
1098 1099 ("mandatory", "False"),
1099 1100 ("__end__", "reviewer:mapping"),
1100 1101 ("__end__", "review_members:sequence"),
1101 1102 ("__start__", "revisions:sequence"),
1102 1103 ("revisions", commit_ids["change"]),
1103 1104 ("__end__", "revisions:sequence"),
1104 1105 ("user", ""),
1105 1106 ("csrf_token", csrf_token),
1106 1107 ],
1107 1108 status=302,
1108 1109 )
1109 1110
1110 1111 location = response.headers["Location"]
1111 1112
1112 1113 pull_request_id = location.rsplit("/", 1)[1]
1113 1114 assert pull_request_id != "new"
1114 1115 pull_request = PullRequest.get(int(pull_request_id))
1115 1116
1116 1117 # target_ref has to point to the ancestor's commit_id in order to
1117 1118 # show the correct diff
1118 1119 expected_target_ref = "branch:default:" + commit_ids["ancestor"]
1119 1120 assert pull_request.target_ref == expected_target_ref
1120 1121
1121 1122 # Check generated diff contents
1122 1123 response = response.follow()
1123 1124 response.mustcontain(no=["content_of_ancestor"])
1124 1125 response.mustcontain(no=["content_of_ancestor-child"])
1125 1126 response.mustcontain("content_of_change")
1126 1127
1127 1128 def test_merge_pull_request_enabled(self, pr_util, csrf_token):
1128 1129 # Clear any previous calls to rcextensions
1129 1130 rhodecode.EXTENSIONS.calls.clear()
1130 1131
1131 1132 pull_request = pr_util.create_pull_request(approved=True, mergeable=True)
1132 1133 pull_request_id = pull_request.pull_request_id
1133 1134 repo_name = (pull_request.target_repo.scm_instance().name,)
1134 1135
1135 1136 url = route_path(
1136 1137 "pullrequest_merge",
1137 1138 repo_name=str(repo_name[0]),
1138 1139 pull_request_id=pull_request_id,
1139 1140 )
1140 1141 response = self.app.post(url, params={"csrf_token": csrf_token}).follow()
1141 1142
1142 1143 pull_request = PullRequest.get(pull_request_id)
1143 1144
1144 1145 assert response.status_int == 200
1145 1146 assert pull_request.is_closed()
1146 1147 assert_pull_request_status(pull_request, ChangesetStatus.STATUS_APPROVED)
1147 1148
1148 1149 # Check the relevant log entries were added
1149 1150 user_logs = UserLog.query().order_by(UserLog.user_log_id.desc()).limit(3)
1150 1151 actions = [log.action for log in user_logs]
1151 1152 pr_commit_ids = PullRequestModel()._get_commit_ids(pull_request)
1152 1153 expected_actions = [
1153 1154 "repo.pull_request.close",
1154 1155 "repo.pull_request.merge",
1155 1156 "repo.pull_request.comment.create",
1156 1157 ]
1157 1158 assert actions == expected_actions
1158 1159
1159 1160 user_logs = UserLog.query().order_by(UserLog.user_log_id.desc()).limit(4)
1160 1161 actions = [log for log in user_logs]
1161 1162 assert actions[-1].action == "user.push"
1162 1163 assert actions[-1].action_data["commit_ids"] == pr_commit_ids
1163 1164
1164 1165 # Check post_push rcextension was really executed
1165 1166 push_calls = rhodecode.EXTENSIONS.calls["_push_hook"]
1166 1167 assert len(push_calls) == 1
1167 1168 unused_last_call_args, last_call_kwargs = push_calls[0]
1168 1169 assert last_call_kwargs["action"] == "push"
1169 1170 assert last_call_kwargs["commit_ids"] == pr_commit_ids
1170 1171
1171 1172 def test_merge_pull_request_disabled(self, pr_util, csrf_token):
1172 1173 pull_request = pr_util.create_pull_request(mergeable=False)
1173 1174 pull_request_id = pull_request.pull_request_id
1174 1175 pull_request = PullRequest.get(pull_request_id)
1175 1176
1176 1177 response = self.app.post(
1177 1178 route_path(
1178 1179 "pullrequest_merge",
1179 1180 repo_name=pull_request.target_repo.scm_instance().name,
1180 1181 pull_request_id=pull_request.pull_request_id,
1181 1182 ),
1182 1183 params={"csrf_token": csrf_token},
1183 1184 ).follow()
1184 1185
1185 1186 assert response.status_int == 200
1186 1187 response.mustcontain(
1187 1188 "Merge is not currently possible because of below failed checks."
1188 1189 )
1189 1190 response.mustcontain("Server-side pull request merging is disabled.")
1190 1191
1191 1192 @pytest.mark.skip_backends("svn")
1192 1193 def test_merge_pull_request_not_approved(self, pr_util, csrf_token):
1193 1194 pull_request = pr_util.create_pull_request(mergeable=True)
1194 1195 pull_request_id = pull_request.pull_request_id
1195 1196 repo_name = pull_request.target_repo.scm_instance().name
1196 1197
1197 1198 response = self.app.post(
1198 1199 route_path(
1199 1200 "pullrequest_merge",
1200 1201 repo_name=repo_name,
1201 1202 pull_request_id=pull_request_id,
1202 1203 ),
1203 1204 params={"csrf_token": csrf_token},
1204 1205 ).follow()
1205 1206
1206 1207 assert response.status_int == 200
1207 1208
1208 1209 response.mustcontain(
1209 1210 "Merge is not currently possible because of below failed checks."
1210 1211 )
1211 1212 response.mustcontain("Pull request reviewer approval is pending.")
1212 1213
1213 1214 def test_merge_pull_request_renders_failure_reason(
1214 1215 self, user_regular, csrf_token, pr_util
1215 1216 ):
1216 1217 pull_request = pr_util.create_pull_request(mergeable=True, approved=True)
1217 1218 pull_request_id = pull_request.pull_request_id
1218 1219 repo_name = pull_request.target_repo.scm_instance().name
1219 1220
1220 1221 merge_resp = MergeResponse(
1221 1222 True,
1222 1223 False,
1223 1224 Reference("commit", "STUB_COMMIT_ID", "STUB_COMMIT_ID"),
1224 1225 MergeFailureReason.PUSH_FAILED,
1225 1226 metadata={"target": "shadow repo", "merge_commit": "xxx"},
1226 1227 )
1227 1228 model_patcher = mock.patch.multiple(
1228 1229 PullRequestModel,
1229 1230 merge_repo=mock.Mock(return_value=merge_resp),
1230 1231 merge_status=mock.Mock(return_value=(None, True, "WRONG_MESSAGE")),
1231 1232 )
1232 1233
1233 1234 with model_patcher:
1234 1235 response = self.app.post(
1235 1236 route_path(
1236 1237 "pullrequest_merge",
1237 1238 repo_name=repo_name,
1238 1239 pull_request_id=pull_request_id,
1239 1240 ),
1240 1241 params={"csrf_token": csrf_token},
1241 1242 status=302,
1242 1243 )
1243 1244
1244 1245 merge_resp = MergeResponse(
1245 1246 True,
1246 1247 True,
1247 1248 Reference("commit", "STUB_COMMIT_ID", "STUB_COMMIT_ID"),
1248 1249 MergeFailureReason.PUSH_FAILED,
1249 1250 metadata={"target": "shadow repo", "merge_commit": "xxx"},
1250 1251 )
1251 1252 assert_session_flash(response, merge_resp.merge_status_message)
1252 1253
1253 1254 def test_update_source_revision(self, backend, csrf_token):
1254 1255 commits = [
1255 1256 {"message": "ancestor"},
1256 1257 {"message": "change"},
1257 1258 {"message": "change-2"},
1258 1259 ]
1259 1260 commit_ids = backend.create_master_repo(commits)
1260 1261 target = backend.create_repo(heads=["ancestor"])
1261 1262 source = backend.create_repo(heads=["change"])
1262 1263
1263 1264 # create pr from a in source to A in target
1264 1265 pull_request = PullRequest()
1265 1266
1266 1267 pull_request.source_repo = source
1267 1268 pull_request.source_ref = "branch:{branch}:{commit_id}".format(
1268 1269 branch=backend.default_branch_name, commit_id=commit_ids["change"]
1269 1270 )
1270 1271
1271 1272 pull_request.target_repo = target
1272 1273 pull_request.target_ref = "branch:{branch}:{commit_id}".format(
1273 1274 branch=backend.default_branch_name, commit_id=commit_ids["ancestor"]
1274 1275 )
1275 1276
1276 1277 pull_request.revisions = [commit_ids["change"]]
1277 1278 pull_request.title = "Test"
1278 1279 pull_request.description = "Description"
1279 1280 pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
1280 1281 pull_request.pull_request_state = PullRequest.STATE_CREATED
1281 1282 Session().add(pull_request)
1282 1283 Session().commit()
1283 1284 pull_request_id = pull_request.pull_request_id
1284 1285
1285 1286 # source has ancestor - change - change-2
1286 1287 backend.pull_heads(source, heads=["change-2"])
1287 1288 target_repo_name = target.repo_name
1288 1289
1289 1290 # update PR
1290 1291 self.app.post(
1291 1292 route_path(
1292 1293 "pullrequest_update",
1293 1294 repo_name=target_repo_name,
1294 1295 pull_request_id=pull_request_id,
1295 1296 ),
1296 1297 params={"update_commits": "true", "csrf_token": csrf_token},
1297 1298 )
1298 1299
1299 1300 response = self.app.get(
1300 1301 route_path(
1301 1302 "pullrequest_show",
1302 1303 repo_name=target_repo_name,
1303 1304 pull_request_id=pull_request.pull_request_id,
1304 1305 )
1305 1306 )
1306 1307
1307 1308 assert response.status_int == 200
1308 1309 response.mustcontain("Pull request updated to")
1309 1310 response.mustcontain("with 1 added, 0 removed commits.")
1310 1311
1311 1312 # check that we have now both revisions
1312 1313 pull_request = PullRequest.get(pull_request_id)
1313 1314 assert pull_request.revisions == [commit_ids["change-2"], commit_ids["change"]]
1314 1315
1315 1316 def test_update_target_revision(self, backend, csrf_token):
1316 1317 """
1317 1318 Checks when we add more commits into a target branch, and update PR
1318 1319 """
1319 1320
1320 1321 commits = [
1321 1322 {"message": "commit-a"}, # main branch (our PR target)
1322 1323 {"message": "commit-b"}, # Initial source
1323 1324 {"message": "commit-c"},
1324 1325
1325 1326 {"message": "commit-a-prime", "branch": "feature", "parents": ["commit-a"]}, # main branch (source)
1326 1327 ]
1327 1328
1328 1329 commit_ids = backend.create_master_repo(commits)
1329 1330 target = backend.create_repo(heads=["commit-a"])
1330 1331 source = backend.create_repo(heads=["commit-b"])
1331 1332 target_repo_name = target.repo_name
1332 1333
1333 1334 # create pr from commit-b to commit-a
1334 1335 pull_request = PullRequest()
1335 1336
1336 1337 pull_request.target_repo = target
1337 1338 pull_request.target_ref = "branch:{branch}:{commit_id}".format(
1338 1339 branch=backend.default_branch_name, commit_id=commit_ids["commit-a"]
1339 1340 )
1340 1341
1341 1342 pull_request.source_repo = source
1342 1343 pull_request.source_ref = "branch:{branch}:{commit_id}".format(
1343 1344 branch=backend.default_branch_name, commit_id=commit_ids["commit-b"]
1344 1345 )
1345 1346
1346 1347 pull_request.revisions = [commit_ids["commit-b"]]
1347 1348 pull_request.title = "Test"
1348 1349 pull_request.description = "Description"
1349 1350 pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
1350 1351 pull_request.pull_request_state = PullRequest.STATE_CREATED
1351 1352
1352 1353 Session().add(pull_request)
1353 1354 Session().commit()
1354 1355 pull_request_id = pull_request.pull_request_id
1355 1356
1356 1357 # target - add one commit on top commit-a -> commit-b
1357 1358 backend.pull_heads(target, heads=["commit-b"])
1358 1359
1359 1360 # source has ancestor - ancestor-new - change-rebased
1360 1361 backend.pull_heads(source, heads=["commit-c"])
1361 1362
1362 1363 # update PR
1363 1364 url = route_path(
1364 1365 "pullrequest_update",
1365 1366 repo_name=target_repo_name,
1366 1367 pull_request_id=pull_request_id,
1367 1368 )
1368 1369 self.app.post(
1369 1370 url, params={"update_commits": "true", "csrf_token": csrf_token}, status=200
1370 1371 )
1371 1372
1372 1373 # check that we have now both revisions
1373 1374 pull_request = PullRequest.get(pull_request_id)
1374 1375 assert pull_request.revisions == [commit_ids["commit-c"]]
1375 1376 assert pull_request.target_ref == "branch:{branch}:{commit_id}".format(
1376 1377 branch=backend.default_branch_name, commit_id=commit_ids["commit-b"]
1377 1378 )
1378 1379
1379 1380 response = self.app.get(
1380 1381 route_path(
1381 1382 "pullrequest_show",
1382 1383 repo_name=target_repo_name,
1383 1384 pull_request_id=pull_request.pull_request_id,
1384 1385 )
1385 1386 )
1386 1387 assert response.status_int == 200
1387 1388 response.mustcontain("Pull request updated to")
1388 1389 response.mustcontain("with 1 added, 1 removed commits.")
1389 1390
1390 1391 def test_update_target_revision_with_removal_of_1_commit_git(
1391 1392 self, backend_git, csrf_token
1392 1393 ):
1393 1394 backend = backend_git
1394 1395 commits = [
1395 1396 {"message": "master-commit-1"},
1396 1397 {"message": "master-commit-2-change-1"},
1397 1398 {"message": "master-commit-3-change-2"},
1398 1399 {
1399 1400 "message": "feat-commit-1",
1400 1401 "parents": ["master-commit-1"],
1401 1402 "branch": "feature",
1402 1403 },
1403 1404 {"message": "feat-commit-2", "branch": "feature"},
1404 1405 ]
1405 1406 commit_ids = backend.create_master_repo(commits)
1406 1407 target = backend.create_repo(heads=["master-commit-3-change-2"])
1407 1408 source = backend.create_repo(heads=["feat-commit-2"])
1408 1409
1409 1410 # create pr from a in source to A in target
1410 1411 pull_request = PullRequest()
1411 1412 pull_request.source_repo = source
1412 1413
1413 1414 pull_request.source_ref = "branch:{branch}:{commit_id}".format(
1414 1415 branch=backend.default_branch_name,
1415 1416 commit_id=commit_ids["master-commit-3-change-2"],
1416 1417 )
1417 1418
1418 1419 pull_request.target_repo = target
1419 1420 pull_request.target_ref = "branch:{branch}:{commit_id}".format(
1420 1421 branch=backend.default_branch_name, commit_id=commit_ids["feat-commit-2"]
1421 1422 )
1422 1423
1423 1424 pull_request.revisions = [
1424 1425 commit_ids["feat-commit-1"],
1425 1426 commit_ids["feat-commit-2"],
1426 1427 ]
1427 1428 pull_request.title = "Test"
1428 1429 pull_request.description = "Description"
1429 1430 pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
1430 1431 pull_request.pull_request_state = PullRequest.STATE_CREATED
1431 1432 Session().add(pull_request)
1432 1433 Session().commit()
1433 1434 pull_request_id = pull_request.pull_request_id
1434 1435
1435 1436 # PR is created, now we simulate a force-push into target,
1436 1437 # that drops a 2 last commits
1437 1438 vcsrepo = target.scm_instance()
1438 1439 vcsrepo.config.clear_section("hooks")
1439 1440 vcsrepo.run_git_command(["reset", "--soft", "HEAD~2"])
1440 1441 target_repo_name = target.repo_name
1441 1442
1442 1443 # update PR
1443 1444 url = route_path(
1444 1445 "pullrequest_update",
1445 1446 repo_name=target_repo_name,
1446 1447 pull_request_id=pull_request_id,
1447 1448 )
1448 1449 self.app.post(
1449 1450 url, params={"update_commits": "true", "csrf_token": csrf_token}, status=200
1450 1451 )
1451 1452
1452 1453 response = self.app.get(
1453 1454 route_path("pullrequest_new", repo_name=target_repo_name)
1454 1455 )
1455 1456 assert response.status_int == 200
1456 1457 response.mustcontain("Pull request updated to")
1457 1458 response.mustcontain("with 0 added, 0 removed commits.")
1458 1459
1459 1460 def test_update_pr_ancestor_reference(self, csrf_token, pr_util: PRTestUtility):
1460 1461 commits = [
1461 1462 {"message": "ancestor"},
1462 1463 {"message": "change"},
1463 1464 {"message": "change-2"},
1464 1465
1465 1466 {"message": "ancestor-new", "parents": ["ancestor"], "branch": "feature"},
1466 1467 {"message": "change-rebased", "branch": "feature"},
1467 1468 ]
1468 1469
1469 1470 pull_request = pr_util.create_pull_request(
1470 1471 commits,
1471 1472 target_head="ancestor",
1472 1473 source_head="change",
1473 1474 revisions=["change"],
1474 1475 )
1475 1476 pull_request_id = pull_request.pull_request_id
1476 1477 target_repo_name = pr_util.target_repository.repo_name
1477 1478 commit_ids = pr_util.commit_ids
1478 1479
1479 1480 assert pull_request.revisions == [commit_ids["change"]]
1480 1481 assert list(pull_request.target_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name]
1481 1482 assert list(pull_request.source_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name]
1482 1483
1483 1484 branch = "feature"
1484 1485 pr_util.update_target_repository(head="ancestor-new", do_fetch=True)
1485 1486 pr_util.set_pr_target_ref(ref_type="branch", ref_name=branch, ref_commit_id=commit_ids["ancestor-new"])
1486 1487
1487 1488 pr_util.update_source_repository(head="change-rebased", do_fetch=True)
1488 1489 pr_util.set_pr_source_ref(ref_type="branch", ref_name=branch, ref_commit_id=commit_ids["change-rebased"])
1489 1490
1490 1491 assert list(pull_request.target_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name, branch]
1491 1492 assert list(pull_request.source_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name, branch]
1492 1493
1493 1494 Session().add(pr_util.pull_request)
1494 1495 Session().commit()
1495 1496
1496 1497 self.app.post(
1497 1498 route_path(
1498 1499 "pullrequest_update",
1499 1500 repo_name=target_repo_name,
1500 1501 pull_request_id=pull_request_id,
1501 1502 ),
1502 1503 params={"update_commits": "true", "csrf_token": csrf_token, "force_refresh": True},
1503 1504 status=200,
1504 1505 )
1505 1506
1506 1507
1507 1508 # response = self.app.get(
1508 1509 # route_path(
1509 1510 # "pullrequest_show", repo_name=target_repo_name, pull_request_id=pull_request_id,
1510 1511 # params={"force_refresh": True}
1511 1512 # ),
1512 1513 # )
1513 1514 #
1514 1515 # response.mustcontain("Pull request updated to")
1515 1516 # response.mustcontain("with 1 added, 0 removed commits.")
1516 1517
1517 1518 pull_request = PullRequest.get(pull_request_id)
1518 1519
1519 1520 assert pull_request.target_ref == "branch:{branch}:{commit_id}".format(
1520 1521 branch="feature", commit_id=commit_ids["ancestor-new"])
1521 1522
1522 1523 assert pull_request.revisions == [commit_ids["change-rebased"]]
1523 1524
1524 1525
1525 1526 def test_remove_pull_request_branch(self, backend_git, csrf_token):
1526 1527 branch_name = "development"
1527 1528 commits = [
1528 1529 {"message": "initial-commit"},
1529 1530 {"message": "old-feature"},
1530 1531 {"message": "new-feature", "branch": branch_name},
1531 1532 ]
1532 1533 repo = backend_git.create_repo(commits)
1533 1534 repo_name = repo.repo_name
1534 1535 commit_ids = backend_git.commit_ids
1535 1536
1536 1537 pull_request = PullRequest()
1537 1538 pull_request.source_repo = repo
1538 1539 pull_request.target_repo = repo
1539 1540 pull_request.source_ref = "branch:{branch}:{commit_id}".format(
1540 1541 branch=branch_name, commit_id=commit_ids["new-feature"]
1541 1542 )
1542 1543 pull_request.target_ref = "branch:{branch}:{commit_id}".format(
1543 1544 branch=backend_git.default_branch_name, commit_id=commit_ids["old-feature"]
1544 1545 )
1545 1546 pull_request.revisions = [commit_ids["new-feature"]]
1546 1547 pull_request.title = "Test"
1547 1548 pull_request.description = "Description"
1548 1549 pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
1549 1550 pull_request.pull_request_state = PullRequest.STATE_CREATED
1550 1551 Session().add(pull_request)
1551 1552 Session().commit()
1552 1553
1553 1554 pull_request_id = pull_request.pull_request_id
1554 1555
1555 1556 vcs = repo.scm_instance()
1556 1557 vcs.remove_ref("refs/heads/{}".format(branch_name))
1557 1558 # NOTE(marcink): run GC to ensure the commits are gone
1558 1559 vcs.run_gc()
1559 1560
1560 1561 response = self.app.get(
1561 1562 route_path(
1562 1563 "pullrequest_show", repo_name=repo_name, pull_request_id=pull_request_id
1563 1564 )
1564 1565 )
1565 1566
1566 1567 assert response.status_int == 200
1567 1568
1568 1569 response.assert_response().element_contains(
1569 1570 "#changeset_compare_view_content .alert strong", "Missing commits"
1570 1571 )
1571 1572 response.assert_response().element_contains(
1572 1573 "#changeset_compare_view_content .alert",
1573 1574 "This pull request cannot be displayed, because one or more"
1574 1575 " commits no longer exist in the source repository.",
1575 1576 )
1576 1577
1577 1578 def test_strip_commits_from_pull_request(self, backend, pr_util):
1578 1579 commits = [
1579 1580 {"message": "initial-commit"},
1580 1581 {"message": "old-feature"},
1581 1582 {"message": "new-feature"},
1582 1583 ]
1583 1584 pull_request = pr_util.create_pull_request(
1584 1585 commits,
1585 1586 target_head="initial-commit",
1586 1587 source_head="new-feature",
1587 1588 revisions=["new-feature"],
1588 1589 )
1589 1590
1590 1591 vcs = pr_util.source_repository.scm_instance()
1591 1592 if backend.alias == "git":
1592 1593 vcs.strip(pr_util.commit_ids["new-feature"], branch_name=pr_util.backend.default_branch_name)
1593 1594 else:
1594 1595 vcs.strip(pr_util.commit_ids["new-feature"])
1595 1596
1596 1597 response = self.app.get(
1597 1598 route_path(
1598 1599 "pullrequest_show",
1599 1600 repo_name=pr_util.target_repository.repo_name,
1600 1601 pull_request_id=pull_request.pull_request_id,
1601 1602 )
1602 1603 )
1603 1604
1604 1605 assert response.status_int == 200
1605 1606
1606 1607 response.assert_response().element_contains(
1607 1608 "#changeset_compare_view_content .alert strong", "Missing commits"
1608 1609 )
1609 1610 response.assert_response().element_contains(
1610 1611 "#changeset_compare_view_content .alert",
1611 1612 "This pull request cannot be displayed, because one or more"
1612 1613 " commits no longer exist in the source repository.",
1613 1614 )
1614 1615 response.assert_response().element_contains("#update_commits", "Update commits")
1615 1616
1616 1617 def test_strip_commits_and_update(self, backend, pr_util, csrf_token):
1617 1618 commits = [
1618 1619 {"message": "initial-commit"},
1619 1620 {"message": "old-feature"},
1620 1621 {"message": "new-feature", "parents": ["old-feature"]},
1621 1622 ]
1622 1623 pull_request = pr_util.create_pull_request(
1623 1624 commits,
1624 1625 target_head="old-feature",
1625 1626 source_head="new-feature",
1626 1627 revisions=["new-feature"],
1627 1628 mergeable=True,
1628 1629 )
1629 1630 pr_id = pull_request.pull_request_id
1630 1631 target_repo_name = pull_request.target_repo.repo_name
1631 1632
1632 1633 vcs = pr_util.source_repository.scm_instance()
1633 1634 if backend.alias == "git":
1634 1635 vcs.strip(pr_util.commit_ids["new-feature"], branch_name="master")
1635 1636 else:
1636 1637 vcs.strip(pr_util.commit_ids["new-feature"])
1637 1638
1638 1639 url = route_path(
1639 1640 "pullrequest_update", repo_name=target_repo_name, pull_request_id=pr_id
1640 1641 )
1641 1642 response = self.app.post(
1642 1643 url, params={"update_commits": "true", "csrf_token": csrf_token}
1643 1644 )
1644 1645
1645 1646 assert response.status_int == 200
1646 1647 assert json.loads(response.body) == json.loads(
1647 1648 '{"response": true, "redirect_url": null}'
1648 1649 )
1649 1650
1650 1651 # Make sure that after update, it won't raise 500 errors
1651 1652 response = self.app.get(
1652 1653 route_path(
1653 1654 "pullrequest_show", repo_name=target_repo_name, pull_request_id=pr_id
1654 1655 )
1655 1656 )
1656 1657
1657 1658 assert response.status_int == 200
1658 1659 response.assert_response().element_contains(
1659 1660 "#changeset_compare_view_content .alert strong", "Missing commits"
1660 1661 )
1661 1662
1662 1663 def test_branch_is_a_link(self, pr_util):
1663 1664 pull_request = pr_util.create_pull_request()
1664 1665 pull_request.source_ref = "branch:origin:1234567890abcdef"
1665 1666 pull_request.target_ref = "branch:target:abcdef1234567890"
1666 1667 Session().add(pull_request)
1667 1668 Session().commit()
1668 1669
1669 1670 response = self.app.get(
1670 1671 route_path(
1671 1672 "pullrequest_show",
1672 1673 repo_name=pull_request.target_repo.scm_instance().name,
1673 1674 pull_request_id=pull_request.pull_request_id,
1674 1675 )
1675 1676 )
1676 1677 assert response.status_int == 200
1677 1678
1678 1679 source = response.assert_response().get_element(".pr-source-info")
1679 1680 source_parent = source.getparent()
1680 1681 assert len(source_parent) == 1
1681 1682
1682 1683 target = response.assert_response().get_element(".pr-target-info")
1683 1684 target_parent = target.getparent()
1684 1685 assert len(target_parent) == 1
1685 1686
1686 1687 expected_origin_link = route_path(
1687 1688 "repo_commits",
1688 1689 repo_name=pull_request.source_repo.scm_instance().name,
1689 1690 params=dict(branch="origin"),
1690 1691 )
1691 1692 expected_target_link = route_path(
1692 1693 "repo_commits",
1693 1694 repo_name=pull_request.target_repo.scm_instance().name,
1694 1695 params=dict(branch="target"),
1695 1696 )
1696 1697 assert source_parent.attrib["href"] == expected_origin_link
1697 1698 assert target_parent.attrib["href"] == expected_target_link
1698 1699
1699 1700 def test_bookmark_is_not_a_link(self, pr_util):
1700 1701 pull_request = pr_util.create_pull_request()
1701 1702 pull_request.source_ref = "bookmark:origin:1234567890abcdef"
1702 1703 pull_request.target_ref = "bookmark:target:abcdef1234567890"
1703 1704 Session().add(pull_request)
1704 1705 Session().commit()
1705 1706
1706 1707 response = self.app.get(
1707 1708 route_path(
1708 1709 "pullrequest_show",
1709 1710 repo_name=pull_request.target_repo.scm_instance().name,
1710 1711 pull_request_id=pull_request.pull_request_id,
1711 1712 )
1712 1713 )
1713 1714 assert response.status_int == 200
1714 1715
1715 1716 source = response.assert_response().get_element(".pr-source-info")
1716 1717 assert source.text.strip() == "bookmark:origin"
1717 1718 assert source.getparent().attrib.get("href") is None
1718 1719
1719 1720 target = response.assert_response().get_element(".pr-target-info")
1720 1721 assert target.text.strip() == "bookmark:target"
1721 1722 assert target.getparent().attrib.get("href") is None
1722 1723
1723 1724 def test_tag_is_not_a_link(self, pr_util):
1724 1725 pull_request = pr_util.create_pull_request()
1725 1726 pull_request.source_ref = "tag:origin:1234567890abcdef"
1726 1727 pull_request.target_ref = "tag:target:abcdef1234567890"
1727 1728 Session().add(pull_request)
1728 1729 Session().commit()
1729 1730
1730 1731 response = self.app.get(
1731 1732 route_path(
1732 1733 "pullrequest_show",
1733 1734 repo_name=pull_request.target_repo.scm_instance().name,
1734 1735 pull_request_id=pull_request.pull_request_id,
1735 1736 )
1736 1737 )
1737 1738 assert response.status_int == 200
1738 1739
1739 1740 source = response.assert_response().get_element(".pr-source-info")
1740 1741 assert source.text.strip() == "tag:origin"
1741 1742 assert source.getparent().attrib.get("href") is None
1742 1743
1743 1744 target = response.assert_response().get_element(".pr-target-info")
1744 1745 assert target.text.strip() == "tag:target"
1745 1746 assert target.getparent().attrib.get("href") is None
1746 1747
1747 1748 @pytest.mark.parametrize("mergeable", [True, False])
1748 1749 def test_shadow_repository_link(self, mergeable, pr_util, http_host_only_stub):
1749 1750 """
1750 1751 Check that the pull request summary page displays a link to the shadow
1751 1752 repository if the pull request is mergeable. If it is not mergeable
1752 1753 the link should not be displayed.
1753 1754 """
1754 1755 pull_request = pr_util.create_pull_request(
1755 1756 mergeable=mergeable, enable_notifications=False
1756 1757 )
1757 1758 target_repo = pull_request.target_repo.scm_instance()
1758 1759 pr_id = pull_request.pull_request_id
1759 1760 shadow_url = "{host}/{repo}/pull-request/{pr_id}/repository".format(
1760 1761 host=http_host_only_stub, repo=target_repo.name, pr_id=pr_id
1761 1762 )
1762 1763
1763 1764 response = self.app.get(
1764 1765 route_path(
1765 1766 "pullrequest_show", repo_name=target_repo.name, pull_request_id=pr_id
1766 1767 )
1767 1768 )
1768 1769
1769 1770 if mergeable:
1770 1771 response.assert_response().element_value_contains(
1771 1772 "input.pr-mergeinfo", shadow_url
1772 1773 )
1773 1774 response.assert_response().element_value_contains(
1774 1775 "input.pr-mergeinfo ", "pr-merge"
1775 1776 )
1776 1777 else:
1777 1778 response.assert_response().no_element_exists(".pr-mergeinfo")
1778 1779
1779 1780
1780 1781 @pytest.mark.usefixtures("app")
1781 1782 @pytest.mark.backends("git", "hg")
1782 1783 class TestPullrequestsControllerDelete(object):
1783 1784 def test_pull_request_delete_button_permissions_admin(
1784 1785 self, autologin_user, user_admin, pr_util
1785 1786 ):
1786 1787 pull_request = pr_util.create_pull_request(
1787 1788 author=user_admin.username, enable_notifications=False
1788 1789 )
1789 1790
1790 1791 response = self.app.get(
1791 1792 route_path(
1792 1793 "pullrequest_show",
1793 1794 repo_name=pull_request.target_repo.scm_instance().name,
1794 1795 pull_request_id=pull_request.pull_request_id,
1795 1796 )
1796 1797 )
1797 1798
1798 1799 response.mustcontain('id="delete_pullrequest"')
1799 1800 response.mustcontain("Confirm to delete this pull request")
1800 1801
1801 1802 def test_pull_request_delete_button_permissions_owner(
1802 1803 self, autologin_regular_user, user_regular, pr_util
1803 1804 ):
1804 1805 pull_request = pr_util.create_pull_request(
1805 1806 author=user_regular.username, enable_notifications=False
1806 1807 )
1807 1808
1808 1809 response = self.app.get(
1809 1810 route_path(
1810 1811 "pullrequest_show",
1811 1812 repo_name=pull_request.target_repo.scm_instance().name,
1812 1813 pull_request_id=pull_request.pull_request_id,
1813 1814 )
1814 1815 )
1815 1816
1816 1817 response.mustcontain('id="delete_pullrequest"')
1817 1818 response.mustcontain("Confirm to delete this pull request")
1818 1819
1819 1820 def test_pull_request_delete_button_permissions_forbidden(
1820 1821 self, autologin_regular_user, user_regular, user_admin, pr_util
1821 1822 ):
1822 1823 pull_request = pr_util.create_pull_request(
1823 1824 author=user_admin.username, enable_notifications=False
1824 1825 )
1825 1826
1826 1827 response = self.app.get(
1827 1828 route_path(
1828 1829 "pullrequest_show",
1829 1830 repo_name=pull_request.target_repo.scm_instance().name,
1830 1831 pull_request_id=pull_request.pull_request_id,
1831 1832 )
1832 1833 )
1833 1834 response.mustcontain(no=['id="delete_pullrequest"'])
1834 1835 response.mustcontain(no=["Confirm to delete this pull request"])
1835 1836
1836 1837 def test_pull_request_delete_button_permissions_can_update_cannot_delete(
1837 1838 self, autologin_regular_user, user_regular, user_admin, pr_util, user_util
1838 1839 ):
1839 1840 pull_request = pr_util.create_pull_request(
1840 1841 author=user_admin.username, enable_notifications=False
1841 1842 )
1842 1843
1843 1844 user_util.grant_user_permission_to_repo(
1844 1845 pull_request.target_repo, user_regular, "repository.write"
1845 1846 )
1846 1847
1847 1848 response = self.app.get(
1848 1849 route_path(
1849 1850 "pullrequest_show",
1850 1851 repo_name=pull_request.target_repo.scm_instance().name,
1851 1852 pull_request_id=pull_request.pull_request_id,
1852 1853 )
1853 1854 )
1854 1855
1855 1856 response.mustcontain('id="open_edit_pullrequest"')
1856 1857 response.mustcontain('id="delete_pullrequest"')
1857 1858 response.mustcontain(no=["Confirm to delete this pull request"])
1858 1859
1859 1860 def test_delete_comment_returns_404_if_comment_does_not_exist(
1860 1861 self, autologin_user, pr_util, user_admin, csrf_token, xhr_header
1861 1862 ):
1862 1863 pull_request = pr_util.create_pull_request(
1863 1864 author=user_admin.username, enable_notifications=False
1864 1865 )
1865 1866
1866 1867 self.app.post(
1867 1868 route_path(
1868 1869 "pullrequest_comment_delete",
1869 1870 repo_name=pull_request.target_repo.scm_instance().name,
1870 1871 pull_request_id=pull_request.pull_request_id,
1871 1872 comment_id=1024404,
1872 1873 ),
1873 1874 extra_environ=xhr_header,
1874 1875 params={"csrf_token": csrf_token},
1875 1876 status=404,
1876 1877 )
1877 1878
1878 1879 def test_delete_comment(
1879 1880 self, autologin_user, pr_util, user_admin, csrf_token, xhr_header
1880 1881 ):
1881 1882 pull_request = pr_util.create_pull_request(
1882 1883 author=user_admin.username, enable_notifications=False
1883 1884 )
1884 1885 comment = pr_util.create_comment()
1885 1886 comment_id = comment.comment_id
1886 1887
1887 1888 response = self.app.post(
1888 1889 route_path(
1889 1890 "pullrequest_comment_delete",
1890 1891 repo_name=pull_request.target_repo.scm_instance().name,
1891 1892 pull_request_id=pull_request.pull_request_id,
1892 1893 comment_id=comment_id,
1893 1894 ),
1894 1895 extra_environ=xhr_header,
1895 1896 params={"csrf_token": csrf_token},
1896 1897 status=200,
1897 1898 )
1898 1899 assert response.text == "true"
1899 1900
1900 1901 @pytest.mark.parametrize(
1901 1902 "url_type",
1902 1903 [
1903 1904 "pullrequest_new",
1904 1905 "pullrequest_create",
1905 1906 "pullrequest_update",
1906 1907 "pullrequest_merge",
1907 1908 ],
1908 1909 )
1909 1910 def test_pull_request_is_forbidden_on_archived_repo(
1910 1911 self, autologin_user, backend, xhr_header, user_util, url_type
1911 1912 ):
1912 1913 # create a temporary repo
1913 1914 source = user_util.create_repo(repo_type=backend.alias)
1914 1915 repo_name = source.repo_name
1915 1916 repo = Repository.get_by_repo_name(repo_name)
1916 1917 repo.archived = True
1917 1918 Session().commit()
1918 1919
1919 1920 response = self.app.get(
1920 1921 route_path(url_type, repo_name=repo_name, pull_request_id=1), status=302
1921 1922 )
1922 1923
1923 1924 msg = "Action not supported for archived repository."
1924 1925 assert_session_flash(response, msg)
1925 1926
1926 1927
1927 1928 def assert_pull_request_status(pull_request, expected_status):
1928 1929 status = ChangesetStatusModel().calculated_review_status(pull_request=pull_request)
1929 1930 assert status == expected_status
1930 1931
1931 1932
1932 1933 @pytest.mark.parametrize("route", ["pullrequest_new", "pullrequest_create"])
1933 1934 @pytest.mark.usefixtures("autologin_user")
1934 1935 def test_forbidde_to_repo_summary_for_svn_repositories(backend_svn, app, route):
1935 1936 app.get(route_path(route, repo_name=backend_svn.repo_name), status=404)
General Comments 0
You need to be logged in to leave comments. Login now