summaryrefslogtreecommitdiffstats
path: root/result_collection_api/tests/unit
diff options
context:
space:
mode:
authorSerenaFeng <feng.xiaowei@zte.com.cn>2016-06-01 15:36:17 +0800
committerSerenaFeng <feng.xiaowei@zte.com.cn>2016-06-01 16:21:46 +0800
commit59cba325d8132186d27aefa959d4bed730e4e68d (patch)
tree3407ccfc8696fde3572e7d9a5b0e8aa1918e9842 /result_collection_api/tests/unit
parentda342687570cd53f2bfb0e1d69db77b1ba664801 (diff)
project-ize testAPI
JIRA: FUNCTEST-284 Change-Id: I219e934bb11f50de84df2aa0345ecc7885223491 Signed-off-by: SerenaFeng <feng.xiaowei@zte.com.cn>
Diffstat (limited to 'result_collection_api/tests/unit')
-rw-r--r--result_collection_api/tests/unit/__init__.py1
-rw-r--r--result_collection_api/tests/unit/fake_pymongo.py144
-rw-r--r--result_collection_api/tests/unit/test_base.py145
-rw-r--r--result_collection_api/tests/unit/test_dashboard.py71
-rw-r--r--result_collection_api/tests/unit/test_fake_pymongo.py68
-rw-r--r--result_collection_api/tests/unit/test_pod.py81
-rw-r--r--result_collection_api/tests/unit/test_project.py132
-rw-r--r--result_collection_api/tests/unit/test_result.py267
-rw-r--r--result_collection_api/tests/unit/test_testcase.py183
-rw-r--r--result_collection_api/tests/unit/test_version.py24
10 files changed, 0 insertions, 1116 deletions
diff --git a/result_collection_api/tests/unit/__init__.py b/result_collection_api/tests/unit/__init__.py
deleted file mode 100644
index 3ed9fd0..0000000
--- a/result_collection_api/tests/unit/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-__author__ = 'root'
diff --git a/result_collection_api/tests/unit/fake_pymongo.py b/result_collection_api/tests/unit/fake_pymongo.py
deleted file mode 100644
index bebb9e8..0000000
--- a/result_collection_api/tests/unit/fake_pymongo.py
+++ /dev/null
@@ -1,144 +0,0 @@
-from bson.objectid import ObjectId
-from concurrent.futures import ThreadPoolExecutor
-
-
-__author__ = 'serena'
-
-
-def thread_execute(method, *args, **kwargs):
- with ThreadPoolExecutor(max_workers=2) as executor:
- result = executor.submit(method, *args, **kwargs)
- return result
-
-
-class MemCursor(object):
- def __init__(self, collection):
- self.collection = collection
- self.count = len(self.collection)
-
- def _is_next_exist(self):
- return self.count != 0
-
- @property
- def fetch_next(self):
- return thread_execute(self._is_next_exist)
-
- def next_object(self):
- self.count -= 1
- return self.collection.pop()
-
-
-class MemDb(object):
-
- def __init__(self):
- self.contents = []
- pass
-
- def _find_one(self, spec_or_id=None, *args):
- if spec_or_id is not None and not isinstance(spec_or_id, dict):
- spec_or_id = {"_id": spec_or_id}
- if '_id' in spec_or_id:
- spec_or_id['_id'] = str(spec_or_id['_id'])
- cursor = self._find(spec_or_id, *args)
- for result in cursor:
- return result
- return None
-
- def find_one(self, spec_or_id=None, *args):
- return thread_execute(self._find_one, spec_or_id, *args)
-
- def _insert(self, doc_or_docs, check_keys=True):
-
- docs = doc_or_docs
- return_one = False
- if isinstance(docs, dict):
- return_one = True
- docs = [docs]
-
- ids = []
- for doc in docs:
- if '_id' not in doc:
- doc['_id'] = str(ObjectId())
- if not check_keys or not self._find_one(doc['_id']):
- ids.append(doc['_id'])
- self.contents.append(doc_or_docs)
-
- if len(ids) == 0:
- return None
- if return_one:
- return ids[0]
- else:
- return ids
-
- def insert(self, doc_or_docs, check_keys=True):
- return thread_execute(self._insert, doc_or_docs, check_keys)
-
- @staticmethod
- def _compare_date(spec, value):
- for k, v in spec.iteritems():
- if k == '$gte' and value >= v:
- return True
- return False
-
- @staticmethod
- def _in(content, *args):
- for arg in args:
- for k, v in arg.iteritems():
- if k == 'start_date':
- if not MemDb._compare_date(v, content.get(k)):
- return False
- elif k == 'trust_indicator':
- if float(content.get(k)) != float(v):
- return False
- elif content.get(k, None) != v:
- return False
-
- return True
-
- def _find(self, *args):
- res = []
- for content in self.contents:
- if self._in(content, *args):
- res.append(content)
-
- return res
-
- def find(self, *args):
- return MemCursor(self._find(*args))
-
- def _update(self, spec, document):
- updated = False
- for index in range(len(self.contents)):
- content = self.contents[index]
- if self._in(content, spec):
- for k, v in document.iteritems():
- updated = True
- content[k] = v
- self.contents[index] = content
- return updated
-
- def update(self, spec, document):
- return thread_execute(self._update, spec, document)
-
- def _remove(self, spec_or_id=None):
- if spec_or_id is None:
- self.contents = []
- if not isinstance(spec_or_id, dict):
- spec_or_id = {'_id': spec_or_id}
- for index in range(len(self.contents)):
- content = self.contents[index]
- if self._in(content, spec_or_id):
- del self.contents[index]
- return True
- return False
-
- def remove(self, spec_or_id=None):
- return thread_execute(self._remove, spec_or_id)
-
- def clear(self):
- self._remove()
-
-pods = MemDb()
-projects = MemDb()
-testcases = MemDb()
-results = MemDb()
diff --git a/result_collection_api/tests/unit/test_base.py b/result_collection_api/tests/unit/test_base.py
deleted file mode 100644
index ef711ae..0000000
--- a/result_collection_api/tests/unit/test_base.py
+++ /dev/null
@@ -1,145 +0,0 @@
-import json
-
-from tornado.web import Application
-from tornado.testing import AsyncHTTPTestCase
-
-from resources.pod_handlers import PodCLHandler, PodGURHandler
-from resources.project_handlers import ProjectCLHandler, ProjectGURHandler
-from resources.handlers import VersionHandler
-from resources.testcase_handlers import TestcaseCLHandler, TestcaseGURHandler
-from resources.result_handlers import ResultsCLHandler, ResultsGURHandler
-from resources.dashboard_handlers import DashboardHandler
-from resources.models import CreateResponse
-import fake_pymongo
-
-
-class TestBase(AsyncHTTPTestCase):
- headers = {'Content-Type': 'application/json; charset=UTF-8'}
-
- def setUp(self):
- self.basePath = ''
- self.create_res = CreateResponse
- self.get_res = None
- self.list_res = None
- self.update_res = None
- self.req_d = None
- self.req_e = None
- self.addCleanup(self._clear)
- super(TestBase, self).setUp()
-
- def get_app(self):
- return Application(
- [
- (r"/versions", VersionHandler),
- (r"/api/v1/pods", PodCLHandler),
- (r"/api/v1/pods/([^/]+)", PodGURHandler),
- (r"/api/v1/projects", ProjectCLHandler),
- (r"/api/v1/projects/([^/]+)", ProjectGURHandler),
- (r"/api/v1/projects/([^/]+)/cases", TestcaseCLHandler),
- (r"/api/v1/projects/([^/]+)/cases/([^/]+)",
- TestcaseGURHandler),
- (r"/api/v1/results", ResultsCLHandler),
- (r"/api/v1/results/([^/]+)", ResultsGURHandler),
- (r"/dashboard/v1/results", DashboardHandler),
- ],
- db=fake_pymongo,
- debug=True,
- )
-
- def create_d(self, *args):
- return self.create(self.req_d, *args)
-
- def create_e(self, *args):
- return self.create(self.req_e, *args)
-
- def create(self, req=None, *args):
- return self.create_help(self.basePath, req, *args)
-
- def create_help(self, uri, req, *args):
- if req:
- req = req.format()
- res = self.fetch(self._update_uri(uri, *args),
- method='POST',
- body=json.dumps(req),
- headers=self.headers)
-
- return self._get_return(res, self.create_res)
-
- def get(self, *args):
- res = self.fetch(self._get_uri(*args),
- method='GET',
- headers=self.headers)
-
- def inner():
- new_args, num = self._get_valid_args(*args)
- return self.get_res \
- if num != self._need_arg_num(self.basePath) else self.list_res
- return self._get_return(res, inner())
-
- def query(self, query):
- res = self.fetch(self._get_query_uri(query),
- method='GET',
- headers=self.headers)
- return self._get_return(res, self.list_res)
-
- def update(self, new=None, *args):
- if new:
- new = new.format()
- res = self.fetch(self._get_uri(*args),
- method='PUT',
- body=json.dumps(new),
- headers=self.headers)
- return self._get_return(res, self.update_res)
-
- def delete(self, *args):
- res = self.fetch(self._get_uri(*args),
- method='DELETE',
- headers=self.headers)
- return res.code, res.body
-
- @staticmethod
- def _get_valid_args(*args):
- new_args = tuple(['%s' % arg for arg in args if arg is not None])
- return new_args, len(new_args)
-
- def _need_arg_num(self, uri):
- return uri.count('%s')
-
- def _get_query_uri(self, query):
- return self.basePath + '?' + query
-
- def _get_uri(self, *args):
- return self._update_uri(self.basePath, *args)
-
- def _update_uri(self, uri, *args):
- r_uri = uri
- new_args, num = self._get_valid_args(*args)
- if num != self._need_arg_num(uri):
- r_uri += '/%s'
-
- return r_uri % tuple(['%s' % arg for arg in new_args])
-
- def _get_return(self, res, cls):
- code = res.code
- body = res.body
- return code, self._get_return_body(code, body, cls)
-
- @staticmethod
- def _get_return_body(code, body, cls):
- return cls.from_dict(json.loads(body)) if code < 300 and cls else body
-
- def assert_href(self, body):
- self.assertIn(self.basePath, body.href)
-
- def assert_create_body(self, body, req=None, *args):
- if not req:
- req = self.req_d
- new_args = args + tuple([req.name])
- self.assertIn(self._get_uri(*new_args), body.href)
-
- @staticmethod
- def _clear():
- fake_pymongo.pods.clear()
- fake_pymongo.projects.clear()
- fake_pymongo.testcases.clear()
- fake_pymongo.results.clear()
diff --git a/result_collection_api/tests/unit/test_dashboard.py b/result_collection_api/tests/unit/test_dashboard.py
deleted file mode 100644
index 5c3f924..0000000
--- a/result_collection_api/tests/unit/test_dashboard.py
+++ /dev/null
@@ -1,71 +0,0 @@
-import unittest
-
-from test_result import TestResultBase
-from common.constants import HTTP_NOT_FOUND, HTTP_OK
-
-__author__ = '__serena__'
-
-
-class TestDashboardBase(TestResultBase):
- def setUp(self):
- super(TestDashboardBase, self).setUp()
- self.basePath = '/dashboard/v1/results'
- self.create_help('/api/v1/results', self.req_d)
- self.create_help('/api/v1/results', self.req_d)
- self.list_res = None
-
-
-class TestDashboardQuery(TestDashboardBase):
- def test_projectMissing(self):
- code, body = self.query(self._set_query(project='missing'))
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn('Project name missing', body)
-
- def test_projectNotReady(self):
- code, body = self.query(self._set_query(project='notReadyProject'))
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn('Project [notReadyProject] not dashboard ready', body)
-
- def test_testcaseMissing(self):
- code, body = self.query(self._set_query(case='missing'))
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn('Test case missing for project [{}]'
- .format(self.project),
- body)
-
- def test_testcaseNotReady(self):
- code, body = self.query(self._set_query(case='notReadyCase'))
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn(
- 'Test case [notReadyCase] not dashboard ready for project [%s]'
- % self.project,
- body)
-
- def test_success(self):
- code, body = self.query(self._set_query())
- self.assertEqual(code, HTTP_OK)
- self.assertIn('{"description": "vPing results for Dashboard"}', body)
-
- def test_caseIsStatus(self):
- code, body = self.query(self._set_query(case='status'))
- self.assertEqual(code, HTTP_OK)
- self.assertIn('{"description": "Functest status"}', body)
-
- def _set_query(self, project=None, case=None):
- uri = ''
- for k, v in list(locals().iteritems()):
- if k == 'self' or k == 'uri':
- continue
- if v is None:
- v = eval('self.' + k)
- if v != 'missing':
- uri += '{}={}&'.format(k, v)
- uri += 'pod={}&'.format(self.pod)
- uri += 'version={}&'.format(self.version)
- uri += 'installer={}&'.format(self.installer)
- uri += 'period={}&'.format(5)
- return uri[0:-1]
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/result_collection_api/tests/unit/test_fake_pymongo.py b/result_collection_api/tests/unit/test_fake_pymongo.py
deleted file mode 100644
index 6920fca..0000000
--- a/result_collection_api/tests/unit/test_fake_pymongo.py
+++ /dev/null
@@ -1,68 +0,0 @@
-import unittest
-from tornado.web import Application
-from tornado import gen
-from tornado.testing import AsyncHTTPTestCase, gen_test
-
-import fake_pymongo
-
-
-class MyTest(AsyncHTTPTestCase):
- def setUp(self):
- super(MyTest, self).setUp()
- self.db = fake_pymongo
- self.addCleanup(self._clear)
- self.io_loop.run_sync(self.fixture_setup)
-
- def get_app(self):
- return Application()
-
- @gen.coroutine
- def fixture_setup(self):
- self.test1 = {'_id': '1', 'name': 'test1'}
- self.test2 = {'name': 'test2'}
- yield self.db.pods.insert({'_id': '1', 'name': 'test1'})
- yield self.db.pods.insert({'name': 'test2'})
-
- @gen_test
- def test_find_one(self):
- user = yield self.db.pods.find_one({'name': 'test1'})
- self.assertEqual(user, self.test1)
- self.db.pods.remove()
-
- @gen_test
- def test_find(self):
- cursor = self.db.pods.find()
- names = []
- while (yield cursor.fetch_next):
- ob = cursor.next_object()
- names.append(ob.get('name'))
- self.assertItemsEqual(names, ['test1', 'test2'])
-
- @gen_test
- def test_update(self):
- yield self.db.pods.update({'_id': '1'}, {'name': 'new_test1'})
- user = yield self.db.pods.find_one({'_id': '1'})
- self.assertEqual(user.get('name', None), 'new_test1')
-
- @gen_test
- def test_remove(self):
- yield self.db.pods.remove({'_id': '1'})
- user = yield self.db.pods.find_one({'_id': '1'})
- self.assertIsNone(user)
-
- @gen_test
- def test_insert_check_keys(self):
- yield self.db.pods.insert({'_id': '1', 'name': 'test1'},
- check_keys=False)
- cursor = self.db.pods.find({'_id': '1'})
- names = []
- while (yield cursor.fetch_next):
- ob = cursor.next_object()
- names.append(ob.get('name'))
- self.assertItemsEqual(names, ['test1', 'test1'])
-
- def _clear(self):
- self.db.pods.clear()
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/result_collection_api/tests/unit/test_pod.py b/result_collection_api/tests/unit/test_pod.py
deleted file mode 100644
index 1a43c05..0000000
--- a/result_collection_api/tests/unit/test_pod.py
+++ /dev/null
@@ -1,81 +0,0 @@
-import unittest
-
-from test_base import TestBase
-from resources.pod_models import PodCreateRequest, Pod, Pods
-from common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_FORBIDDEN, HTTP_NOT_FOUND
-
-
-class TestPodBase(TestBase):
- def setUp(self):
- super(TestPodBase, self).setUp()
- self.req_d = PodCreateRequest('zte-1', 'virtual',
- 'zte pod 1', 'ci-pod')
- self.req_e = PodCreateRequest('zte-2', 'metal', 'zte pod 2')
- self.get_res = Pod
- self.list_res = Pods
- self.basePath = '/api/v1/pods'
-
- def assert_get_body(self, pod, req=None):
- if not req:
- req = self.req_d
- self.assertEqual(pod.name, req.name)
- self.assertEqual(pod.mode, req.mode)
- self.assertEqual(pod.details, req.details)
- self.assertEqual(pod.role, req.role)
- self.assertIsNotNone(pod.creation_date)
- self.assertIsNotNone(pod._id)
-
-
-class TestPodCreate(TestPodBase):
- def test_withoutBody(self):
- (code, body) = self.create()
- self.assertEqual(code, HTTP_BAD_REQUEST)
-
- def test_emptyName(self):
- req_empty = PodCreateRequest('')
- (code, body) = self.create(req_empty)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('name missing', body)
-
- def test_noneName(self):
- req_none = PodCreateRequest(None)
- (code, body) = self.create(req_none)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('name missing', body)
-
- def test_success(self):
- code, body = self.create_d()
- self.assertEqual(code, HTTP_OK)
- self.assert_create_body(body)
-
- def test_alreadyExist(self):
- self.create_d()
- code, body = self.create_d()
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn('already exists', body)
-
-
-class TestPodGet(TestPodBase):
- def test_notExist(self):
- code, body = self.get('notExist')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_getOne(self):
- self.create_d()
- code, body = self.get(self.req_d.name)
- self.assert_get_body(body)
-
- def test_list(self):
- self.create_d()
- self.create_e()
- code, body = self.get()
- self.assertEqual(len(body.pods), 2)
- for pod in body.pods:
- if self.req_d.name == pod.name:
- self.assert_get_body(pod)
- else:
- self.assert_get_body(pod, self.req_e)
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/result_collection_api/tests/unit/test_project.py b/result_collection_api/tests/unit/test_project.py
deleted file mode 100644
index 4f5bd9d..0000000
--- a/result_collection_api/tests/unit/test_project.py
+++ /dev/null
@@ -1,132 +0,0 @@
-import unittest
-
-from test_base import TestBase
-from resources.project_models import ProjectCreateRequest, Project, Projects
-from common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_FORBIDDEN, HTTP_NOT_FOUND
-
-
-class TestProjectBase(TestBase):
- def setUp(self):
- super(TestProjectBase, self).setUp()
- self.req_d = ProjectCreateRequest('vping', 'vping-ssh test')
- self.req_e = ProjectCreateRequest('doctor', 'doctor test')
- self.get_res = Project
- self.list_res = Projects
- self.update_res = Project
- self.basePath = '/api/v1/projects'
-
- def assert_body(self, project, req=None):
- if not req:
- req = self.req_d
- self.assertEqual(project.name, req.name)
- self.assertEqual(project.description, req.description)
- self.assertIsNotNone(project._id)
- self.assertIsNotNone(project.creation_date)
-
-
-class TestProjectCreate(TestProjectBase):
- def test_withoutBody(self):
- (code, body) = self.create()
- self.assertEqual(code, HTTP_BAD_REQUEST)
-
- def test_emptyName(self):
- req_empty = ProjectCreateRequest('')
- (code, body) = self.create(req_empty)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('name missing', body)
-
- def test_noneName(self):
- req_none = ProjectCreateRequest(None)
- (code, body) = self.create(req_none)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('name missing', body)
-
- def test_success(self):
- (code, body) = self.create_d()
- self.assertEqual(code, HTTP_OK)
- self.assert_create_body(body)
-
- def test_alreadyExist(self):
- self.create_d()
- (code, body) = self.create_d()
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn('already exists', body)
-
-
-class TestProjectGet(TestProjectBase):
- def test_notExist(self):
- code, body = self.get('notExist')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_getOne(self):
- self.create_d()
- code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
- self.assert_body(body)
-
- def test_list(self):
- self.create_d()
- self.create_e()
- code, body = self.get()
- for project in body.projects:
- if self.req_d.name == project.name:
- self.assert_body(project)
- else:
- self.assert_body(project, self.req_e)
-
-
-class TestProjectUpdate(TestProjectBase):
- def test_withoutBody(self):
- code, _ = self.update(None, 'noBody')
- self.assertEqual(code, HTTP_BAD_REQUEST)
-
- def test_notFound(self):
- code, _ = self.update(self.req_e, 'notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_newNameExist(self):
- self.create_d()
- self.create_e()
- code, body = self.update(self.req_e, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn("already exists", body)
-
- def test_noUpdate(self):
- self.create_d()
- code, body = self.update(self.req_d, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn("Nothing to update", body)
-
- def test_success(self):
- self.create_d()
- code, body = self.get(self.req_d.name)
- _id = body._id
-
- req = ProjectCreateRequest('newName', 'new description')
- code, body = self.update(req, self.req_d.name)
- self.assertEqual(code, HTTP_OK)
- self.assertEqual(_id, body._id)
- self.assert_body(body, req)
-
- _, new_body = self.get(req.name)
- self.assertEqual(_id, new_body._id)
- self.assert_body(new_body, req)
-
-
-class TestProjectDelete(TestProjectBase):
- def test_notFound(self):
- code, body = self.delete('notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_success(self):
- self.create_d()
- code, body = self.delete(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
- self.assertEqual(body, '')
-
- code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_NOT_FOUND)
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/result_collection_api/tests/unit/test_result.py b/result_collection_api/tests/unit/test_result.py
deleted file mode 100644
index 40daab8..0000000
--- a/result_collection_api/tests/unit/test_result.py
+++ /dev/null
@@ -1,267 +0,0 @@
-import unittest
-import copy
-
-from test_base import TestBase
-from resources.pod_models import PodCreateRequest
-from resources.project_models import ProjectCreateRequest
-from resources.testcase_models import TestcaseCreateRequest
-from resources.result_models import ResultCreateRequest, \
- TestResult, TestResults
-from common.constants import HTTP_OK, HTTP_BAD_REQUEST, HTTP_NOT_FOUND
-
-__author__ = '__serena__'
-
-
-class Details(object):
- def __init__(self, timestart=None, duration=None, status=None):
- self.timestart = timestart
- self.duration = duration
- self.status = status
-
- def format(self):
- return {
- "timestart": self.timestart,
- "duration": self.duration,
- "status": self.status
- }
-
- @staticmethod
- def from_dict(a_dict):
-
- if a_dict is None:
- return None
-
- t = Details()
- t.timestart = a_dict.get('timestart')
- t.duration = a_dict.get('duration')
- t.status = a_dict.get('status')
- return t
-
-
-class TestResultBase(TestBase):
- def setUp(self):
- self.pod = 'zte-pod1'
- self.project = 'functest'
- self.case = 'vPing'
- self.installer = 'fuel'
- self.version = 'C'
- self.build_tag = 'v3.0'
- self.scenario = 'odl-l2'
- self.criteria = 'passed'
- self.trust_indicator = 0.7
- self.start_date = "2016-05-23 07:16:09.477097"
- self.stop_date = "2016-05-23 07:16:19.477097"
- super(TestResultBase, self).setUp()
- self.details = Details(timestart='0', duration='9s', status='OK')
- self.req_d = ResultCreateRequest(pod_name=self.pod,
- project_name=self.project,
- case_name=self.case,
- installer=self.installer,
- version=self.version,
- start_date=self.start_date,
- stop_date=self.stop_date,
- details=self.details.format(),
- build_tag=self.build_tag,
- scenario=self.scenario,
- criteria=self.criteria,
- trust_indicator=self.trust_indicator)
- self.get_res = TestResult
- self.list_res = TestResults
- self.basePath = '/api/v1/results'
- self.req_pod = PodCreateRequest(self.pod, 'metal', 'zte pod 1')
- self.req_project = ProjectCreateRequest(self.project, 'vping test')
- self.req_testcase = TestcaseCreateRequest(self.case,
- '/cases/vping',
- 'vping-ssh test')
- self.create_help('/api/v1/pods', self.req_pod)
- self.create_help('/api/v1/projects', self.req_project)
- self.create_help('/api/v1/projects/%s/cases',
- self.req_testcase,
- self.project)
-
- def assert_res(self, code, result, req=None):
- self.assertEqual(code, HTTP_OK)
- if req is None:
- req = self.req_d
- self.assertEqual(result.pod_name, req.pod_name)
- self.assertEqual(result.project_name, req.project_name)
- self.assertEqual(result.case_name, req.case_name)
- self.assertEqual(result.installer, req.installer)
- self.assertEqual(result.version, req.version)
- details_req = Details.from_dict(req.details)
- details_res = Details.from_dict(result.details)
- self.assertEqual(details_res.duration, details_req.duration)
- self.assertEqual(details_res.timestart, details_req.timestart)
- self.assertEqual(details_res.status, details_req.status)
- self.assertEqual(result.build_tag, req.build_tag)
- self.assertEqual(result.scenario, req.scenario)
- self.assertEqual(result.criteria, req.criteria)
- self.assertEqual(result.trust_indicator, req.trust_indicator)
- self.assertEqual(result.start_date, req.start_date)
- self.assertEqual(result.stop_date, req.stop_date)
- self.assertIsNotNone(result._id)
-
-
-class TestResultCreate(TestResultBase):
- def test_nobody(self):
- (code, body) = self.create(None)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('no body', body)
-
- def test_podNotProvided(self):
- req = self.req_d
- req.pod_name = None
- (code, body) = self.create(req)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('pod_name missing', body)
-
- def test_projectNotProvided(self):
- req = self.req_d
- req.project_name = None
- (code, body) = self.create(req)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('project_name missing', body)
-
- def test_testcaseNotProvided(self):
- req = self.req_d
- req.case_name = None
- (code, body) = self.create(req)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('case_name missing', body)
-
- def test_noPod(self):
- req = self.req_d
- req.pod_name = 'notExistPod'
- (code, body) = self.create(req)
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn('Could not find pod', body)
-
- def test_noProject(self):
- req = self.req_d
- req.project_name = 'notExistProject'
- (code, body) = self.create(req)
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn('Could not find project', body)
-
- def test_noTestcase(self):
- req = self.req_d
- req.case_name = 'notExistTestcase'
- (code, body) = self.create(req)
- self.assertEqual(code, HTTP_NOT_FOUND)
- self.assertIn('Could not find testcase', body)
-
- def test_success(self):
- (code, body) = self.create_d()
- self.assertEqual(code, HTTP_OK)
- self.assert_href(body)
-
- def test_createSameResults(self):
- req_again = copy.deepcopy(self.req_d)
- req_again.start_date = "2016-05-23 08:16:09.477097"
- req_again.stop_date = "2016-05-23 08:16:19.477097"
-
- (code, body) = self.create(req_again)
- self.assertEqual(code, HTTP_OK)
- self.assert_href(body)
-
-
-class TestResultGet(TestResultBase):
- def test_getOne(self):
- _, res = self.create_d()
- _id = res.href.split('/')[-1]
- code, body = self.get(_id)
- self.assert_res(code, body)
-
- def test_queryPod(self):
- self._query_and_assert(self._set_query('pod'))
-
- def test_queryProject(self):
- self._query_and_assert(self._set_query('project'))
-
- def test_queryTestcase(self):
- self._query_and_assert(self._set_query('case'))
-
- def test_queryVersion(self):
- self._query_and_assert(self._set_query('version'))
-
- def test_queryInstaller(self):
- self._query_and_assert(self._set_query('installer'))
-
- def test_queryBuildTag(self):
- self._query_and_assert(self._set_query('build_tag'))
-
- def test_queryScenario(self):
- self._query_and_assert(self._set_query('scenario'))
-
- def test_queryTrustIndicator(self):
- self._query_and_assert(self._set_query('trust_indicator'))
-
- def test_queryCriteria(self):
- self._query_and_assert(self._set_query('criteria'))
-
- def test_queryPeriodFail(self):
- self._query_and_assert(self._set_query('period=1'),
- aheadof=True,
- found=False)
-
- def test_queryPeriodSuccess(self):
- self._query_and_assert(self._set_query('period=1'),
- aheadof=False,
- found=True)
-
- def test_combination(self):
- self._query_and_assert(self._set_query('pod',
- 'project',
- 'case',
- 'version',
- 'installer',
- 'build_tag',
- 'scenario',
- 'trust_indicator',
- 'criteria',
- 'period=1'))
-
- def test_notFound(self):
- self._query_and_assert(self._set_query('pod=notExistPod',
- 'project',
- 'case',
- 'version',
- 'installer',
- 'build_tag',
- 'scenario',
- 'trust_indicator',
- 'criteria',
- 'period=1'),
- found=False)
-
- def _query_and_assert(self, query, aheadof=False, found=True):
- import copy
- from datetime import datetime, timedelta
- req = copy.deepcopy(self.req_d)
- if aheadof:
- req.start_date = datetime.now() - timedelta(days=10)
- else:
- req.start_date = datetime.now()
- req.stop_date = str(req.start_date + timedelta(minutes=10))
- req.start_date = str(req.start_date)
- _, res = self.create(req)
- code, body = self.query(query)
- if not found:
- self.assertEqual(code, HTTP_OK)
- self.assertEqual(0, len(body.results))
- else:
- self.assertEqual(1, len(body.results))
- for result in body.results:
- self.assert_res(code, result, req)
-
- def _set_query(self, *args):
- uri = ''
- for arg in args:
- if '=' in arg:
- uri += arg + '&'
- else:
- uri += '{}={}&'.format(arg, eval('self.' + arg))
- return uri[0: -1]
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/result_collection_api/tests/unit/test_testcase.py b/result_collection_api/tests/unit/test_testcase.py
deleted file mode 100644
index 2371472..0000000
--- a/result_collection_api/tests/unit/test_testcase.py
+++ /dev/null
@@ -1,183 +0,0 @@
-import unittest
-
-from test_base import TestBase
-from resources.testcase_models import TestcaseCreateRequest, \
- Testcase, Testcases, TestcaseUpdateRequest
-from resources.project_models import ProjectCreateRequest
-from common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_FORBIDDEN, HTTP_NOT_FOUND
-
-
-__author__ = '__serena__'
-
-
-class TestCaseBase(TestBase):
- def setUp(self):
- super(TestCaseBase, self).setUp()
- self.req_d = TestcaseCreateRequest('vping_1',
- '/cases/vping_1',
- 'vping-ssh test')
- self.req_e = TestcaseCreateRequest('doctor_1',
- '/cases/doctor_1',
- 'create doctor')
- self.update_d = TestcaseUpdateRequest('vping_1',
- 'vping-ssh test',
- 'functest')
- self.update_e = TestcaseUpdateRequest('doctor_1',
- 'create doctor',
- 'functest')
- self.get_res = Testcase
- self.list_res = Testcases
- self.update_res = Testcase
- self.basePath = '/api/v1/projects/%s/cases'
- self.create_project()
-
- def assert_body(self, case, req=None):
- if not req:
- req = self.req_d
- self.assertEqual(case.name, req.name)
- self.assertEqual(case.description, req.description)
- self.assertEqual(case.url, req.url)
- self.assertIsNotNone(case._id)
- self.assertIsNotNone(case.creation_date)
-
- def assert_update_body(self, old, new, req=None):
- if not req:
- req = self.req_d
- self.assertEqual(new.name, req.name)
- self.assertEqual(new.description, req.description)
- self.assertEqual(new.url, old.url)
- self.assertIsNotNone(new._id)
- self.assertIsNotNone(new.creation_date)
-
- def create_project(self):
- req_p = ProjectCreateRequest('functest', 'vping-ssh test')
- self.create_help('/api/v1/projects', req_p)
- self.project = req_p.name
-
- def create_d(self):
- return super(TestCaseBase, self).create_d(self.project)
-
- def create_e(self):
- return super(TestCaseBase, self).create_e(self.project)
-
- def get(self, case=None):
- return super(TestCaseBase, self).get(self.project, case)
-
- def update(self, new=None, case=None):
- return super(TestCaseBase, self).update(new, self.project, case)
-
- def delete(self, case):
- return super(TestCaseBase, self).delete(self.project, case)
-
-
-class TestCaseCreate(TestCaseBase):
- def test_noBody(self):
- (code, body) = self.create(None, 'vping')
- self.assertEqual(code, HTTP_BAD_REQUEST)
-
- def test_noProject(self):
- code, body = self.create(self.req_d, 'noProject')
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn('Could not find project', body)
-
- def test_emptyName(self):
- req_empty = TestcaseCreateRequest('')
- (code, body) = self.create(req_empty, self.project)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('name missing', body)
-
- def test_noneName(self):
- req_none = TestcaseCreateRequest(None)
- (code, body) = self.create(req_none, self.project)
- self.assertEqual(code, HTTP_BAD_REQUEST)
- self.assertIn('name missing', body)
-
- def test_success(self):
- code, body = self.create_d()
- self.assertEqual(code, HTTP_OK)
- self.assert_create_body(body, None, self.project)
-
- def test_alreadyExist(self):
- self.create_d()
- code, body = self.create_d()
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn('already exists', body)
-
-
-class TestCaseGet(TestCaseBase):
- def test_notExist(self):
- code, body = self.get('notExist')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_getOne(self):
- self.create_d()
- code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
- self.assert_body(body)
-
- def test_list(self):
- self.create_d()
- self.create_e()
- code, body = self.get()
- for case in body.testcases:
- if self.req_d.name == case.name:
- self.assert_body(case)
- else:
- self.assert_body(case, self.req_e)
-
-
-class TestCaseUpdate(TestCaseBase):
- def test_noBody(self):
- code, _ = self.update(case='noBody')
- self.assertEqual(code, HTTP_BAD_REQUEST)
-
- def test_notFound(self):
- code, _ = self.update(self.update_e, 'notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_newNameExist(self):
- self.create_d()
- self.create_e()
- code, body = self.update(self.update_e, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn("already exists", body)
-
- def test_noUpdate(self):
- self.create_d()
- code, body = self.update(self.update_d, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
- self.assertIn("Nothing to update", body)
-
- def test_success(self):
- self.create_d()
- code, body = self.get(self.req_d.name)
- _id = body._id
-
- code, body = self.update(self.update_e, self.req_d.name)
- self.assertEqual(code, HTTP_OK)
- self.assertEqual(_id, body._id)
- self.assert_update_body(self.req_d, body, self.update_e)
-
- _, new_body = self.get(self.req_e.name)
- self.assertEqual(_id, new_body._id)
- self.assert_update_body(self.req_d, new_body, self.update_e)
-
-
-class TestCaseDelete(TestCaseBase):
- def test_notFound(self):
- code, body = self.delete('notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
-
- def test_success(self):
- self.create_d()
- code, body = self.delete(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
- self.assertEqual(body, '')
-
- code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_NOT_FOUND)
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/result_collection_api/tests/unit/test_version.py b/result_collection_api/tests/unit/test_version.py
deleted file mode 100644
index 8e544e0..0000000
--- a/result_collection_api/tests/unit/test_version.py
+++ /dev/null
@@ -1,24 +0,0 @@
-import unittest
-
-from test_base import TestBase
-from resources.models import Versions
-__author__ = 'serena'
-
-
-class TestVersionbBase(TestBase):
- def setUp(self):
- super(TestVersionbBase, self).setUp()
- self.list_res = Versions
- self.basePath = '/versions'
-
-
-class TestVersion(TestVersionbBase):
- def test_success(self):
- code, body = self.get()
- self.assertEqual(200, code)
- self.assertEqual(len(body.versions), 1)
- self.assertEqual(body.versions[0].version, 'v1.0')
- self.assertEqual(body.versions[0].description, 'basics')
-
-if __name__ == '__main__':
- unittest.main()