)
)
+ try:
+ waitcount = 0
+ while waitcount <= 5:
+ if is_sshd_running(container):
+ break
+ waitcount += 1
+ time.sleep(1)
+ if waitcount >= 5:
+ debug("couldn't detect sshd running")
+ raise Exception("Unable to verify container sshd")
+
+ except Exception as ex:
+ debug(
+ "Error checking sshd status on {}: {}".format(
+ test_machine,
+ ex,
+ )
+ )
+
# HACK: We need to give sshd a chance to bind to the interface,
# and pylxd's container.execute seems to be broken and fails and/or
# hangs trying to properly check if the service is up.
return container
+def is_sshd_running(container):
+ """Check if sshd is running in the container.
+
+ Check to see if the sshd process is running and listening on port 22.
+
+ :param container: The container to check
+ :return boolean: True if sshd is running.
+ """
+ debug("Container: {}".format(container))
+ try:
+ (rc, stdout, stderr) = container.execute(
+ ["service", "ssh", "status"]
+ )
+ # If the status is a) found and b) running, the exit code will be 0
+ if rc == 0:
+ return True
+ except Exception as ex:
+ debug("Failed to check sshd service status: {}".format(ex))
+
+ return False
+
+
def destroy_lxd_container(container):
"""Stop and delete a LXD container.
self._running = False
self._stopping = True
+ # Destroy the network service
+ try:
+ await self.n2vc.DestroyNetworkService(self.ns_name)
+ except Exception as e:
+ debug(
+ "Error Destroying Network Service \"{}\": {}".format(
+ self.ns_name,
+ e,
+ )
+ )
+
+ # Wait for the applications to be removed and delete the containers
for application in self.charms:
try:
- await self.n2vc.RemoveCharms(self.ns_name, application)
-
- await self.n2vc.DestroyNetworkService(self.ns_name)
while True:
# Wait for the application to be removed
application,
):
break
- await self.n2vc.DestroyNetworkService(self.ns_name)
# Need to wait for the charm to finish, because native charms
if self.state[application]['container']:
debug("{} is done".format(application))
return
+ if status in ['error']:
+ # To test broken charms, if a charm enters an error state we should
+ # end the test
+ debug("{} is in an error state, stop the test.".format(application))
+ # asyncio.ensure_future(self.stop())
+ self.state[application]['done'] = True
+ assert False
+
if status in ["blocked"] and self.isproxy(application):
if self.state[application]['phase'] == "deploy":
debug("Configuring proxy charm for {}".format(application))
+++ /dev/null
-#!/bin/bash
-
-sudo add-apt-repository ppa:juju/stable -y
-sudo apt-get update
-sudo apt-get install amulet python-requests -y
+++ /dev/null
-#!/usr/bin/python3
-
-import amulet
-import requests
-import unittest
-
-
-class TestCharm(unittest.TestCase):
- def setUp(self):
- self.d = amulet.Deployment()
-
- self.d.add('simple')
- self.d.expose('simple')
-
- self.d.setup(timeout=900)
- self.d.sentry.wait()
-
- self.unit = self.d.sentry['simple'][0]
-
- def test_service(self):
- # test we can access over http
- page = requests.get('http://{}'.format(self.unit.info['public-address']))
- self.assertEqual(page.status_code, 200)
- # Now you can use self.d.sentry[SERVICE][UNIT] to address each of the units and perform
- # more in-depth steps. Each self.d.sentry[SERVICE][UNIT] has the following methods:
- # - .info - An array of the information of that unit from Juju
- # - .file(PATH) - Get the details of a file on that unit
- # - .file_contents(PATH) - Get plain text output of PATH file from that unit
- # - .directory(PATH) - Get details of directory
- # - .directory_contents(PATH) - List files and folders in PATH on that unit
- # - .relation(relation, service:rel) - Get relation data from return service
-
-
-if __name__ == '__main__':
- unittest.main()
+++ /dev/null
-#!/bin/bash
-
-sudo add-apt-repository ppa:juju/stable -y
-sudo apt-get update
-sudo apt-get install amulet python-requests -y
+++ /dev/null
-#!/usr/bin/python3
-
-import amulet
-import requests
-import unittest
-
-
-class TestCharm(unittest.TestCase):
- def setUp(self):
- self.d = amulet.Deployment()
-
- self.d.add('metrics-demo')
- self.d.expose('metrics-demo')
-
- self.d.setup(timeout=900)
- self.d.sentry.wait()
-
- self.unit = self.d.sentry['metrics-demo'][0]
-
- def test_service(self):
- # test we can access over http
- page = requests.get('http://{}'.format(self.unit.info['public-address']))
- self.assertEqual(page.status_code, 200)
- # Now you can use self.d.sentry[SERVICE][UNIT] to address each of the units and perform
- # more in-depth steps. Each self.d.sentry[SERVICE][UNIT] has the following methods:
- # - .info - An array of the information of that unit from Juju
- # - .file(PATH) - Get the details of a file on that unit
- # - .file_contents(PATH) - Get plain text output of PATH file from that unit
- # - .directory(PATH) - Get details of directory
- # - .directory_contents(PATH) - List files and folders in PATH on that unit
- # - .relation(relation, service:rel) - Get relation data from return service
-
-
-if __name__ == '__main__':
- unittest.main()
+++ /dev/null
-#!/bin/bash
-
-sudo add-apt-repository ppa:juju/stable -y
-sudo apt-get update
-sudo apt-get install amulet python-requests -y
+++ /dev/null
-#!/usr/bin/python3
-
-import amulet
-import requests
-import unittest
-
-
-class TestCharm(unittest.TestCase):
- def setUp(self):
- self.d = amulet.Deployment()
-
- self.d.add('metrics-demo')
- self.d.expose('metrics-demo')
-
- self.d.setup(timeout=900)
- self.d.sentry.wait()
-
- self.unit = self.d.sentry['metrics-demo'][0]
-
- def test_service(self):
- # test we can access over http
- page = requests.get('http://{}'.format(self.unit.info['public-address']))
- self.assertEqual(page.status_code, 200)
- # Now you can use self.d.sentry[SERVICE][UNIT] to address each of the units and perform
- # more in-depth steps. Each self.d.sentry[SERVICE][UNIT] has the following methods:
- # - .info - An array of the information of that unit from Juju
- # - .file(PATH) - Get the details of a file on that unit
- # - .file_contents(PATH) - Get plain text output of PATH file from that unit
- # - .directory(PATH) - Get details of directory
- # - .directory_contents(PATH) - List files and folders in PATH on that unit
- # - .relation(relation, service:rel) - Get relation data from return service
-
-
-if __name__ == '__main__':
- unittest.main()
+++ /dev/null
-#!/bin/bash
-
-sudo add-apt-repository ppa:juju/stable -y
-sudo apt-get update
-sudo apt-get install amulet python-requests -y
+++ /dev/null
-#!/usr/bin/python3
-
-import amulet
-import requests
-import unittest
-
-
-class TestCharm(unittest.TestCase):
- def setUp(self):
- self.d = amulet.Deployment()
-
- self.d.add('simple')
- self.d.expose('simple')
-
- self.d.setup(timeout=900)
- self.d.sentry.wait()
-
- self.unit = self.d.sentry['simple'][0]
-
- def test_service(self):
- # test we can access over http
- page = requests.get('http://{}'.format(self.unit.info['public-address']))
- self.assertEqual(page.status_code, 200)
- # Now you can use self.d.sentry[SERVICE][UNIT] to address each of the units and perform
- # more in-depth steps. Each self.d.sentry[SERVICE][UNIT] has the following methods:
- # - .info - An array of the information of that unit from Juju
- # - .file(PATH) - Get the details of a file on that unit
- # - .file_contents(PATH) - Get plain text output of PATH file from that unit
- # - .directory(PATH) - Get details of directory
- # - .directory_contents(PATH) - List files and folders in PATH on that unit
- # - .relation(relation, service:rel) - Get relation data from return service
-
-
-if __name__ == '__main__':
- unittest.main()
await asyncio.sleep(15)
print("test_charm_native stopped")
-
return 'ok'
juju:
charm: metrics-proxy-ci
proxy: true
- """
+ initial-config-primitive:
+ - seq: '1'
+ name: run
+ parameter:
+ - name: command
+ data-type: STRING
+ value: hostname
+"""
# @pytest.mark.serial
@pytest.mark.asyncio
NSD_YAML = """
nsd:nsd-catalog:
nsd:
- - id: charmnative-ns
- name: charmnative-ns
- short-name: charmnative-ns
+ - id: nonstring-ns
+ name: nonstring-ns
+ short-name: nonstring-ns
description: NS with 1 VNFs charmnative-vnf connected by datanet and mgmtnet VLs
version: '1.0'
logo: osm.png
print("Waiting for test to finish...")
await asyncio.sleep(15)
logging.debug("test_charm_non_string_parameter stopped")
-
+ # assert False
return 'ok'
basepython=python3
usedevelop=True
# for testing with other python versions
-commands = py.test --ignore modules/ --tb native -ra -v -s -n auto -k 'not integration' -m 'not serial' {posargs}
+commands = py.test --ignore modules/ --ignore tests/charms/ --tb native -ra -v -s -n auto -k 'not integration' -m 'not serial' {posargs}
passenv =
HOME
VCA_HOST
[testenv:py3]
# default tox env, excludes integration and serial tests
commands =
- pytest --ignore modules/ --tb native -ra -v -s -n auto -k 'not integration' -m 'not serial' {posargs}
+ pytest --ignore modules/ --ignore tests/charms/ --tb native -ra -v -s -n auto -k 'not integration' -m 'not serial' {posargs}
[testenv:lint]
envdir = {toxworkdir}/py3
[testenv:integration]
envdir = {toxworkdir}/py3
-commands = py.test --ignore modules/ --tb native -ra -v -s -n 1 -k 'integration' -m 'serial' {posargs}
+commands = py.test --ignore modules/ --ignore tests/charms/ --tb native -ra -v -s -n 1 -k 'integration' -m 'serial' {posargs}
[testenv:build]
deps =
projects / osm / N2VC.git / commitdiff