<a href="#" ng-click="showPod = !showPod">{{pod.name}}</a>
<div class="show-pod" ng-class="{ 'hidden': ! showPod }" style="margin-left:24px;">
<p>
+ owner: {{pod.owner}}<br>
role: {{pod.role}}<br>
mode: {{pod.mode}}<br>
create_date: {{pod.creation_date}}<br>
from tornado import gen
+from opnfv_testapi.common import constants
from opnfv_testapi.common import message
from opnfv_testapi.common import raises
from opnfv_testapi.db import api as dbapi
+def is_authorized(method):
+ @functools.wraps(method)
+ def wrapper(self, *args, **kwargs):
+ if self.table in ['pods']:
+ testapi_id = self.get_secure_cookie(constants.TESTAPI_ID)
+ if not testapi_id:
+ raises.Unauthorized(message.not_login())
+ user_info = yield dbapi.db_find_one('users', {'user': testapi_id})
+ if not user_info:
+ raises.Unauthorized(message.not_lfid())
+ kwargs['owner'] = testapi_id
+ ret = yield gen.coroutine(method)(self, *args, **kwargs)
+ raise gen.Return(ret)
+ return wrapper
+
+
def valid_token(method):
@functools.wraps(method)
def wrapper(self, *args, **kwargs):
return 'Invalid Token'
+def not_login():
+ return 'TestAPI id is not provided'
+
+
+def not_lfid():
+ return 'Not a valid Linux Foundation Account'
+
+
def no_update():
return 'Nothing to update'
@web.asynchronous
@gen.coroutine
+ @check.is_authorized
@check.valid_token
@check.no_body
@check.miss_fields
class Pod(models.ModelBase):
def __init__(self,
name='', mode='', details='',
- role="", _id='', create_date=''):
+ role="", _id='', create_date='', owner=''):
self.name = name
self.mode = mode
self.details = details
self.role = role
self._id = _id
self.creation_date = create_date
+ self.owner = owner
@swagger.model()
import functools
import httplib
+from concurrent.futures import ThreadPoolExecutor
+import mock
+
+
+O_get_secure_cookie = (
+ 'opnfv_testapi.resources.handlers.GenericApiHandler.get_secure_cookie')
+
+
+def thread_execute(method, *args, **kwargs):
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ result = executor.submit(method, *args, **kwargs)
+ return result
+
+
+def mock_invalid_lfid():
+ def _mock_invalid_lfid(xstep):
+ def wrap(self, *args, **kwargs):
+ with mock.patch(O_get_secure_cookie) as m_cookie:
+ m_cookie.return_value = 'InvalidUser'
+ return xstep(self, *args, **kwargs)
+ return wrap
+ return _mock_invalid_lfid
+
+
+def mock_valid_lfid():
+ def _mock_valid_lfid(xstep):
+ def wrap(self, *args, **kwargs):
+ with mock.patch(O_get_secure_cookie) as m_cookie:
+ m_cookie.return_value = 'ValidUser'
+ return xstep(self, *args, **kwargs)
+ return wrap
+ return _mock_valid_lfid
+
def upload(excepted_status, excepted_response):
def _upload(create_request):
results = MemDb('results')
scenarios = MemDb('scenarios')
tokens = MemDb('tokens')
+users = MemDb('users')
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+from datetime import datetime
import json
from os import path
+from bson.objectid import ObjectId
import mock
from tornado import testing
from opnfv_testapi.resources import models
+from opnfv_testapi.resources import pod_models
from opnfv_testapi.tests.unit import fake_pymongo
self.get_res = None
self.list_res = None
self.update_res = None
+ self.pod_d = pod_models.Pod(name='zte-pod1',
+ mode='virtual',
+ details='zte pod 1',
+ role='community-ci',
+ _id=str(ObjectId()),
+ owner='ValidUser',
+ create_date=str(datetime.now()))
+ self.pod_e = pod_models.Pod(name='zte-pod2',
+ mode='metal',
+ details='zte pod 2',
+ role='production-ci',
+ _id=str(ObjectId()),
+ owner='ValidUser',
+ create_date=str(datetime.now()))
self.req_d = None
self.req_e = None
self.addCleanup(self._clear)
super(TestBase, self).setUp()
+ fake_pymongo.users.insert({"user": "ValidUser",
+ 'email': 'validuser@lf.com',
+ 'fullname': 'Valid User',
+ 'groups': [
+ 'opnfv-testapi-users',
+ 'opnfv-gerrit-functest-submitters',
+ 'opnfv-gerrit-qtip-contributors']
+ })
def tearDown(self):
self.db_patcher.stop()
from opnfv_testapi.common import message
from opnfv_testapi.resources import pod_models
from opnfv_testapi.tests.unit import executor
+from opnfv_testapi.tests.unit import fake_pymongo
from opnfv_testapi.tests.unit.resources import test_base as base
class TestPodBase(base.TestBase):
def setUp(self):
super(TestPodBase, self).setUp()
- self.req_d = pod_models.PodCreateRequest('zte-1', 'virtual',
- 'zte pod 1', 'ci-pod')
- self.req_e = pod_models.PodCreateRequest('zte-2', 'metal', 'zte pod 2')
- self.req_f = pod_models.PodCreateRequest('Zte-1', 'virtual',
- 'zte pod 1', 'ci-pod')
self.get_res = pod_models.Pod
self.list_res = pod_models.Pods
self.basePath = '/api/v1/pods'
+ self.req_d = pod_models.PodCreateRequest(name=self.pod_d.name,
+ mode=self.pod_d.mode,
+ details=self.pod_d.details,
+ role=self.pod_d.role)
+ self.req_e = pod_models.PodCreateRequest(name=self.pod_e.name,
+ mode=self.pod_e.mode,
+ details=self.pod_e.details,
+ role=self.pod_e.role)
def assert_get_body(self, pod, req=None):
if not req:
req = self.req_d
+ self.assertEqual(pod.owner, 'ValidUser')
self.assertEqual(pod.name, req.name)
self.assertEqual(pod.mode, req.mode)
self.assertEqual(pod.details, req.details)
class TestPodCreate(TestPodBase):
+ @executor.create(httplib.BAD_REQUEST, message.not_login())
+ def test_notlogin(self):
+ return self.req_d
+
+ @executor.mock_invalid_lfid()
+ @executor.create(httplib.BAD_REQUEST, message.not_lfid())
+ def test_invalidLfid(self):
+ return self.req_d
+
+ @executor.mock_valid_lfid()
@executor.create(httplib.BAD_REQUEST, message.no_body())
def test_withoutBody(self):
return None
+ @executor.mock_valid_lfid()
@executor.create(httplib.BAD_REQUEST, message.missing('name'))
def test_emptyName(self):
return pod_models.PodCreateRequest('')
+ @executor.mock_valid_lfid()
@executor.create(httplib.BAD_REQUEST, message.missing('name'))
def test_noneName(self):
return pod_models.PodCreateRequest(None)
+ @executor.mock_valid_lfid()
@executor.create(httplib.OK, 'assert_create_body')
def test_success(self):
return self.req_d
+ @executor.mock_valid_lfid()
@executor.create(httplib.FORBIDDEN, message.exist_base)
def test_alreadyExist(self):
- self.create_d()
+ fake_pymongo.pods.insert(self.pod_d.format())
return self.req_d
+ @executor.mock_valid_lfid()
@executor.create(httplib.FORBIDDEN, message.exist_base)
def test_alreadyExistCaseInsensitive(self):
- self.create(self.req_f)
+ fake_pymongo.pods.insert(self.pod_d.format())
+ self.req_d.name = self.req_d.name.upper()
return self.req_d
class TestPodGet(TestPodBase):
def setUp(self):
super(TestPodGet, self).setUp()
- self.create_d()
- self.create_e()
+ fake_pymongo.pods.insert(self.pod_d.format())
+ fake_pymongo.pods.insert(self.pod_e.format())
@executor.get(httplib.NOT_FOUND, message.not_found_base)
def test_notExist(self):
@executor.get(httplib.OK, 'assert_get_body')
def test_getOne(self):
- return self.req_d.name
+ return self.pod_d.name
@executor.get(httplib.OK, '_assert_list')
def test_list(self):
def _assert_list(self, body):
self.assertEqual(len(body.pods), 2)
for pod in body.pods:
- if self.req_d.name == pod.name:
+ if self.pod_d.name == pod.name:
self.assert_get_body(pod)
else:
- self.assert_get_body(pod, self.req_e)
+ self.assert_get_body(pod, self.pod_e)
if __name__ == '__main__':
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
import copy
+from datetime import datetime
+from datetime import timedelta
import httplib
-import unittest
-from datetime import datetime, timedelta
import json
+import unittest
from opnfv_testapi.common import message
-from opnfv_testapi.resources import pod_models
from opnfv_testapi.resources import project_models
from opnfv_testapi.resources import result_models
from opnfv_testapi.resources import testcase_models
from opnfv_testapi.tests.unit import executor
+from opnfv_testapi.tests.unit import fake_pymongo
from opnfv_testapi.tests.unit.resources import test_base as base
class TestResultBase(base.TestBase):
def setUp(self):
- self.pod = 'zte-pod1'
+ super(TestResultBase, self).setUp()
+ self.pod = self.pod_d.name
self.project = 'functest'
self.case = 'vPing'
self.installer = 'fuel'
self.stop_date = str(datetime.now() + timedelta(minutes=1))
self.update_date = str(datetime.now() + timedelta(days=1))
self.update_step = -0.05
- super(TestResultBase, self).setUp()
self.details = Details(timestart='0', duration='9s', status='OK')
self.req_d = result_models.ResultCreateRequest(
pod_name=self.pod,
self.list_res = result_models.TestResults
self.update_res = result_models.TestResult
self.basePath = '/api/v1/results'
- self.req_pod = pod_models.PodCreateRequest(
- self.pod,
- 'metal',
- 'zte pod 1')
self.req_project = project_models.ProjectCreateRequest(
self.project,
'vping test')
self.case,
'/cases/vping',
'vping-ssh test')
- self.create_help('/api/v1/pods', self.req_pod)
+ fake_pymongo.pods.insert(self.pod_d.format())
self.create_help('/api/v1/projects', self.req_project)
self.create_help('/api/v1/projects/%s/cases',
self.req_testcase,
Code Review / releng.git / commitdiff