openecomp-be/tools/build/scripts/action_library_client/test/test_action_library_client_integration.py

   1 import sys
   2 import os
   3 import unittest
   4 import uuid
   5 import json
   6 import tempfile
   7 import action_library_client
   8
   9 class IntegrationTest(unittest.TestCase):
  10
  11     HTTP = "http://10.147.97.199:8080"
  12     HTTPS = "https://10.147.97.199:8443"
  13
  14     def setUp(self):
  15         os.environ["ALC_HTTP_USER"] = "AUTH-DELETE"
  16         os.environ["ALC_HTTP_PASS"] = "test"
  17
  18     def tearDown(self):
  19         os.environ["ALC_HTTP_INSECURE"] = ""
  20         os.environ["ALC_HTTP_USER"] = ""
  21         os.environ["ALC_HTTP_PASS"] = ""
  22
  23     @staticmethod
  24     def __prepare(testcase, name):
  25         with open(testcase, 'r') as fin:
  26             jsonk = json.loads(fin.read())
  27             jsonk['name'] = name
  28             with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp:
  29                 temp.write(json.dumps(jsonk))
  30                 temp.flush()
  31                 return temp.name
  32
  33     @staticmethod
  34     def __get_sequence():
  35         with open(r'./seq.txt', 'r+') as f:
  36             value = int(f.read())
  37             f.seek(0)
  38             f.write(str(value + 1))
  39             return value
  40
  41     def __print_separator(self):
  42         logger = action_library_client.Runner.get_logger()
  43         logger.info("==================================================")
  44
  45     def __list(self, stdargs):
  46         logger = action_library_client.Runner.get_logger()
  47         list_response = action_library_client.execute(["--list"] + stdargs)
  48         logger.info("--list response: {}".format(list_response))
  49         self.assertTrue(isinstance(list_response, dict))
  50         return list_response
  51
  52     def __get_action(self, list_response, ai_uuid):
  53         for action in list_response['actionList']:
  54             if action['actionInvariantUUID'] == ai_uuid:
  55                 return action
  56
  57     def __create_delete(self, extraargs):
  58
  59         logger = action_library_client.Runner.get_logger()
  60
  61         # Setup.
  62
  63         seq = IntegrationTest.__get_sequence()
  64         name = "Backout{}".format(seq)
  65         path = IntegrationTest.__prepare("scenarios/Backout.json", name)
  66         stdargs = ["--url", self.HTTP, "--verbose"]
  67         if extraargs:
  68             stdargs.extend(extraargs)
  69
  70         # List actions.
  71
  72         self.__print_separator()
  73         list_response1 = self.__list(stdargs)
  74         self.assertTrue(isinstance(list_response1, dict))
  75
  76         # CREATE action.
  77
  78         self.__print_separator()
  79         create_response = action_library_client.execute(["--create", "--in", path] + stdargs)
  80         logger.info("--create response: {}".format(create_response))
  81         self.assertTrue(isinstance(create_response, dict))
  82         ai_uuid = create_response['actionInvariantUUID']
  83         self.assertTrue(ai_uuid)
  84         self.assertEquals(create_response['status'], 'Locked')
  85         self.assertEquals(create_response['version'], '0.1')
  86
  87         # UPDATE action #1.
  88
  89         self.__print_separator()
  90         update_response1 = action_library_client.execute(["--update", "--in", path, "--uuid", ai_uuid] + stdargs)
  91         logger.info("--update response: {}".format(update_response1))
  92         self.assertTrue(isinstance(update_response1, dict))
  93
  94         # UPDATE action #2.
  95
  96         self.__print_separator()
  97         update_response2 = action_library_client.execute(["--update", "--in", path, "--uuid", ai_uuid] + stdargs)
  98         logger.info("--update response: {}".format(update_response2))
  99         self.assertTrue(isinstance(update_response2, dict))
 100
 101         # CHECKOUT action (usage unknown).
 102
 103         self.__print_separator()
 104         try:
 105             action_library_client.execute(["--checkout", "--uuid", ai_uuid] + stdargs)
 106             self.fail("--checkout should fail")
 107         except Exception as err:
 108             print(err)
 109
 110         # CHECKIN action.
 111
 112         self.__print_separator()
 113         checkin_response = action_library_client.execute(["--checkin", "--in", path, "--uuid", ai_uuid] + stdargs)
 114         logger.info("--checkin response: {}".format(checkin_response))
 115         self.assertTrue(isinstance(checkin_response, dict))
 116         self.assertEquals(checkin_response['status'], 'Available')
 117         self.assertEquals(checkin_response['version'], '0.1')
 118
 119         # SUBMIT action.
 120
 121         self.__print_separator()
 122         submit_response = action_library_client.execute(["--submit", "--in", path, "--uuid", ai_uuid] + stdargs)
 123         logger.info("--submit response: {}".format(submit_response))
 124         self.assertTrue(isinstance(submit_response, dict))
 125         self.assertEquals(submit_response['status'], 'Final')
 126         self.assertEquals(submit_response['version'], '1.0')
 127
 128         # LIST again
 129
 130         self.__print_separator()
 131         list_response2 = self.__list(stdargs)
 132         action_found2 = self.__get_action(list_response2, ai_uuid)
 133         self.assertTrue(action_found2)
 134
 135         # DELETE action.
 136
 137         self.__print_separator()
 138         delete_response = action_library_client.execute(["--delete", "--uuid", ai_uuid] + stdargs)
 139         logger.info("--delete response: {}".format(delete_response))
 140         self.assertEqual(delete_response, action_library_client.ResponseCodes.OK)
 141
 142         # LIST yet again
 143
 144         self.__print_separator()
 145         list_response3 = self.__list(stdargs)
 146         action_found3 = self.__get_action(list_response3, ai_uuid)
 147         self.assertFalse(action_found3)
 148
 149     def __create_undo(self, extraargs):
 150
 151         # Setup
 152
 153         logger = action_library_client.Runner.get_logger()
 154         seq = IntegrationTest.__get_sequence()
 155         name = "Backout{}".format(seq)
 156         path = IntegrationTest.__prepare("scenarios/Backout.json", name)
 157         stdargs = ["--url", self.HTTP, "--verbose"]
 158
 159         # CREATE action.
 160
 161         self.__print_separator()
 162         create_response = action_library_client.execute(["--create", "--in", path] + stdargs + extraargs)
 163         logger.info("--create response: {}".format(create_response))
 164         self.assertTrue(isinstance(create_response, dict))
 165         ai_uuid = create_response['actionInvariantUUID']
 166         self.assertTrue(ai_uuid)
 167         self.assertEquals(create_response['status'], 'Locked')
 168         self.assertEquals(create_response['version'], '0.1')
 169
 170         # UNDOCHECKOUT action
 171
 172         self.__print_separator()
 173         undocheckout_response = action_library_client.execute(["--undocheckout", "--uuid", ai_uuid] + stdargs + extraargs)
 174         self.assertTrue(isinstance(undocheckout_response, dict))
 175
 176     def __create_list(self, extraargs):
 177         # Setup
 178
 179         logger = action_library_client.Runner.get_logger()
 180         seq = IntegrationTest.__get_sequence()
 181         name = "Backout{}".format(seq)
 182         path = IntegrationTest.__prepare("scenarios/Backout.json", name)
 183         stdargs = ["--url", self.HTTP, "--verbose"]
 184
 185         # CREATE action.
 186
 187         self.__print_separator()
 188         create_response = action_library_client.execute(["--create", "--in", path] + stdargs + extraargs)
 189         logger.info("--create response: {}".format(create_response))
 190         self.assertTrue(isinstance(create_response, dict))
 191         ai_uuid = create_response['actionInvariantUUID']
 192         self.assertTrue(ai_uuid)
 193         self.assertEquals(create_response['status'], 'Locked')
 194         self.assertEquals(create_response['version'], '0.1')
 195
 196         # CHECKIN action.
 197
 198         self.__print_separator()
 199         checkin_response = action_library_client.execute(["--checkin", "--in", path, "--uuid", ai_uuid] +
 200                                                          stdargs + extraargs)
 201         logger.info("--checkin response: {}".format(checkin_response))
 202         self.assertTrue(isinstance(checkin_response, dict))
 203         self.assertEquals(checkin_response['status'], 'Available')
 204         self.assertEquals(checkin_response['version'], '0.1')
 205
 206         try:
 207             # LIST.
 208
 209             self.__print_separator()
 210             list_response1 = self.__list(stdargs + extraargs)
 211             action_found1 = self.__get_action(list_response1, ai_uuid)
 212             self.assertTrue(action_found1)
 213
 214             # LIST with UUID.
 215
 216             self.__print_separator()
 217             list_response2 = self.__list(stdargs + extraargs + ["--uuid", ai_uuid])
 218             self.assertFalse(hasattr(list_response2, 'actionList'))
 219             self.assertEquals(len(list_response2['versions']), 1)
 220
 221             # LIST with bad UUID.
 222
 223             self.__print_separator()
 224             list_response3 = action_library_client.execute(["--list"] + stdargs + extraargs +
 225                                                            ["--uuid", "where_the_wind_blows"])
 226             if isinstance(list_response3, int):
 227                 self.assertEquals(action_library_client.ResponseCodes.HTTP_NOT_FOUND_ERROR, list_response3)
 228             else:
 229                 self.assertEquals("ACT1045", list_response3["code"])
 230
 231         finally:
 232
 233             # DELETE action
 234
 235             self.__print_separator()
 236             action_library_client.execute(["--delete", "--uuid", ai_uuid] + stdargs + extraargs)
 237
 238     def __http_secure(self, extraargs):
 239         os.environ["ALC_HTTP_INSECURE"] = ""
 240         try:
 241             self.__list(["--url", self.HTTPS, "--verbose"] + extraargs)
 242             if not (sys.version_info[0] == 2 and sys.version_info[1] == 6):
 243                 self.fail("Should fail (non-2.6) for TLS + secure")
 244         except Exception:
 245             pass
 246
 247     def __http_insecure(self, extraargs):
 248         os.environ["ALC_HTTP_INSECURE"] = True
 249         self.__list(["--url", self.HTTPS, "--verbose"] + extraargs)
 250
 251     def __no_credentials(self, extraargs):
 252
 253         args = ["--url", self.HTTP] + extraargs
 254         self.__list(args)
 255         print("OK")
 256
 257         os.environ["ALC_HTTP_USER"] = ""
 258         os.environ["ALC_HTTP_PASS"] = ""
 259         try:
 260             action_library_client.execute(["--list"] + args)
 261             self.fail("Should fail for missing credentials")
 262         except Exception as e:
 263             self.assertEquals("REST service credentials not found", e.message)
 264
 265     def __bad_credentials(self, extraargs):
 266
 267         args = ["--url", self.HTTP] + extraargs
 268         self.__list(args)
 269
 270         os.environ["ALC_HTTP_USER"] = "wakey_wakey"
 271         os.environ["ALC_HTTP_PASS"] = "rise_and_shine"
 272         code = action_library_client.execute(["--list"] + args)
 273         self.assertEquals(action_library_client.ResponseCodes.HTTP_FORBIDDEN_ERROR, code)
 274
 275     ################################################################################
 276
 277     def test_https_insecure_local_fail(self):
 278         self.__http_secure([])
 279
 280     def test_https_insecure_remote_fail(self):
 281         self.__http_secure(["--curl"])
 282
 283     def test_https_native(self):
 284         self.__http_secure([])
 285
 286     def test_https_curl(self):
 287         self.__http_secure(["--curl"])
 288
 289     def test_undo_checkout_native(self):
 290         self.__create_undo([])
 291
 292     def test_undo_checkout_curl(self):
 293         self.__create_undo(["--curl"])
 294
 295     def test_create_delete_native(self):
 296         self.__create_delete([])
 297
 298     def test_create_delete_curl(self):
 299         self.__create_delete(["--curl"])
 300
 301     def test_create_list_native(self):
 302         self.__create_list([])
 303
 304     def test_create_list_curl(self):
 305         self.__create_list(["--curl"])
 306
 307     def test_bad_credentials_native(self):
 308         self.__bad_credentials([])
 309
 310     def test_bad_credentials_curl(self):
 311         self.__bad_credentials(["--curl"])
 312     #
 313     def test_no_credentials_native(self):
 314         self.__no_credentials([])
 315
 316     def test_no_credentials_curl(self):
 317         self.__no_credentials(["--curl"])
 318
 319     def test_create_to_delete_dryrun(self):
 320         ai_uuid = str(uuid.uuid4())
 321         path = IntegrationTest.__prepare("scenarios/Backout.json", "Backout{}".format("001"))
 322         stdargs = ["--url", self.HTTP, "--verbose", "--dryrun"]
 323         action_library_client.execute(["--create", "--in", path] + stdargs)
 324         action_library_client.execute(["--update", "--in", path, "--uuid", ai_uuid] + stdargs)
 325         action_library_client.execute(["--checkout", "--uuid", ai_uuid] + stdargs)
 326         action_library_client.execute(["--undocheckout", "--uuid", ai_uuid] + stdargs)
 327         action_library_client.execute(["--checkin", "--uuid", ai_uuid] + stdargs)
 328         action_library_client.execute(["--submit", "--uuid", ai_uuid] + stdargs)
 329         action_library_client.execute(["--list"] + stdargs)