azure/aria/aria-extension-cloudify/src/aria/tests/end2end/test_hello_world.py

   1 # Licensed to the Apache Software Foundation (ASF) under one or more
   2 # contributor license agreements.  See the NOTICE file distributed with
   3 # this work for additional information regarding copyright ownership.
   4 # The ASF licenses this file to You under the Apache License, Version 2.0
   5 # (the "License"); you may not use this file except in compliance with
   6 # the License.  You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import requests
  17
  18 from .testenv import testenv  # pylint: disable=unused-import
  19 from .. import helpers
  20
  21
  22 def test_hello_world(testenv):
  23     hello_world_template_uri = helpers.get_example_uri('hello-world', 'hello-world.yaml')
  24     service_name = testenv.install_service(hello_world_template_uri)
  25
  26     try:
  27         _verify_deployed_service_in_storage(service_name, testenv.model_storage)
  28         _verify_webserver_up('http://localhost:9090')
  29     finally:
  30         # Even if some assertions failed, attempt to execute uninstall so the
  31         # webserver process doesn't stay up once the test is finished
  32         testenv.uninstall_service()
  33
  34     _verify_webserver_down('http://localhost:9090')
  35     testenv.verify_clean_storage()
  36
  37
  38 def _verify_webserver_up(http_endpoint):
  39     server_response = requests.get(http_endpoint, timeout=10)
  40     assert server_response.status_code == 200
  41
  42
  43 def _verify_webserver_down(http_endpoint):
  44     try:
  45         requests.get(http_endpoint, timeout=10)
  46         assert False
  47     except requests.exceptions.ConnectionError:
  48         pass
  49
  50
  51 def _verify_deployed_service_in_storage(service_name, model_storage):
  52     service_templates = model_storage.service_template.list()
  53     assert len(service_templates) == 1
  54     assert len(service_templates[0].services) == 1
  55     service = service_templates[0].services[service_name]
  56     assert service.name == service_name
  57     assert len(service.executions) == 1
  58     assert len(service.nodes) == 2
  59     assert service.outputs['port'].value == 9090
  60     assert all(node.state == node.STARTED for node in service.nodes.itervalues())
  61     assert len(service.executions[0].logs) > 0