How to use daphne - 10 common examples

# Clear result storage
        TestApplication.delete_setup()
        TestApplication.delete_result()
        # Option Daphne features
        kwargs = {}
        # Optionally enable X-Forwarded-For support.
        if self.xff:
            kwargs["proxy_forwarded_address_header"] = "X-Forwarded-For"
            kwargs["proxy_forwarded_port_header"] = "X-Forwarded-Port"
            kwargs["proxy_forwarded_proto_header"] = "X-Forwarded-Proto"
        if self.http_timeout:
            kwargs["http_timeout"] = self.http_timeout
        # Start up process
        self.process = DaphneProcess(
            host=self.host,
            application=TestApplication,
            kwargs=kwargs,
            setup=self.process_setup,
            teardown=self.process_teardown,
        )
        self.process.start()
        # Wait for the port
        if self.process.ready.wait(self.startup_timeout):
            self.port = self.process.port.value
            return self
        else:
            if self.process.errors.empty():
                raise RuntimeError("Daphne did not start up, no error caught")
            else:
                error, traceback = self.process.errors.get(False)
                raise RuntimeError("Daphne did not start up:\n%s" % traceback)

def __enter__(self):
        # Clear result storage
        TestApplication.delete_setup()
        TestApplication.delete_result()
        # Option Daphne features
        kwargs = {}
        # Optionally enable X-Forwarded-For support.
        if self.xff:
            kwargs["proxy_forwarded_address_header"] = "X-Forwarded-For"
            kwargs["proxy_forwarded_port_header"] = "X-Forwarded-Port"
            kwargs["proxy_forwarded_proto_header"] = "X-Forwarded-Proto"
        if self.http_timeout:
            kwargs["http_timeout"] = self.http_timeout
        # Start up process
        self.process = DaphneProcess(
            host=self.host,
            application=TestApplication,
            kwargs=kwargs,
            setup=self.process_setup,
            teardown=self.process_teardown,
        )
        self.process.start()
        # Wait for the port
        if self.process.ready.wait(self.startup_timeout):
            self.port = self.process.port.value
            return self
        else:
            if self.process.errors.empty():
                raise RuntimeError("Daphne did not start up, no error caught")
            else:
                error, traceback = self.process.errors.get(False)

def test_no_original(self):
        headers = {}
        self.assertEqual(parse_x_forwarded_for(headers), (None, None))

def test_address_only(self):
        headers = Headers({b"X-Forwarded-For": [b"10.1.2.3"]})
        self.assertEqual(parse_x_forwarded_for(headers), (["10.1.2.3", 0], None))

def test_multiple_proxys(self):
        headers = Headers({b"X-Forwarded-For": [b"10.1.2.3, 10.1.2.4"]})
        self.assertEqual(parse_x_forwarded_for(headers), (["10.1.2.3", 0], None))

def test_original(self):
        headers = Headers({})
        self.assertEqual(
            parse_x_forwarded_for(headers, original_addr=["127.0.0.1", 80]),
            (["127.0.0.1", 80], None),
        )

def test_basic(self):
        headers = {
            b"X-Forwarded-For": b"10.1.2.3",
            b"X-Forwarded-Port": b"1234",
            b"X-Forwarded-Proto": b"https",
        }
        self.assertEqual(parse_x_forwarded_for(headers), (["10.1.2.3", 1234], "https"))

def test_multiple_proxies(self):
        headers = {b"X-Forwarded-For": b"10.1.2.3, 10.1.2.4"}
        self.assertEqual(parse_x_forwarded_for(headers), (["10.1.2.3", 0], None))

def test_v6_address(self):
        headers = {b"X-Forwarded-For": [b"1043::a321:0001, 10.0.5.6"]}
        self.assertEqual(parse_x_forwarded_for(headers), (["1043::a321:0001", 0], None))

def test_address_only(self):
        headers = {b"X-Forwarded-For": b"10.1.2.3"}
        self.assertEqual(parse_x_forwarded_for(headers), (["10.1.2.3", 0], None))