import time
import asyncio
import importlib
from tempfile import NamedTemporaryFile
from micro import di
from micro.web.testing import AppTestCase
from micro.rest.swag import swagger
from .service import Service
from .discover import discover_api, initialize
[docs]class ServiceTestCase(AppTestCase):
"""
The base test class which is capable of running API (see ``run_api()``)
"""
def setUp(self):
super().setUp()
di.container.reset()
swagger.reset()
@asyncio.coroutine
[docs] def run_api(self, *args, **kwargs):
"""
Reload the service module before running it.
This is important to not to share microservice application state between test cases.
:param args: The arguments to pass to your API initializer
:param kwargs: The keyworded arguments to pass to your API initializer
"""
module = importlib.import_module('service')
importlib.reload(module)
api_initializer = discover_api(module)
api_instance = initialize(api_initializer, *args, **kwargs)
if asyncio.iscoroutine(api_instance):
api_instance = yield from api_instance
if isinstance(api_instance, Service):
app = api_instance.production()
else:
app = api_instance.app
return (yield from self.run_app(app))
def make_tmpfile(default_content=None):
tfile = NamedTemporaryFile(delete=False)
name = tfile.name
if default_content:
if isinstance(default_content, str):
default_content = default_content.encode('utf-8')
with open(name, "w"):
tfile.write(default_content)
return name
@asyncio.coroutine
def wait_until(secs, condition):
end_time = time.time() + secs
while True:
result = condition()
if result:
return condition
elif time.time() > end_time:
raise TimeoutError("Timeout waiting for {}".format(condition)) # pragma: no cover
else:
yield from asyncio.sleep(.1)
__all__ = ["ServiceTestCase", "wait_until", "make_tmpfile"]