Mercurial > kallithea
comparison rhodecode/tests/scripts/test_concurency.py @ 2527:95624ce4465f beta
orginized test module
- separated test models into separate files
- added scripts for script tests
- few fixes and renames for readability
author | Marcin Kuzminski <marcin@python-works.com> |
---|---|
date | Sun, 01 Jul 2012 16:24:15 +0200 |
parents | rhodecode/tests/_test_concurency.py@89efedac4e6c |
children | f5dc0417d6b1 |
comparison
equal
deleted
inserted
replaced
2526:473794943022 | 2527:95624ce4465f |
---|---|
1 # -*- coding: utf-8 -*- | |
2 """ | |
3 rhodecode.tests.test_hg_operations | |
4 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
5 | |
6 Test suite for making push/pull operations | |
7 | |
8 :created_on: Dec 30, 2010 | |
9 :author: marcink | |
10 :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com> | |
11 :license: GPLv3, see COPYING for more details. | |
12 """ | |
13 # This program is free software: you can redistribute it and/or modify | |
14 # it under the terms of the GNU General Public License as published by | |
15 # the Free Software Foundation, either version 3 of the License, or | |
16 # (at your option) any later version. | |
17 # | |
18 # This program is distributed in the hope that it will be useful, | |
19 # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
20 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
21 # GNU General Public License for more details. | |
22 # | |
23 # You should have received a copy of the GNU General Public License | |
24 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
25 | |
26 import os | |
27 import sys | |
28 import shutil | |
29 import logging | |
30 from os.path import join as jn | |
31 from os.path import dirname as dn | |
32 | |
33 from tempfile import _RandomNameSequence | |
34 from subprocess import Popen, PIPE | |
35 | |
36 from paste.deploy import appconfig | |
37 from pylons import config | |
38 from sqlalchemy import engine_from_config | |
39 | |
40 from rhodecode.lib.utils import add_cache | |
41 from rhodecode.model import init_model | |
42 from rhodecode.model import meta | |
43 from rhodecode.model.db import User, Repository | |
44 from rhodecode.lib.auth import get_crypt_password | |
45 | |
46 from rhodecode.tests import TESTS_TMP_PATH, NEW_HG_REPO, HG_REPO | |
47 from rhodecode.config.environment import load_environment | |
48 | |
49 rel_path = dn(dn(dn(os.path.abspath(__file__)))) | |
50 conf = appconfig('config:development.ini', relative_to=rel_path) | |
51 load_environment(conf.global_conf, conf.local_conf) | |
52 | |
53 add_cache(conf) | |
54 | |
55 USER = 'test_admin' | |
56 PASS = 'test12' | |
57 HOST = 'hg.local' | |
58 METHOD = 'pull' | |
59 DEBUG = True | |
60 log = logging.getLogger(__name__) | |
61 | |
62 | |
63 class Command(object): | |
64 | |
65 def __init__(self, cwd): | |
66 self.cwd = cwd | |
67 | |
68 def execute(self, cmd, *args): | |
69 """Runs command on the system with given ``args``. | |
70 """ | |
71 | |
72 command = cmd + ' ' + ' '.join(args) | |
73 log.debug('Executing %s' % command) | |
74 if DEBUG: | |
75 print command | |
76 p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd) | |
77 stdout, stderr = p.communicate() | |
78 if DEBUG: | |
79 print stdout, stderr | |
80 return stdout, stderr | |
81 | |
82 | |
83 def get_session(): | |
84 engine = engine_from_config(conf, 'sqlalchemy.db1.') | |
85 init_model(engine) | |
86 sa = meta.Session | |
87 return sa | |
88 | |
89 | |
90 def create_test_user(force=True): | |
91 print 'creating test user' | |
92 sa = get_session() | |
93 | |
94 user = sa.query(User).filter(User.username == USER).scalar() | |
95 | |
96 if force and user is not None: | |
97 print 'removing current user' | |
98 for repo in sa.query(Repository).filter(Repository.user == user).all(): | |
99 sa.delete(repo) | |
100 sa.delete(user) | |
101 sa.commit() | |
102 | |
103 if user is None or force: | |
104 print 'creating new one' | |
105 new_usr = User() | |
106 new_usr.username = USER | |
107 new_usr.password = get_crypt_password(PASS) | |
108 new_usr.email = 'mail@mail.com' | |
109 new_usr.name = 'test' | |
110 new_usr.lastname = 'lasttestname' | |
111 new_usr.active = True | |
112 new_usr.admin = True | |
113 sa.add(new_usr) | |
114 sa.commit() | |
115 | |
116 print 'done' | |
117 | |
118 | |
119 def create_test_repo(force=True): | |
120 print 'creating test repo' | |
121 from rhodecode.model.repo import RepoModel | |
122 sa = get_session() | |
123 | |
124 user = sa.query(User).filter(User.username == USER).scalar() | |
125 if user is None: | |
126 raise Exception('user not found') | |
127 | |
128 repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar() | |
129 | |
130 if repo is None: | |
131 print 'repo not found creating' | |
132 | |
133 form_data = {'repo_name':HG_REPO, | |
134 'repo_type':'hg', | |
135 'private':False, | |
136 'clone_uri':'' } | |
137 rm = RepoModel(sa) | |
138 rm.base_path = '/home/hg' | |
139 rm.create(form_data, user) | |
140 | |
141 print 'done' | |
142 | |
143 | |
144 def set_anonymous_access(enable=True): | |
145 sa = get_session() | |
146 user = sa.query(User).filter(User.username == 'default').one() | |
147 user.active = enable | |
148 sa.add(user) | |
149 sa.commit() | |
150 | |
151 | |
152 def get_anonymous_access(): | |
153 sa = get_session() | |
154 return sa.query(User).filter(User.username == 'default').one().active | |
155 | |
156 | |
157 #============================================================================== | |
158 # TESTS | |
159 #============================================================================== | |
160 def test_clone_with_credentials(no_errors=False, repo=HG_REPO, method=METHOD, | |
161 seq=None): | |
162 cwd = path = jn(TESTS_TMP_PATH, repo) | |
163 | |
164 if seq == None: | |
165 seq = _RandomNameSequence().next() | |
166 | |
167 try: | |
168 shutil.rmtree(path, ignore_errors=True) | |
169 os.makedirs(path) | |
170 #print 'made dirs %s' % jn(path) | |
171 except OSError: | |
172 raise | |
173 | |
174 clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \ | |
175 {'user':USER, | |
176 'pass':PASS, | |
177 'host':HOST, | |
178 'cloned_repo':repo, } | |
179 | |
180 dest = path + seq | |
181 if method == 'pull': | |
182 stdout, stderr = Command(cwd).execute('hg', method, '--cwd', dest, clone_url) | |
183 else: | |
184 stdout, stderr = Command(cwd).execute('hg', method, clone_url, dest) | |
185 | |
186 if no_errors is False: | |
187 assert """adding file changes""" in stdout, 'no messages about cloning' | |
188 assert """abort""" not in stderr , 'got error from clone' | |
189 | |
190 if __name__ == '__main__': | |
191 try: | |
192 create_test_user(force=False) | |
193 seq = None | |
194 import time | |
195 | |
196 try: | |
197 METHOD = sys.argv[3] | |
198 except: | |
199 pass | |
200 | |
201 if METHOD == 'pull': | |
202 seq = _RandomNameSequence().next() | |
203 test_clone_with_credentials(repo=sys.argv[1], method='clone', | |
204 seq=seq) | |
205 s = time.time() | |
206 for i in range(1, int(sys.argv[2]) + 1): | |
207 print 'take', i | |
208 test_clone_with_credentials(repo=sys.argv[1], method=METHOD, | |
209 seq=seq) | |
210 print 'time taken %.3f' % (time.time() - s) | |
211 except Exception, e: | |
212 raise | |
213 sys.exit('stop on %s' % e) |