##// END OF EJS Templates
tests: Add new attribute `is_shadow_repo` to test dummy extra data.
Martin Bornhold -
r909:22de0b38 default
parent child Browse files
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