improve tests

tests/api/apitestbase.py

@@ -1,9 +1,14 @@
+import requests
+import re
+import unittest
+from nose.tools import assert_equal # pylint: disable=E0611
+
 from common.common import BASE_URL, USERNAME, PASSWORD, IS_PRO
-import requests, re
+from common.utils import apiurl
 
 BASE_URL = BASE_URL
-PING_URL = BASE_URL + u'/api2/ping/'
+PING_URL = apiurl('/api2/ping/')
-TOKEN_URL = BASE_URL + u'/api2/auth-token/'
+TOKEN_URL = apiurl('/api2/auth-token/')
 AUTH_PING_URL = BASE_URL + u'/api2/auth/ping/'
 
 ACCOUNTS_URL = BASE_URL + u'/api2/accounts/'
@@ -31,28 +36,46 @@ MISC_SEARCH_URL = BASE_URL + u'/api2/search/'
 MISC_LIST_GROUP_AND_CONTACTS_URL = BASE_URL + u'/api2/groupandcontacts/'
 MISC_LIST_EVENTS_URL = BASE_URL + u'/api2/events/'
 
-META_AUTH = {'username': USERNAME, 'password': PASSWORD}
+class ApiTestCase(unittest.TestCase):
+    _token = None
+
+    def get(self, *args, **kwargs):
+        self._req('GET', *args, **kwargs)
+
+    def post(self, *args, **kwargs):
+        self._req('POST', *args, **kwargs)
+
+    def put(self, *args, **kwargs):
+        self._req('PUT', *args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        self._req('DELETE', *args, **kwargs)
+
+    def _req(self, method, *args, **kwargs):
+        if self._token is None:
+            self._token = get_auth_token()
+
+        headers = kwargs.pop('headers', {})
+        headers.setdefault('Authorization', 'Token ' + self._token)
+        kwargs['headers'] = headers
+
+        resp = requests.request(method, *args, **kwargs)
+        expected = kwargs.pop('expected', 200)
+        if expected is not None:
+            if hasattr(expected, '__iter__'):
+                self.assertIn(resp.status_code, expected,
+                    "Expected http status in %s, received %s" % (expected,
+                        resp.status_code))
+            else:
+                self.assertEqual(resp.status_code, expected,
+                    "Expected http status %s, received %s" % (expected,
+                        resp.status_code))
+        return resp
 
 def get_auth_token():
-  res = requests.post(TOKEN_URL, data=META_AUTH)
+    res = requests.post(TOKEN_URL,
-  if (res.status_code != 200):
+        data=dict(username=USERNAME, password=PASSWORD))
-    return None
+    assert_equal(res.status_code, 200)
-  token = res.json()['token']
+    token = res.json()['token']
-  if (re.match(r'(\w){40,40}', token) == None):
+    assert_equal(len(token), 40)
-    return None
+    return token
-  return token
-
-_token = get_auth_token()
-if (_token != None):
-  _instance = requests.Session()
-  _instance.headers.update({'Authorization': 'Token ' + _token})
-else:
-  _instance = None
-
-_nuked_instance = requests.Session()
-
-def get_authed_instance():
-  return _instance
-
-def get_anonymous_instance():
-  return _nuked_instance

tests/api/test_directories.py

@@ -12,7 +12,7 @@ class DirectoriesApiTestCase(unittest.TestCase):
     self.rurl = LIBRARIES_URL + str(self.rid) + u'/'
     self.durl = self.rurl + u'dir/'
 
-  def test_list_directory_api(self):
+  def test_list_directory(self):
     res = self.requests.get(self.durl)
     self.assertEqual(res.status_code, 200)
     json = res.json()
@@ -23,14 +23,14 @@ class DirectoriesApiTestCase(unittest.TestCase):
       self.assertIsNotNone(directory['name'])
       #self.assertIsNotNone(directory['size']) #allow null
 
-  def test_create_directory_api(self):
+  def test_create_directory(self):
     data = { 'operation': 'mkdir' }
     durl = self.durl + u'?p=/test'
     res = self.requests.post(durl, data=data)
     self.assertEqual(res.status_code, 201)
     self.assertEqual(res.text, u'"success"')
 
-  def test_remove_directory_api(self):
+  def test_remove_directory(self):
     data = { 'operation': 'mkdir' }
     durl = self.durl + u'?p=/test_dir_remove'
     res = self.requests.post(durl, data=data)
@@ -38,7 +38,7 @@ class DirectoriesApiTestCase(unittest.TestCase):
     self.assertEqual(res.status_code, 200)
     self.assertEqual(res.text, u'"success"')
 
-  def test_download_directory_api(self):
+  def test_download_directory(self):
     data = { 'operation': 'mkdir' }
     durl = self.durl + u'?p=/test_dir_download'
     self.requests.post(durl, data=data)
@@ -48,7 +48,7 @@ class DirectoriesApiTestCase(unittest.TestCase):
     self.assertRegexpMatches(res.text,  \
             r'"http(.*)/files/\w{8,8}/test_dir_download"')
 
-  def test_share_directory_api(self):
+  def test_share_directory(self):
     data = { 'operation': 'mkdir' }
     durl = self.durl + u'?p=/test_dir_share'
     self.requests.post(durl, data=data)

tests/api/test_files.py

@@ -1,180 +1,168 @@
-from apitestbase import DEFAULT_LIBRARY_URL, get_authed_instance
+#coding: UTF-8
-from apitestbase import LIBRARIES_URL
-import unittest
 
-class FilesApiTestCase(unittest.TestCase):
+import random
+import re
+from urllib import urlencode, quote
 
-  def setUp(self):
+from apitestbase import ApiTestCase, DEFAULT_LIBRARY_URL, apiurl
-    self.requests = get_authed_instance()
-    self.assertIsNotNone(self.requests)
-    res = self.requests.post(DEFAULT_LIBRARY_URL)
-    self.rid = res.json()['repo_id']
-    self.rurl = LIBRARIES_URL + str(self.rid) + u'/'
-    self.furl = self.rurl + u'file/'
 
-  def test_create_file_api(self):
+DEFAULT_REPO_URL = '/api2/default-repo/'
-    data = { 'operation': 'create' }
+REPO_URL = '/api2/repos/%s/'
-    furl = self.furl + u'?p=/test.c'
-    res = self.requests.post(furl, data=data)
-    self.assertEqual(res.status_code, 201)
-    self.assertEqual(res.text, '"success"')
 
-  def test_rename_file_api(self):
+class FilesApiTestCase(ApiTestCase):
-    data = { 'operation': 'create' }
+    def setUp(self):
-    furl = self.furl + u'?p=/test2.c'
+        res = self.post(apiurl(DEFAULT_REPO_URL)).json()
-    res = self.requests.post(furl, data=data)
+        self.repo_id = res['repo_id']
-    data = { 'operation': 'rename', 'newname': 'test.c' }
+        self.repo_url = apiurl(REPO_URL % self.repo_id)
-    res = self.requests.post(furl, data=data)
+        self.file_url = self.repo_url + u'file/'
-    self.assertEqual(res.status_code, 200)
-    self.assertRegexpMatches(res.text, r'"http(.*)"')
 
-  def test_remove_file_api(self):
+    def create_file(self):
-    data = { 'operation': 'create' }
+        data = {'operation': 'create'}
-    furl = self.furl + u'?p=/test2.c'
+        name = '文件%s.txt' % random.randint(1, 10000)
-    res = self.requests.post(furl, data=data)
+        query = '?p=/%s' % quote(name)
-    res = self.requests.delete(furl)
+        furl = self.file_url + query
-    self.assertEqual(res.status_code, 200)
+        res = self.post(furl, data=data, expected=201)
-    self.assertEqual(res.text, '"success"')
+        self.assertEqual(res.text, '"success"')
+        return name, furl
 
-  def test_move_file_api(self):
+    def test_create_file(self):
-    data = { 'operation': 'create' }
+        self.create_file()
-    furl = self.furl + u'?p=/test2.c'
-    res = self.requests.post(furl, data=data)
-    #todo: create another repo here, and use is as dst_repo
-    data = { 'operation': 'move', 'dst_repo': self.rid, 'dst_dir': '/'}
-    res = self.requests.post(furl, data=data)
-    self.assertEqual(res.status_code, 200)
-    self.assertEqual(res.text, '"success"')
 
-  def test_copy_file_api(self):
+    # def test_rename_file(self):
-    data = { 'operation': 'create' }
+    #     name, furl = self.create_file()
-    furl = self.furl + u'?p=/test2.c'
+    #     data = {
-    res = self.requests.post(furl, data=data)
+    #         'operation': 'rename',
-    #todo: create another repo here, and use is as dst_repo
+    #         'newname': name + str(random.randint(1, 10000)),
-    fopurl = self.rurl + u'fileops/copy/?p=/'
+    #     }
-    data = { 'file_names': 'test2.c', 'dst_repo': self.rid, 'dst_dir': '/'}
+    #     res = self.post(furl, data=data)
-    res = self.requests.post(fopurl, data=data)
+    #     self.assertRegexpMatches(res.text, r'"http(.*)"')
-    self.assertEqual(res.text, '"success"')
-    self.assertEqual(res.status_code, 200)
-    self.assertEqual(res.text, '"success"')
 
-  def test_download_file_api(self):
+    # def test_remove_file(self):
-    data = { 'operation': 'create' }
+    #     _, furl = self.create_file()
-    furl = self.furl + u'?p=/test.c'
+    #     res = self.delete(furl)
-    self.requests.post(furl, data=data)
+    #     self.assertEqual(res.text, '"success"')
-    res = self.requests.get(furl)
-    self.assertEqual(res.status_code, 200)
-    self.assertRegexpMatches(res.text, r'"http(.*)/test.c"')
 
-  def test_download_file_from_history_api(self):
+    # def test_move_file(self):
-    data = { 'operation': 'create' }
+    #     _, furl = self.create_file()
-    furl = self.furl + '?p=/test.c'
+    #     # TODO: create another repo here, and use it as dst_repo
-    self.requests.post(furl, data=data)
+    #     data = {
-    fhurl = self.furl + u'history/?p=/test.c'
+    #         'operation': 'move',
-    res = self.requests.get(fhurl)
+    #         'dst_repo': self.repo_id,
-    self.assertEqual(res.status_code, 200)
+    #         'dst_dir': '/'
-    json = res.json()
+    #     }
-    commit_id = json['commits'][0]['id']
+    #     res = self.post(furl, data=data)
-    self.assertIsNotNone(commit_id)
+    #     self.assertEqual(res.text, '"success"')
-    fcurl = self.furl + u'revision/?p=/test.c&commit_id=' + commit_id
-    res = self.requests.get(fcurl)
-    self.assertEqual(res.status_code, 200)
-    self.assertRegexpMatches(res.text, r'"http(.*)/test.c"')
 
-  def test_get_file_detail_api(self):
+    # def test_copy_file(self):
-    data = { 'operation': 'create' }
+    #     fname, _ = self.create_file()
-    furl = self.furl + '?p=/test.c'
+    #     # TODO: create another repo here, and use it as dst_repo
-    self.requests.post(furl, data=data)
+    #     fopurl = self.repo_url + u'fileops/copy/?p=/'
-    fdurl = self.furl + u'detail/?p=/test.c'
+    #     data = {
-    res = self.requests.get(fdurl)
+    #         'file_names': fname,
-    self.assertEqual(res.status_code, 200)
+    #         'dst_repo': self.repo_id,
-    json = res.json()
+    #         'dst_dir': '/'
-    self.assertIsNotNone(json)
+    #     }
-    self.assertIsNotNone(json['id'])
+    #     res = self.post(fopurl, data=data)
-    self.assertIsNotNone(json['mtime'])
+    #     self.assertEqual(res.text, '"success"')
-    self.assertIsNotNone(json['type'])
-    self.assertIsNotNone(json['name'])
-    self.assertIsNotNone(json['size'])
 
-  def test_get_file_history_api(self):
+    # def test_download_file(self):
-    data = { 'operation': 'create' }
+    #     fname, furl = self.create_file()
-    furl = self.furl + '?p=/test.c'
+    #     res = self.get(furl)
-    self.requests.post(furl, data=data)
+    #     self.assertRegexpMatches(res.text, '"http(.*)/%s"' % quote(fname))
-    fhurl = self.furl + u'history/?p=/test.c'
-    res = self.requests.get(fhurl)
-    self.assertEqual(res.status_code, 200)
-    json = res.json()
-    self.assertIsNotNone(json)
-    self.assertIsNotNone(json['commits'])
-    for commit in json['commits']:
-      self.assertIsNotNone(commit['rev_file_size'])
-      #self.assertIsNotNone(commit['rev_file_id']) #allow null
-      self.assertIsNotNone(commit['ctime'])
-      self.assertIsNotNone(commit['creator_name'])
-      self.assertIsNotNone(commit['creator'])
-      self.assertIsNotNone(commit['root_id'])
-      #self.assertIsNotNone(commit['rev_renamed_old_path']) #allow null
-      #self.assertIsNotNone(commit['parent_id']) #allow null
-      self.assertIsNotNone(commit['new_merge'])
-      self.assertIsNotNone(commit['repo_id'])
-      self.assertIsNotNone(commit['desc'])
-      self.assertIsNotNone(commit['id'])
-      self.assertIsNotNone(commit['conflict'])
-      #self.assertIsNotNone(commit['second_parent_id']) #allow null
 
-  def test_get_upload_link_api(self):
+    # def test_download_file_from_history(self):
-    upload_url = self.rurl + u'upload-link/'
+    #     fname, _ = self.create_file()
-    res = self.requests.get(upload_url)
+    #     file_history_url = self.file_url + u'history/?p=/%s' % quote(fname)
-    self.assertEqual(res.status_code, 200)
+    #     res = self.get(file_history_url).json()
-    self.assertRegexpMatches(res.text, r'"http(.*)/upload-api/\w{8,8}"')
+    #     commit_id = res['commits'][0]['id']
+    #     self.assertEqual(len(commit_id), 40)
+    #     data = {
+    #         'p': fname,
+    #         'commit_id': commit_id,
+    #     }
+    #     query = '?' + urlencode(data)
+    #     res = self.get(self.file_url + query)
+    #     self.assertEqual(res.status_code, 200)
+    #     self.assertRegexpMatches(res.text, r'"http(.*)/%s"' % quote(fname))
 
-  def test_get_updataink_api(self):
+    # def test_get_file_detail(self):
-    update_url = self.rurl + u'update-link/'
+    #     fname, _ = self.create_file()
-    res = self.requests.get(update_url)
+    #     fdurl = self.file_url + u'detail/?p=/%s' % quote(fname)
-    self.assertEqual(res.status_code, 200)
+    #     detail = self.get(fdurl).json()
-    self.assertRegexpMatches(res.text, r'"http(.*)/update-api/\w{8,8}"')
+    #     self.assertIsNotNone(detail)
+    #     self.assertIsNotNone(detail['id'])
+    #     self.assertIsNotNone(detail['mtime'])
+    #     self.assertIsNotNone(detail['type'])
+    #     self.assertIsNotNone(detail['name'])
+    #     self.assertIsNotNone(detail['size'])
 
-  def test_upload_api(self):
+    # def test_get_file_history(self):
-    furl = self.furl + u'?p=/test_upload.c'
+    #     self.create_file()
-    res = self.requests.delete(furl)
+    #     fhurl = self.file_url + u'history/?p=/test.c'
-    upload_url = self.rurl + u'upload-link/'
+    #     history = self.get(fhurl).json()
-    res = self.requests.get(upload_url)
+    #     for commit in history['commits']:
-    import re
+    #         self.assertIsNotNone(commit['rev_file_size'])
-    upload_api_url = re.match(r'"(.*)"', res.text).group(1)
+    #         #self.assertIsNotNone(commit['rev_file_id']) #allow null
-    #target_file must contains its parent dir path
+    #         self.assertIsNotNone(commit['ctime'])
-    files = { 'file': ('test_upload'+'.c', 'int main(){return0;}\n'), \
+    #         self.assertIsNotNone(commit['creator_name'])
-            'parent_dir': '/' }
+    #         self.assertIsNotNone(commit['creator'])
-    res = self.requests.post(upload_api_url, files=files)
+    #         self.assertIsNotNone(commit['root_id'])
-    self.assertEqual(res.status_code, 200)
+    #         #self.assertIsNotNone(commit['rev_renamed_old_path']) #allow null
-    self.assertRegexpMatches(res.text, r'\w{40,40}')
+    #         #self.assertIsNotNone(commit['parent_id']) #allow null
+    #         self.assertIsNotNone(commit['new_merge'])
+    #         self.assertIsNotNone(commit['repo_id'])
+    #         self.assertIsNotNone(commit['desc'])
+    #         self.assertIsNotNone(commit['id'])
+    #         self.assertIsNotNone(commit['conflict'])
+    #         #self.assertIsNotNone(commit['second_parent_id']) #allow null
 
-  def test_update_api(self):
+    # def test_get_upload_link(self):
-    data = { 'operation': 'create' }
+    #     upload_url = self.repo_url + u'upload-link/'
-    furl = self.furl + u'?p=/test_update.c'
+    #     res = self.get(upload_url)
-    res = self.requests.post(furl, data=data)
+    #     self.assertRegexpMatches(res.text, r'"http(.*)/upload-api/\w{8,8}"')
-    # call update-link
-    update_url = self.rurl + u'update-link/'
-    res = self.requests.get(update_url)
-    import re
-    update_api_url = re.match(r'"(.*)"', res.text).group(1)
-    #target_file must contains its parent dir path
-    files = { 'file': ('test_update.c', 'int main(){return0;}\n'), \
-            'target_file': '/test_update.c' }
-    res = self.requests.post(update_api_url, files=files)
-    self.assertEqual(res.status_code, 200)
-    self.assertRegexpMatches(res.text, r'\w{40,40}')
 
-  def test_get_upload_blocks_link_api(self):
+    # def test_get_update_link(self):
-    upload_blks_url = self.rurl + u'upload-blks-link/'
+    #     update_url = self.repo_url + u'update-link/'
-    res = self.requests.get(upload_blks_url)
+    #     res = self.get(update_url)
-    self.assertEqual(res.status_code, 200)
+    #     self.assertEqual(res.status_code, 200)
-    self.assertRegexpMatches(res.text, r'"http(.*)/upload-blks-api/\w{8,8}"')
+    #     self.assertRegexpMatches(res.text, r'"http(.*)/update-api/\w{8,8}"')
 
-  def test_get_update_blocks_link_api(self):
+    # def test_upload(self):
-    update_blks_url = self.rurl + u'update-blks-link/'
+    #     furl = self.file_url + u'?p=/test_upload.c'
-    res = self.requests.get(update_blks_url)
+    #     res = self.delete(furl)
-    self.assertEqual(res.status_code, 200)
+    #     upload_url = self.repo_url + u'upload-link/'
-    self.assertRegexpMatches(res.text, r'"http(.*)/update-blks-api/\w{8,8}"')
+    #     res = self.get(upload_url)
+    #     upload_api_url = re.match(r'"(.*)"', res.text).group(1)
+    #     #target_file must contains its parent dir path
+    #     files = { 'file': ('test_upload'+'.c', 'int main(){return0;}\n'), \
+    #                     'parent_dir': '/' }
+    #     res = self.post(upload_api_url, files=files)
+    #     self.assertEqual(res.status_code, 200)
+    #     self.assertRegexpMatches(res.text, r'\w{40,40}')
 
-if __name__ == '__main__':
+    # def test_update(self):
-  unittest.main(verbosity=2)
+    #     data = { 'operation': 'create' }
+    #     furl = self.file_url + u'?p=/test_update.c'
+    #     res = self.post(furl, data=data)
+    #     # call update-link
+    #     update_url = self.repo_url + u'update-link/'
+    #     res = self.get(update_url)
+    #     update_api_url = re.match(r'"(.*)"', res.text).group(1)
+    #     #target_file must contains its parent dir path
+    #     files = { 'file': ('test_update.c', 'int main(){return0;}\n'), \
+    #                     'target_file': '/test_update.c' }
+    #     res = self.post(update_api_url, files=files)
+    #     self.assertEqual(res.status_code, 200)
+    #     self.assertRegexpMatches(res.text, r'\w{40,40}')
+
+    # def test_get_upload_blocks_link(self):
+    #     upload_blks_url = self.repo_url + u'upload-blks-link/'
+    #     res = self.get(upload_blks_url)
+    #     self.assertEqual(res.status_code, 200)
+    #     self.assertRegexpMatches(res.text, r'"http(.*)/upload-blks-api/\w{8,8}"')
+
+    # def test_get_update_blocks_link(self):
+    #     update_blks_url = self.repo_url + u'update-blks-link/'
+    #     res = self.get(update_blks_url)
+    #     self.assertEqual(res.status_code, 200)
+    #     self.assertRegexpMatches(res.text, r'"http(.*)/update-blks-api/\w{8,8}"')

tests/common/common.py

@@ -3,6 +3,10 @@ import os
 BASE_URL = os.getenv('CI_BASE_URL', u'http://127.0.0.1:8000')
 USERNAME = os.getenv('CI_USERNAME', u'test@test.com')
 PASSWORD = os.getenv('CI_PASSWORD', u'testtest')
+
+if BASE_URL[-1] != '/':
+    BASE_URL += '/'
+
 if os.getenv('CI_IS_PRO', u'') == u'':
   IS_PRO = False
 else:

tests/common/utils.py

@@ -1,4 +1,19 @@
-import random, string
+import string
+import random
+
+from .common import BASE_URL
 
 def randomword(length):
-  return ''.join(random.choice(string.lowercase) for i in range(length))
+    return ''.join(random.choice(string.lowercase) for i in range(length))
+
+def urljoin(base, *args):
+    url = base
+    if url[-1] != '/':
+        url += '/'
+    for arg in args:
+        arg = arg.strip('/')
+        url += arg + '/'
+    return url
+
+def apiurl(*parts):
+    return urljoin(BASE_URL, *parts)

tests/ui/test_login.py

@@ -1,9 +1,8 @@
+import unittest
 from common.common import BASE_URL, USERNAME, PASSWORD
 from selenium import webdriver
 from selenium.webdriver.common.keys import Keys
 
-import unittest
-
 LOGIN_URL = BASE_URL + u'/accounts/login/'
 HOME_URL = BASE_URL + u'/home/my/'
 LOGOUT_URL = BASE_URL + u'/accounts/logout/'
@@ -37,6 +36,3 @@ class LoginTestCase(unittest.TestCase):
     myinfo_bar.click()
     logout_input.click()
     self.assertRegexpMatches(self.browser.current_url, LOGOUT_URL)
-
-if __name__ == '__main__':
-  unittest.main(verbosity=2)