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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from .testenv import testenv # pylint: disable=unused-import
19 from .. import helpers
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)
27 _verify_deployed_service_in_storage(service_name, testenv.model_storage)
28 _verify_webserver_up('http://localhost:9090')
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()
34 _verify_webserver_down('http://localhost:9090')
35 testenv.verify_clean_storage()
38 def _verify_webserver_up(http_endpoint):
39 server_response = requests.get(http_endpoint, timeout=10)
40 assert server_response.status_code == 200
43 def _verify_webserver_down(http_endpoint):
45 requests.get(http_endpoint, timeout=10)
47 except requests.exceptions.ConnectionError:
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