import pytest
import soyokaze
from soyokaze import ALPN, BodyKind, HeaderCase, Headers, Method, Port, Role, TransportKind, URL, Version
from soyokaze.helpers import fields, scan, sync, text
def test_versions_carry_their_rfc_alpn_identifiers():
assert Version.V1_1.alpn() == "http/1.1"
assert Version.V2_0.alpn() == "h2"
assert Version.V3_0.alpn() == "h3"
assert Version.from_alpn(b"h2") == Version.V2_0
assert Version.from_alpn(b"h3") == Version.V3_0
assert Version.from_alpn(b"spdy/3") is None
assert Version.V1_0.major() == 1 and Version.V1_1.major() == 1
assert Version.V2_0.major() == 2 and Version.V3_0.major() == 3
def test_a_version_runs_over_the_transport_its_alpn_implies():
assert Version.V1_1.transport() == TransportKind.STREAM
assert Version.V2_0.transport() == TransportKind.STREAM
assert Version.V3_0.transport() == TransportKind.QUIC
def test_a_version_round_trips_through_its_written_form():
for version in Version:
assert Version.parse(version.as_str()) == version
assert Version.parse("HTTP/9") is None
def test_a_port_carries_exactly_the_versions_its_transport_carries():
assert Port.TCP(80).transport() == TransportKind.STREAM
assert Port.QUIC(443).transport() == TransportKind.QUIC
assert Port.UDS("/tmp/s").transport() == TransportKind.STREAM
assert Port.TCP(80).carries(Version.V2_0)
assert not Port.TCP(80).carries(Version.V3_0)
assert Port.QUIC(443).carries(Version.V3_0)
assert not Port.QUIC(443).carries(Version.V2_0)
offered = Port.TCP(80).offers([Version.V3_0, Version.V2_0, Version.V1_1])
assert offered == [Version.V2_0, Version.V1_1], "the order the caller gave must be kept"
def test_alpn_is_written_as_length_prefixed_identifiers():
assert ALPN.wire([Version.V2_0]) == b"\x02h2"
assert ALPN.wire([Version.V3_0, Version.V2_0]) == b"\x02h3\x02h2"
assert ALPN.wire([Version.V1_1]) == b"\x08http/1.1"
def test_alpn_selection_prefers_this_ends_order_not_the_clients():
offered = [Version.V3_0, Version.V2_0]
assert ALPN.select(offered, ALPN.wire([Version.V2_0, Version.V3_0])) == b"h3"
assert ALPN.select([Version.V1_1], ALPN.wire([Version.V2_0])) is None
def test_a_handshake_that_agreed_on_nothing_settles_on_http_1_1_only_if_offered():
assert ALPN.negotiated(None, [Version.V1_1, Version.V2_0]) == Version.V1_1
with pytest.raises(soyokaze.VersionError):
ALPN.negotiated(None, [Version.V2_0])
with pytest.raises(soyokaze.VersionError):
ALPN.negotiated(b"spdy/3", [Version.V2_0])
def test_methods_match_rfc_9110_on_safety_and_idempotence():
safe = {Method.GET, Method.HEAD, Method.OPTIONS, Method.TRACE}
idempotent = safe | {Method.PUT, Method.DELETE}
for method in Method:
assert method.safe() == (method in safe)
assert method.idempotent() == (method in idempotent)
assert Method.parse(method.as_str()) == method
assert Method.parse("BREW") is None
def test_roles_split_into_the_side_that_asks_and_the_side_that_answers():
assert Role.USER_AGENT.is_client() and Role.PROXY.is_client()
assert Role.ORIGIN.is_server() and Role.GATEWAY.is_server()
assert not Role.TUNNEL.is_client() and not Role.TUNNEL.is_server()
def test_header_case_follows_the_version_that_will_write_it():
assert HeaderCase.from_version(Version.V1_0) == HeaderCase.TITLE
assert HeaderCase.from_version(Version.V1_1) == HeaderCase.TITLE
assert HeaderCase.from_version(Version.V2_0) == HeaderCase.LOWER
assert HeaderCase.from_version(Version.V3_0) == HeaderCase.LOWER
assert HeaderCase.TITLE.apply("content-length") == "Content-Length"
assert HeaderCase.TITLE.apply("CONTENT-LENGTH") == "Content-Length"
assert HeaderCase.LOWER.apply("Content-Length") == "content-length"
assert HeaderCase.TITLE.apply("te") == "Te"
def test_a_url_defaults_its_port_from_its_scheme():
assert URL.default_port("http") == 80
assert URL.default_port("https") == 443
assert URL.default_port("ws") == 80
assert URL.default_port("wss") == 443
def test_an_authority_omits_the_port_only_when_it_is_the_default():
assert URL.authority_of("https", "example.com", 443) == "example.com"
assert URL.authority_of("https", "example.com", 8443) == "example.com:8443"
assert URL.authority_of("http", "example.com", 80) == "example.com"
assert URL.authority_of("http", "example.com", 443) == "example.com:443"
def test_a_field_section_keeps_order_and_never_folds_repeats():
headers = Headers()
headers.append("Set-Cookie", "a=1")
headers.append("set-cookie", "b=2")
assert len(headers) == 2, "repeats must never be folded together"
assert headers.get_all("set-cookie") == ["a=1", "b=2"]
assert headers.get("SET-COOKIE") == "a=1", "lookup is case-insensitive"
assert list(headers) == [("set-cookie", "a=1"), ("set-cookie", "b=2")]
headers.insert("set-cookie", "c=3")
assert list(headers) == [("set-cookie", "c=3")], "inserting must drop what was there"
assert headers.contains("set-cookie") and not headers.absent("set-cookie")
assert headers.remove("set-cookie") and headers.is_empty()
assert not headers.remove("set-cookie")
def test_a_well_known_name_has_a_presence_bit_and_an_unknown_one_does_not():
assert Headers.well_known("cookie") != 0
assert Headers.well_known("content-length") != 0
assert Headers.well_known("x-custom") == 0
assert Headers.well_known("Cookie") == 0, "the bit is only defined for a lowercase name"
assert Headers.named("content-length", "Content-Length")
assert not Headers.named("content-length", "content-type")
def test_a_messages_section_is_borrowed_from_it_and_writes_through():
message = soyokaze.Message.response(200)
message.headers.append("x-a", "1")
assert message.header("x-a") == "1", "the borrowed section must be the message's own"
assert list(message.trailers) == []
def test_a_body_reports_which_of_the_three_kinds_it_is():
message = soyokaze.Message.response(200)
assert message.body_kind() == BodyKind.NONE and message.body_is_empty()
message.set_body(b"abc")
assert message.body_kind() == BodyKind.DATA
assert message.body_inline() == b"abc" and message.body_len() == 3
message.set_body("abc")
assert message.body_kind() == BodyKind.TEXT and message.body_inline() == b"abc"
message.clear_body()
assert message.body_kind() == BodyKind.NONE and message.body_len() is None
def test_a_message_is_rewritten_through_its_own_setters():
message = soyokaze.Message.request(Method.GET, "/a")
message.method = Method.POST
message.target = "/b"
message.version = Version.V2_0
message.connection_id = b"cid"
assert (message.method, message.target, message.version) == (Method.POST, "/b", Version.V2_0)
assert message.connection_id == b"cid"
message.connection_id = None
assert message.connection_id is None
def test_a_prefixed_integer_matches_rfc_7541_appendix_c():
assert fields.Integer.encode(10, 5) == bytes([0b000_01010])
assert fields.Integer.encode(1337, 5) == bytes([0b000_11111, 0b1001_1010, 0b0000_1010])
assert fields.Integer.decode(fields.Integer.encode(1337, 5), 5) == (3, 1337)
assert fields.Integer.limit(5) == 31
with pytest.raises(soyokaze.ProtocolError):
fields.Integer.decode(b"\x1f", 5)
def test_a_string_literal_round_trips_under_either_coding():
for huffman in (False, True):
encoded = fields.StringLiteral.encode(b"custom-key", 7, huffman=huffman)
assert fields.StringLiteral.decode(encoded, 7)[1] == b"custom-key"
assert fields.StringLiteral.prefers_huffman(b"www.example.com")
def test_a_field_costs_its_octets_plus_the_rfc_7541_overhead():
assert fields.HeaderField.OVERHEAD == 32
assert fields.HeaderField("ab", "cde").size() == 2 + 3 + 32
assert fields.HeaderField("cookie", "x").sensitive()
assert fields.HeaderField("authorization", "x").sensitive()
assert not fields.HeaderField("accept", "x").sensitive()
def test_a_timeout_of_zero_or_less_arms_no_deadline():
assert not sync.Timeout.armed(0.0)
assert not sync.Timeout.armed(-1.0)
assert not sync.Timeout.armed(float("nan"))
assert not sync.Timeout.armed(float("inf"))
assert sync.Timeout.armed(0.5)
assert sync.Timeout.duration(0.0) is None
assert sync.Timeout.duration(1.5) == pytest.approx(1.5)
def test_scanning_finds_an_octet_and_classifies_a_field_value():
assert scan.find(b"abcdef", ord("d")) == 3
assert scan.find(b"abcdef", ord("z")) is None
assert scan.copy(b"abcdef") == b"abcdef"
assert scan.is_field_value(b"plain text")
assert not scan.is_field_value(b"with\nnewline")
assert scan.classify_field_value(b"\x01") & scan.VALUE_CONTROL
assert scan.classify_field_value(b"\x80") & scan.VALUE_OBS_TEXT
def test_text_stays_inline_until_it_outgrows_the_inline_capacity():
short = text.Text.from_str("a" * text.INLINE)
long = text.Text.from_str("a" * (text.INLINE + 1))
assert short.is_inline() and not long.is_inline()
assert short.as_str() == "a" * text.INLINE
assert len(long) == text.INLINE + 1
lowered = text.Text.from_ascii_lowercase(b"Content-Length")
assert lowered.as_str() == "content-length"
assert text.Text.from_str("abc") == "abc"