languageregexp languageregexp post post http1 1 http1 1 200 ok flagsre verbose bodyendings messageseparator crlf not really but useful for testing just remember not to include it in a body tokenregexmap expect requestheaders makeheadersreexpect requestheaders expect responseheaders makeheadersreexpect responseheaders expect responsemeta keyvaluere expect separator separatorre class outputmatchingerrorvalueerror pass def expecttokenstokens iterableexpect s str for token in tokens s expecttokentoken s if s raise outputmatchingerrorf unmatched remaining output for tokens in s r def expecttokentoken expect s str str if token is expect body s expectbodys else s expectregextoken s return s def expectregextoken expect s str str match tokenregexmaptoken matchs if not match raise outputmatchingerrorf no match for token in s r return smatch end def expectbodys str str if contentdisposition in s lower multipart body heuristic finalboundaryre re compile rn rn match finalboundaryre searchs if match return smatch end endings s indexsep for sep in bodyendings if sep in s if not endings s only body else end minendings if end 0 raise outputmatchingerrorf empty body s r s send return s language regexp language regexp post post http 1 1 http 1 1 200 ok start_line_re non_crlf non_crlf crlf crlf not really but useful for testing just remember not to include it in a body we require some text and continue to read until we find an ending or until the end of the string multipart body heuristic only body
import re
from typing import Iterable

from httpie.output.writer import MESSAGE_SEPARATOR
from .tokens import Expect
from ...utils import CRLF


SEPARATOR_RE = re.compile(f'^{MESSAGE_SEPARATOR}')
KEY_VALUE_RE = re.compile(r'[\n]*((.*?):(.+)[\n]?)+[\n]*')


def make_headers_re(message_type: Expect):
    assert message_type in {Expect.REQUEST_HEADERS, Expect.RESPONSE_HEADERS}

    
    crlf = r'[\r][\n]'
    non_crlf = rf'[^{CRLF}]'

    
    http_version = r'HTTP/\d+\.\d+'
    if message_type is Expect.REQUEST_HEADERS:
        
        start_line_re = fr'{non_crlf}*{http_version}{crlf}'
    else:
        
        start_line_re = fr'{http_version}{non_crlf}*{crlf}'

    return re.compile(
        fr,
        flags=re.VERBOSE
    )


BODY_ENDINGS = [
    MESSAGE_SEPARATOR,
    CRLF,  
]
TOKEN_REGEX_MAP = {
    Expect.REQUEST_HEADERS: make_headers_re(Expect.REQUEST_HEADERS),
    Expect.RESPONSE_HEADERS: make_headers_re(Expect.RESPONSE_HEADERS),
    Expect.RESPONSE_META: KEY_VALUE_RE,
    Expect.SEPARATOR: SEPARATOR_RE,
}


class OutputMatchingError(ValueError):
    pass


def expect_tokens(tokens: Iterable[Expect], s: str):
    for token in tokens:
        s = expect_token(token, s)
    if s:
        raise OutputMatchingError(f'Unmatched remaining output for {tokens} in {s!r}')


def expect_token(token: Expect, s: str) -> str:
    if token is Expect.BODY:
        s = expect_body(s)
    else:
        s = expect_regex(token, s)
    return s


def expect_regex(token: Expect, s: str) -> str:
    match = TOKEN_REGEX_MAP[token].match(s)
    if not match:
        raise OutputMatchingError(f'No match for {token} in {s!r}')
    return s[match.end():]


def expect_body(s: str) -> str:
    
    if 'content-disposition:' in s.lower():
        
        final_boundary_re = re.compile('\r\n--[^-]+?--\r\n')
        match = final_boundary_re.search(s)
        if match:
            return s[match.end():]

    endings = [s.index(sep) for sep in BODY_ENDINGS if sep in s]
    if not endings:
        s = ''  
    else:
        end = min(endings)
        if end == 0:
            raise OutputMatchingError(f'Empty body: {s!r}')
        s = s[end:]
    return s