OLD | NEW |
| (Empty) |
1 # Copyright 2016 The Chromium Authors. All rights reserved. | |
2 # Use of this source code is governed by a BSD-style license that can be | |
3 # found in the LICENSE file. | |
4 | |
5 import textwrap | |
6 | |
7 from testing_utils import testing | |
8 | |
9 from common import retry_http_client | |
10 from common.rietveld import Rietveld | |
11 | |
12 | |
13 class DummyHttpClient(retry_http_client.RetryHttpClient): | |
14 | |
15 def __init__(self): | |
16 super(DummyHttpClient, self).__init__() | |
17 self.responses = {} | |
18 self.requests = [] | |
19 | |
20 def SetResponse(self, url, result): | |
21 self.responses[url] = result | |
22 | |
23 def GetBackoff(self, *_): # pragma: no cover | |
24 """Override to avoid sleep.""" | |
25 return 0 | |
26 | |
27 def _Get(self, *_): # pragma: no cover | |
28 pass | |
29 | |
30 def _Post(self, url, data, _, headers): | |
31 self.requests.append((url, data, headers)) | |
32 return self.responses.get(url, (404, 'Not Found')) | |
33 | |
34 def _Put(self, *_): # pragma: no cover | |
35 pass | |
36 | |
37 | |
38 class RietveldTest(testing.AppengineTestCase): | |
39 | |
40 def setUp(self): | |
41 super(RietveldTest, self).setUp() | |
42 self.http_client = DummyHttpClient() | |
43 self.rietveld = Rietveld() | |
44 self.rietveld.HTTP_CLIENT = self.http_client | |
45 | |
46 def testGetXsrfTokenSuccess(self): | |
47 rietveld_url = 'https://test' | |
48 self.http_client.SetResponse('%s/xsrf_token' % rietveld_url, (200, 'abc')) | |
49 self.assertEqual('abc', self.rietveld._GetXsrfToken(rietveld_url)) | |
50 self.assertEqual(1, len(self.http_client.requests)) | |
51 _, _, headers = self.http_client.requests[0] | |
52 self.assertTrue('X-Requesting-XSRF-Token' in headers) | |
53 | |
54 def testGetXsrfTokenFailure(self): | |
55 rietveld_url = 'https://test' | |
56 self.http_client.SetResponse('%s/xsrf_token' % rietveld_url, (302, 'login')) | |
57 self.assertIsNone(self.rietveld._GetXsrfToken(rietveld_url)) | |
58 | |
59 def testGetRietveldUrlAndIssueNumber(self): | |
60 cases = { | |
61 'http://abc/123': '123', | |
62 'https://abc/456/': '456', | |
63 'http://abc/789/diff': '789', | |
64 } | |
65 for issue_url, expected_issue_number in cases.iteritems(): | |
66 rietveld_url, issue_number = self.rietveld._GetRietveldUrlAndIssueNumber( | |
67 issue_url) | |
68 self.assertEqual('https://abc', rietveld_url) | |
69 self.assertEqual(expected_issue_number, issue_number) | |
70 | |
71 def testEncodeMultipartFormData(self): | |
72 content_type, body = self.rietveld._EncodeMultipartFormData({'a':'b'}) | |
73 expected_content_type = ( | |
74 'multipart/form-data; boundary=-F-I-N-D-I-T-M-E-S-S-A-G-E-') | |
75 expected_body = textwrap.dedent(""" | |
76 ---F-I-N-D-I-T-M-E-S-S-A-G-E-\r | |
77 Content-Disposition: form-data; name="a"\r | |
78 \r | |
79 b\r | |
80 ---F-I-N-D-I-T-M-E-S-S-A-G-E---\r | |
81 """)[1:] | |
82 self.assertEqual(expected_content_type, content_type) | |
83 self.assertEqual(expected_body, body) | |
84 | |
85 def testPostMessageSuccess(self): | |
86 rietveld_url = 'https://test' | |
87 issue_url = '%s/123' % rietveld_url | |
88 message_publish_url = '%s/publish' % issue_url | |
89 self.http_client.SetResponse('%s/xsrf_token' % rietveld_url, (200, 'abc')) | |
90 self.http_client.SetResponse(message_publish_url, (200, 'OK')) | |
91 self.assertTrue(self.rietveld.PostMessage(issue_url, 'message')) | |
92 self.assertEqual(2, len(self.http_client.requests)) | |
93 | |
94 def testPostMessageFailOnXsrfToken(self): | |
95 rietveld_url = 'https://test' | |
96 issue_url = '%s/123' % rietveld_url | |
97 message_publish_url = '%s/publish' % issue_url | |
98 self.http_client.SetResponse('%s/xsrf_token' % rietveld_url, (302, 'login')) | |
99 self.http_client.SetResponse(message_publish_url, (200, 'OK')) | |
100 self.assertFalse(self.rietveld.PostMessage(issue_url, 'message')) | |
101 self.assertEqual(1, len(self.http_client.requests)) | |
102 | |
103 def testPostMessageFailOnPublish(self): | |
104 rietveld_url = 'https://test' | |
105 issue_url = '%s/123' % rietveld_url | |
106 message_publish_url = '%s/publish' % issue_url | |
107 self.http_client.SetResponse('%s/xsrf_token' % rietveld_url, (302, 'login')) | |
108 self.http_client.SetResponse(message_publish_url, (429, 'Error')) | |
109 self.assertFalse(self.rietveld.PostMessage(issue_url, 'message')) | |
110 self.assertEqual(1, len(self.http_client.requests)) | |
OLD | NEW |