async def application(scope, receive, send):
    event = await receive()
    ...
    await send({"type": "websocket.send", ...})

{
    "type": "http.request",        # HTTP 요청임을 나타내는 타입
    "body": b"Hello World",        # 요청 본문 (바이트 형식)
    "more_body": False,            # 추가 본문이 없음을 표시
}

{
    "type": "websocket.send",      # WebSocket 메시지 전송을 나타내는 타입
    "text": "Hello world!",        # 전송할 텍스트 메시지
}

async def test_get_request(http_protocol_cls: HTTPProtocol):
    app = Response("Hello, world", media_type="text/plain")

    protocol = get_connected_protocol(app, http_protocol_cls)
    protocol.data_received(SIMPLE_GET_REQUEST)
    await protocol.loop.run_one()
    assert b"HTTP/1.1 200 OK" in protocol.transport.buffer
    assert b"Hello, world" in protocol.transport.buffer

def get_connected_protocol(
    app: ASGIApplication,
    http_protocol_cls: HTTPProtocol,  # 이건 무슨타입일까요.
    lifespan: LifespanOff | LifespanOn | None = None,
    **kwargs: Any,
):
    ...

if TYPE_CHECKING:
    # (생략)
    HTTPProtocol: TypeAlias = "type[HttpToolsProtocol | H11Protocol]"   # 이 코드에 집중합시다!
    WSProtocol: TypeAlias = "type[WebSocketProtocol | _WSProtocol]"

@pytest.fixture(
    params=[
        pytest.param(
            "uvicorn.protocols.http.httptools_impl:HttpToolsProtocol",
            marks=pytest.mark.skipif(
                not importlib.util.find_spec("httptools"),
                reason="httptools not installed.",
            ),
            id="httptools",
        ),
        pytest.param("uvicorn.protocols.http.h11_impl:H11Protocol", id="h11"),
    ]
)
def http_protocol_cls(request: pytest.FixtureRequest):
    return import_from_string(request.param)  # 이건 아래 4번 각주를 참고해주세요.

# uvicorn/protocols/http/h11_impl.py
class H11Protocol(asyncio.Protocol):
    def __init__(
        self,
        config: Config,
        server_state: ServerState,
        app_state: dict[str, Any],
        _loop: asyncio.AbstractEventLoop | None = None,
    ) -> None:
        ...

# uvicorn/protocols/http/httptools_impl.py
class HttpToolsProtocol(asyncio.Protocol):
    def __init__(
        self,
        config: Config,
        server_state: ServerState,
        app_state: dict[str, Any],
        _loop: asyncio.AbstractEventLoop | None = None,
    ) -> None:
        ...

[클라이언트] HTTP 요청
       ↓
[OS 소켓/TCP]
       ↓
[이벤트 루프] (create_server로 생성된)
       ↓
[Transport] (TCP/소켓 추상화)
    │  ↑ (1:1 관계)
    ↓  │
[Protocol] (HTTP 파싱/처리)
       ↓
[ASGI 애플리케이션]

import asyncio
from typing import Callable


class MyAsgiProtocol(asyncio.Protocol):
    def __init__(self, app: Callable):
        self.app = app
        self.transport = None
        self._buffer = b''

def connection_made(self, transport):
    self.transport = transport

def data_received(self, data: bytes):
    self._buffer += data

    # 올바른 HTTP 요청 헤더인지 확인
    if b"\r\n\r\n" in self._buffer:
        asyncio.create_task(self.handle_request())  # 후술합니다

def connection_lost(self, exc):
    if exc:
        print(f"Connection lost: {exc}")
    else:
        print(f"Connection closed")

async def handle_request(self):
    request_data = self._buffer.decode()

    # HTTP 요청 자체를 헤더, 바디로 파싱
    headers, _, body = request_data.partition("\r\n\r\n")

    # HTTP 요청 첫 줄 파싱 (예: "GET / HTTP/1.1")
    request_line, *_ = headers.split("\r\n")
    # HTTP method와 경로를 파싱
    method, path, _ = request_line.split(" ")

    scope = {
        "type": "http",
        "method": method,
        "path": path,
        "headers": [],  # 헤더를 추가할 수 있음
    }

    async def receive():
        """ 애플리케이션이 클라이언트로부터 메시지를 받게 해줌

        :return:
        """
        return {"type": "http.request", "body": body.encode(), "more_body": False}

    async def send(message):
        """ 애플리케이션이 이벤트 메시지를 클라이언트에 전송해줌

        :param message:
        :return:
        """
        if message["type"] == "http.response.start":
            response_headers = b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\n"
            self.transport.write(response_headers)
        elif message["type"] == "http.response.body":
            self.transport.write(message["body"])
            self.transport.close()  # 응답 완료 후 연결 종료

    await self.app(scope, receive, send)

@pytest.fixture
async def server():
    """서버 fixture: 테스트용 서버를 시작하고 정리"""
    loop = asyncio.get_event_loop()
    server = await loop.create_server(
        lambda: MyAsgiProtocol(app),
        "127.0.0.1",
        8000
    )

    try:
        await asyncio.sleep(0.1)    # 서버 인스턴스 생성을 위해 일시대기
        yield server
    finally:
        server.close()
        await server.wait_closed()


@pytest.fixture
async def client():
    """클라이언트 fixture: aiohttp 클라이언트 세션 생성 및 정리"""
    async with aiohttp.ClientSession() as session:
        yield session

@pytest.mark.asyncio
async def test_get_request_returns_hello_world(server, client):
    """기본 HTTP GET 요청 테스트"""
    async with client.get("http://127.0.0.1:8000") as response:
        assert response.status == 200
        text = await response.text()
        assert text == "Hello, World!"

$ pytest -v
================================================================ test session starts =================================================================
platform darwin -- Python 3.12.6, pytest-8.3.4, pluggy-1.5.0 -- /Users/seongeun.yu/Library/Caches/pypoetry/virtualenvs/asgi-explained-S4wxQJIT-py3.12/bin/python
cachedir: .pytest_cache
rootdir: /Users/seongeun.yu/workspace/asgi_explained
configfile: pyproject.toml
plugins: cov-6.0.0, asyncio-0.25.3
asyncio: mode=Mode.AUTO, asyncio_default_fixture_loop_scope=function
collected 1 item                                                                                                                                     

tests/test_server.py::test_get_request_returns_hello_world PASSED                                                                              [100%]

================================================================= 1 passed in 0.16s ==================================================================

async def app(scope, receive, send):
    """ ASGI 프로토콜 서버에서 데이터를 받고, 이를 가공

    :param scope:
    :param receive:
    :param send:
    :return:
    """
    if scope["method"] == "POST":
        request = await receive()
        body = request.get("body", b"")  # 요청 데이터 가져오기
    else:
        body = b"Hello, World!"  # GET 요청에서는 바디가 필요 없음
    
    # 생략

@pytest.mark.asyncio
async def test_post_request_returns_hello_world(server, client):
    """기본 HTTP POST 요청 테스트"""
    async with client.post("http://127.0.0.1:8000", data="Hello, ASGI!") as response:
        assert response.status == 200
        text = await response.text()
        assert text == "Hello, ASGI!"

$ pytest -v
================================================================ test session starts =================================================================
platform darwin -- Python 3.12.6, pytest-8.3.4, pluggy-1.5.0 -- /Users/seongeun.yu/Library/Caches/pypoetry/virtualenvs/asgi-explained-S4wxQJIT-py3.12/bin/python
cachedir: .pytest_cache
rootdir: /Users/seongeun.yu/workspace/asgi_explained
configfile: pyproject.toml
plugins: cov-6.0.0, asyncio-0.25.3
asyncio: mode=Mode.AUTO, asyncio_default_fixture_loop_scope=function
collected 2 items                                                                                                                                    

tests/test_server.py::test_get_request_returns_hello_world PASSED                                                                              [ 50%]
tests/test_server.py::test_post_request_returns_hello_world PASSED                                                                             [100%]

================================================================= 2 passed in 0.27s ==================================================================