Show More
@@ -1,116 +1,117 b'' | |||
|
1 | 1 | # -*- coding: utf-8 -*- |
|
2 | 2 | |
|
3 | 3 | # Copyright (C) 2010-2016 RhodeCode GmbH |
|
4 | 4 | # |
|
5 | 5 | # This program is free software: you can redistribute it and/or modify |
|
6 | 6 | # it under the terms of the GNU Affero General Public License, version 3 |
|
7 | 7 | # (only), as published by the Free Software Foundation. |
|
8 | 8 | # |
|
9 | 9 | # This program is distributed in the hope that it will be useful, |
|
10 | 10 | # but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 | 11 | # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 | 12 | # GNU General Public License for more details. |
|
13 | 13 | # |
|
14 | 14 | # You should have received a copy of the GNU Affero General Public License |
|
15 | 15 | # along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | 16 | # |
|
17 | 17 | # This program is dual-licensed. If you wish to learn more about the |
|
18 | 18 | # RhodeCode Enterprise Edition, including its added features, Support services, |
|
19 | 19 | # and proprietary license terms, please see https://rhodecode.com/licenses/ |
|
20 | 20 | |
|
21 | 21 | import pytest |
|
22 | 22 | |
|
23 | 23 | from rhodecode.tests.events.conftest import EventCatcher |
|
24 | 24 | |
|
25 | 25 | from rhodecode.lib import hooks_base, utils2 |
|
26 | 26 | from rhodecode.model.repo import RepoModel |
|
27 | 27 | from rhodecode.events.repo import ( |
|
28 | 28 | RepoPrePullEvent, RepoPullEvent, |
|
29 | 29 | RepoPrePushEvent, RepoPushEvent, |
|
30 | 30 | RepoPreCreateEvent, RepoCreateEvent, |
|
31 | 31 | RepoPreDeleteEvent, RepoDeleteEvent, |
|
32 | 32 | ) |
|
33 | 33 | |
|
34 | 34 | |
|
35 | 35 | @pytest.fixture |
|
36 | 36 | def scm_extras(user_regular, repo_stub): |
|
37 | 37 | extras = utils2.AttributeDict({ |
|
38 | 38 | 'ip': '127.0.0.1', |
|
39 | 39 | 'username': user_regular.username, |
|
40 | 40 | 'action': '', |
|
41 | 41 | 'repository': repo_stub.repo_name, |
|
42 | 42 | 'scm': repo_stub.scm_instance().alias, |
|
43 | 43 | 'config': '', |
|
44 | 44 | 'server_url': 'http://example.com', |
|
45 | 45 | 'make_lock': None, |
|
46 | 46 | 'locked_by': [None], |
|
47 | 47 | 'commit_ids': ['a' * 40] * 3, |
|
48 | 'is_shadow_repo': False, | |
|
48 | 49 | }) |
|
49 | 50 | return extras |
|
50 | 51 | |
|
51 | 52 | |
|
52 | 53 | # TODO: dan: make the serialization tests complete json comparisons |
|
53 | 54 | @pytest.mark.parametrize('EventClass', [ |
|
54 | 55 | RepoPreCreateEvent, RepoCreateEvent, |
|
55 | 56 | RepoPreDeleteEvent, RepoDeleteEvent, |
|
56 | 57 | ]) |
|
57 | 58 | def test_repo_events_serialized(repo_stub, EventClass): |
|
58 | 59 | event = EventClass(repo_stub) |
|
59 | 60 | data = event.as_dict() |
|
60 | 61 | assert data['name'] == EventClass.name |
|
61 | 62 | assert data['repo']['repo_name'] == repo_stub.repo_name |
|
62 | 63 | assert data['repo']['url'] |
|
63 | 64 | |
|
64 | 65 | |
|
65 | 66 | @pytest.mark.parametrize('EventClass', [ |
|
66 | 67 | RepoPrePullEvent, RepoPullEvent, RepoPrePushEvent |
|
67 | 68 | ]) |
|
68 | 69 | def test_vcs_repo_events_serialize(repo_stub, scm_extras, EventClass): |
|
69 | 70 | event = EventClass(repo_name=repo_stub.repo_name, extras=scm_extras) |
|
70 | 71 | data = event.as_dict() |
|
71 | 72 | assert data['name'] == EventClass.name |
|
72 | 73 | assert data['repo']['repo_name'] == repo_stub.repo_name |
|
73 | 74 | assert data['repo']['url'] |
|
74 | 75 | |
|
75 | 76 | |
|
76 | 77 | |
|
77 | 78 | @pytest.mark.parametrize('EventClass', [RepoPushEvent]) |
|
78 | 79 | def test_vcs_repo_push_event_serialize(repo_stub, scm_extras, EventClass): |
|
79 | 80 | event = EventClass(repo_name=repo_stub.repo_name, |
|
80 | 81 | pushed_commit_ids=scm_extras['commit_ids'], |
|
81 | 82 | extras=scm_extras) |
|
82 | 83 | data = event.as_dict() |
|
83 | 84 | assert data['name'] == EventClass.name |
|
84 | 85 | assert data['repo']['repo_name'] == repo_stub.repo_name |
|
85 | 86 | assert data['repo']['url'] |
|
86 | 87 | |
|
87 | 88 | |
|
88 | 89 | def test_create_delete_repo_fires_events(backend): |
|
89 | 90 | with EventCatcher() as event_catcher: |
|
90 | 91 | repo = backend.create_repo() |
|
91 | 92 | assert event_catcher.events_types == [RepoPreCreateEvent, RepoCreateEvent] |
|
92 | 93 | |
|
93 | 94 | with EventCatcher() as event_catcher: |
|
94 | 95 | RepoModel().delete(repo) |
|
95 | 96 | assert event_catcher.events_types == [RepoPreDeleteEvent, RepoDeleteEvent] |
|
96 | 97 | |
|
97 | 98 | |
|
98 | 99 | def test_pull_fires_events(scm_extras): |
|
99 | 100 | with EventCatcher() as event_catcher: |
|
100 | 101 | hooks_base.pre_push(scm_extras) |
|
101 | 102 | assert event_catcher.events_types == [RepoPrePushEvent] |
|
102 | 103 | |
|
103 | 104 | with EventCatcher() as event_catcher: |
|
104 | 105 | hooks_base.post_push(scm_extras) |
|
105 | 106 | assert event_catcher.events_types == [RepoPushEvent] |
|
106 | 107 | |
|
107 | 108 | |
|
108 | 109 | def test_push_fires_events(scm_extras): |
|
109 | 110 | with EventCatcher() as event_catcher: |
|
110 | 111 | hooks_base.pre_pull(scm_extras) |
|
111 | 112 | assert event_catcher.events_types == [RepoPrePullEvent] |
|
112 | 113 | |
|
113 | 114 | with EventCatcher() as event_catcher: |
|
114 | 115 | hooks_base.post_pull(scm_extras) |
|
115 | 116 | assert event_catcher.events_types == [RepoPullEvent] |
|
116 | 117 |
@@ -1,310 +1,311 b'' | |||
|
1 | 1 | # -*- coding: utf-8 -*- |
|
2 | 2 | |
|
3 | 3 | # Copyright (C) 2010-2016 RhodeCode GmbH |
|
4 | 4 | # |
|
5 | 5 | # This program is free software: you can redistribute it and/or modify |
|
6 | 6 | # it under the terms of the GNU Affero General Public License, version 3 |
|
7 | 7 | # (only), as published by the Free Software Foundation. |
|
8 | 8 | # |
|
9 | 9 | # This program is distributed in the hope that it will be useful, |
|
10 | 10 | # but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 | 11 | # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 | 12 | # GNU General Public License for more details. |
|
13 | 13 | # |
|
14 | 14 | # You should have received a copy of the GNU Affero General Public License |
|
15 | 15 | # along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | 16 | # |
|
17 | 17 | # This program is dual-licensed. If you wish to learn more about the |
|
18 | 18 | # RhodeCode Enterprise Edition, including its added features, Support services, |
|
19 | 19 | # and proprietary license terms, please see https://rhodecode.com/licenses/ |
|
20 | 20 | |
|
21 | 21 | import base64 |
|
22 | 22 | |
|
23 | 23 | import mock |
|
24 | 24 | import pytest |
|
25 | 25 | import webtest.app |
|
26 | 26 | |
|
27 | 27 | from rhodecode.lib.caching_query import FromCache |
|
28 | 28 | from rhodecode.lib.hooks_daemon import DummyHooksCallbackDaemon |
|
29 | 29 | from rhodecode.lib.middleware import simplevcs |
|
30 | 30 | from rhodecode.lib.middleware.https_fixup import HttpsFixup |
|
31 | 31 | from rhodecode.lib.middleware.utils import scm_app |
|
32 | 32 | from rhodecode.model.db import User, _hash_key |
|
33 | 33 | from rhodecode.model.meta import Session |
|
34 | 34 | from rhodecode.tests import ( |
|
35 | 35 | HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS) |
|
36 | 36 | from rhodecode.tests.lib.middleware import mock_scm_app |
|
37 | 37 | from rhodecode.tests.utils import set_anonymous_access |
|
38 | 38 | |
|
39 | 39 | |
|
40 | 40 | class StubVCSController(simplevcs.SimpleVCS): |
|
41 | 41 | |
|
42 | 42 | SCM = 'hg' |
|
43 | 43 | stub_response_body = tuple() |
|
44 | 44 | |
|
45 | 45 | def __init__(self, *args, **kwargs): |
|
46 | 46 | super(StubVCSController, self).__init__(*args, **kwargs) |
|
47 | 47 | self.acl_repo_name = HG_REPO |
|
48 | 48 | self.url_repo_name = HG_REPO |
|
49 | 49 | self.vcs_repo_name = HG_REPO |
|
50 | self.is_shadow_repo = False | |
|
50 | 51 | |
|
51 | 52 | def _get_repository_name(self, environ): |
|
52 | 53 | return HG_REPO |
|
53 | 54 | |
|
54 | 55 | def _get_action(self, environ): |
|
55 | 56 | return "pull" |
|
56 | 57 | |
|
57 | 58 | def _create_wsgi_app(self, repo_path, repo_name, config): |
|
58 | 59 | def fake_app(environ, start_response): |
|
59 | 60 | start_response('200 OK', []) |
|
60 | 61 | return self.stub_response_body |
|
61 | 62 | return fake_app |
|
62 | 63 | |
|
63 | 64 | def _create_config(self, extras, repo_name): |
|
64 | 65 | return None |
|
65 | 66 | |
|
66 | 67 | |
|
67 | 68 | @pytest.fixture |
|
68 | 69 | def vcscontroller(pylonsapp, config_stub): |
|
69 | 70 | config_stub.testing_securitypolicy() |
|
70 | 71 | config_stub.include('rhodecode.authentication') |
|
71 | 72 | |
|
72 | 73 | set_anonymous_access(True) |
|
73 | 74 | controller = StubVCSController(pylonsapp, pylonsapp.config, None) |
|
74 | 75 | app = HttpsFixup(controller, pylonsapp.config) |
|
75 | 76 | app = webtest.app.TestApp(app) |
|
76 | 77 | |
|
77 | 78 | _remove_default_user_from_query_cache() |
|
78 | 79 | |
|
79 | 80 | # Sanity checks that things are set up correctly |
|
80 | 81 | app.get('/' + HG_REPO, status=200) |
|
81 | 82 | |
|
82 | 83 | app.controller = controller |
|
83 | 84 | return app |
|
84 | 85 | |
|
85 | 86 | |
|
86 | 87 | def _remove_default_user_from_query_cache(): |
|
87 | 88 | user = User.get_default_user(cache=True) |
|
88 | 89 | query = Session().query(User).filter(User.username == user.username) |
|
89 | 90 | query = query.options(FromCache( |
|
90 | 91 | "sql_cache_short", "get_user_%s" % _hash_key(user.username))) |
|
91 | 92 | query.invalidate() |
|
92 | 93 | Session().expire(user) |
|
93 | 94 | |
|
94 | 95 | |
|
95 | 96 | @pytest.fixture |
|
96 | 97 | def disable_anonymous_user(request, pylonsapp): |
|
97 | 98 | set_anonymous_access(False) |
|
98 | 99 | |
|
99 | 100 | @request.addfinalizer |
|
100 | 101 | def cleanup(): |
|
101 | 102 | set_anonymous_access(True) |
|
102 | 103 | |
|
103 | 104 | |
|
104 | 105 | def test_handles_exceptions_during_permissions_checks( |
|
105 | 106 | vcscontroller, disable_anonymous_user): |
|
106 | 107 | user_and_pass = '%s:%s' % (TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS) |
|
107 | 108 | auth_password = base64.encodestring(user_and_pass).strip() |
|
108 | 109 | extra_environ = { |
|
109 | 110 | 'AUTH_TYPE': 'Basic', |
|
110 | 111 | 'HTTP_AUTHORIZATION': 'Basic %s' % auth_password, |
|
111 | 112 | 'REMOTE_USER': TEST_USER_ADMIN_LOGIN, |
|
112 | 113 | } |
|
113 | 114 | |
|
114 | 115 | # Verify that things are hooked up correctly |
|
115 | 116 | vcscontroller.get('/', status=200, extra_environ=extra_environ) |
|
116 | 117 | |
|
117 | 118 | # Simulate trouble during permission checks |
|
118 | 119 | with mock.patch('rhodecode.model.db.User.get_by_username', |
|
119 | 120 | side_effect=Exception) as get_user: |
|
120 | 121 | # Verify that a correct 500 is returned and check that the expected |
|
121 | 122 | # code path was hit. |
|
122 | 123 | vcscontroller.get('/', status=500, extra_environ=extra_environ) |
|
123 | 124 | assert get_user.called |
|
124 | 125 | |
|
125 | 126 | |
|
126 | 127 | def test_returns_forbidden_if_no_anonymous_access( |
|
127 | 128 | vcscontroller, disable_anonymous_user): |
|
128 | 129 | vcscontroller.get('/', status=401) |
|
129 | 130 | |
|
130 | 131 | |
|
131 | 132 | class StubFailVCSController(simplevcs.SimpleVCS): |
|
132 | 133 | def _handle_request(self, environ, start_response): |
|
133 | 134 | raise Exception("BOOM") |
|
134 | 135 | |
|
135 | 136 | |
|
136 | 137 | @pytest.fixture(scope='module') |
|
137 | 138 | def fail_controller(pylonsapp): |
|
138 | 139 | controller = StubFailVCSController(pylonsapp, pylonsapp.config, None) |
|
139 | 140 | controller = HttpsFixup(controller, pylonsapp.config) |
|
140 | 141 | controller = webtest.app.TestApp(controller) |
|
141 | 142 | return controller |
|
142 | 143 | |
|
143 | 144 | |
|
144 | 145 | def test_handles_exceptions_as_internal_server_error(fail_controller): |
|
145 | 146 | fail_controller.get('/', status=500) |
|
146 | 147 | |
|
147 | 148 | |
|
148 | 149 | def test_provides_traceback_for_appenlight(fail_controller): |
|
149 | 150 | response = fail_controller.get( |
|
150 | 151 | '/', status=500, extra_environ={'appenlight.client': 'fake'}) |
|
151 | 152 | assert 'appenlight.__traceback' in response.request.environ |
|
152 | 153 | |
|
153 | 154 | |
|
154 | 155 | def test_provides_utils_scm_app_as_scm_app_by_default(pylonsapp): |
|
155 | 156 | controller = StubVCSController(pylonsapp, pylonsapp.config, None) |
|
156 | 157 | assert controller.scm_app is scm_app |
|
157 | 158 | |
|
158 | 159 | |
|
159 | 160 | def test_allows_to_override_scm_app_via_config(pylonsapp): |
|
160 | 161 | config = pylonsapp.config.copy() |
|
161 | 162 | config['vcs.scm_app_implementation'] = ( |
|
162 | 163 | 'rhodecode.tests.lib.middleware.mock_scm_app') |
|
163 | 164 | controller = StubVCSController(pylonsapp, config, None) |
|
164 | 165 | assert controller.scm_app is mock_scm_app |
|
165 | 166 | |
|
166 | 167 | |
|
167 | 168 | @pytest.mark.parametrize('query_string, expected', [ |
|
168 | 169 | ('cmd=stub_command', True), |
|
169 | 170 | ('cmd=listkeys', False), |
|
170 | 171 | ]) |
|
171 | 172 | def test_should_check_locking(query_string, expected): |
|
172 | 173 | result = simplevcs._should_check_locking(query_string) |
|
173 | 174 | assert result == expected |
|
174 | 175 | |
|
175 | 176 | |
|
176 | 177 | @mock.patch.multiple( |
|
177 | 178 | 'Pyro4.config', SERVERTYPE='multiplex', POLLTIMEOUT=0.01) |
|
178 | 179 | class TestGenerateVcsResponse: |
|
179 | 180 | |
|
180 | 181 | def test_ensures_that_start_response_is_called_early_enough(self): |
|
181 | 182 | self.call_controller_with_response_body(iter(['a', 'b'])) |
|
182 | 183 | assert self.start_response.called |
|
183 | 184 | |
|
184 | 185 | def test_invalidates_cache_after_body_is_consumed(self): |
|
185 | 186 | result = self.call_controller_with_response_body(iter(['a', 'b'])) |
|
186 | 187 | assert not self.was_cache_invalidated() |
|
187 | 188 | # Consume the result |
|
188 | 189 | list(result) |
|
189 | 190 | assert self.was_cache_invalidated() |
|
190 | 191 | |
|
191 | 192 | @mock.patch('rhodecode.lib.middleware.simplevcs.HTTPLockedRC') |
|
192 | 193 | def test_handles_locking_exception(self, http_locked_rc): |
|
193 | 194 | result = self.call_controller_with_response_body( |
|
194 | 195 | self.raise_result_iter(vcs_kind='repo_locked')) |
|
195 | 196 | assert not http_locked_rc.called |
|
196 | 197 | # Consume the result |
|
197 | 198 | list(result) |
|
198 | 199 | assert http_locked_rc.called |
|
199 | 200 | |
|
200 | 201 | @mock.patch('rhodecode.lib.middleware.simplevcs.HTTPRequirementError') |
|
201 | 202 | def test_handles_requirement_exception(self, http_requirement): |
|
202 | 203 | result = self.call_controller_with_response_body( |
|
203 | 204 | self.raise_result_iter(vcs_kind='requirement')) |
|
204 | 205 | assert not http_requirement.called |
|
205 | 206 | # Consume the result |
|
206 | 207 | list(result) |
|
207 | 208 | assert http_requirement.called |
|
208 | 209 | |
|
209 | 210 | @mock.patch('rhodecode.lib.middleware.simplevcs.HTTPLockedRC') |
|
210 | 211 | def test_handles_locking_exception_in_app_call(self, http_locked_rc): |
|
211 | 212 | app_factory_patcher = mock.patch.object( |
|
212 | 213 | StubVCSController, '_create_wsgi_app') |
|
213 | 214 | with app_factory_patcher as app_factory: |
|
214 | 215 | app_factory().side_effect = self.vcs_exception() |
|
215 | 216 | result = self.call_controller_with_response_body(['a']) |
|
216 | 217 | list(result) |
|
217 | 218 | assert http_locked_rc.called |
|
218 | 219 | |
|
219 | 220 | def test_raises_unknown_exceptions(self): |
|
220 | 221 | result = self.call_controller_with_response_body( |
|
221 | 222 | self.raise_result_iter(vcs_kind='unknown')) |
|
222 | 223 | with pytest.raises(Exception): |
|
223 | 224 | list(result) |
|
224 | 225 | |
|
225 | 226 | def test_prepare_callback_daemon_is_called(self): |
|
226 | 227 | def side_effect(extras): |
|
227 | 228 | return DummyHooksCallbackDaemon(), extras |
|
228 | 229 | |
|
229 | 230 | prepare_patcher = mock.patch.object( |
|
230 | 231 | StubVCSController, '_prepare_callback_daemon') |
|
231 | 232 | with prepare_patcher as prepare_mock: |
|
232 | 233 | prepare_mock.side_effect = side_effect |
|
233 | 234 | self.call_controller_with_response_body(iter(['a', 'b'])) |
|
234 | 235 | assert prepare_mock.called |
|
235 | 236 | assert prepare_mock.call_count == 1 |
|
236 | 237 | |
|
237 | 238 | def call_controller_with_response_body(self, response_body): |
|
238 | 239 | settings = { |
|
239 | 240 | 'base_path': 'fake_base_path', |
|
240 | 241 | 'vcs.hooks.protocol': 'http', |
|
241 | 242 | 'vcs.hooks.direct_calls': False, |
|
242 | 243 | } |
|
243 | 244 | controller = StubVCSController(None, settings, None) |
|
244 | 245 | controller._invalidate_cache = mock.Mock() |
|
245 | 246 | controller.stub_response_body = response_body |
|
246 | 247 | self.start_response = mock.Mock() |
|
247 | 248 | result = controller._generate_vcs_response( |
|
248 | 249 | environ={}, start_response=self.start_response, |
|
249 | 250 | repo_path='fake_repo_path', |
|
250 | 251 | repo_name='fake_repo_name', |
|
251 | 252 | extras={}, action='push') |
|
252 | 253 | self.controller = controller |
|
253 | 254 | return result |
|
254 | 255 | |
|
255 | 256 | def raise_result_iter(self, vcs_kind='repo_locked'): |
|
256 | 257 | """ |
|
257 | 258 | Simulates an exception due to a vcs raised exception if kind vcs_kind |
|
258 | 259 | """ |
|
259 | 260 | raise self.vcs_exception(vcs_kind=vcs_kind) |
|
260 | 261 | yield "never_reached" |
|
261 | 262 | |
|
262 | 263 | def vcs_exception(self, vcs_kind='repo_locked'): |
|
263 | 264 | locked_exception = Exception('TEST_MESSAGE') |
|
264 | 265 | locked_exception._vcs_kind = vcs_kind |
|
265 | 266 | return locked_exception |
|
266 | 267 | |
|
267 | 268 | def was_cache_invalidated(self): |
|
268 | 269 | return self.controller._invalidate_cache.called |
|
269 | 270 | |
|
270 | 271 | |
|
271 | 272 | class TestInitializeGenerator: |
|
272 | 273 | |
|
273 | 274 | def test_drains_first_element(self): |
|
274 | 275 | gen = self.factory(['__init__', 1, 2]) |
|
275 | 276 | result = list(gen) |
|
276 | 277 | assert result == [1, 2] |
|
277 | 278 | |
|
278 | 279 | @pytest.mark.parametrize('values', [ |
|
279 | 280 | [], |
|
280 | 281 | [1, 2], |
|
281 | 282 | ]) |
|
282 | 283 | def test_raises_value_error(self, values): |
|
283 | 284 | with pytest.raises(ValueError): |
|
284 | 285 | self.factory(values) |
|
285 | 286 | |
|
286 | 287 | @simplevcs.initialize_generator |
|
287 | 288 | def factory(self, iterable): |
|
288 | 289 | for elem in iterable: |
|
289 | 290 | yield elem |
|
290 | 291 | |
|
291 | 292 | |
|
292 | 293 | class TestPrepareHooksDaemon(object): |
|
293 | 294 | def test_calls_imported_prepare_callback_daemon(self, app_settings): |
|
294 | 295 | expected_extras = {'extra1': 'value1'} |
|
295 | 296 | daemon = DummyHooksCallbackDaemon() |
|
296 | 297 | |
|
297 | 298 | controller = StubVCSController(None, app_settings, None) |
|
298 | 299 | prepare_patcher = mock.patch.object( |
|
299 | 300 | simplevcs, 'prepare_callback_daemon', |
|
300 | 301 | return_value=(daemon, expected_extras)) |
|
301 | 302 | with prepare_patcher as prepare_mock: |
|
302 | 303 | callback_daemon, extras = controller._prepare_callback_daemon( |
|
303 | 304 | expected_extras.copy()) |
|
304 | 305 | prepare_mock.assert_called_once_with( |
|
305 | 306 | expected_extras, |
|
306 | 307 | protocol=app_settings['vcs.hooks.protocol'], |
|
307 | 308 | use_direct_calls=app_settings['vcs.hooks.direct_calls']) |
|
308 | 309 | |
|
309 | 310 | assert callback_daemon == daemon |
|
310 | 311 | assert extras == extras |
@@ -1,53 +1,54 b'' | |||
|
1 | 1 | # -*- coding: utf-8 -*- |
|
2 | 2 | |
|
3 | 3 | # Copyright (C) 2010-2016 RhodeCode GmbH |
|
4 | 4 | # |
|
5 | 5 | # This program is free software: you can redistribute it and/or modify |
|
6 | 6 | # it under the terms of the GNU Affero General Public License, version 3 |
|
7 | 7 | # (only), as published by the Free Software Foundation. |
|
8 | 8 | # |
|
9 | 9 | # This program is distributed in the hope that it will be useful, |
|
10 | 10 | # but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 | 11 | # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 | 12 | # GNU General Public License for more details. |
|
13 | 13 | # |
|
14 | 14 | # You should have received a copy of the GNU Affero General Public License |
|
15 | 15 | # along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | 16 | # |
|
17 | 17 | # This program is dual-licensed. If you wish to learn more about the |
|
18 | 18 | # RhodeCode Enterprise Edition, including its added features, Support services, |
|
19 | 19 | # and proprietary license terms, please see https://rhodecode.com/licenses/ |
|
20 | 20 | |
|
21 | 21 | import mock |
|
22 | 22 | |
|
23 | 23 | from rhodecode.lib import hooks_base, utils2 |
|
24 | 24 | |
|
25 | 25 | |
|
26 | 26 | @mock.patch.multiple( |
|
27 | 27 | hooks_base, |
|
28 | 28 | action_logger=mock.Mock(), |
|
29 | 29 | post_push_extension=mock.Mock(), |
|
30 | 30 | Repository=mock.Mock()) |
|
31 | 31 | def test_post_push_truncates_commits(user_regular, repo_stub): |
|
32 | 32 | extras = { |
|
33 | 33 | 'ip': '127.0.0.1', |
|
34 | 34 | 'username': user_regular.username, |
|
35 | 35 | 'action': 'push_local', |
|
36 | 36 | 'repository': repo_stub.repo_name, |
|
37 | 37 | 'scm': 'git', |
|
38 | 38 | 'config': '', |
|
39 | 39 | 'server_url': 'http://example.com', |
|
40 | 40 | 'make_lock': None, |
|
41 | 41 | 'locked_by': [None], |
|
42 | 42 | 'commit_ids': ['abcde12345' * 4] * 30000, |
|
43 | 'is_shadow_repo': False, | |
|
43 | 44 | } |
|
44 | 45 | extras = utils2.AttributeDict(extras) |
|
45 | 46 | |
|
46 | 47 | hooks_base.post_push(extras) |
|
47 | 48 | |
|
48 | 49 | # Calculate appropriate action string here |
|
49 | 50 | expected_action = 'push_local:%s' % ','.join(extras.commit_ids[:29000]) |
|
50 | 51 | |
|
51 | 52 | hooks_base.action_logger.assert_called_with( |
|
52 | 53 | extras.username, expected_action, extras.repository, extras.ip, |
|
53 | 54 | commit=True) |
General Comments 0
You need to be logged in to leave comments.
Login now