diff --git a/frontik/debug.py b/frontik/debug.py
index 5e023f13c..7058c6f63 100644
--- a/frontik/debug.py
+++ b/frontik/debug.py
@@ -31,6 +31,7 @@
import frontik.xml_util
from frontik import media_types, request_context
from frontik.auth import check_debug_auth
+from frontik.frontik_response import FrontikResponse
from frontik.loggers import BufferedHandler
from frontik.options import options
from frontik.util import get_cookie_or_param_from_request
@@ -366,7 +367,7 @@ def _produce_one(self, record: logging.LogRecord) -> etree.Element:
DEBUG_XSL = os.path.join(os.path.dirname(__file__), 'debug/debug.xsl')
-def _data_to_chunk(data: Any, headers: HTTPHeaders) -> bytes:
+def _data_to_chunk(data: Any) -> bytes:
result: bytes = b''
if data is None:
return result
@@ -375,7 +376,6 @@ def _data_to_chunk(data: Any, headers: HTTPHeaders) -> bytes:
elif isinstance(data, dict):
chunk = json.dumps(data).replace(' bool:
return self.debug_mode.inherited
def transform_chunk(
- self, tornado_request: httputil.HTTPServerRequest, status_code: int, original_headers: HTTPHeaders, data: bytes
- ) -> tuple[int, HTTPHeaders, bytes]:
- chunk = _data_to_chunk(data, original_headers)
-
+ self, tornado_request: httputil.HTTPServerRequest, response: FrontikResponse
+ ) -> FrontikResponse:
if not self.is_enabled():
- return status_code, original_headers, chunk
+ return response
if not self.is_inherited():
wrap_headers = {'Content-Type': media_types.TEXT_HTML}
else:
wrap_headers = {'Content-Type': media_types.APPLICATION_XML, DEBUG_HEADER_NAME: 'true'}
+ chunk = b'Streamable response' if response.data_written else _data_to_chunk(response.body)
start_time = time.time()
+ handler_name = request_context.get_handler_name()
debug_log_data = request_context.get_log_handler().produce_all() # type: ignore
- debug_log_data.set('code', str(int(status_code)))
- debug_log_data.set('handler-name', request_context.get_handler_name())
+ debug_log_data.set('code', str(int(response.status_code)))
+ debug_log_data.set('handler-name', handler_name if handler_name else 'unknown handler')
debug_log_data.set('started', _format_number(tornado_request._start_time))
debug_log_data.set('request-id', str(tornado_request.request_id)) # type: ignore
debug_log_data.set('stages-total', _format_number((time.time() - tornado_request._start_time) * 1000))
@@ -437,12 +437,12 @@ def transform_chunk(
),
)
- debug_log_data.append(E.response(_headers_to_xml(original_headers), _cookies_to_xml(original_headers)))
+ debug_log_data.append(E.response(_headers_to_xml(response.headers), _cookies_to_xml(response.headers)))
original_response = {
'buffer': base64.b64encode(chunk),
- 'headers': dict(original_headers),
- 'code': int(status_code),
+ 'headers': dict(response.headers),
+ 'code': int(response.status_code),
}
debug_log_data.append(dict_to_xml(original_response, 'original-response'))
@@ -471,7 +471,7 @@ def transform_chunk(
else:
log_document = etree.tostring(debug_log_data, encoding='UTF-8', xml_declaration=True)
- return 200, HTTPHeaders(wrap_headers), log_document
+ return FrontikResponse(status_code=200, headers=wrap_headers, body=log_document)
class DebugMode:
diff --git a/frontik/frontik_response.py b/frontik/frontik_response.py
new file mode 100644
index 000000000..e532be4dc
--- /dev/null
+++ b/frontik/frontik_response.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import Mapping
+
+from tornado import httputil
+from tornado.httputil import HTTPHeaders
+
+from frontik import request_context
+from frontik.version import version as frontik_version
+
+
+class FrontikResponse:
+ def __init__(
+ self,
+ status_code: int,
+ headers: dict[str, str] | None | HTTPHeaders = None,
+ body: bytes = b'',
+ ):
+ self.headers = HTTPHeaders(get_default_headers()) # type: ignore
+ if headers is not None:
+ self.headers.update(headers)
+ self.status_code = status_code
+ self.body = body
+ self.data_written = False
+
+ @property
+ def reason(self) -> str:
+ return httputil.responses.get(self.status_code, 'Unknown')
+
+
+def get_default_headers() -> Mapping[str, str | None]:
+ request_id = request_context.get_request_id() or ''
+ return {
+ 'Server': f'Frontik/{frontik_version}',
+ 'X-Request-Id': request_id,
+ }
diff --git a/frontik/handler.py b/frontik/handler.py
index 545ed150f..8702312aa 100644
--- a/frontik/handler.py
+++ b/frontik/handler.py
@@ -1064,11 +1064,3 @@ async def handler_getter(request: Request) -> PageHandlerT:
return request['handler']
return Depends(handler_getter)
-
-
-def get_default_headers() -> dict[str, str]:
- request_id = request_context.get_request_id() or ''
- return {
- 'Server': f'Frontik/{frontik_version}',
- 'X-Request-Id': request_id,
- }
diff --git a/frontik/handler_asgi.py b/frontik/handler_asgi.py
index 592c63f4e..fd8623e38 100644
--- a/frontik/handler_asgi.py
+++ b/frontik/handler_asgi.py
@@ -4,17 +4,17 @@
import http.client
import logging
from functools import partial
-from typing import TYPE_CHECKING, Any, Callable, Optional
+from typing import TYPE_CHECKING, Optional
from tornado import httputil
-from tornado.httputil import HTTPHeaders, HTTPServerRequest
+from tornado.httputil import HTTPServerRequest
from frontik import media_types, request_context
from frontik.debug import DebugMode, DebugTransform
-from frontik.handler import PageHandler, get_default_headers, log_request
+from frontik.frontik_response import FrontikResponse
+from frontik.handler import PageHandler, log_request
from frontik.handler_active_limit import request_limiter
from frontik.http_status import CLIENT_CLOSED_REQUEST
-from frontik.json_builder import JsonBuilder
from frontik.options import options
from frontik.routing import find_route, get_allowed_methods, method_not_allowed_router, not_found_router
from frontik.util import check_request_id, generate_uniq_timestamp_request_id
@@ -46,8 +46,8 @@ async def serve_tornado_request(
partial(_on_connection_close, tornado_request, process_request_task)
)
- status, reason, headers, data = await process_request_task
- log_request(tornado_request, status)
+ response = await process_request_task
+ log_request(tornado_request, response.status_code)
assert tornado_request.connection is not None
tornado_request.connection.set_close_callback(None) # type: ignore
@@ -55,34 +55,35 @@ async def serve_tornado_request(
if getattr(tornado_request, 'canceled', False):
return None
- start_line = httputil.ResponseStartLine('', status, reason)
- future = tornado_request.connection.write_headers(start_line, headers, data)
+ if not response.data_written:
+ start_line = httputil.ResponseStartLine('', response.status_code, response.reason)
+ await tornado_request.connection.write_headers(start_line, response.headers, response.body)
+
tornado_request.connection.finish()
- return await future
async def process_request(
frontik_app: FrontikApplication,
asgi_app: FrontikAsgiApp,
tornado_request: httputil.HTTPServerRequest,
-) -> tuple[int, str, HTTPHeaders, bytes]:
+) -> FrontikResponse:
with request_limiter(frontik_app.statsd_client) as accepted:
if not accepted:
- status, reason, headers, data = make_not_accepted_response()
+ response = make_not_accepted_response()
else:
- status, reason, headers, data = await execute_page(frontik_app, asgi_app, tornado_request)
- headers.add(
+ response = await execute_page(frontik_app, asgi_app, tornado_request)
+ response.headers.add(
'Server-Timing', f'frontik;desc="frontik execution time";dur={tornado_request.request_time()!s}'
)
- return status, reason, headers, data
+ return response
async def execute_page(
frontik_app: FrontikApplication,
asgi_app: FrontikAsgiApp,
tornado_request: HTTPServerRequest,
-) -> tuple[int, str, HTTPHeaders, bytes]:
+) -> FrontikResponse:
debug_mode = make_debug_mode(frontik_app, tornado_request)
if debug_mode.auth_failed():
assert debug_mode.failed_auth_header is not None
@@ -92,26 +93,19 @@ async def execute_page(
assert tornado_request.protocol == 'http'
scope = find_route(tornado_request.path, tornado_request.method)
- data: bytes
if scope['route'] is None:
- status, reason, headers, data = await make_not_found_response(frontik_app, tornado_request, debug_mode)
+ response = await make_not_found_response(frontik_app, tornado_request, debug_mode)
elif scope['page_cls'] is not None:
- status, reason, headers, data = await execute_tornado_page(frontik_app, tornado_request, scope, debug_mode)
+ response = await execute_tornado_page(frontik_app, tornado_request, scope, debug_mode)
else:
- status, reason, headers, data = await execute_asgi_page(
- asgi_app,
- tornado_request,
- scope,
- debug_mode,
- )
+ response = await execute_asgi_page(asgi_app, tornado_request, scope, debug_mode)
- if debug_mode.enabled:
+ if debug_mode.enabled and not response.data_written:
debug_transform = DebugTransform(frontik_app, debug_mode)
- status, headers, data = debug_transform.transform_chunk(tornado_request, status, headers, data)
- reason = httputil.responses.get(status, 'Unknown')
+ response = debug_transform.transform_chunk(tornado_request, response)
- return status, reason, headers, data
+ return response
async def execute_asgi_page(
@@ -119,45 +113,73 @@ async def execute_asgi_page(
tornado_request: HTTPServerRequest,
scope: dict,
debug_mode: DebugMode,
-) -> tuple[int, str, HTTPHeaders, bytes]:
+) -> FrontikResponse:
request_context.set_handler_name(scope['route'])
- result: dict = {'headers': get_default_headers()}
- scope, receive, send = convert_tornado_request_to_asgi(
- asgi_app,
- tornado_request,
- scope,
- debug_mode,
- result,
- )
- await asgi_app(scope, receive, send)
- status: int = result['status']
- reason = httputil.responses.get(status, 'Unknown')
- headers = HTTPHeaders(result['headers'])
- data = result['data']
+ response = FrontikResponse(status_code=200)
- if not scope['json_builder'].is_empty():
- if data != b'null':
- raise RuntimeError('Cant have "return" and "json.put" at the same time')
+ request_headers = [
+ (header.encode(CHARSET).lower(), value.encode(CHARSET))
+ for header in tornado_request.headers
+ for value in tornado_request.headers.get_list(header)
+ ]
- headers['Content-Type'] = media_types.APPLICATION_JSON
- data = scope['json_builder'].to_bytes()
- headers['Content-Length'] = str(len(data))
+ scope.update({
+ 'http_version': tornado_request.version,
+ 'query_string': tornado_request.query.encode(CHARSET),
+ 'headers': request_headers,
+ 'client': (tornado_request.remote_ip, 0),
+ 'http_client_factory': asgi_app.http_client_factory,
+ 'debug_mode': debug_mode,
+ 'start_time': tornado_request._start_time,
+ })
- return status, reason, headers, data
+ async def receive():
+ await asyncio.sleep(0)
+ return {
+ 'body': tornado_request.body,
+ 'type': 'http.request',
+ 'more_body': False,
+ }
+
+ async def send(data):
+ assert tornado_request.connection is not None
+
+ if data['type'] == 'http.response.start':
+ response.status_code = int(data['status'])
+ for h in data['headers']:
+ if len(h) == 2:
+ response.headers.add(h[0].decode(CHARSET), h[1].decode(CHARSET))
+ elif data['type'] == 'http.response.body':
+ chunk = data['body']
+ if debug_mode.enabled or not data.get('more_body'):
+ response.body += chunk
+ elif not response.data_written:
+ await tornado_request.connection.write_headers(
+ start_line=httputil.ResponseStartLine('', response.status_code, response.reason),
+ headers=response.headers,
+ chunk=chunk,
+ )
+ response.data_written = True
+ else:
+ await tornado_request.connection.write(chunk)
+ else:
+ raise RuntimeError(f'Unsupported response type "{data["type"]}" for asgi app')
+
+ await asgi_app(scope, receive, send)
+
+ return response
async def make_not_found_response(
frontik_app: FrontikApplication,
tornado_request: httputil.HTTPServerRequest,
debug_mode: DebugMode,
-) -> tuple[int, str, HTTPHeaders, bytes]:
+) -> FrontikResponse:
allowed_methods = get_allowed_methods(tornado_request.path)
- default_headers = get_default_headers()
- headers: Any
if allowed_methods and len(method_not_allowed_router.routes) != 0:
- status, _, headers, data = await execute_tornado_page(
+ return await execute_tornado_page(
frontik_app,
tornado_request,
{
@@ -167,24 +189,18 @@ async def make_not_found_response(
},
debug_mode,
)
- elif allowed_methods:
- status = 405
- headers = {'Allow': ', '.join(allowed_methods)}
- data = b''
- elif len(not_found_router.routes) != 0:
- status, _, headers, data = await execute_tornado_page(
+
+ if allowed_methods:
+ return FrontikResponse(status_code=405, headers={'Allow': ', '.join(allowed_methods)})
+
+ if len(not_found_router.routes) != 0:
+ return await execute_tornado_page(
frontik_app,
tornado_request,
{'route': not_found_router.routes[0], 'page_cls': not_found_router._cls, 'path_params': {}},
debug_mode,
)
- else:
- status, headers, data = build_error_data(404, 'Not Found')
-
- default_headers.update(headers)
-
- reason = httputil.responses.get(status, 'Unknown')
- return status, reason, HTTPHeaders(headers), data
+ return build_error_data(404, 'Not Found')
def make_debug_mode(frontik_app: FrontikApplication, tornado_request: HTTPServerRequest) -> DebugMode:
@@ -201,28 +217,18 @@ def make_debug_mode(frontik_app: FrontikApplication, tornado_request: HTTPServer
return debug_mode
-def make_debug_auth_failed_response(auth_header: str) -> tuple[int, str, HTTPHeaders, bytes]:
- status = http.client.UNAUTHORIZED
- reason = httputil.responses.get(status, 'Unknown')
- headers = get_default_headers()
- headers['WWW-Authenticate'] = auth_header
-
- return status, reason, HTTPHeaders(headers), b''
+def make_debug_auth_failed_response(auth_header: str) -> FrontikResponse:
+ return FrontikResponse(status_code=http.client.UNAUTHORIZED, headers={'WWW-Authenticate': auth_header})
-def make_not_accepted_response() -> tuple[int, str, HTTPHeaders, bytes]:
- status = http.client.SERVICE_UNAVAILABLE
- reason = httputil.responses.get(status, 'Unknown')
- headers = get_default_headers()
- return status, reason, HTTPHeaders(headers), b''
+def make_not_accepted_response() -> FrontikResponse:
+ return FrontikResponse(status_code=http.client.SERVICE_UNAVAILABLE)
-def build_error_data(
- status_code: int = 500, message: Optional[str] = 'Internal Server Error'
-) -> tuple[int, dict, bytes]:
+def build_error_data(status_code: int = 500, message: Optional[str] = 'Internal Server Error') -> FrontikResponse:
headers = {'Content-Type': media_types.TEXT_HTML}
data = f'{status_code}: {message}{status_code}: {message}'.encode()
- return status_code, headers, data
+ return FrontikResponse(status_code=status_code, headers=headers, body=data)
async def execute_tornado_page(
@@ -230,57 +236,12 @@ async def execute_tornado_page(
tornado_request: httputil.HTTPServerRequest,
scope: dict,
debug_mode: DebugMode,
-) -> tuple[int, str, HTTPHeaders, bytes]:
+) -> FrontikResponse:
route, page_cls, path_params = scope['route'], scope['page_cls'], scope['path_params']
request_context.set_handler_name(route)
handler: PageHandler = page_cls(frontik_app, tornado_request, route, debug_mode, path_params)
- return await handler.execute()
-
-
-def convert_tornado_request_to_asgi(
- asgi_app: FrontikAsgiApp,
- tornado_request: httputil.HTTPServerRequest,
- scope: dict,
- debug_mode: DebugMode,
- result: dict[str, Any],
-) -> tuple[dict, Callable, Callable]:
- headers = [
- (header.encode(CHARSET).lower(), value.encode(CHARSET))
- for header in tornado_request.headers
- for value in tornado_request.headers.get_list(header)
- ]
-
- scope.update({
- 'http_version': tornado_request.version,
- 'query_string': tornado_request.query.encode(CHARSET),
- 'headers': headers,
- 'client': (tornado_request.remote_ip, 0),
- 'http_client_factory': asgi_app.http_client_factory,
- 'debug_mode': debug_mode,
- 'start_time': tornado_request._start_time,
- 'json_builder': JsonBuilder(),
- })
-
- async def receive():
- return {
- 'body': tornado_request.body,
- 'type': 'http.request',
- 'more_body': False,
- }
-
- async def send(data):
- if data['type'] == 'http.response.start':
- result['status'] = data['status']
- for h in data['headers']:
- if len(h) == 2:
- result['headers'][h[0].decode(CHARSET)] = h[1].decode(CHARSET)
- elif data['type'] == 'http.response.body':
- assert isinstance(data['body'], bytes)
- result['data'] = data['body']
- else:
- raise RuntimeError(f'Unsupported response type "{data["type"]}" for asgi app')
-
- return scope, receive, send
+ status_code, _, headers, body = await handler.execute()
+ return FrontikResponse(status_code=status_code, headers=headers, body=body)
def _on_connection_close(tornado_request, process_request_task):
diff --git a/tests/projects/test_app/pages/debug.py b/tests/projects/test_app/pages/debug.py
index 95913ee0f..f2883287d 100644
--- a/tests/projects/test_app/pages/debug.py
+++ b/tests/projects/test_app/pages/debug.py
@@ -1,12 +1,56 @@
+import logging
+
+from fastapi import Request
from lxml.builder import E
from frontik import media_types
+from frontik.balancing_client import HttpClientT
from frontik.handler import PageHandler, get_current_handler
-from frontik.routing import plain_router
+from frontik.routing import plain_router, router
+
+logger = logging.getLogger('handler')
+
+
+@router.get('/debug_asgi')
+async def get_page_asgi(http_client: HttpClientT, request: Request, no_recursion: str = 'false') -> None:
+ logger.debug('debug: starting debug page')
+
+ def _exception_trace() -> None:
+ def _inner() -> None:
+ msg = 'Testing an exception юникод'
+ raise ValueError(msg)
+
+ _inner()
+
+ try:
+ _exception_trace()
+ except ValueError:
+ logger.exception('exception catched')
+
+ logger.warning('warning: testing simple inherited debug')
+ await http_client.post_url(request.headers.get('host', 'no_host'), '/debug')
+
+ logger.error('error: testing failing urls')
+ await http_client.get_url('invalid_host', 'invalid_url')
+
+ logger.info('info: testing responses')
+ await http_client.put_url(request.headers.get('host', 'no_host'), 'debug?type=html')
+ await http_client.put_url(request.headers.get('host', 'no_host'), 'debug?type=protobuf')
+ await http_client.put_url(request.headers.get('host', 'no_host'), 'debug?type=xml')
+ await http_client.put_url(request.headers.get('host', 'no_host'), 'debug?type=javascript')
+ await http_client.put_url(request.headers.get('host', 'no_host'), 'debug?type=text')
+
+ logger.debug('testing xml output', extra={'_xml': E.root(E.child1(param='тест'), E.child2('тест'))})
+ logger.debug('testing utf-8 text output', extra={'_text': 'some\nmultiline\nюникод\ndebug'})
+ logger.debug('testing unicode text output', extra={'_text': 'some\nmultiline\nюникод\ndebug'})
+
+ if no_recursion != 'true':
+ logger.debug('testing complex inherited debug')
+ await http_client.get_url(request.headers.get('host', 'no_host'), 'debug?no_recursion=true&debug=xslt')
@plain_router.get('/debug', cls=PageHandler)
-async def get_page(handler=get_current_handler()):
+async def get_page(handler: PageHandler = get_current_handler()) -> None:
handler.log.debug('debug: starting debug page')
def _exception_trace() -> None:
@@ -22,21 +66,21 @@ def _inner() -> None:
handler.log.exception('exception catched')
handler.log.warning('warning: testing simple inherited debug')
- await handler.post_url(handler.get_header('host'), handler.path)
+ await handler.post_url(handler.get_header('host', 'no_host'), handler.path)
handler.log.error('error: testing failing urls')
await handler.get_url('invalid_host', 'invalid_url')
handler.log.info('info: testing responses')
- await handler.put_url(handler.get_header('host'), handler.path + '?type=html')
- await handler.put_url(handler.get_header('host'), handler.path + '?type=protobuf')
- await handler.put_url(handler.get_header('host'), handler.path + '?type=xml')
- await handler.put_url(handler.get_header('host'), handler.path + '?type=javascript')
- await handler.put_url(handler.get_header('host'), handler.path + '?type=text')
+ await handler.put_url(handler.get_header('host', 'no_host'), handler.path + '?type=html')
+ await handler.put_url(handler.get_header('host', 'no_host'), handler.path + '?type=protobuf')
+ await handler.put_url(handler.get_header('host', 'no_host'), handler.path + '?type=xml')
+ await handler.put_url(handler.get_header('host', 'no_host'), handler.path + '?type=javascript')
+ await handler.put_url(handler.get_header('host', 'no_host'), handler.path + '?type=text')
if handler.get_query_argument('no_recursion', 'false') != 'true':
handler.log.debug('testing complex inherited debug')
- await handler.get_url(handler.get_header('host'), handler.path + '?no_recursion=true&debug=xslt')
+ await handler.get_url(handler.get_header('host', 'no_host'), handler.path + '?no_recursion=true&debug=xslt')
else:
handler.log.debug('testing xslt profiling')
handler.set_xsl('simple.xsl')
@@ -47,14 +91,14 @@ def _inner() -> None:
@plain_router.post('/debug', cls=PageHandler)
-async def post_page(handler=get_current_handler()):
+async def post_page(handler: PageHandler = get_current_handler()) -> None:
handler.log.debug('this page returns json')
handler.json.put({'param1': 'value', 'param2': 'тест', 'тест': 'value'})
@plain_router.put('/debug', cls=PageHandler)
-async def put_page(handler=get_current_handler()):
+async def put_page(handler: PageHandler = get_current_handler()) -> None:
content_type = handler.get_query_argument('type')
if content_type == 'html':
diff --git a/tests/projects/test_app/pages/handler/delete.py b/tests/projects/test_app/pages/handler/delete.py
index 0f655c8ca..25862fc8c 100644
--- a/tests/projects/test_app/pages/handler/delete.py
+++ b/tests/projects/test_app/pages/handler/delete.py
@@ -1,26 +1,25 @@
+from typing import Any
+
from fastapi import Request
from frontik.balancing_client import HttpClientT
-from frontik.json_builder import JsonBuilderT
from frontik.routing import plain_router
@plain_router.get('/handler/delete')
-async def get_page(request: Request, http_client: HttpClientT, json_builder: JsonBuilderT) -> None:
+async def get_page(request: Request, http_client: HttpClientT) -> Any:
result = await http_client.delete_url(
'http://' + request.headers.get('host', ''), request.url.path, data={'data': 'true'}
)
- if not result.failed:
- json_builder.put(result.data)
+ return result.data
@plain_router.post('/handler/delete')
-async def post_page(request: Request, http_client: HttpClientT, json_builder: JsonBuilderT) -> None:
+async def post_page(request: Request, http_client: HttpClientT) -> Any:
result = await http_client.delete_url('http://backend', request.url.path, fail_fast=True)
- if not result.failed:
- json_builder.put(result.data)
+ return result.data
@plain_router.delete('/handler/delete')
-async def delete_page(data: str, json_builder: JsonBuilderT) -> None:
- json_builder.put({'delete': data})
+async def delete_page(data: str) -> dict:
+ return {'delete': data}
diff --git a/tests/projects/test_app/pages/streaming.py b/tests/projects/test_app/pages/streaming.py
new file mode 100644
index 000000000..6142780cf
--- /dev/null
+++ b/tests/projects/test_app/pages/streaming.py
@@ -0,0 +1,15 @@
+from typing import AsyncIterable
+
+from fastapi.responses import StreamingResponse
+
+from frontik.media_types import TEXT_PLAIN
+from frontik.routing import plain_router as router
+
+
+@router.get('/stream')
+async def get_page():
+ async def iterable() -> AsyncIterable:
+ yield b'response+'
+ yield b'second_part'
+
+ return StreamingResponse(content=iterable(), headers={'Content-type': TEXT_PLAIN})
diff --git a/tests/test_debug.py b/tests/test_debug.py
index ccc3261cf..1f8adacb2 100644
--- a/tests/test_debug.py
+++ b/tests/test_debug.py
@@ -16,6 +16,72 @@
class TestDebug(unittest.TestCase):
DEBUG_BASIC_AUTH = create_basic_auth_header('user:god')
+ def test_asgi_debug_page(self):
+ response = frontik_test_app.get_page('debug_asgi?debug')
+ response_content = to_unicode(response.content)
+
+ self.assertEqual(response.status_code, 200)
+
+ # Basic debug messages
+
+ basic_messages = (
+ 'debug: starting debug page',
+ 'warning: testing simple inherited debug',
+ 'error: testing failing urls',
+ 'info: testing responses',
+ )
+
+ for msg in basic_messages:
+ assert msg in response_content
+
+ # Extra output and different types of content
+
+ extra_output = (
+ '<child2>тест</child2>',
+ 'юникод\ndebug',
+ '"тест": "value"',
+ 'SomeProtobufObject()',
+ '<response>some xml</response>',
+ 'document.body.write("Привет")',
+ 'привет charset',
+ )
+
+ for msg in extra_output:
+ assert msg in response_content
+
+ # Check that all http requests are present
+
+ self.assertEqual(response_content.count(''), 17)
+
+ # Inherited debug
+
+ assert_occurs_twice = (
+ 'ValueError: Testing an exception',
+ 'Exception traceback',
+ 'testing xml output',
+ 'testing utf-8 text output',
+ 'testing unicode text output',
+ )
+
+ for msg in assert_occurs_twice:
+ self.assertEqual(response_content.count(msg), 2)
+
+ # Check that everything went right
+
+ assert_not_found = (
+ 'cannot parse request body',
+ 'cannot parse response body',
+ 'cannot append time info',
+ 'cannot log response info',
+ 'cannot decode parameter name or value',
+ 'cannot add traceback lines',
+ 'error creating log entry with attrs',
+ 'XSLT debug file error',
+ )
+
+ for msg in assert_not_found:
+ self.assertNotIn(msg, response_content)
+
def test_complex_debug_page(self):
response = frontik_test_app.get_page('debug?debug')
response_content = to_unicode(response.content)
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
new file mode 100644
index 000000000..68be676ab
--- /dev/null
+++ b/tests/test_streaming.py
@@ -0,0 +1,8 @@
+from tests.instances import frontik_test_app
+
+
+class TestStreaming:
+ def test_streaming_response(self):
+ response = frontik_test_app.get_page('stream')
+ assert response.headers['content-type'] == 'text/plain'
+ assert response.content == b'response+second_part'