Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import tomodachi
from tomodachi.discovery.dummy_registry import DummyRegistry
from tomodachi.protocol.json_base import JsonBase
from .code import test_func
@tomodachi.service
class DummyService(tomodachi.Service):
name = 'test_dummy'
discovery = [DummyRegistry]
message_protocol = JsonBase
options = {
'aws_sns_sqs': {
'region_name': 'eu-west-1',
'aws_access_key_id': 'XXXXXXXXX',
'aws_secret_access_key': 'XXXXXXXXX'
},
'amqp': {
'port': 54321,
'login': 'invalid',
'password': 'invalid'
}
}
start = False
started = False
import asyncio
import os
import signal
import tomodachi
from tomodachi.discovery.dummy_registry import DummyRegistry
from tomodachi.protocol.json_base import JsonBase
@tomodachi.service
class AutoClosingService(tomodachi.Service):
name = 'test_auto_closing'
discovery = [DummyRegistry]
message_protocol = JsonBase
start = False
started = False
stop = False
async def _start_service(self) -> None:
self.start = True
async def _started_service(self) -> None:
self.started = True
await asyncio.sleep(0.1)
os.kill(os.getpid(), signal.SIGTERM)
async def _stop_service(self) -> None:
self.stop = True
import tomodachi
from tomodachi.discovery.dummy_registry import DummyRegistry
from tomodachi.protocol.json_base import JsonBase
@tomodachi.service
class DummyService(tomodachi.Service):
name = 'test_dummy'
discovery = [DummyRegistry]
message_protocol = JsonBase
options = {
'aws_sns_sqs': {
'region_name': 'eu-west-1',
'aws_access_key_id': 'XXXXXXXXX',
'aws_secret_access_key': 'XXXXXXXXX'
},
'amqp': {
'port': 54321,
'login': 'invalid',
'password': 'invalid'
}
}
start = False
started = False
if request.headers.get('X-Use-Middleware') == 'Before':
return 'before'
return_value = await func()
if request.headers.get('X-Use-Middleware') == 'After':
return 'after'
return return_value
@tomodachi.service
class HttpService(tomodachi.Service):
name = 'test_http'
discovery = [DummyRegistry]
options = {
'http': {
'port': None,
'access_log': True,
'real_ip_from': '127.0.0.1'
}
}
uuid = None
closer = asyncio.Future() # type: Any
http_middleware = [middleware_function]
slow_request = False
middleware_called = False
function_triggered = False
websocket_connected = False
websocket_received_data = None
websocket_header = None
import asyncio
import os
import signal
import tomodachi
from tomodachi.discovery.dummy_registry import DummyRegistry
from tomodachi.protocol.json_base import JsonBase
@tomodachi.service
class AutoClosingService(tomodachi.Service):
name = 'test_auto_closing'
discovery = [DummyRegistry]
message_protocol = JsonBase
start = False
started = False
stop = False
async def _start_service(self) -> None:
self.start = True
async def _started_service(self) -> None:
self.started = True
await asyncio.sleep(0.1)
os.kill(os.getpid(), signal.SIGINT)
async def _stop_service(self) -> None:
self.stop = True
import asyncio
import os
import signal
import tomodachi
from tomodachi.discovery.dummy_registry import DummyRegistry
from tomodachi.protocol.json_base import JsonBase
@tomodachi.service
class AutoClosingService(tomodachi.Service):
name = 'test_auto_closing'
discovery = [DummyRegistry]
message_protocol = JsonBase
start = False
started = False
stop = False
async def _start_service(self) -> None:
self.log('_start_service() function called')
self.start = True
async def _started_service(self) -> None:
self.log('_started_service() function called')
self.started = True
await asyncio.sleep(0.1)
os.kill(os.getpid(), signal.SIGTERM)
import tomodachi
from tomodachi.discovery.dummy_registry import DummyRegistry
from tomodachi.protocol.protobuf_base import ProtobufBase
@tomodachi.service
class DummyService(tomodachi.Service):
name = 'test_dummy_protobuf'
discovery = [DummyRegistry]
message_protocol = ProtobufBase
options = {
'aws_sns_sqs': {
'region_name': 'eu-west-1',
'aws_access_key_id': 'XXXXXXXXX',
'aws_secret_access_key': 'XXXXXXXXX'
},
'amqp': {
'port': 54321,
'login': 'invalid',
'password': 'invalid'
}
}
start = False
started = False