patx/mrhttp-asgi

import socket

import requests

import tests.common
from tests.common import eq, stop_server


server = None


def setup():
  global server
  server = tests.common.start_server("tests/s_asgi.py")
  if not server:
    return 1
  return 0


def setup_module(module):
  assert setup() == 0


def test_asgi_http():
  r = requests.get("http://localhost:8080/")
  eq(r.status_code, 200)
  eq(r.text, "hello asgi")
  eq(r.headers["content-type"], "text/plain")

  r = requests.post("http://localhost:8080/echo", data=b"posted")
  eq(r.status_code, 200)
  eq(r.content, b"posted")

  sock = socket.create_connection(("127.0.0.1", 8080), timeout=3)
  try:
    sock.sendall(
      b"GET /scope?x=1 HTTP/1.1\r\n"
      b"Host: localhost\r\n"
      b"X-Dup: one\r\n"
      b"X-Dup: two\r\n"
      b"\r\n"
    )
    data = sock.recv(4096)
    assert b"\r\n\r\nGET|/scope|/scope|x=1|2|True" in data, data
  finally:
    sock.close()


def test_chunked_request_body():
  sock = socket.create_connection(("127.0.0.1", 8080), timeout=3)
  try:
    sock.sendall(
      b"POST /echo HTTP/1.1\r\n"
      b"Host: localhost\r\n"
      b"Transfer-Encoding: chunked\r\n"
      b"\r\n"
      b"4\r\nWiki\r\n"
      b"5\r\npedia\r\n"
      b"0\r\n\r\n"
    )
    data = sock.recv(4096)
    assert b"\r\n\r\nWikipedia" in data, data
  finally:
    sock.close()


def teardown():
  global server
  proc = server
  server = None
  stop_server(proc)
  if proc is not None:
    try:
      proc.wait(timeout=2)
    except Exception:
      pass
    if proc.stdout:
      proc.stdout.close()
    if proc.stderr:
      proc.stderr.close()


def teardown_module(module):
  teardown()