Project

General

Profile

Fix #8145

Updated by Christina Meno almost 10 years ago

<pre> bc.. ______________________________________________________________________ TestApi.test_touch_urls _______________________________________________________________________

self = <tests.test_rest_api.TestApi testMethod=test_touch_urls>

def test_touch_urls(self):
"""
This has two purposes:

- To check that the touched URLs service GETs successfully
- To store the returned structures for use as examples in the API documentation

This is a slight intrusion of docs/build on the tests, but it's handy to have this inside
the tests because the infrastructure for setting up a cluster and talking to it is already here.
"""

# Unlike most of the integration tests, this one reaches inside the calamari
# server code to introspect things. Do this inside function scope so that
# import errors are test errors rather than world-breakers.
os.environ.setdefault("CALAMARI_CONFIG", "dev/calamari.conf")
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "calamari_web.settings")
from calamari_common.types import SYNC_OBJECT_TYPES
from calamari_rest.management.commands.api_docs import ApiIntrospector

url_patterns = ApiIntrospector("calamari_rest.urls.v2").get_url_list()
from pprint import pprint
pprint(url_patterns)

# Spin up a running Calamari+Ceph environment
self.ceph_ctl.configure(3)
self.calamari_ctl.configure()
fsid = self._wait_for_cluster()
# Unlike most tests we really do need all the servers to be there in case
# the one we're checking for isn't.
> self._wait_for_servers()

tests/test_rest_api.py:40:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <tests.test_rest_api.TestApi testMethod=test_touch_urls>

def _wait_for_servers(self):
"""
Wait for all the expected servers to appear in the REST API
"""
expected_servers = self.ceph_ctl.get_server_fqdns()
> wait_until_true(lambda: set([s['fqdn'] for s in self.api.get("server").json()]) == set(expected_servers), timeout=30)

tests/server_testcase.py:102:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

condition = <function <lambda> at 0x4d21f50>, timeout = 30

def wait_until_true(condition, timeout=get_timeout()):
elapsed = 0
period = 1
while not condition():
if elapsed >= timeout:
> raise WaitTimeout("After %s seconds (at %s)" % (elapsed, datetime.datetime.utcnow().isoformat()))
E WaitTimeout: After 30 seconds (at 2014-04-18T11:38:47.184349)

tests/utils.py:25: WaitTimeout
</pre> p.

Back