How to use klein - 10 common examples

else:
        raise ValueError("Unexpected return value: %r" % (result,))

# Unfortunately Klein imposes this strange requirement that the request object
# be adaptable to KleinRequest.  Klein only registers an adapter from
# twisted.web.server.Request - despite the fact that the adapter doesn't
# actually use the adaptee for anything.
#
# Here, register an adapter from the dummy request type so that tests can
# exercise Klein-based code without trying to use the real request type.
#
# See https://github.com/twisted/klein/issues/31
from twisted.python.components import registerAdapter
from klein.app import KleinRequest
from klein.interfaces import IKleinRequest
registerAdapter(KleinRequest, _DummyRequest, IKleinRequest)


def build_schema_test(name, schema, schema_store,
                      failing_instances, passing_instances):
    """
    Create test case verifying that various instances pass and fail
    verification with a given JSON Schema.

    :param bytes name: Name of test case to create.
    :param dict schema: Schema to test.
    :param dict schema_store: The schema definitions.
    :param list failing_instances: Instances which should fail validation.
    :param list passing_instances: Instances which should pass validation.

    :returns: The test case; a ``SynchronousTestCase`` subclass.
    """

def test_register_notification_api_endpoints(self):
    app = klein.Klein()
    auth = DummyAuthProvider()
    result_foo = [{'_source': {'event_timestamp': 0}}]
    result_bar = [{'_source': {'event_timestamp': 1}}]

    mock_hook = mock.Mock()
    mock_hook.obj.transform.side_effect = lambda x: x
    mock_hook_manager = stevedore.ExtensionManager.make_test_instance([mock_hook])
    mock_extension_manager = stevedore.ExtensionManager.make_test_instance(
        [get_mock_ext(result_foo, 'foo'), get_mock_ext(result_bar, 'bar')])

    with mock.patch('stethoscope.plugins.utils.instantiate_plugins') as \
        mock_instantiate_plugins:
      mock_instantiate_plugins.side_effect = [mock_extension_manager, mock_hook_manager]
      stethoscope.api.endpoints.notifications.register_notification_api_endpoints(app, config, auth)

    # see klein's test_app.py

def test_register_device_api_endpoints(self):
    app = klein.Klein()
    auth = DummyAuthProvider()
    result = ['foobar']

    with mock.patch('stethoscope.plugins.utils.instantiate_plugins') as \
        mock_instantiate_plugins:
      mock_instantiate_plugins.return_value = stevedore.ExtensionManager.make_test_instance(
          [get_mock_ext(result)])
      stethoscope.api.endpoints.devices.register_device_api_endpoints(app, config, auth)

    # see klein's test_app.py
    self.assertEqual(
        app.url_map.bind('devices-foo-email').match("/devices/foo/email/user@example.com"),
        ('devices-foo-email', {'email': "user@example.com"})
    )
    self.assertEqual(
        app.url_map.bind('devices-foo-macaddr').match("/devices/foo/macaddr/de:ca:fb:ad:00:00"),

def setUp(self):
    self.app = app = klein.Klein()
    self.auth = auth = DummyAuthProvider()
    self.config = config = {
      'DEBUG': True,
      'TESTING': True,
    }

    stethoscope.api.factory.register_error_handlers(app, config, auth)

def test_9_by_tx(self):
        mock_req = requestMock(path=b'/tx/0x4c927a7f365cb842ea3576eae474a89183c9e43970a8509b23570a86cb4f5121')
        res = self.app.get_by_tx(mock_req, '0x4c927a7f365cb842ea3576eae474a89183c9e43970a8509b23570a86cb4f5121')
        jsn = json.loads(res)
        self.assertEqual(jsn['total'], 1)
        results = jsn['results']
        self.assertEqual(len(results), 1)

def mock_post_request(body):
    return requestMock(path=b'/', method=b"POST", body=body)

def test_get_host_nonstandard_https_port(hostchecker):
  request = requestMock(b"/foo", host=b'example.com', port=8443, isSecure=True)
  request.requestHeaders.removeHeader(b'host')  # force hostchecker to use transport attributes
  assert hostchecker.get_host(request) == 'example.com:8443'

def test_route_match_required_with_cookie(self):
    request = requestMock(b"/api/user@example.com")
    request.received_cookies = {b'token': mock.sentinel}
    deferred = _render(self.kr, request)

    self.assertEqual(self.successResultOf(deferred), None)
    self.assertEqual(request.getWrittenData(), json.dumps(self.userinfo).encode('ascii'))

    self.auth.decode_token.assert_called_once_with(mock.sentinel)

def test_invalid_serial(self):
    request = requestMock(b"/api/not a valid serial")
    deferred = _render(KleinResource(self.app), request)

    self.assertEqual(self.successResultOf(deferred), None)
    self.assertEqual(request.code, 400)
    self.assertTrue(b"Invalid serial number: 'not a valid serial'." in request.getWrittenData())

def check_result(self, app, path, expected, callback=None):
    # see klein's test_resource.py
    request = klein.test.test_resource.requestMock(path)
    deferred = klein.test.test_resource._render(KleinResource(app), request)
    self.assertEqual(self.successResultOf(deferred), None)
    self.assertEqual(request.getWrittenData(), six.b(json.dumps(expected)))

    if callback is not None:
      callback.assert_called_once_with(expected)