How to use the pysoa.server.server.Server function in pysoa

def test_server_settings_generic_with_defaults(self):
        """The server is successfully instantiated with generic settings, Redis Transport"""

        class _TestServer(Server):
            service_name = 'tag_geo'

        settings_dict = {
            'transport': {
                'path': 'pysoa.common.transport.redis_gateway.server:RedisServerTransport',
                'kwargs': {
                    'backend_type': REDIS_BACKEND_TYPE_STANDARD,
                }
            },
        }

        server = _TestServer(ServerSettings(settings_dict))

        assert isinstance(server.transport, RedisServerTransport)
        assert server.transport._receive_queue_name == 'service.tag_geo'
        assert isinstance(server.transport.core, RedisTransportCore)

def harakiri_loop_forceful(*_, **__):
    def action(*___):
        i = 0
        while True:
            try:
                while True:
                    i += math.factorial(10)
            except HarakiriInterrupt:
                pass  # muuuuaaaaaaahahaha I won't let go!

    return action


class Server(BaseServer):
    service_name = 'echo'
    action_class_map = {
        'harakiri_loop_graceful': harakiri_loop_graceful,
        'harakiri_loop_forceful': harakiri_loop_forceful,
    }

class SwitchedActionOne(SwitchedAction):
    switch_to_action_map = (
        (5, FakeActionTwo),
        (3, FakeActionOne),
    )


class SwitchedActionTwo(SwitchedAction):
    switch_to_action_map = (
        (4, FakeActionOne),
        (SwitchedAction.DEFAULT_ACTION, FakeActionTwo),
    )


class FakeServerOne(Server):
    """This is the documentation we should get"""

    action_class_map = {
        'status': StatusActionFactory('1.2.3'),
        'one': FakeActionOne,
    }

    # noinspection PyMissingConstructor
    def __init__(self):
        # Do not call super
        self.settings = cast(ServerSettings, {})


class FakeServerTwo(Server):
    """This is NOT the documentation we should get"""

except Exception:
                self.error_count += 1
                raise
            finally:
                self.request_count += 1
        return handler


def _job_error(*_, **__):
    def a(*_, **__):
        raise JobError(errors=[Error(code='BAD_JOB', message='You are a bad job')])

    return a


class ErrorServer(Server):
    service_name = 'error_service'

    # noinspection PyTypeChecker
    action_class_map = {
        'job_error': _job_error,
        'okay_action': lambda *_, **__: lambda *_, **__: ActionResponse(action='okay_action', body={'no_error': True}),
    }


@fields.ClassConfigurationSchema.provider(fields.Dictionary({}))
class SendErrorTransport(ClientTransport):
    def send_request_message(self, request_id, meta, body, message_expiry_in_seconds=None):
        raise MessageSendError('The message failed to send')

    def receive_response_message(self, receive_timeout_in_seconds=None):
        raise AssertionError('Something weird happened; receive should not have been called')

def test_make_default_status_action_class(self):
        class TestServer1(Server):
            pass

        class TestServer2(Server):
            pass

        module1 = mock.MagicMock()
        module1.__name__ = 'neat_service'
        module1.__version__ = '1.6.3'
        module2 = mock.MagicMock()
        module2.__name__ = 'cooler_service'
        module2.__version__ = '3.15.5'
        module2.__build__ = 'cooler_service-3.15.5-af7ed3c'

        TestServer1.__module__ = 'neat_service.server'
        TestServer2.__module__ = 'cooler_service.further.lower.server'

class FakeServerOne(Server):
    """This is the documentation we should get"""

    action_class_map = {
        'status': StatusActionFactory('1.2.3'),
        'one': FakeActionOne,
    }

    # noinspection PyMissingConstructor
    def __init__(self):
        # Do not call super
        self.settings = cast(ServerSettings, {})


class FakeServerTwo(Server):
    """This is NOT the documentation we should get"""

    description = 'Instead, we should get this documentation'

    action_class_map = {
        'introspect': IntrospectionAction,
        'one': FakeActionOne,
        'two': FakeActionTwo,
    }

    # noinspection PyMissingConstructor
    def __init__(self):
        # Do not call super
        self.settings = cast(ServerSettings, {})

from __future__ import (
    absolute_import,
    unicode_literals,
)

from unittest import TestCase

from pysoa.common.constants import ERROR_CODE_INVALID
from pysoa.common.errors import Error
from pysoa.server.errors import ActionError
from pysoa.server.middleware import ServerMiddleware
from pysoa.server.server import Server
from pysoa.test import factories


class ProcessJobServer(Server):
    """
    Stub server to test against.
    """
    service_name = 'test_service'
    action_class_map = {
        'respond_actionerror': factories.ActionFactory(exception=ActionError(errors=[Error(
            code=ERROR_CODE_INVALID,
            message='This field is invalid.',
            field='body.field',
        )])),
        'respond_empty': factories.ActionFactory(),
    }


class ProcessJobMiddleware(ServerMiddleware):
    """

unicode_literals,
)

from typing import Mapping
from unittest import TestCase

from conformity import fields
import six

from pysoa.common.transport.base import ServerTransport
from pysoa.server.server import Server
from pysoa.server.types import ActionType
from pysoa.test import factories


class HandleNextRequestServer(Server):
    """
    Stub server to test against.
    """
    service_name = 'test_service'
    action_class_map = {}  # type: Mapping[six.text_type, ActionType]


@fields.ClassConfigurationSchema.provider(fields.Dictionary({}))
class SimplePassthroughServerTransport(ServerTransport):
    def set_request(self, request):
        self._request = request

    def receive_request_message(self):
        return (0, {}, self._request)

    def send_response_message(self, request_id, meta, body):

from pysoa.server.server import Server as BaseServer
from pysoa.server.action.base import Action


class VeryLargeResponseAction(Action):
    def run(self, request):
        return {'key-{}'.format(i): 'value-{}'.format(i) for i in range(10000, 47000)}


class Server(BaseServer):
    service_name = 'meta'
    action_class_map = {
        'very_large_response': VeryLargeResponseAction,
    }

)


_server_settings = fields.SchemalessDictionary(
    key_type=fields.UnicodeString(),
    description='A dictionary of settings for the server (which will further validate them).',
)


class LocalClientTransportSchema(fields.Dictionary):
    contents = {
        # Server class can be an import path or a class object
        'server_class': fields.Any(
            fields.TypePath(
                description='The importable Python path to the `Server`-extending class.',
                base_classes=Server,
            ),
            fields.TypeReference(
                description='A reference to the `Server`-extending class',
                base_classes=Server,
            ),
            description='The path to the `Server` class to use locally (as a library), or a reference to the '
                        '`Server`-extending class/type itself.',
        ),
        # No deeper validation than "schemaless dictionary" because the Server will perform its own validation
        'server_settings': fields.Any(
            fields.PythonPath(
                value_schema=_server_settings,
                description='The importable Python path to the settings dict, in the format "module.name:VARIABLE".',
            ),
            _server_settings,
            description='The settings to use when instantiating the `server_class`.',