Advanced Features

class PushPromiseFrame(Padding, Frame):
    def __init__(self, stream_id: int, promised_stream_id: int = 0, 
                 data: bytes = b"", **kwargs) -> None:
        """
        Create a PUSH_PROMISE frame.
        
        Parameters:
        - stream_id (int): Stream identifier (must be non-zero)
        - promised_stream_id (int): Stream ID being promised (must be even and non-zero)
        - data (bytes): HPACK-encoded header block for promised request
        - pad_length (int): Padding length if PADDED flag set
        - flags (Iterable[str]): Frame flags ("END_HEADERS", "PADDED")
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is zero or promised_stream_id is invalid
        """
    
    @property
    def promised_stream_id(self) -> int:
        """Stream ID that will be used for the promised stream."""
    
    @property
    def data(self) -> bytes:
        """HPACK-encoded header block for the promised request."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

class AltSvcFrame(Frame):
    def __init__(self, stream_id: int, origin: bytes = b"", field: bytes = b"", **kwargs) -> None:
        """
        Create an ALTSVC frame.
        
        Parameters:
        - stream_id (int): Stream identifier (0 for connection, non-zero for stream)
        - origin (bytes): Origin for alternate service (empty if stream_id != 0)
        - field (bytes): Alt-Svc field value
        - **kwargs: Additional arguments for parent classes
        
        Note: Either stream_id must be 0 XOR origin must be empty (not both)
        
        Raises:
        - InvalidDataError: If origin or field are not bytes
        """
    
    @property
    def origin(self) -> bytes:
        """Origin that the alternate service applies to."""
    
    @property
    def field(self) -> bytes:
        """Alt-Svc field value describing the alternate service."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

class ExtensionFrame(Frame):
    def __init__(self, type: int, stream_id: int, flag_byte: int = 0x0, 
                 body: bytes = b"", **kwargs) -> None:
        """
        Create an extension frame for unknown frame types.
        
        Parameters:
        - type (int): Frame type identifier
        - stream_id (int): Stream identifier
        - flag_byte (int): Raw flag byte value
        - body (bytes): Frame body data
        - **kwargs: Additional arguments for parent classes
        """
    
    @property
    def type(self) -> int:
        """Frame type identifier."""
    
    @property
    def flag_byte(self) -> int:
        """Raw flag byte from frame header."""
    
    @property
    def body(self) -> bytes:
        """Raw frame body data."""
    
    def parse_flags(self, flag_byte: int) -> None:
        """
        Store raw flag byte instead of parsing individual flags.
        
        Parameters:
        - flag_byte (int): 8-bit flag value from frame header
        """
    
    def serialize(self) -> bytes:
        """
        Serialize extension frame exactly as received.
        
        Returns:
        bytes: Complete frame with original type and flag byte
        """
    
    # Inherited from Frame
    def parse_body(self, data: memoryview) -> None: ...

class Flag:
    def __init__(self, name: str, bit: int) -> None:
        """
        Create a flag definition.
        
        Parameters:
        - name (str): Flag name (e.g., "END_STREAM")
        - bit (int): Bit value for the flag (e.g., 0x01)
        """
    
    @property
    def name(self) -> str:
        """Flag name."""
    
    @property
    def bit(self) -> int:
        """Flag bit value."""

class Flags:
    def __init__(self, defined_flags: Iterable[Flag]) -> None:
        """
        Create a flags container for a specific frame type.
        
        Parameters:
        - defined_flags (Iterable[Flag]): Valid flags for this frame type
        """
    
    def add(self, value: str) -> None:
        """
        Add a flag to the set.
        
        Parameters:
        - value (str): Flag name to add
        
        Raises:
        - ValueError: If flag is not defined for this frame type
        """
    
    def discard(self, value: str) -> None:
        """
        Remove a flag from the set if present.
        
        Parameters:
        - value (str): Flag name to remove
        """
    
    def __contains__(self, flag: str) -> bool:
        """
        Check if flag is set.
        
        Parameters:
        - flag (str): Flag name to check
        
        Returns:
        bool: True if flag is set
        """
    
    def __iter__(self) -> Iterator[str]:
        """
        Iterate over set flags.
        
        Returns:
        Iterator[str]: Iterator over flag names
        """
    
    def __len__(self) -> int:
        """
        Get number of set flags.
        
        Returns:
        int: Number of flags currently set
        """

from hyperframe.frame import PushPromiseFrame

# Server promising to push a resource
push_promise = PushPromiseFrame(
    stream_id=1,              # Original request stream
    promised_stream_id=2,     # Stream for pushed response
    data=b"\\x00\\x05:path\\x0A/style.css",  # HPACK headers for promised request
    flags=["END_HEADERS"]
)

print(f"Promising stream {push_promise.promised_stream_id} from stream {push_promise.stream_id}")

# PUSH_PROMISE with padding
padded_promise = PushPromiseFrame(
    stream_id=1,
    promised_stream_id=4,
    data=b"\\x00\\x05:path\\x0B/script.js",
    pad_length=8,
    flags=["END_HEADERS", "PADDED"]
)

from hyperframe.frame import AltSvcFrame

# Connection-level alternate service
connection_altsvc = AltSvcFrame(
    stream_id=0,  # Connection level
    origin=b"example.com",
    field=b'h2="alt.example.com:443"; ma=3600'
)

# Stream-specific alternate service
stream_altsvc = AltSvcFrame(
    stream_id=1,  # Stream specific
    origin=b"",   # Empty for stream-specific
    field=b'h2="alt2.example.com:443"'
)

print(f"Connection Alt-Svc: {connection_altsvc.field}")
print(f"Stream Alt-Svc: {stream_altsvc.field}")

from hyperframe.frame import Frame, ExtensionFrame

# Parse unknown frame type
unknown_frame_data = b'\\x00\\x00\\x05\\xEE\\x0F\\x00\\x00\\x00\\x01Hello'  # Type 0xEE
frame, length = Frame.parse_frame_header(unknown_frame_data[:9], strict=False)

if isinstance(frame, ExtensionFrame):
    frame.parse_body(memoryview(unknown_frame_data[9:9 + length]))
    print(f"Unknown frame type: 0x{frame.type:02X}")
    print(f"Flag byte: 0x{frame.flag_byte:02X}")
    print(f"Body: {frame.body}")
    
    # Round-trip serialization
    serialized = frame.serialize()
    assert serialized == unknown_frame_data

# Create extension frame manually
custom_frame = ExtensionFrame(
    type=0xCC,
    stream_id=3,
    flag_byte=0x05,
    body=b"Custom frame data"
)

from hyperframe.flags import Flag, Flags
from hyperframe.frame import DataFrame

# Working with frame flags
data_frame = DataFrame(stream_id=1, data=b"Test data")

# Add flags
data_frame.flags.add("END_STREAM")
print(f"END_STREAM set: {'END_STREAM' in data_frame.flags}")

# Try to add invalid flag
try:
    data_frame.flags.add("INVALID_FLAG")
except ValueError as e:
    print(f"Flag error: {e}")

# Check all set flags
print(f"Set flags: {list(data_frame.flags)}")

# Remove flag
data_frame.flags.discard("END_STREAM")
print(f"Flags after discard: {list(data_frame.flags)}")

# Create custom flag set
custom_flags = [
    Flag("CUSTOM_FLAG_1", 0x01),
    Flag("CUSTOM_FLAG_2", 0x02)
]
flag_set = Flags(custom_flags)
flag_set.add("CUSTOM_FLAG_1")
print(f"Custom flags: {list(flag_set)}")

from hyperframe.frame import HeadersFrame, ContinuationFrame, PushPromiseFrame

# Server push scenario with large headers
large_headers = b"\\x00\\x07:method\\x03GET" + b"\\x00" * 2000

# Initial PUSH_PROMISE without END_HEADERS
push_start = PushPromiseFrame(
    stream_id=1,
    promised_stream_id=2,
    data=large_headers[:1000]
    # No END_HEADERS flag
)

# CONTINUATION to complete the promise
push_continuation = ContinuationFrame(
    stream_id=1,
    data=large_headers[1000:],
    flags=["END_HEADERS"]
)

print(f"Push promise starts: {len(push_start.data)} bytes")
print(f"Continuation completes: {len(push_continuation.data)} bytes")

# Later, server sends response on promised stream
response_headers = HeadersFrame(
    stream_id=2,  # Promised stream
    data=b"\\x00\\x07:status\\x03200",
    flags=["END_HEADERS"]
)

from hyperframe.frame import PushPromiseFrame, AltSvcFrame
from hyperframe.exceptions import InvalidDataError, InvalidFrameError

# Invalid PUSH_PROMISE stream ID
try:
    PushPromiseFrame(stream_id=1, promised_stream_id=1)  # Must be even
except InvalidDataError as e:
    print(f"Push promise error: {e}")

# Invalid ALTSVC parameters
try:
    AltSvcFrame(stream_id=1, origin="not bytes", field=b"valid")  # Origin must be bytes
except InvalidDataError as e:
    print(f"AltSvc error: {e}")

# Parse extension frame with malformed data
try:
    bad_data = b'\\x00\\x00\\x05\\xFF\\x00\\x00\\x00\\x00\\x01ABC'  # Only 3 bytes, claims 5
    frame, length = Frame.parse_frame_header(bad_data[:9])
    frame.parse_body(memoryview(bad_data[9:]))  # Will be truncated
    print(f"Extension frame body: {frame.body}")  # Only gets available data
except Exception as e:
    print(f"Parse error: {e}")