You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

71 lines
3.1 KiB

import logging
import socket
import unittest
from collections.abc import Generator
from tornado_py3 import gen as gen, netutil as netutil
from tornado_py3.httpclient import AsyncHTTPClient as AsyncHTTPClient, HTTPResponse as HTTPResponse
from tornado_py3.httpserver import HTTPServer as HTTPServer
from tornado_py3.ioloop import IOLoop as IOLoop, TimeoutError as TimeoutError
from tornado_py3.log import app_log as app_log
from tornado_py3.platform.asyncio import AsyncIOMainLoop as AsyncIOMainLoop
from tornado_py3.process import Subprocess as Subprocess
from tornado_py3.util import basestring_type as basestring_type, raise_exc_info as raise_exc_info
from tornado_py3.web import Application as Application
from types import TracebackType
from typing import Any, Callable, Coroutine, Dict, Optional, Tuple, Type, Union
def bind_unused_port(reuse_port: bool=...) -> Tuple[socket.socket, int]: ...
def get_async_test_timeout() -> float: ...
class _TestMethodWrapper:
orig_method: Any = ...
def __init__(self, orig_method: Callable) -> None: ...
def __call__(self, *args: Any, **kwargs: Any) -> None: ...
def __getattr__(self, name: str) -> Any: ...
class AsyncTestCase(unittest.TestCase):
def __init__(self, methodName: str=...) -> None: ...
io_loop: Any = ...
def setUp(self) -> None: ...
def tearDown(self) -> None: ...
def get_new_ioloop(self) -> IOLoop: ...
def run(self, result: unittest.TestResult=...) -> unittest.TestCase: ...
def stop(self, _arg: Any=..., **kwargs: Any) -> None: ...
def wait(self, condition: Callable[..., bool]=..., timeout: float=...) -> None: ...
class AsyncHTTPTestCase(AsyncTestCase):
http_client: Any = ...
http_server: Any = ...
def setUp(self) -> None: ...
def get_http_client(self) -> AsyncHTTPClient: ...
def get_http_server(self) -> HTTPServer: ...
def get_app(self) -> Application: ...
def fetch(self, path: str, raise_error: bool=..., **kwargs: Any) -> HTTPResponse: ...
def get_httpserver_options(self) -> Dict[str, Any]: ...
def get_http_port(self) -> int: ...
def get_protocol(self) -> str: ...
def get_url(self, path: str) -> str: ...
def tearDown(self) -> None: ...
class AsyncHTTPSTestCase(AsyncHTTPTestCase):
def get_http_client(self) -> AsyncHTTPClient: ...
def get_httpserver_options(self) -> Dict[str, Any]: ...
def get_ssl_options(self) -> Dict[str, Any]: ...
@staticmethod
def default_ssl_options() -> Dict[str, Any]: ...
def get_protocol(self) -> str: ...
def gen_test(*, timeout: float=...) -> Callable[[Callable[..., Union[Generator, Coroutine]]], Callable[..., None]]: ...
class ExpectLog(logging.Filter):
logger: Any = ...
regex: Any = ...
required: Any = ...
matched: bool = ...
logged_stack: bool = ...
def __init__(self, logger: Union[logging.Logger, basestring_type], regex: str, required: bool=...) -> None: ...
def filter(self, record: logging.LogRecord) -> bool: ...
def __enter__(self) -> ExpectLog: ...
def __exit__(self, typ: Optional[Type[BaseException]], value: Optional[BaseException], tb: Optional[TracebackType]) -> None: ...
def main(**kwargs: Any) -> None: ...