Merge "Compass4NFV CI verify support centos7"
[releng.git] / utils / test / testapi / opnfv_testapi / tests / unit / test_scenario.py
1 import json
2 import os
3
4 from opnfv_testapi.common.constants import HTTP_BAD_REQUEST
5 from opnfv_testapi.common.constants import HTTP_FORBIDDEN
6 from opnfv_testapi.common.constants import HTTP_OK
7 from opnfv_testapi.resources.scenario_models import Scenario
8 from opnfv_testapi.resources.scenario_models import ScenarioCreateRequest
9 from opnfv_testapi.resources.scenario_models import Scenarios
10 from test_testcase import TestBase
11
12
13 class TestScenarioBase(TestBase):
14     def setUp(self):
15         super(TestScenarioBase, self).setUp()
16         self.get_res = Scenario
17         self.list_res = Scenarios
18         self.basePath = '/api/v1/scenarios'
19         self.req_d = self._load_request('scenario-c1.json')
20         self.req_2 = self._load_request('scenario-c2.json')
21
22     def tearDown(self):
23         pass
24
25     def assert_body(self, project, req=None):
26         pass
27
28     @staticmethod
29     def _load_request(f_req):
30         abs_file = os.path.join(os.path.dirname(__file__), f_req)
31         with open(abs_file, 'r') as f:
32             loader = json.load(f)
33             f.close()
34         return loader
35
36     def create_return_name(self, req):
37         _, res = self.create(req)
38         return res.href.split('/')[-1]
39
40     def assert_res(self, code, scenario, req=None):
41         self.assertEqual(code, HTTP_OK)
42         if req is None:
43             req = self.req_d
44         scenario_dict = scenario.format_http()
45         self.assertIsNotNone(scenario_dict['_id'])
46         self.assertIsNotNone(scenario_dict['creation_date'])
47         self.assertDictContainsSubset(req, scenario_dict)
48
49
50 class TestScenarioCreate(TestScenarioBase):
51     def test_withoutBody(self):
52         (code, body) = self.create()
53         self.assertEqual(code, HTTP_BAD_REQUEST)
54
55     def test_emptyName(self):
56         req_empty = ScenarioCreateRequest('')
57         (code, body) = self.create(req_empty)
58         self.assertEqual(code, HTTP_BAD_REQUEST)
59         self.assertIn('name missing', body)
60
61     def test_noneName(self):
62         req_none = ScenarioCreateRequest(None)
63         (code, body) = self.create(req_none)
64         self.assertEqual(code, HTTP_BAD_REQUEST)
65         self.assertIn('name missing', body)
66
67     def test_success(self):
68         (code, body) = self.create_d()
69         self.assertEqual(code, HTTP_OK)
70         self.assert_create_body(body)
71
72     def test_alreadyExist(self):
73         self.create_d()
74         (code, body) = self.create_d()
75         self.assertEqual(code, HTTP_FORBIDDEN)
76         self.assertIn('already exists', body)
77
78
79 class TestScenarioGet(TestScenarioBase):
80     def setUp(self):
81         super(TestScenarioGet, self).setUp()
82         self.scenario_1 = self.create_return_name(self.req_d)
83         self.scenario_2 = self.create_return_name(self.req_2)
84
85     def test_getByName(self):
86         code, body = self.get(self.scenario_1)
87         self.assert_res(code, body, req=self.req_d)
88
89     def test_getAll(self):
90         self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
91
92     def test_queryName(self):
93         query = self._set_query('name=nosdn-nofeature-ha')
94         self._query_and_assert(query, reqs=[self.req_d])
95
96     def test_queryInstaller(self):
97         query = self._set_query('installer=apex')
98         self._query_and_assert(query, reqs=[self.req_d])
99
100     def test_queryVersion(self):
101         query = self._set_query('version=master')
102         self._query_and_assert(query, reqs=[self.req_d])
103
104     def test_queryProject(self):
105         query = self._set_query('project=functest')
106         self._query_and_assert(query, reqs=[self.req_d, self.req_2])
107
108     def test_queryCombination(self):
109         query = self._set_query('name=nosdn-nofeature-ha',
110                                 'installer=apex',
111                                 'version=master',
112                                 'project=functest')
113
114         self._query_and_assert(query, reqs=[self.req_d])
115
116     @staticmethod
117     def _set_query(*args):
118         uri = ''
119         for arg in args:
120             uri += arg + '&'
121         return uri[0: -1]
122
123     def _query_and_assert(self, query, found=True, reqs=None):
124         code, body = self.query(query)
125         if not found:
126             self.assertEqual(code, HTTP_OK)
127             self.assertEqual(0, len(body.scenarios))
128         else:
129             self.assertEqual(len(reqs), len(body.scenarios))
130             for req in reqs:
131                 for scenario in body.scenarios:
132                     if req['name'] == scenario.name:
133                         self.assert_res(code, scenario, req)