diff options
| -rw-r--r-- | Doc/library/email.policy.rst | 323 | ||||
| -rw-r--r-- | Lib/email/_encoded_words.py | 211 | ||||
| -rw-r--r-- | Lib/email/_header_value_parser.py | 2145 | ||||
| -rw-r--r-- | Lib/email/_headerregistry.py | 456 | ||||
| -rw-r--r-- | Lib/email/_policybase.py | 12 | ||||
| -rw-r--r-- | Lib/email/errors.py | 43 | ||||
| -rw-r--r-- | Lib/email/generator.py | 11 | ||||
| -rw-r--r-- | Lib/email/policy.py | 173 | ||||
| -rw-r--r-- | Lib/email/utils.py | 7 | ||||
| -rw-r--r-- | Lib/test/test_email/__init__.py | 6 | ||||
| -rw-r--r-- | Lib/test/test_email/test__encoded_words.py | 187 | ||||
| -rw-r--r-- | Lib/test/test_email/test__header_value_parser.py | 2466 | ||||
| -rw-r--r-- | Lib/test/test_email/test__headerregistry.py | 717 | ||||
| -rw-r--r-- | Lib/test/test_email/test_generator.py | 168 | ||||
| -rw-r--r-- | Lib/test/test_email/test_pickleable.py | 57 | ||||
| -rw-r--r-- | Lib/test/test_email/test_policy.py | 128 |
16 files changed, 6994 insertions, 116 deletions
diff --git a/Doc/library/email.policy.rst b/Doc/library/email.policy.rst index 73cfba1..c65130b 100644 --- a/Doc/library/email.policy.rst +++ b/Doc/library/email.policy.rst @@ -306,3 +306,326 @@ added matters. To illustrate:: ``7bit``, non-ascii binary data is CTE encoded using the ``unknown-8bit`` charset. Otherwise the original source header is used, with its existing line breaks and and any (RFC invalid) binary data it may contain. + + +.. note:: + + The remainder of the classes documented below are included in the standard + library on a :term:`provisional basis <provisional package>`. Backwards + incompatible changes (up to and including removal of the feature) may occur + if deemed necessary by the core developers. + + +.. class:: EmailPolicy(**kw) + + This concrete :class:`Policy` provides behavior that is intended to be fully + compliant with the current email RFCs. These include (but are not limited + to) :rfc:`5322`, :rfc:`2047`, and the current MIME RFCs. + + This policy adds new header parsing and folding algorithms. Instead of + simple strings, headers are custom objects with custom attributes depending + on the type of the field. The parsing and folding algorithm fully implement + :rfc:`2047` and :rfc:`5322`. + + In addition to the settable attributes listed above that apply to all + policies, this policy adds the following additional attributes: + + .. attribute:: refold_source + + If the value for a header in the ``Message`` object originated from a + :mod:`~email.parser` (as opposed to being set by a program), this + attribute indicates whether or not a generator should refold that value + when transforming the message back into stream form. The possible values + are: + + ======== =============================================================== + ``none`` all source values use original folding + + ``long`` source values that have any line that is longer than + ``max_line_length`` will be refolded + + ``all`` all values are refolded. + ======== =============================================================== + + The default is ``long``. + + .. attribute:: header_factory + + A callable that takes two arguments, ``name`` and ``value``, where + ``name`` is a header field name and ``value`` is an unfolded header field + value, and returns a string-like object that represents that header. A + default ``header_factory`` is provided that understands some of the + :RFC:`5322` header field types. (Currently address fields and date + fields have special treatment, while all other fields are treated as + unstructured. This list will be completed before the extension is marked + stable.) + + The class provides the following concrete implementations of the abstract + methods of :class:`Policy`: + + .. method:: header_source_parse(sourcelines) + + The implementation of this method is the same as that for the + :class:`Compat32` policy. + + .. method:: header_store_parse(name, value) + + The name is returned unchanged. If the input value has a ``name`` + attribute and it matches *name* ignoring case, the value is returned + unchanged. Otherwise the *name* and *value* are passed to + ``header_factory``, and the resulting custom header object is returned as + the value. In this case a ``ValueError`` is raised if the input value + contains CR or LF characters. + + .. method:: header_fetch_parse(name, value) + + If the value has a ``name`` attribute, it is returned to unmodified. + Otherwise the *name*, and the *value* with any CR or LF characters + removed, are passed to the ``header_factory``, and the resulting custom + header object is returned. Any surrogateescaped bytes get turned into + the unicode unknown-character glyph. + + .. method:: fold(name, value) + + Header folding is controlled by the :attr:`refold_source` policy setting. + A value is considered to be a 'source value' if and only if it does not + have a ``name`` attribute (having a ``name`` attribute means it is a + header object of some sort). If a source value needs to be refolded + according to the policy, it is converted into a custom header object by + passing the *name* and the *value* with any CR and LF characters removed + to the ``header_factory``. Folding of a custom header object is done by + calling its ``fold`` method with the current policy. + + Source values are split into lines using :meth:`~str.splitlines`. If + the value is not to be refolded, the lines are rejoined using the + ``linesep`` from the policy and returned. The exception is lines + containing non-ascii binary data. In that case the value is refolded + regardless of the ``refold_source`` setting, which causes the binary data + to be CTE encoded using the ``unknown-8bit`` charset. + + .. method:: fold_binary(name, value) + + The same as :meth:`fold` if :attr:`cte_type` is ``7bit``, except that + the returned value is bytes. + + If :attr:`cte_type` is ``8bit``, non-ASCII binary data is converted back + into bytes. Headers with binary data are not refolded, regardless of the + ``refold_header`` setting, since there is no way to know whether the + binary data consists of single byte characters or multibyte characters. + +The following instances of :class:`EmailPolicy` provide defaults suitable for +specific application domains. Note that in the future the behavior of these +instances (in particular the ``HTTP` instance) may be adjusted to conform even +more closely to the RFCs relevant to their domains. + +.. data:: default + + An instance of ``EmailPolicy`` with all defaults unchanged. This policy + uses the standard Python ``\n`` line endings rather than the RFC-correct + ``\r\n``. + +.. data:: SMTP + + Suitable for serializing messages in conformance with the email RFCs. + Like ``default``, but with ``linesep`` set to ``\r\n``, which is RFC + compliant. + +.. data:: HTTP + + Suitable for serializing headers with for use in HTTP traffic. Like + ``SMTP`` except that ``max_line_length`` is set to ``None`` (unlimited). + +.. data:: strict + + Convenience instance. The same as ``default`` except that + ``raise_on_defect`` is set to ``True``. This allows any policy to be made + strict by writing:: + + somepolicy + policy.strict + +With all of these :class:`EmailPolicies <.EmailPolicy>`, the effective API of +the email package is changed from the Python 3.2 API in the following ways: + + * Setting a header on a :class:`~email.message.Message` results in that + header being parsed and a custom header object created. + + * Fetching a header value from a :class:`~email.message.Message` results + in that header being parsed and a custom header object created and + returned. + + * Any custom header object, or any header that is refolded due to the + policy settings, is folded using an algorithm that fully implements the + RFC folding algorithms, including knowing where encoded words are required + and allowed. + +From the application view, this means that any header obtained through the +:class:`~email.message.Message` is a custom header object with custom +attributes, whose string value is the fully decoded unicode value of the +header. Likewise, a header may be assigned a new value, or a new header +created, using a unicode string, and the policy will take care of converting +the unicode string into the correct RFC encoded form. + +The custom header objects and their attributes are described below. All custom +header objects are string subclasses, and their string value is the fully +decoded value of the header field (the part of the field after the ``:``) + + +.. class:: BaseHeader + + This is the base class for all custom header objects. It provides the + following attributes: + + .. attribute:: name + + The header field name (the portion of the field before the ':'). + + .. attribute:: defects + + A possibly empty list of :class:`~email.errors.MessageDefect` objects + that record any RFC violations found while parsing the header field. + + .. method:: fold(*, policy) + + Return a string containing :attr:`~email.policy.Policy.linesep` + characters as required to correctly fold the header according + to *policy*. A :attr:`~email.policy.Policy.cte_type` of + ``8bit`` will be treated as if it were ``7bit``, since strings + may not contain binary data. + + +.. class:: UnstructuredHeader + + The class used for any header that does not have a more specific + type. (The :mailheader:`Subject` header is an example of an + unstructured header.) It does not have any additional attributes. + + +.. class:: DateHeader + + The value of this type of header is a single date and time value. The + primary example of this type of header is the :mailheader:`Date` header. + + .. attribute:: datetime + + A :class:`~datetime.datetime` encoding the date and time from the + header value. + + The ``datetime`` will be a naive ``datetime`` if the value either does + not have a specified timezone (which would be a violation of the RFC) or + if the timezone is specified as ``-0000``. This timezone value indicates + that the date and time is to be considered to be in UTC, but with no + indication of the local timezone in which it was generated. (This + contrasts to ``+0000``, which indicates a date and time that really is in + the UTC ``0000`` timezone.) + + If the header value contains a valid timezone that is not ``-0000``, the + ``datetime`` will be an aware ``datetime`` having a + :class:`~datetime.tzinfo` set to the :class:`~datetime.timezone` + indicated by the header value. + + A ``datetime`` may also be assigned to a :mailheader:`Date` type header. + The resulting string value will use a timezone of ``-0000`` if the + ``datetime`` is naive, and the appropriate UTC offset if the ``datetime`` is + aware. + + +.. class:: AddressHeader + + This class is used for all headers that can contain addresses, whether they + are supposed to be singleton addresses or a list. + + .. attribute:: addresses + + A list of :class:`.Address` objects listing all of the addresses that + could be parsed out of the field value. + + .. attribute:: groups + + A list of :class:`.Group` objects. Every address in :attr:`.addresses` + appears in one of the group objects in the tuple. Addresses that are not + syntactically part of a group are represented by ``Group`` objects whose + ``name`` is ``None``. + + In addition to addresses in string form, any combination of + :class:`.Address` and :class:`.Group` objects, singly or in a list, may be + assigned to an address header. + + +.. class:: Address(display_name='', username='', domain='', addr_spec=None): + + The class used to represent an email address. The general form of an + address is:: + + [display_name] <username@domain> + + or:: + + username@domain + + where each part must conform to specific syntax rules spelled out in + :rfc:`5322`. + + As a convenience *addr_spec* can be specified instead of *username* and + *domain*, in which case *username* and *domain* will be parsed from the + *addr_spec*. An *addr_spec* must be a properly RFC quoted string; if it is + not ``Address`` will raise an error. Unicode characters are allowed and + will be property encoded when serialized. However, per the RFCs, unicode is + *not* allowed in the username portion of the address. + + .. attribute:: display_name + + The display name portion of the address, if any, with all quoting + removed. If the address does not have a display name, this attribute + will be an empty string. + + .. attribute:: username + + The ``username`` portion of the address, with all quoting removed. + + .. attribute:: domain + + The ``domain`` portion of the address. + + .. attribute:: addr_spec + + The ``username@domain`` portion of the address, correctly quoted + for use as a bare address (the second form shown above). This + attribute is not mutable. + + .. method:: __str__() + + The ``str`` value of the object is the address quoted according to + :rfc:`5322` rules, but with no Content Transfer Encoding of any non-ASCII + characters. + + +.. class:: Group(display_name=None, addresses=None) + + The class used to represent an address group. The general form of an + address group is:: + + display_name: [address-list]; + + As a convenience for processing lists of addresses that consist of a mixture + of groups and single addresses, a ``Group`` may also be used to represent + single addresses that are not part of a group by setting *display_name* to + ``None`` and providing a list of the single address as *addresses*. + + .. attribute:: display_name + + The ``display_name`` of the group. If it is ``None`` and there is + exactly one ``Address`` in ``addresses``, then the ``Group`` represents a + single address that is not in a group. + + .. attribute:: addresses + + A possibly empty tuple of :class:`.Address` objects representing the + addresses in the group. + + .. method:: __str__() + + The ``str`` value of a ``Group`` is formatted according to :rfc:`5322`, + but with no Content Transfer Encoding of any non-ASCII characters. If + ``display_name`` is none and there is a single ``Address`` in the + ``addresses` list, the ``str`` value will be the same as the ``str`` of + that single ``Address``. diff --git a/Lib/email/_encoded_words.py b/Lib/email/_encoded_words.py new file mode 100644 index 0000000..01fe42f --- /dev/null +++ b/Lib/email/_encoded_words.py @@ -0,0 +1,211 @@ +""" Routines for manipulating RFC2047 encoded words. + +This is currently a package-private API, but will be considered for promotion +to a public API if there is demand. + +""" + +# An ecoded word looks like this: +# +# =?charset[*lang]?cte?encoded_string?= +# +# for more information about charset see the charset module. Here it is one +# of the preferred MIME charset names (hopefully; you never know when parsing). +# cte (Content Transfer Encoding) is either 'q' or 'b' (ignoring case). In +# theory other letters could be used for other encodings, but in practice this +# (almost?) never happens. There could be a public API for adding entries +# to to the CTE tables, but YAGNI for now. 'q' is Quoted Printable, 'b' is +# Base64. The meaning of encoded_string should be obvious. 'lang' is optional +# as indicated by the brackets (they are not part of the syntax) but is almost +# never encountered in practice. +# +# The general interface for a CTE decoder is that it takes the encoded_string +# as its argument, and returns a tuple (cte_decoded_string, defects). The +# cte_decoded_string is the original binary that was encoded using the +# specified cte. 'defects' is a list of MessageDefect instances indicating any +# problems encountered during conversion. 'charset' and 'lang' are the +# corresponding strings extracted from the EW, case preserved. +# +# The general interface for a CTE encoder is that it takes a binary sequence +# as input and returns the cte_encoded_string, which is an ascii-only string. +# +# Each decoder must also supply a length function that takes the binary +# sequence as its argument and returns the length of the resulting encoded +# string. +# +# The main API functions for the module are decode, which calls the decoder +# referenced by the cte specifier, and encode, which adds the appropriate +# RFC 2047 "chrome" to the encoded string, and can optionally automatically +# select the shortest possible encoding. See their docstrings below for +# details. + +import re +import base64 +import binascii +import functools +from string import ascii_letters, digits +from email import errors + +# +# Quoted Printable +# + +# regex based decoder. +_q_byte_subber = functools.partial(re.compile(br'=([a-fA-F0-9]{2})').sub, + lambda m: bytes([int(m.group(1), 16)])) + +def decode_q(encoded): + encoded = encoded.replace(b'_', b' ') + return _q_byte_subber(encoded), [] + + +# dict mapping bytes to their encoded form +class QByteMap(dict): + + safe = b'-!*+/' + ascii_letters.encode('ascii') + digits.encode('ascii') + + def __missing__(self, key): + if key in self.safe: + self[key] = chr(key) + else: + self[key] = "={:02X}".format(key) + return self[key] + +_q_byte_map = QByteMap() + +# In headers spaces are mapped to '_'. +_q_byte_map[ord(' ')] = '_' + +def encode_q(bstring): + return ''.join(_q_byte_map[x] for x in bstring) + +def len_q(bstring): + return sum(len(_q_byte_map[x]) for x in bstring) + + +# +# Base64 +# + +def decode_b(encoded): + defects = [] + pad_err = len(encoded) % 4 + if pad_err: + defects.append(errors.InvalidBase64PaddingDefect()) + padded_encoded = encoded + b'==='[:4-pad_err] + else: + padded_encoded = encoded + try: + return base64.b64decode(padded_encoded, validate=True), defects + except binascii.Error: + # Since we had correct padding, this must an invalid char error. + defects = [errors.InvalidBase64CharactersDefect()] + # The non-alphabet characters are ignored as far as padding + # goes, but we don't know how many there are. So we'll just + # try various padding lengths until something works. + for i in 0, 1, 2, 3: + try: + return base64.b64decode(encoded+b'='*i, validate=False), defects + except binascii.Error: + if i==0: + defects.append(errors.InvalidBase64PaddingDefect()) + else: + # This should never happen. + raise AssertionError("unexpected binascii.Error") + +def encode_b(bstring): + return base64.b64encode(bstring).decode('ascii') + +def len_b(bstring): + groups_of_3, leftover = divmod(len(bstring), 3) + # 4 bytes out for each 3 bytes (or nonzero fraction thereof) in. + return groups_of_3 * 4 + (4 if leftover else 0) + + +_cte_decoders = { + 'q': decode_q, + 'b': decode_b, + } + +def decode(ew): + """Decode encoded word and return (string, charset, lang, defects) tuple. + + An RFC 2047/2243 encoded word has the form: + + =?charset*lang?cte?encoded_string?= + + where '*lang' may be omitted but the other parts may not be. + + This function expects exactly such a string (that is, it does not check the + syntax and may raise errors if the string is not well formed), and returns + the encoded_string decoded first from its Content Transfer Encoding and + then from the resulting bytes into unicode using the specified charset. If + the cte-decoded string does not successfully decode using the specified + character set, a defect is added to the defects list and the unknown octets + are replaced by the unicode 'unknown' character \uFDFF. + + The specified charset and language are returned. The default for language, + which is rarely if ever encountered, is the empty string. + + """ + _, charset, cte, cte_string, _ = ew.split('?') + charset, _, lang = charset.partition('*') + cte = cte.lower() + # Recover the original bytes and do CTE decoding. + bstring = cte_string.encode('ascii', 'surrogateescape') + bstring, defects = _cte_decoders[cte](bstring) + # Turn the CTE decoded bytes into unicode. + try: + string = bstring.decode(charset) + except UnicodeError: + defects.append(errors.UndecodableBytesDefect("Encoded word " + "contains bytes not decodable using {} charset".format(charset))) + string = bstring.decode(charset, 'surrogateescape') + except LookupError: + string = bstring.decode('ascii', 'surrogateescape') + if charset.lower() != 'unknown-8bit': + defects.append(errors.CharsetError("Unknown charset {} " + "in encoded word; decoded as unknown bytes".format(charset))) + return string, charset, lang, defects + + +_cte_encoders = { + 'q': encode_q, + 'b': encode_b, + } + +_cte_encode_length = { + 'q': len_q, + 'b': len_b, + } + +def encode(string, charset='utf-8', encoding=None, lang=''): + """Encode string using the CTE encoding that produces the shorter result. + + Produces an RFC 2047/2243 encoded word of the form: + + =?charset*lang?cte?encoded_string?= + + where '*lang' is omitted unless the 'lang' parameter is given a value. + Optional argument charset (defaults to utf-8) specifies the charset to use + to encode the string to binary before CTE encoding it. Optional argument + 'encoding' is the cte specifier for the encoding that should be used ('q' + or 'b'); if it is None (the default) the encoding which produces the + shortest encoded sequence is used, except that 'q' is preferred if it is up + to five characters longer. Optional argument 'lang' (default '') gives the + RFC 2243 language string to specify in the encoded word. + + """ + if charset == 'unknown-8bit': + bstring = string.encode('ascii', 'surrogateescape') + else: + bstring = string.encode(charset) + if encoding is None: + qlen = _cte_encode_length['q'](bstring) + blen = _cte_encode_length['b'](bstring) + # Bias toward q. 5 is arbitrary. + encoding = 'q' if qlen - blen < 5 else 'b' + encoded = _cte_encoders[encoding](bstring) + if lang: + lang = '*' + lang + return "=?{}{}?{}?{}?=".format(charset, lang, encoding, encoded) diff --git a/Lib/email/_header_value_parser.py b/Lib/email/_header_value_parser.py new file mode 100644 index 0000000..87d8f68 --- /dev/null +++ b/Lib/email/_header_value_parser.py @@ -0,0 +1,2145 @@ +"""Header value parser implementing various email-related RFC parsing rules. + +The parsing methods defined in this module implement various email related +parsing rules. Principal among them is RFC 5322, which is the followon +to RFC 2822 and primarily a clarification of the former. It also implements +RFC 2047 encoded word decoding. + +RFC 5322 goes to considerable trouble to maintain backward compatibility with +RFC 822 in the parse phase, while cleaning up the structure on the generation +phase. This parser supports correct RFC 5322 generation by tagging white space +as folding white space only when folding is allowed in the non-obsolete rule +sets. Actually, the parser is even more generous when accepting input than RFC +5322 mandates, following the spirit of Postel's Law, which RFC 5322 encourages. +Where possible deviations from the standard are annotated on the 'defects' +attribute of tokens that deviate. + +The general structure of the parser follows RFC 5322, and uses its terminology +where there is a direct correspondence. Where the implementation requires a +somewhat different structure than that used by the formal grammar, new terms +that mimic the closest existing terms are used. Thus, it really helps to have +a copy of RFC 5322 handy when studying this code. + +Input to the parser is a string that has already been unfolded according to +RFC 5322 rules. According to the RFC this unfolding is the very first step, and +this parser leaves the unfolding step to a higher level message parser, which +will have already detected the line breaks that need unfolding while +determining the beginning and end of each header. + +The output of the parser is a TokenList object, which is a list subclass. A +TokenList is a recursive data structure. The terminal nodes of the structure +are Terminal objects, which are subclasses of str. These do not correspond +directly to terminal objects in the formal grammar, but are instead more +practical higher level combinations of true terminals. + +All TokenList and Terminal objects have a 'value' attribute, which produces the +semantically meaningful value of that part of the parse subtree. The value of +all whitespace tokens (no matter how many sub-tokens they may contain) is a +single space, as per the RFC rules. This includes 'CFWS', which is herein +included in the general class of whitespace tokens. There is one exception to +the rule that whitespace tokens are collapsed into single spaces in values: in +the value of a 'bare-quoted-string' (a quoted-string with no leading or +trailing whitespace), any whitespace that appeared between the quotation marks +is preserved in the returned value. Note that in all Terminal strings quoted +pairs are turned into their unquoted values. + +All TokenList and Terminal objects also have a string value, which attempts to +be a "canonical" representation of the RFC-compliant form of the substring that +produced the parsed subtree, including minimal use of quoted pair quoting. +Whitespace runs are not collapsed. + +Comment tokens also have a 'content' attribute providing the string found +between the parens (including any nested comments) with whitespace preserved. + +All TokenList and Terminal objects have a 'defects' attribute which is a +possibly empty list all of the defects found while creating the token. Defects +may appear on any token in the tree, and a composite list of all defects in the +subtree is available through the 'all_defects' attribute of any node. (For +Terminal notes x.defects == x.all_defects.) + +Each object in a parse tree is called a 'token', and each has a 'token_type' +attribute that gives the name from the RFC 5322 grammar that it represents. +Not all RFC 5322 nodes are produced, and there is one non-RFC 5322 node that +may be produced: 'ptext'. A 'ptext' is a string of printable ascii characters. +It is returned in place of lists of (ctext/quoted-pair) and +(qtext/quoted-pair). + +XXX: provide complete list of token types. +""" + +import re +from email import _encoded_words as _ew +from email import errors +from email import utils + +# +# Useful constants and functions +# + +WSP = set(' \t') +CFWS_LEADER = WSP | set('(') +SPECIALS = set(r'()<>@,:;.\"[]') +ATOM_ENDS = SPECIALS | WSP +DOT_ATOM_ENDS = ATOM_ENDS - set('.') +# '.', '"', and '(' do not end phrases in order to support obs-phrase +PHRASE_ENDS = SPECIALS - set('."(') + +def quote_string(value): + return '"'+str(value).replace('\\', '\\\\').replace('"', r'\"')+'"' + +# +# Accumulator for header folding +# + +class _Folded: + + def __init__(self, maxlen, policy): + self.maxlen = maxlen + self.policy = policy + self.lastlen = 0 + self.stickyspace = None + self.firstline = True + self.done = [] + self.current = [] + + def newline(self): + self.done.extend(self.current) + self.done.append(self.policy.linesep) + self.current.clear() + self.lastlen = 0 + + def finalize(self): + if self.current: + self.newline() + + def __str__(self): + return ''.join(self.done) + + def append(self, stoken): + self.current.append(stoken) + + def append_if_fits(self, token, stoken=None): + if stoken is None: + stoken = str(token) + l = len(stoken) + if self.stickyspace is not None: + stickyspace_len = len(self.stickyspace) + if self.lastlen + stickyspace_len + l <= self.maxlen: + self.current.append(self.stickyspace) + self.lastlen += stickyspace_len + self.current.append(stoken) + self.lastlen += l + self.stickyspace = None + self.firstline = False + return True + if token.has_fws: + ws = token.pop_leading_fws() + if ws is not None: + self.stickyspace += str(ws) + stickyspace_len += len(ws) + token._fold(self) + return True + if stickyspace_len and l + 1 <= self.maxlen: + margin = self.maxlen - l + if 0 < margin < stickyspace_len: + trim = stickyspace_len - margin + self.current.append(self.stickyspace[:trim]) + self.stickyspace = self.stickyspace[trim:] + stickyspace_len = trim + self.newline() + self.current.append(self.stickyspace) + self.current.append(stoken) + self.lastlen = l + stickyspace_len + self.stickyspace = None + self.firstline = False + return True + if not self.firstline: + self.newline() + self.current.append(self.stickyspace) + self.current.append(stoken) + self.stickyspace = None + self.firstline = False + return True + if self.lastlen + l <= self.maxlen: + self.current.append(stoken) + self.lastlen += l + return True + if l < self.maxlen: + self.newline() + self.current.append(stoken) + self.lastlen = l + return True + return False + +# +# TokenList and its subclasses +# + +class TokenList(list): + + token_type = None + + def __init__(self, *args, **kw): + super().__init__(*args, **kw) + self.defects = [] + + def __str__(self): + return ''.join(str(x) for x in self) + + def __repr__(self): + return '{}({})'.format(self.__class__.__name__, + super().__repr__()) + + @property + def value(self): + return ''.join(x.value for x in self if x.value) + + @property + def all_defects(self): + return sum((x.all_defects for x in self), self.defects) + + # + # Folding API + # + # parts(): + # + # return a list of objects that constitute the "higher level syntactic + # objects" specified by the RFC as the best places to fold a header line. + # The returned objects must include leading folding white space, even if + # this means mutating the underlying parse tree of the object. Each object + # is only responsible for returning *its* parts, and should not drill down + # to any lower level except as required to meet the leading folding white + # space constraint. + # + # _fold(folded): + # + # folded: the result accumulator. This is an instance of _Folded. + # (XXX: I haven't finished factoring this out yet, the folding code + # pretty much uses this as a state object.) When the folded.current + # contains as much text as will fit, the _fold method should call + # folded.newline. + # folded.lastlen: the current length of the test stored in folded.current. + # folded.maxlen: The maximum number of characters that may appear on a + # folded line. Differs from the policy setting in that "no limit" is + # represented by +inf, which means it can be used in the trivially + # logical fashion in comparisons. + # + # Currently no subclasses implement parts, and I think this will remain + # true. A subclass only needs to implement _fold when the generic version + # isn't sufficient. _fold will need to be implemented primarily when it is + # possible for encoded words to appear in the specialized token-list, since + # there is no generic algorithm that can know where exactly the encoded + # words are allowed. A _fold implementation is responsible for filling + # lines in the same general way that the top level _fold does. It may, and + # should, call the _fold method of sub-objects in a similar fashion to that + # of the top level _fold. + # + # XXX: I'm hoping it will be possible to factor the existing code further + # to reduce redundancy and make the logic clearer. + + @property + def parts(self): + klass = self.__class__ + this = [] + for token in self: + if token.startswith_fws(): + if this: + yield this[0] if len(this)==1 else klass(this) + this.clear() + end_ws = token.pop_trailing_ws() + this.append(token) + if end_ws: + yield klass(this) + this = [end_ws] + if this: + yield this[0] if len(this)==1 else klass(this) + + def startswith_fws(self): + return self[0].startswith_fws() + + def pop_leading_fws(self): + if self[0].token_type == 'fws': + return self.pop(0) + return self[0].pop_leading_fws() + + def pop_trailing_ws(self): + if self[-1].token_type == 'cfws': + return self.pop(-1) + return self[-1].pop_trailing_ws() + + @property + def has_fws(self): + for part in self: + if part.has_fws: + return True + return False + + def has_leading_comment(self): + return self[0].has_leading_comment() + + @property + def comments(self): + comments = [] + for token in self: + comments.extend(token.comments) + return comments + + def fold(self, *, policy): + # max_line_length 0/None means no limit, ie: infinitely long. + maxlen = policy.max_line_length or float("+inf") + folded = _Folded(maxlen, policy) + self._fold(folded) + folded.finalize() + return str(folded) + + def as_encoded_word(self, charset): + # This works only for things returned by 'parts', which include + # the leading fws, if any, that should be used. + res = [] + ws = self.pop_leading_fws() + if ws: + res.append(ws) + trailer = self.pop(-1) if self[-1].token_type=='fws' else '' + res.append(_ew.encode(str(self), charset)) + res.append(trailer) + return ''.join(res) + + def cte_encode(self, charset, policy): + res = [] + for part in self: + res.append(part.cte_encode(charset, policy)) + return ''.join(res) + + def _fold(self, folded): + for part in self.parts: + tstr = str(part) + tlen = len(tstr) + try: + str(part).encode('us-ascii') + except UnicodeEncodeError: + if any(isinstance(x, errors.UndecodableBytesDefect) + for x in part.all_defects): + charset = 'unknown-8bit' + else: + # XXX: this should be a policy setting + charset = 'utf-8' + tstr = part.cte_encode(charset, folded.policy) + tlen = len(tstr) + if folded.append_if_fits(part, tstr): + continue + # Peel off the leading whitespace if any and make it sticky, to + # avoid infinite recursion. + ws = part.pop_leading_fws() + if ws is not None: + # Peel off the leading whitespace and make it sticky, to + # avoid infinite recursion. + folded.stickyspace = str(part.pop(0)) + if folded.append_if_fits(part): + continue + if part.has_fws: + part._fold(folded) + continue + # There are no fold points in this one; it is too long for a single + # line and can't be split...we just have to put it on its own line. + folded.append(tstr) + folded.newline() + + def pprint(self, indent=''): + print('\n'.join(self._pp(indent=''))) + + def ppstr(self, indent=''): + return '\n'.join(self._pp(indent='')) + + def _pp(self, indent=''): + yield '{}{}/{}('.format( + indent, + self.__class__.__name__, + self.token_type) + for token in self: + for line in token._pp(indent+' '): + yield line + if self.defects: + extra = ' Defects: {}'.format(self.defects) + else: + extra = '' + yield '{}){}'.format(indent, extra) + + +class WhiteSpaceTokenList(TokenList): + + @property + def value(self): + return ' ' + + @property + def comments(self): + return [x.content for x in self if x.token_type=='comment'] + + +class UnstructuredTokenList(TokenList): + + token_type = 'unstructured' + + def _fold(self, folded): + if any(x.token_type=='encoded-word' for x in self): + return self._fold_encoded(folded) + # Here we can have either a pure ASCII string that may or may not + # have surrogateescape encoded bytes, or a unicode string. + last_ew = None + for part in self.parts: + tstr = str(part) + is_ew = False + try: + str(part).encode('us-ascii') + except UnicodeEncodeError: + if any(isinstance(x, errors.UndecodableBytesDefect) + for x in part.all_defects): + charset = 'unknown-8bit' + else: + charset = 'utf-8' + if last_ew is not None: + # We've already done an EW, combine this one with it + # if there's room. + chunk = get_unstructured( + ''.join(folded.current[last_ew:]+[tstr])).as_encoded_word(charset) + oldlastlen = sum(len(x) for x in folded.current[:last_ew]) + schunk = str(chunk) + lchunk = len(schunk) + if oldlastlen + lchunk <= folded.maxlen: + del folded.current[last_ew:] + folded.append(schunk) + folded.lastlen = oldlastlen + lchunk + continue + tstr = part.as_encoded_word(charset) + is_ew = True + if folded.append_if_fits(part, tstr): + if is_ew: + last_ew = len(folded.current) - 1 + continue + if is_ew or last_ew: + # It's too big to fit on the line, but since we've + # got encoded words we can use encoded word folding. + part._fold_as_ew(folded) + continue + # Peel off the leading whitespace if any and make it sticky, to + # avoid infinite recursion. + ws = part.pop_leading_fws() + if ws is not None: + folded.stickyspace = str(ws) + if folded.append_if_fits(part): + continue + if part.has_fws: + part.fold(folded) + continue + # It can't be split...we just have to put it on its own line. + folded.append(tstr) + folded.newline() + last_ew = None + + def cte_encode(self, charset, policy): + res = [] + last_ew = None + for part in self: + spart = str(part) + try: + spart.encode('us-ascii') + res.append(spart) + except UnicodeEncodeError: + if last_ew is None: + res.append(part.cte_encode(charset, policy)) + last_ew = len(res) + else: + tl = get_unstructured(''.join(res[last_ew:] + [spart])) + res.append(tl.as_encoded_word()) + return ''.join(res) + + +class Phrase(TokenList): + + token_type = 'phrase' + + def _fold(self, folded): + # As with Unstructured, we can have pure ASCII with or without + # surrogateescape encoded bytes, or we could have unicode. But this + # case is more complicated, since we have to deal with the various + # sub-token types and how they can be composed in the face of + # unicode-that-needs-CTE-encoding, and the fact that if a token a + # comment that becomes a barrier across which we can't compose encoded + # words. + last_ew = None + for part in self.parts: + tstr = str(part) + tlen = len(tstr) + has_ew = False + try: + str(part).encode('us-ascii') + except UnicodeEncodeError: + if any(isinstance(x, errors.UndecodableBytesDefect) + for x in part.all_defects): + charset = 'unknown-8bit' + else: + charset = 'utf-8' + if last_ew is not None and not part.has_leading_comment(): + # We've already done an EW, let's see if we can combine + # this one with it. The last_ew logic ensures that all we + # have at this point is atoms, no comments or quoted + # strings. So we can treat the text between the last + # encoded word and the content of this token as + # unstructured text, and things will work correctly. But + # we have to strip off any trailing comment on this token + # first, and if it is a quoted string we have to pull out + # the content (we're encoding it, so it no longer needs to + # be quoted). + if part[-1].token_type == 'cfws' and part.comments: + remainder = part.pop(-1) + else: + remainder = '' + for i, token in enumerate(part): + if token.token_type == 'bare-quoted-string': + part[i] = UnstructuredTokenList(token[:]) + chunk = get_unstructured( + ''.join(folded.current[last_ew:]+[tstr])).as_encoded_word(charset) + schunk = str(chunk) + lchunk = len(schunk) + if last_ew + lchunk <= folded.maxlen: + del folded.current[last_ew:] + folded.append(schunk) + folded.lastlen = sum(len(x) for x in folded.current) + continue + tstr = part.as_encoded_word(charset) + tlen = len(tstr) + has_ew = True + if folded.append_if_fits(part, tstr): + if has_ew and not part.comments: + last_ew = len(folded.current) - 1 + elif part.comments or part.token_type == 'quoted-string': + # If a comment is involved we can't combine EWs. And if a + # quoted string is involved, it's not worth the effort to + # try to combine them. + last_ew = None + continue + part._fold(folded) + + def cte_encode(self, charset, policy): + res = [] + last_ew = None + is_ew = False + for part in self: + spart = str(part) + try: + spart.encode('us-ascii') + res.append(spart) + except UnicodeEncodeError: + is_ew = True + if last_ew is None: + if not part.comments: + last_ew = len(res) + res.append(part.cte_encode(charset, policy)) + elif not part.has_leading_comment(): + if part[-1].token_type == 'cfws' and part.comments: + remainder = part.pop(-1) + else: + remainder = '' + for i, token in enumerate(part): + if token.token_type == 'bare-quoted-string': + part[i] = UnstructuredTokenList(token[:]) + tl = get_unstructured(''.join(res[last_ew:] + [spart])) + res[last_ew:] = [tl.as_encoded_word(charset)] + if part.comments or (not is_ew and part.token_type == 'quoted-string'): + last_ew = None + return ''.join(res) + +class Word(TokenList): + + token_type = 'word' + + +class CFWSList(WhiteSpaceTokenList): + + token_type = 'cfws' + + def has_leading_comment(self): + return bool(self.comments) + + +class Atom(TokenList): + + token_type = 'atom' + + +class EncodedWord(TokenList): + + token_type = 'encoded-word' + cte = None + charset = None + lang = None + + @property + def encoded(self): + if self.cte is not None: + return self.cte + _ew.encode(str(self), self.charset) + + + +class QuotedString(TokenList): + + token_type = 'quoted-string' + + @property + def content(self): + for x in self: + if x.token_type == 'bare-quoted-string': + return x.value + + @property + def quoted_value(self): + res = [] + for x in self: + if x.token_type == 'bare-quoted-string': + res.append(str(x)) + else: + res.append(x.value) + return ''.join(res) + + +class BareQuotedString(QuotedString): + + token_type = 'bare-quoted-string' + + def __str__(self): + return quote_string(''.join(self)) + + @property + def value(self): + return ''.join(str(x) for x in self) + + +class Comment(WhiteSpaceTokenList): + + token_type = 'comment' + + def __str__(self): + return ''.join(sum([ + ["("], + [self.quote(x) for x in self], + [")"], + ], [])) + + def quote(self, value): + if value.token_type == 'comment': + return str(value) + return str(value).replace('\\', '\\\\').replace( + '(', '\(').replace( + ')', '\)') + + @property + def content(self): + return ''.join(str(x) for x in self) + + @property + def comments(self): + return [self.content] + +class AddressList(TokenList): + + token_type = 'address-list' + + @property + def addresses(self): + return [x for x in self if x.token_type=='address'] + + @property + def mailboxes(self): + return sum((x.mailboxes + for x in self if x.token_type=='address'), []) + + @property + def all_mailboxes(self): + return sum((x.all_mailboxes + for x in self if x.token_type=='address'), []) + + +class Address(TokenList): + + token_type = 'address' + + @property + def display_name(self): + if self[0].token_type == 'group': + return self[0].display_name + + @property + def mailboxes(self): + if self[0].token_type == 'mailbox': + return [self[0]] + elif self[0].token_type == 'invalid-mailbox': + return [] + return self[0].mailboxes + + @property + def all_mailboxes(self): + if self[0].token_type == 'mailbox': + return [self[0]] + elif self[0].token_type == 'invalid-mailbox': + return [self[0]] + return self[0].all_mailboxes + +class MailboxList(TokenList): + + token_type = 'mailbox-list' + + @property + def mailboxes(self): + return [x for x in self if x.token_type=='mailbox'] + + @property + def all_mailboxes(self): + return [x for x in self + if x.token_type in ('mailbox', 'invalid-mailbox')] + + +class GroupList(TokenList): + + token_type = 'group-list' + + @property + def mailboxes(self): + if not self or self[0].token_type != 'mailbox-list': + return [] + return self[0].mailboxes + + @property + def all_mailboxes(self): + if not self or self[0].token_type != 'mailbox-list': + return [] + return self[0].all_mailboxes + + +class Group(TokenList): + + token_type = "group" + + @property + def mailboxes(self): + if self[2].token_type != 'group-list': + return [] + return self[2].mailboxes + + @property + def all_mailboxes(self): + if self[2].token_type != 'group-list': + return [] + return self[2].all_mailboxes + + @property + def display_name(self): + return self[0].display_name + + +class NameAddr(TokenList): + + token_type = 'name-addr' + + @property + def display_name(self): + if len(self) == 1: + return None + return self[0].display_name + + @property + def local_part(self): + return self[-1].local_part + + @property + def domain(self): + return self[-1].domain + + @property + def route(self): + return self[-1].route + + @property + def addr_spec(self): + return self[-1].addr_spec + + +class AngleAddr(TokenList): + + token_type = 'angle-addr' + + @property + def local_part(self): + for x in self: + if x.token_type == 'addr-spec': + return x.local_part + + @property + def domain(self): + for x in self: + if x.token_type == 'addr-spec': + return x.domain + + @property + def route(self): + for x in self: + if x.token_type == 'obs-route': + return x.domains + + @property + def addr_spec(self): + for x in self: + if x.token_type == 'addr-spec': + return x.addr_spec + + +class ObsRoute(TokenList): + + token_type = 'obs-route' + + @property + def domains(self): + return [x.domain for x in self if x.token_type == 'domain'] + + +class Mailbox(TokenList): + + token_type = 'mailbox' + + @property + def display_name(self): + if self[0].token_type == 'name-addr': + return self[0].display_name + + @property + def local_part(self): + return self[0].local_part + + @property + def domain(self): + return self[0].domain + + @property + def route(self): + if self[0].token_type == 'name-addr': + return self[0].route + + @property + def addr_spec(self): + return self[0].addr_spec + + +class InvalidMailbox(TokenList): + + token_type = 'invalid-mailbox' + + @property + def display_name(self): + return None + + local_part = domain = route = addr_spec = display_name + + +class Domain(TokenList): + + token_type = 'domain' + + @property + def domain(self): + return ''.join(super().value.split()) + + +class DotAtom(TokenList): + + token_type = 'dot-atom' + + +class DotAtomText(TokenList): + + token_type = 'dot-atom-text' + + +class AddrSpec(TokenList): + + token_type = 'addr-spec' + + @property + def local_part(self): + return self[0].local_part + + @property + def domain(self): + if len(self) < 3: + return None + return self[-1].domain + + @property + def value(self): + if len(self) < 3: + return self[0].value + return self[0].value.rstrip()+self[1].value+self[2].value.lstrip() + + @property + def addr_spec(self): + nameset = set(self.local_part) + if len(nameset) > len(nameset-DOT_ATOM_ENDS): + lp = quote_string(self.local_part) + else: + lp = self.local_part + if self.domain is not None: + return lp + '@' + self.domain + return lp + + +class ObsLocalPart(TokenList): + + token_type = 'obs-local-part' + + +class DisplayName(Phrase): + + token_type = 'display-name' + + @property + def display_name(self): + res = TokenList(self) + if res[0].token_type == 'cfws': + res.pop(0) + else: + if res[0][0].token_type == 'cfws': + res[0] = TokenList(res[0][1:]) + if res[-1].token_type == 'cfws': + res.pop() + else: + if res[-1][-1].token_type == 'cfws': + res[-1] = TokenList(res[-1][:-1]) + return res.value + + @property + def value(self): + quote = False + if self.defects: + quote = True + else: + for x in self: + if x.token_type == 'quoted-string': + quote = True + if quote: + pre = post = '' + if self[0].token_type=='cfws' or self[0][0].token_type=='cfws': + pre = ' ' + if self[-1].token_type=='cfws' or self[-1][-1].token_type=='cfws': + post = ' ' + return pre+quote_string(self.display_name)+post + else: + return super().value + + +class LocalPart(TokenList): + + token_type = 'local-part' + + @property + def value(self): + if self[0].token_type == "quoted-string": + return self[0].quoted_value + else: + return self[0].value + + @property + def local_part(self): + # Strip whitespace from front, back, and around dots. + res = [DOT] + last = DOT + last_is_tl = False + for tok in self[0] + [DOT]: + if tok.token_type == 'cfws': + continue + if (last_is_tl and tok.token_type == 'dot' and + last[-1].token_type == 'cfws'): + res[-1] = TokenList(last[:-1]) + is_tl = isinstance(tok, TokenList) + if (is_tl and last.token_type == 'dot' and + tok[0].token_type == 'cfws'): + res.append(TokenList(tok[1:])) + else: + res.append(tok) + last = res[-1] + last_is_tl = is_tl + res = TokenList(res[1:-1]) + return res.value + + +class DomainLiteral(TokenList): + + token_type = 'domain-literal' + + @property + def domain(self): + return ''.join(super().value.split()) + + @property + def ip(self): + for x in self: + if x.token_type == 'ptext': + return x.value + + +class HeaderLabel(TokenList): + + token_type = 'header-label' + + +class Header(TokenList): + + token_type = 'header' + + def _fold(self, folded): + folded.append(str(self.pop(0))) + folded.lastlen = len(folded.current[0]) + # The first line of the header is different from all others: we don't + # want to start a new object on a new line if it has any fold points in + # it that would allow part of it to be on the first header line. + # Further, if the first fold point would fit on the new line, we want + # to do that, but if it doesn't we want to put it on the first line. + # Folded supports this via the stickyspace attribute. If this + # attribute is not None, it does the special handling. + folded.stickyspace = str(self.pop(0)) if self[0].token_type == 'cfws' else '' + rest = self.pop(0) + if self: + raise ValueError("Malformed Header token list") + rest._fold(folded) + + +# +# Terminal classes and instances +# + +class Terminal(str): + + def __new__(cls, value, token_type): + self = super().__new__(cls, value) + self.token_type = token_type + self.defects = [] + return self + + def __repr__(self): + return "{}({})".format(self.__class__.__name__, super().__repr__()) + + @property + def all_defects(self): + return list(self.defects) + + def _pp(self, indent=''): + return ["{}{}/{}({}){}".format( + indent, + self.__class__.__name__, + self.token_type, + super().__repr__(), + '' if not self.defects else ' {}'.format(self.defects), + )] + + def cte_encode(self, charset, policy): + value = str(self) + try: + value.encode('us-ascii') + return value + except UnicodeEncodeError: + return _ew.encode(value, charset) + + def pop_trailing_ws(self): + # This terminates the recursion. + return None + + def pop_leading_fws(self): + # This terminates the recursion. + return None + + @property + def comments(self): + return [] + + def has_leading_comment(self): + return False + + def __getnewargs__(self): + return(str(self), self.token_type) + + +class WhiteSpaceTerminal(Terminal): + + @property + def value(self): + return ' ' + + def startswith_fws(self): + return True + + has_fws = True + + +class ValueTerminal(Terminal): + + @property + def value(self): + return self + + def startswith_fws(self): + return False + + has_fws = False + + def as_encoded_word(self, charset): + return _ew.encode(str(self), charset) + + +class EWWhiteSpaceTerminal(WhiteSpaceTerminal): + + @property + def value(self): + return '' + + @property + def encoded(self): + return self[:] + + def __str__(self): + return '' + + has_fws = True + + +# XXX these need to become classes and used as instances so +# that a program can't change them in a parse tree and screw +# up other parse trees. Maybe should have tests for that, too. +DOT = ValueTerminal('.', 'dot') +ListSeparator = ValueTerminal(',', 'list-separator') +RouteComponentMarker = ValueTerminal('@', 'route-component-marker') + +# +# Parser +# + +"""Parse strings according to RFC822/2047/2822/5322 rules. + +This is a stateless parser. Each get_XXX function accepts a string and +returns either a Terminal or a TokenList representing the RFC object named +by the method and a string containing the remaining unparsed characters +from the input. Thus a parser method consumes the next syntactic construct +of a given type and returns a token representing the construct plus the +unparsed remainder of the input string. + +For example, if the first element of a structured header is a 'phrase', +then: + + phrase, value = get_phrase(value) + +returns the complete phrase from the start of the string value, plus any +characters left in the string after the phrase is removed. + +""" + +_wsp_splitter = re.compile(r'([{}]+)'.format(''.join(WSP))).split +_non_atom_end_matcher = re.compile(r"[^{}]+".format( + ''.join(ATOM_ENDS).replace('\\','\\\\').replace(']','\]'))).match +_non_printable_finder = re.compile(r"[\x00-\x20\x7F]").findall + +def _validate_xtext(xtext): + """If input token contains ASCII non-printables, register a defect.""" + + non_printables = _non_printable_finder(xtext) + if non_printables: + xtext.defects.append(errors.NonPrintableDefect(non_printables)) + if utils._has_surrogates(xtext): + xtext.defects.append(errors.UndecodableBytesDefect( + "Non-ASCII characters found in header token")) + +def _get_ptext_to_endchars(value, endchars): + """Scan printables/quoted-pairs until endchars and return unquoted ptext. + + This function turns a run of qcontent, ccontent-without-comments, or + dtext-with-quoted-printables into a single string by unquoting any + quoted printables. It returns the string, the remaining value, and + a flag that is True iff there were any quoted printables decoded. + + """ + fragment, *remainder = _wsp_splitter(value, 1) + vchars = [] + escape = False + had_qp = False + for pos in range(len(fragment)): + if fragment[pos] == '\\': + if escape: + escape = False + had_qp = True + else: + escape = True + continue + if escape: + escape = False + elif fragment[pos] in endchars: + break + vchars.append(fragment[pos]) + else: + pos = pos + 1 + return ''.join(vchars), ''.join([fragment[pos:]] + remainder), had_qp + +def _decode_ew_run(value): + """ Decode a run of RFC2047 encoded words. + + _decode_ew_run(value) -> (text, value, defects) + + Scans the supplied value for a run of tokens that look like they are RFC + 2047 encoded words, decodes those words into text according to RFC 2047 + rules (whitespace between encoded words is discarded), and returns the text + and the remaining value (including any leading whitespace on the remaining + value), as well as a list of any defects encountered while decoding. The + input value may not have any leading whitespace. + + """ + res = [] + defects = [] + last_ws = '' + while value: + try: + tok, ws, value = _wsp_splitter(value, 1) + except ValueError: + tok, ws, value = value, '', '' + if not (tok.startswith('=?') and tok.endswith('?=')): + return ''.join(res), last_ws + tok + ws + value, defects + text, charset, lang, new_defects = _ew.decode(tok) + res.append(text) + defects.extend(new_defects) + last_ws = ws + return ''.join(res), last_ws, defects + +def get_fws(value): + """FWS = 1*WSP + + This isn't the RFC definition. We're using fws to represent tokens where + folding can be done, but when we are parsing the *un*folding has already + been done so we don't need to watch out for CRLF. + + """ + newvalue = value.lstrip() + fws = WhiteSpaceTerminal(value[:len(value)-len(newvalue)], 'fws') + return fws, newvalue + +def get_encoded_word(value): + """ encoded-word = "=?" charset "?" encoding "?" encoded-text "?=" + + """ + ew = EncodedWord() + if not value.startswith('=?'): + raise errors.HeaderParseError( + "expected encoded word but found {}".format(value)) + tok, *remainder = value[2:].split('?=', 1) + if tok == value[2:]: + raise errors.HeaderParseError( + "expected encoded word but found {}".format(value)) + remstr = ''.join(remainder) + if remstr[:2].isdigit(): + rest, *remainder = remstr.split('?=', 1) + tok = tok + '?=' + rest + if len(tok.split()) > 1: + ew.defects.append(errors.InvalidHeaderDefect( + "whitespace inside encoded word")) + ew.cte = value + value = ''.join(remainder) + try: + text, charset, lang, defects = _ew.decode('=?' + tok + '?=') + except ValueError: + raise errors.HeaderParseError( + "encoded word format invalid: '{}'".format(ew.cte)) + ew.charset = charset + ew.lang = lang + ew.defects.extend(defects) + while text: + if text[0] in WSP: + token, text = get_fws(text) + ew.append(token) + continue + chars, *remainder = _wsp_splitter(text, 1) + vtext = ValueTerminal(chars, 'vtext') + _validate_xtext(vtext) + ew.append(vtext) + text = ''.join(remainder) + return ew, value + +def get_unstructured(value): + """unstructured = (*([FWS] vchar) *WSP) / obs-unstruct + obs-unstruct = *((*LF *CR *(obs-utext) *LF *CR)) / FWS) + obs-utext = %d0 / obs-NO-WS-CTL / LF / CR + + obs-NO-WS-CTL is control characters except WSP/CR/LF. + + So, basically, we have printable runs, plus control characters or nulls in + the obsolete syntax, separated by whitespace. Since RFC 2047 uses the + obsolete syntax in its specification, but requires whitespace on either + side of the encoded words, I can see no reason to need to separate the + non-printable-non-whitespace from the printable runs if they occur, so we + parse this into xtext tokens separated by WSP tokens. + + Because an 'unstructured' value must by definition constitute the entire + value, this 'get' routine does not return a remaining value, only the + parsed TokenList. + + """ + # XXX: but what about bare CR and LF? They might signal the start or + # end of an encoded word. YAGNI for now, since out current parsers + # will never send us strings with bard CR or LF. + + unstructured = UnstructuredTokenList() + while value: + if value[0] in WSP: + token, value = get_fws(value) + unstructured.append(token) + continue + if value.startswith('=?'): + try: + token, value = get_encoded_word(value) + except errors.HeaderParseError: + pass + else: + have_ws = True + if len(unstructured) > 0: + if unstructured[-1].token_type != 'fws': + unstructured.defects.append(errors.InvalidHeaderDefect( + "missing whitespace before encoded word")) + have_ws = False + if have_ws and len(unstructured) > 1: + if unstructured[-2].token_type == 'encoded-word': + unstructured[-1] = EWWhiteSpaceTerminal( + unstructured[-1], 'fws') + unstructured.append(token) + continue + tok, *remainder = _wsp_splitter(value, 1) + vtext = ValueTerminal(tok, 'vtext') + _validate_xtext(vtext) + unstructured.append(vtext) + value = ''.join(remainder) + return unstructured + +def get_qp_ctext(value): + """ctext = <printable ascii except \ ( )> + + This is not the RFC ctext, since we are handling nested comments in comment + and unquoting quoted-pairs here. We allow anything except the '()' + characters, but if we find any ASCII other than the RFC defined printable + ASCII an NonPrintableDefect is added to the token's defects list. Since + quoted pairs are converted to their unquoted values, what is returned is + a 'ptext' token. In this case it is a WhiteSpaceTerminal, so it's value + is ' '. + + """ + ptext, value, _ = _get_ptext_to_endchars(value, '()') + ptext = WhiteSpaceTerminal(ptext, 'ptext') + _validate_xtext(ptext) + return ptext, value + +def get_qcontent(value): + """qcontent = qtext / quoted-pair + + We allow anything except the DQUOTE character, but if we find any ASCII + other than the RFC defined printable ASCII an NonPrintableDefect is + added to the token's defects list. Any quoted pairs are converted to their + unquoted values, so what is returned is a 'ptext' token. In this case it + is a ValueTerminal. + + """ + ptext, value, _ = _get_ptext_to_endchars(value, '"') + ptext = ValueTerminal(ptext, 'ptext') + _validate_xtext(ptext) + return ptext, value + +def get_atext(value): + """atext = <matches _atext_matcher> + + We allow any non-ATOM_ENDS in atext, but add an InvalidATextDefect to + the token's defects list if we find non-atext characters. + """ + m = _non_atom_end_matcher(value) + if not m: + raise errors.HeaderParseError( + "expected atext but found '{}'".format(value)) + atext = m.group() + value = value[len(atext):] + atext = ValueTerminal(atext, 'atext') + _validate_xtext(atext) + return atext, value + +def get_bare_quoted_string(value): + """bare-quoted-string = DQUOTE *([FWS] qcontent) [FWS] DQUOTE + + A quoted-string without the leading or trailing white space. Its + value is the text between the quote marks, with whitespace + preserved and quoted pairs decoded. + """ + if value[0] != '"': + raise errors.HeaderParseError( + "expected '\"' but found '{}'".format(value)) + bare_quoted_string = BareQuotedString() + value = value[1:] + while value and value[0] != '"': + if value[0] in WSP: + token, value = get_fws(value) + else: + token, value = get_qcontent(value) + bare_quoted_string.append(token) + if not value: + bare_quoted_string.defects.append(errors.InvalidHeaderDefect( + "end of header inside quoted string")) + return bare_quoted_string, value + return bare_quoted_string, value[1:] + +def get_comment(value): + """comment = "(" *([FWS] ccontent) [FWS] ")" + ccontent = ctext / quoted-pair / comment + + We handle nested comments here, and quoted-pair in our qp-ctext routine. + """ + if value and value[0] != '(': + raise errors.HeaderParseError( + "expected '(' but found '{}'".format(value)) + comment = Comment() + value = value[1:] + while value and value[0] != ")": + if value[0] in WSP: + token, value = get_fws(value) + elif value[0] == '(': + token, value = get_comment(value) + else: + token, value = get_qp_ctext(value) + comment.append(token) + if not value: + comment.defects.append(errors.InvalidHeaderDefect( + "end of header inside comment")) + return comment, value + return comment, value[1:] + +def get_cfws(value): + """CFWS = (1*([FWS] comment) [FWS]) / FWS + + """ + cfws = CFWSList() + while value and value[0] in CFWS_LEADER: + if value[0] in WSP: + token, value = get_fws(value) + else: + token, value = get_comment(value) + cfws.append(token) + return cfws, value + +def get_quoted_string(value): + """quoted-string = [CFWS] <bare-quoted-string> [CFWS] + + 'bare-quoted-string' is an intermediate class defined by this + parser and not by the RFC grammar. It is the quoted string + without any attached CFWS. + """ + quoted_string = QuotedString() + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + quoted_string.append(token) + token, value = get_bare_quoted_string(value) + quoted_string.append(token) + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + quoted_string.append(token) + return quoted_string, value + +def get_atom(value): + """atom = [CFWS] 1*atext [CFWS] + + """ + atom = Atom() + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + atom.append(token) + if value and value[0] in ATOM_ENDS: + raise errors.HeaderParseError( + "expected atom but found '{}'".format(value)) + token, value = get_atext(value) + atom.append(token) + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + atom.append(token) + return atom, value + +def get_dot_atom_text(value): + """ dot-text = 1*atext *("." 1*atext) + + """ + dot_atom_text = DotAtomText() + if not value or value[0] in ATOM_ENDS: + raise errors.HeaderParseError("expected atom at a start of " + "dot-atom-text but found '{}'".format(value)) + while value and value[0] not in ATOM_ENDS: + token, value = get_atext(value) + dot_atom_text.append(token) + if value and value[0] == '.': + dot_atom_text.append(DOT) + value = value[1:] + if dot_atom_text[-1] is DOT: + raise errors.HeaderParseError("expected atom at end of dot-atom-text " + "but found '{}'".format('.'+value)) + return dot_atom_text, value + +def get_dot_atom(value): + """ dot-atom = [CFWS] dot-atom-text [CFWS] + + """ + dot_atom = DotAtom() + if value[0] in CFWS_LEADER: + token, value = get_cfws(value) + dot_atom.append(token) + token, value = get_dot_atom_text(value) + dot_atom.append(token) + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + dot_atom.append(token) + return dot_atom, value + +def get_word(value): + """word = atom / quoted-string + + Either atom or quoted-string may start with CFWS. We have to peel off this + CFWS first to determine which type of word to parse. Afterward we splice + the leading CFWS, if any, into the parsed sub-token. + + If neither an atom or a quoted-string is found before the next special, a + HeaderParseError is raised. + + The token returned is either an Atom or a QuotedString, as appropriate. + This means the 'word' level of the formal grammar is not represented in the + parse tree; this is because having that extra layer when manipulating the + parse tree is more confusing than it is helpful. + + """ + if value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + else: + leader = None + if value[0]=='"': + token, value = get_quoted_string(value) + elif value[0] in SPECIALS: + raise errors.HeaderParseError("Expected 'atom' or 'quoted-string' " + "but found '{}'".format(value)) + else: + token, value = get_atom(value) + if leader is not None: + token[:0] = [leader] + return token, value + +def get_phrase(value): + """ phrase = 1*word / obs-phrase + obs-phrase = word *(word / "." / CFWS) + + This means a phrase can be a sequence of words, periods, and CFWS in any + order as long as it starts with at least one word. If anything other than + words is detected, an ObsoleteHeaderDefect is added to the token's defect + list. We also accept a phrase that starts with CFWS followed by a dot; + this is registered as an InvalidHeaderDefect, since it is not supported by + even the obsolete grammar. + + """ + phrase = Phrase() + try: + token, value = get_word(value) + phrase.append(token) + except errors.HeaderParseError: + phrase.defects.append(errors.InvalidHeaderDefect( + "phrase does not start with word")) + while value and value[0] not in PHRASE_ENDS: + if value[0]=='.': + phrase.append(DOT) + phrase.defects.append(errors.ObsoleteHeaderDefect( + "period in 'phrase'")) + value = value[1:] + else: + try: + token, value = get_word(value) + except errors.HeaderParseError: + if value[0] in CFWS_LEADER: + token, value = get_cfws(value) + phrase.defects.append(errors.ObsoleteHeaderDefect( + "comment found without atom")) + else: + raise + phrase.append(token) + return phrase, value + +def get_local_part(value): + """ local-part = dot-atom / quoted-string / obs-local-part + + """ + local_part = LocalPart() + leader = None + if value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + if not value: + raise errors.HeaderParseError( + "expected local-part but found '{}'".format(value)) + try: + token, value = get_dot_atom(value) + except errors.HeaderParseError: + try: + token, value = get_word(value) + except errors.HeaderParseError: + if value[0] != '\\' and value[0] in PHRASE_ENDS: + raise + token = TokenList() + if leader is not None: + token[:0] = [leader] + local_part.append(token) + if value and (value[0]=='\\' or value[0] not in PHRASE_ENDS): + obs_local_part, value = get_obs_local_part(str(local_part) + value) + if obs_local_part.token_type == 'invalid-obs-local-part': + local_part.defects.append(errors.InvalidHeaderDefect( + "local-part is not dot-atom, quoted-string, or obs-local-part")) + else: + local_part.defects.append(errors.ObsoleteHeaderDefect( + "local-part is not a dot-atom (contains CFWS)")) + local_part[0] = obs_local_part + try: + local_part.value.encode('ascii') + except UnicodeEncodeError: + local_part.defects.append(errors.NonASCIILocalPartDefect( + "local-part contains non-ASCII characters)")) + return local_part, value + +def get_obs_local_part(value): + """ obs-local-part = word *("." word) + """ + obs_local_part = ObsLocalPart() + last_non_ws_was_dot = False + while value and (value[0]=='\\' or value[0] not in PHRASE_ENDS): + if value[0] == '.': + if last_non_ws_was_dot: + obs_local_part.defects.append(errors.InvalidHeaderDefect( + "invalid repeated '.'")) + obs_local_part.append(DOT) + last_non_ws_was_dot = True + value = value[1:] + continue + elif value[0]=='\\': + obs_local_part.append(ValueTerminal(value[0], + 'misplaced-special')) + value = value[1:] + obs_local_part.defects.append(errors.InvalidHeaderDefect( + "'\\' character outside of quoted-string/ccontent")) + last_non_ws_was_dot = False + continue + if obs_local_part and obs_local_part[-1].token_type != 'dot': + obs_local_part.defects.append(errors.InvalidHeaderDefect( + "missing '.' between words")) + try: + token, value = get_word(value) + last_non_ws_was_dot = False + except errors.HeaderParseError: + if value[0] not in CFWS_LEADER: + raise + token, value = get_cfws(value) + obs_local_part.append(token) + if (obs_local_part[0].token_type == 'dot' or + obs_local_part[0].token_type=='cfws' and + obs_local_part[1].token_type=='dot'): + obs_local_part.defects.append(errors.InvalidHeaderDefect( + "Invalid leading '.' in local part")) + if (obs_local_part[-1].token_type == 'dot' or + obs_local_part[-1].token_type=='cfws' and + obs_local_part[-2].token_type=='dot'): + obs_local_part.defects.append(errors.InvalidHeaderDefect( + "Invalid trailing '.' in local part")) + if obs_local_part.defects: + obs_local_part.token_type = 'invalid-obs-local-part' + return obs_local_part, value + +def get_dtext(value): + """ dtext = <printable ascii except \ [ ]> / obs-dtext + obs-dtext = obs-NO-WS-CTL / quoted-pair + + We allow anything except the excluded characters, but but if we find any + ASCII other than the RFC defined printable ASCII an NonPrintableDefect is + added to the token's defects list. Quoted pairs are converted to their + unquoted values, so what is returned is a ptext token, in this case a + ValueTerminal. If there were quoted-printables, an ObsoleteHeaderDefect is + added to the returned token's defect list. + + """ + ptext, value, had_qp = _get_ptext_to_endchars(value, '[]') + ptext = ValueTerminal(ptext, 'ptext') + if had_qp: + ptext.defects.append(errors.ObsoleteHeaderDefect( + "quoted printable found in domain-literal")) + _validate_xtext(ptext) + return ptext, value + +def _check_for_early_dl_end(value, domain_literal): + if value: + return False + domain_literal.append(errors.InvalidHeaderDefect( + "end of input inside domain-literal")) + domain_literal.append(ValueTerminal(']', 'domain-literal-end')) + return True + +def get_domain_literal(value): + """ domain-literal = [CFWS] "[" *([FWS] dtext) [FWS] "]" [CFWS] + + """ + domain_literal = DomainLiteral() + if value[0] in CFWS_LEADER: + token, value = get_cfws(value) + domain_literal.append(token) + if not value: + raise errors.HeaderParseError("expected domain-literal") + if value[0] != '[': + raise errors.HeaderParseError("expected '[' at start of domain-literal " + "but found '{}'".format(value)) + value = value[1:] + if _check_for_early_dl_end(value, domain_literal): + return domain_literal, value + domain_literal.append(ValueTerminal('[', 'domain-literal-start')) + if value[0] in WSP: + token, value = get_fws(value) + domain_literal.append(token) + token, value = get_dtext(value) + domain_literal.append(token) + if _check_for_early_dl_end(value, domain_literal): + return domain_literal, value + if value[0] in WSP: + token, value = get_fws(value) + domain_literal.append(token) + if _check_for_early_dl_end(value, domain_literal): + return domain_literal, value + if value[0] != ']': + raise errors.HeaderParseError("expected ']' at end of domain-literal " + "but found '{}'".format(value)) + domain_literal.append(ValueTerminal(']', 'domain-literal-end')) + value = value[1:] + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + domain_literal.append(token) + return domain_literal, value + +def get_domain(value): + """ domain = dot-atom / domain-literal / obs-domain + obs-domain = atom *("." atom)) + + """ + domain = Domain() + leader = None + if value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + if not value: + raise errors.HeaderParseError( + "expected domain but found '{}'".format(value)) + if value[0] == '[': + token, value = get_domain_literal(value) + if leader is not None: + token[:0] = [leader] + domain.append(token) + return domain, value + try: + token, value = get_dot_atom(value) + except errors.HeaderParseError: + token, value = get_atom(value) + if leader is not None: + token[:0] = [leader] + domain.append(token) + if value and value[0] == '.': + domain.defects.append(errors.ObsoleteHeaderDefect( + "domain is not a dot-atom (contains CFWS)")) + if domain[0].token_type == 'dot-atom': + domain[:] = domain[0] + while value and value[0] == '.': + domain.append(DOT) + token, value = get_atom(value[1:]) + domain.append(token) + return domain, value + +def get_addr_spec(value): + """ addr-spec = local-part "@" domain + + """ + addr_spec = AddrSpec() + token, value = get_local_part(value) + addr_spec.append(token) + if not value or value[0] != '@': + addr_spec.defects.append(errors.InvalidHeaderDefect( + "add-spec local part with no domain")) + return addr_spec, value + addr_spec.append(ValueTerminal('@', 'address-at-symbol')) + token, value = get_domain(value[1:]) + addr_spec.append(token) + return addr_spec, value + +def get_obs_route(value): + """ obs-route = obs-domain-list ":" + obs-domain-list = *(CFWS / ",") "@" domain *("," [CFWS] ["@" domain]) + + Returns an obs-route token with the appropriate sub-tokens (that is, + there is no obs-domain-list in the parse tree). + """ + obs_route = ObsRoute() + while value and (value[0]==',' or value[0] in CFWS_LEADER): + if value[0] in CFWS_LEADER: + token, value = get_cfws(value) + obs_route.append(token) + elif value[0] == ',': + obs_route.append(ListSeparator) + value = value[1:] + if not value or value[0] != '@': + raise errors.HeaderParseError( + "expected obs-route domain but found '{}'".format(value)) + obs_route.append(RouteComponentMarker) + token, value = get_domain(value[1:]) + obs_route.append(token) + while value and value[0]==',': + obs_route.append(ListSeparator) + value = value[1:] + if not value: + break + if value[0] in CFWS_LEADER: + token, value = get_cfws(value) + obs_route.append(token) + if value[0] == '@': + obs_route.append(RouteComponentMarker) + token, value = get_domain(value[1:]) + obs_route.append(token) + if not value: + raise errors.HeaderParseError("end of header while parsing obs-route") + if value[0] != ':': + raise errors.HeaderParseError( "expected ':' marking end of " + "obs-route but found '{}'".format(value)) + obs_route.append(ValueTerminal(':', 'end-of-obs-route-marker')) + return obs_route, value[1:] + +def get_angle_addr(value): + """ angle-addr = [CFWS] "<" addr-spec ">" [CFWS] / obs-angle-addr + obs-angle-addr = [CFWS] "<" obs-route addr-spec ">" [CFWS] + + """ + angle_addr = AngleAddr() + if value[0] in CFWS_LEADER: + token, value = get_cfws(value) + angle_addr.append(token) + if not value or value[0] != '<': + raise errors.HeaderParseError( + "expected angle-addr but found '{}'".format(value)) + angle_addr.append(ValueTerminal('<', 'angle-addr-start')) + value = value[1:] + try: + token, value = get_addr_spec(value) + except errors.HeaderParseError: + try: + token, value = get_obs_route(value) + angle_addr.defects.append(errors.ObsoleteHeaderDefect( + "obsolete route specification in angle-addr")) + except errors.HeaderParseError: + raise errors.HeaderParseError( + "expected addr-spec or but found '{}'".format(value)) + angle_addr.append(token) + token, value = get_addr_spec(value) + angle_addr.append(token) + if value and value[0] == '>': + value = value[1:] + else: + angle_addr.defects.append(errors.InvalidHeaderDefect( + "missing trailing '>' on angle-addr")) + angle_addr.append(ValueTerminal('>', 'angle-addr-end')) + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + angle_addr.append(token) + return angle_addr, value + +def get_display_name(value): + """ display-name = phrase + + Because this is simply a name-rule, we don't return a display-name + token containing a phrase, but rather a display-name token with + the content of the phrase. + + """ + display_name = DisplayName() + token, value = get_phrase(value) + display_name.extend(token[:]) + display_name.defects = token.defects[:] + return display_name, value + + +def get_name_addr(value): + """ name-addr = [display-name] angle-addr + + """ + name_addr = NameAddr() + # Both the optional display name and the angle-addr can start with cfws. + leader = None + if value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + if not value: + raise errors.HeaderParseError( + "expected name-addr but found '{}'".format(leader)) + if value[0] != '<': + if value[0] in PHRASE_ENDS: + raise errors.HeaderParseError( + "expected name-addr but found '{}'".format(value)) + token, value = get_display_name(value) + if not value: + raise errors.HeaderParseError( + "expected name-addr but found '{}'".format(token)) + if leader is not None: + token[0][:0] = [leader] + leader = None + name_addr.append(token) + token, value = get_angle_addr(value) + if leader is not None: + token[:0] = [leader] + name_addr.append(token) + return name_addr, value + +def get_mailbox(value): + """ mailbox = name-addr / addr-spec + + """ + # The only way to figure out if we are dealing with a name-addr or an + # addr-spec is to try parsing each one. + mailbox = Mailbox() + try: + token, value = get_name_addr(value) + except errors.HeaderParseError: + try: + token, value = get_addr_spec(value) + except errors.HeaderParseError: + raise errors.HeaderParseError( + "expected mailbox but found '{}'".format(value)) + if any(isinstance(x, errors.InvalidHeaderDefect) + for x in token.all_defects): + mailbox.token_type = 'invalid-mailbox' + mailbox.append(token) + return mailbox, value + +def get_invalid_mailbox(value, endchars): + """ Read everything up to one of the chars in endchars. + + This is outside the formal grammar. The InvalidMailbox TokenList that is + returned acts like a Mailbox, but the data attributes are None. + + """ + invalid_mailbox = InvalidMailbox() + while value and value[0] not in endchars: + if value[0] in PHRASE_ENDS: + invalid_mailbox.append(ValueTerminal(value[0], + 'misplaced-special')) + value = value[1:] + else: + token, value = get_phrase(value) + invalid_mailbox.append(token) + return invalid_mailbox, value + +def get_mailbox_list(value): + """ mailbox-list = (mailbox *("," mailbox)) / obs-mbox-list + obs-mbox-list = *([CFWS] ",") mailbox *("," [mailbox / CFWS]) + + For this routine we go outside the formal grammar in order to improve error + handling. We recognize the end of the mailbox list only at the end of the + value or at a ';' (the group terminator). This is so that we can turn + invalid mailboxes into InvalidMailbox tokens and continue parsing any + remaining valid mailboxes. We also allow all mailbox entries to be null, + and this condition is handled appropriately at a higher level. + + """ + mailbox_list = MailboxList() + while value and value[0] != ';': + try: + token, value = get_mailbox(value) + mailbox_list.append(token) + except errors.HeaderParseError: + leader = None + if value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + if not value or value[0] in ',;': + mailbox_list.append(leader) + mailbox_list.defects.append(errors.ObsoleteHeaderDefect( + "empty element in mailbox-list")) + else: + token, value = get_invalid_mailbox(value, ',;') + if leader is not None: + token[:0] = [leader] + mailbox_list.append(token) + mailbox_list.defects.append(errors.InvalidHeaderDefect( + "invalid mailbox in mailbox-list")) + elif value[0] == ',': + mailbox_list.defects.append(errors.ObsoleteHeaderDefect( + "empty element in mailbox-list")) + else: + token, value = get_invalid_mailbox(value, ',;') + if leader is not None: + token[:0] = [leader] + mailbox_list.append(token) + mailbox_list.defects.append(errors.InvalidHeaderDefect( + "invalid mailbox in mailbox-list")) + if value and value[0] not in ',;': + # Crap after mailbox; treat it as an invalid mailbox. + # The mailbox info will still be available. + mailbox = mailbox_list[-1] + mailbox.token_type = 'invalid-mailbox' + token, value = get_invalid_mailbox(value, ',;') + mailbox.extend(token) + mailbox_list.defects.append(errors.InvalidHeaderDefect( + "invalid mailbox in mailbox-list")) + if value and value[0] == ',': + mailbox_list.append(ListSeparator) + value = value[1:] + return mailbox_list, value + + +def get_group_list(value): + """ group-list = mailbox-list / CFWS / obs-group-list + obs-group-list = 1*([CFWS] ",") [CFWS] + + """ + group_list = GroupList() + if not value: + group_list.defects.append(errors.InvalidHeaderDefect( + "end of header before group-list")) + return group_list, value + leader = None + if value and value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + if not value: + # This should never happen in email parsing, since CFWS-only is a + # legal alternative to group-list in a group, which is the only + # place group-list appears. + group_list.defects.append(errors.InvalidHeaderDefect( + "end of header in group-list")) + group_list.append(leader) + return group_list, value + if value[0] == ';': + group_list.append(leader) + return group_list, value + token, value = get_mailbox_list(value) + if len(token.all_mailboxes)==0: + if leader is not None: + group_list.append(leader) + group_list.extend(token) + group_list.defects.append(errors.ObsoleteHeaderDefect( + "group-list with empty entries")) + return group_list, value + if leader is not None: + token[:0] = [leader] + group_list.append(token) + return group_list, value + +def get_group(value): + """ group = display-name ":" [group-list] ";" [CFWS] + + """ + group = Group() + token, value = get_display_name(value) + if not value or value[0] != ':': + raise errors.HeaderParseError("expected ':' at end of group " + "display name but found '{}'".format(value)) + group.append(token) + group.append(ValueTerminal(':', 'group-display-name-terminator')) + value = value[1:] + if value and value[0] == ';': + group.append(ValueTerminal(';', 'group-terminator')) + return group, value[1:] + token, value = get_group_list(value) + group.append(token) + if not value: + group.defects.append(errors.InvalidHeaderDefect( + "end of header in group")) + if value[0] != ';': + raise errors.HeaderParseError( + "expected ';' at end of group but found {}".format(value)) + group.append(ValueTerminal(';', 'group-terminator')) + value = value[1:] + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + group.append(token) + return group, value + +def get_address(value): + """ address = mailbox / group + + Note that counter-intuitively, an address can be either a single address or + a list of addresses (a group). This is why the returned Address object has + a 'mailboxes' attribute which treats a single address as a list of length + one. When you need to differentiate between to two cases, extract the single + element, which is either a mailbox or a group token. + + """ + # The formal grammar isn't very helpful when parsing an address. mailbox + # and group, especially when allowing for obsolete forms, start off very + # similarly. It is only when you reach one of @, <, or : that you know + # what you've got. So, we try each one in turn, starting with the more + # likely of the two. We could perhaps make this more efficient by looking + # for a phrase and then branching based on the next character, but that + # would be a premature optimization. + address = Address() + try: + token, value = get_group(value) + except errors.HeaderParseError: + try: + token, value = get_mailbox(value) + except errors.HeaderParseError: + raise errors.HeaderParseError( + "expected address but found '{}'".format(value)) + address.append(token) + return address, value + +def get_address_list(value): + """ address_list = (address *("," address)) / obs-addr-list + obs-addr-list = *([CFWS] ",") address *("," [address / CFWS]) + + We depart from the formal grammar here by continuing to parse until the end + of the input, assuming the input to be entirely composed of an + address-list. This is always true in email parsing, and allows us + to skip invalid addresses to parse additional valid ones. + + """ + address_list = AddressList() + while value: + try: + token, value = get_address(value) + address_list.append(token) + except errors.HeaderParseError as err: + leader = None + if value[0] in CFWS_LEADER: + leader, value = get_cfws(value) + if not value or value[0] == ',': + address_list.append(leader) + address_list.defects.append(errors.ObsoleteHeaderDefect( + "address-list entry with no content")) + else: + token, value = get_invalid_mailbox(value, ',') + if leader is not None: + token[:0] = [leader] + address_list.append(Address([token])) + address_list.defects.append(errors.InvalidHeaderDefect( + "invalid address in address-list")) + elif value[0] == ',': + address_list.defects.append(errors.ObsoleteHeaderDefect( + "empty element in address-list")) + else: + token, value = get_invalid_mailbox(value, ',') + if leader is not None: + token[:0] = [leader] + address_list.append(Address([token])) + address_list.defects.append(errors.InvalidHeaderDefect( + "invalid address in address-list")) + if value and value[0] != ',': + # Crap after address; treat it as an invalid mailbox. + # The mailbox info will still be available. + mailbox = address_list[-1][0] + mailbox.token_type = 'invalid-mailbox' + token, value = get_invalid_mailbox(value, ',') + mailbox.extend(token) + address_list.defects.append(errors.InvalidHeaderDefect( + "invalid address in address-list")) + if value: # Must be a , at this point. + address_list.append(ValueTerminal(',', 'list-separator')) + value = value[1:] + return address_list, value diff --git a/Lib/email/_headerregistry.py b/Lib/email/_headerregistry.py new file mode 100644 index 0000000..6588546 --- /dev/null +++ b/Lib/email/_headerregistry.py @@ -0,0 +1,456 @@ +"""Representing and manipulating email headers via custom objects. + +This module provides an implementation of the HeaderRegistry API. +The implementation is designed to flexibly follow RFC5322 rules. + +Eventually HeaderRegistry will be a public API, but it isn't yet, +and will probably change some before that happens. + +""" + +from email import utils +from email import errors +from email import _header_value_parser as parser + +class Address: + + def __init__(self, display_name='', username='', domain='', addr_spec=None): + """Create an object represeting a full email address. + + An address can have a 'display_name', a 'username', and a 'domain'. In + addition to specifying the username and domain separately, they may be + specified together by using the addr_spec keyword *instead of* the + username and domain keywords. If an addr_spec string is specified it + must be properly quoted according to RFC 5322 rules; an error will be + raised if it is not. + + An Address object has display_name, username, domain, and addr_spec + attributes, all of which are read-only. The addr_spec and the string + value of the object are both quoted according to RFC5322 rules, but + without any Content Transfer Encoding. + + """ + # This clause with its potential 'raise' may only happen when an + # application program creates an Address object using an addr_spec + # keyword. The email library code itself must always supply username + # and domain. + if addr_spec is not None: + if username or domain: + raise TypeError("addrspec specified when username and/or " + "domain also specified") + a_s, rest = parser.get_addr_spec(addr_spec) + if rest: + raise ValueError("Invalid addr_spec; only '{}' " + "could be parsed from '{}'".format( + a_s, addr_spec)) + if a_s.all_defects: + raise a_s.all_defects[0] + username = a_s.local_part + domain = a_s.domain + self._display_name = display_name + self._username = username + self._domain = domain + + @property + def display_name(self): + return self._display_name + + @property + def username(self): + return self._username + + @property + def domain(self): + return self._domain + + @property + def addr_spec(self): + """The addr_spec (username@domain) portion of the address, quoted + according to RFC 5322 rules, but with no Content Transfer Encoding. + """ + nameset = set(self.username) + if len(nameset) > len(nameset-parser.DOT_ATOM_ENDS): + lp = parser.quote_string(self.username) + else: + lp = self.username + if self.domain: + return lp + '@' + self.domain + if not lp: + return '<>' + return lp + + def __repr__(self): + return "Address(display_name={!r}, username={!r}, domain={!r})".format( + self.display_name, self.username, self.domain) + + def __str__(self): + nameset = set(self.display_name) + if len(nameset) > len(nameset-parser.SPECIALS): + disp = parser.quote_string(self.display_name) + else: + disp = self.display_name + if disp: + addr_spec = '' if self.addr_spec=='<>' else self.addr_spec + return "{} <{}>".format(disp, addr_spec) + return self.addr_spec + + def __eq__(self, other): + if type(other) != type(self): + return False + return (self.display_name == other.display_name and + self.username == other.username and + self.domain == other.domain) + + +class Group: + + def __init__(self, display_name=None, addresses=None): + """Create an object representing an address group. + + An address group consists of a display_name followed by colon and an + list of addresses (see Address) terminated by a semi-colon. The Group + is created by specifying a display_name and a possibly empty list of + Address objects. A Group can also be used to represent a single + address that is not in a group, which is convenient when manipulating + lists that are a combination of Groups and individual Addresses. In + this case the display_name should be set to None. In particular, the + string representation of a Group whose display_name is None is the same + as the Address object, if there is one and only one Address object in + the addresses list. + + """ + self._display_name = display_name + self._addresses = tuple(addresses) if addresses else tuple() + + @property + def display_name(self): + return self._display_name + + @property + def addresses(self): + return self._addresses + + def __repr__(self): + return "Group(display_name={!r}, addresses={!r}".format( + self.display_name, self.addresses) + + def __str__(self): + if self.display_name is None and len(self.addresses)==1: + return str(self.addresses[0]) + disp = self.display_name + if disp is not None: + nameset = set(disp) + if len(nameset) > len(nameset-parser.SPECIALS): + disp = parser.quote_string(disp) + adrstr = ", ".join(str(x) for x in self.addresses) + adrstr = ' ' + adrstr if adrstr else adrstr + return "{}:{};".format(disp, adrstr) + + def __eq__(self, other): + if type(other) != type(self): + return False + return (self.display_name == other.display_name and + self.addresses == other.addresses) + + +# Header Classes # + +class BaseHeader(str): + + """Base class for message headers. + + Implements generic behavior and provides tools for subclasses. + + A subclass must define a classmethod named 'parse' that takes an unfolded + value string and a dictionary as its arguments. The dictionary will + contain one key, 'defects', initialized to an empty list. After the call + the dictionary must contain two additional keys: parse_tree, set to the + parse tree obtained from parsing the header, and 'decoded', set to the + string value of the idealized representation of the data from the value. + (That is, encoded words are decoded, and values that have canonical + representations are so represented.) + + The defects key is intended to collect parsing defects, which the message + parser will subsequently dispose of as appropriate. The parser should not, + insofar as practical, raise any errors. Defects should be added to the + list instead. The standard header parsers register defects for RFC + compliance issues, for obsolete RFC syntax, and for unrecoverable parsing + errors. + + The parse method may add additional keys to the dictionary. In this case + the subclass must define an 'init' method, which will be passed the + dictionary as its keyword arguments. The method should use (usually by + setting them as the value of similarly named attributes) and remove all the + extra keys added by its parse method, and then use super to call its parent + class with the remaining arguments and keywords. + + The subclass should also make sure that a 'max_count' attribute is defined + that is either None or 1. XXX: need to better define this API. + + """ + + def __new__(cls, name, value): + kwds = {'defects': []} + cls.parse(value, kwds) + if utils._has_surrogates(kwds['decoded']): + kwds['decoded'] = utils._sanitize(kwds['decoded']) + self = str.__new__(cls, kwds['decoded']) + del kwds['decoded'] + self.init(name, **kwds) + return self + + def init(self, name, *, parse_tree, defects): + self._name = name + self._parse_tree = parse_tree + self._defects = defects + + @property + def name(self): + return self._name + + @property + def defects(self): + return tuple(self._defects) + + def __reduce__(self): + return ( + _reconstruct_header, + ( + self.__class__.__name__, + self.__class__.__bases__, + str(self), + ), + self.__dict__) + + @classmethod + def _reconstruct(cls, value): + return str.__new__(cls, value) + + def fold(self, *, policy): + """Fold header according to policy. + + The parsed representation of the header is folded according to + RFC5322 rules, as modified by the policy. If the parse tree + contains surrogateescaped bytes, the bytes are CTE encoded using + the charset 'unknown-8bit". + + Any non-ASCII characters in the parse tree are CTE encoded using + charset utf-8. XXX: make this a policy setting. + + The returned value is an ASCII-only string possibly containing linesep + characters, and ending with a linesep character. The string includes + the header name and the ': ' separator. + + """ + # At some point we need to only put fws here if it was in the source. + header = parser.Header([ + parser.HeaderLabel([ + parser.ValueTerminal(self.name, 'header-name'), + parser.ValueTerminal(':', 'header-sep')]), + parser.CFWSList([parser.WhiteSpaceTerminal(' ', 'fws')]), + self._parse_tree]) + return header.fold(policy=policy) + + +def _reconstruct_header(cls_name, bases, value): + return type(cls_name, bases, {})._reconstruct(value) + + +class UnstructuredHeader: + + max_count = None + value_parser = staticmethod(parser.get_unstructured) + + @classmethod + def parse(cls, value, kwds): + kwds['parse_tree'] = cls.value_parser(value) + kwds['decoded'] = str(kwds['parse_tree']) + + +class UniqueUnstructuredHeader(UnstructuredHeader): + + max_count = 1 + + +class DateHeader: + + """Header whose value consists of a single timestamp. + + Provides an additional attribute, datetime, which is either an aware + datetime using a timezone, or a naive datetime if the timezone + in the input string is -0000. Also accepts a datetime as input. + The 'value' attribute is the normalized form of the timestamp, + which means it is the output of format_datetime on the datetime. + """ + + max_count = None + + # This is used only for folding, not for creating 'decoded'. + value_parser = staticmethod(parser.get_unstructured) + + @classmethod + def parse(cls, value, kwds): + if not value: + kwds['defects'].append(errors.HeaderMissingRequiredValue()) + kwds['datetime'] = None + kwds['decoded'] = '' + kwds['parse_tree'] = parser.TokenList() + return + if isinstance(value, str): + value = utils.parsedate_to_datetime(value) + kwds['datetime'] = value + kwds['decoded'] = utils.format_datetime(kwds['datetime']) + kwds['parse_tree'] = cls.value_parser(kwds['decoded']) + + def init(self, *args, **kw): + self._datetime = kw.pop('datetime') + super().init(*args, **kw) + + @property + def datetime(self): + return self._datetime + + +class UniqueDateHeader(DateHeader): + + max_count = 1 + + +class AddressHeader: + + max_count = None + + @staticmethod + def value_parser(value): + address_list, value = parser.get_address_list(value) + assert not value, 'this should not happen' + return address_list + + @classmethod + def parse(cls, value, kwds): + if isinstance(value, str): + # We are translating here from the RFC language (address/mailbox) + # to our API language (group/address). + kwds['parse_tree'] = address_list = cls.value_parser(value) + groups = [] + for addr in address_list.addresses: + groups.append(Group(addr.display_name, + [Address(mb.display_name or '', + mb.local_part or '', + mb.domain or '') + for mb in addr.all_mailboxes])) + defects = list(address_list.all_defects) + else: + # Assume it is Address/Group stuff + if not hasattr(value, '__iter__'): + value = [value] + groups = [Group(None, [item]) if not hasattr(item, 'addresses') + else item + for item in value] + defects = [] + kwds['groups'] = groups + kwds['defects'] = defects + kwds['decoded'] = ', '.join([str(item) for item in groups]) + if 'parse_tree' not in kwds: + kwds['parse_tree'] = cls.value_parser(kwds['decoded']) + + def init(self, *args, **kw): + self._groups = tuple(kw.pop('groups')) + self._addresses = None + super().init(*args, **kw) + + @property + def groups(self): + return self._groups + + @property + def addresses(self): + if self._addresses is None: + self._addresses = tuple([address for group in self._groups + for address in group.addresses]) + return self._addresses + + +class UniqueAddressHeader(AddressHeader): + + max_count = 1 + + +class SingleAddressHeader(AddressHeader): + + @property + def address(self): + if len(self.addresses)!=1: + raise ValueError(("value of single address header {} is not " + "a single address").format(self.name)) + return self.addresses[0] + + +class UniqueSingleAddressHeader(SingleAddressHeader): + + max_count = 1 + + +# The header factory # + +_default_header_map = { + 'subject': UniqueUnstructuredHeader, + 'date': UniqueDateHeader, + 'resent-date': DateHeader, + 'orig-date': UniqueDateHeader, + 'sender': UniqueSingleAddressHeader, + 'resent-sender': SingleAddressHeader, + 'to': UniqueAddressHeader, + 'resent-to': AddressHeader, + 'cc': UniqueAddressHeader, + 'resent-cc': AddressHeader, + 'bcc': UniqueAddressHeader, + 'resent-bcc': AddressHeader, + 'from': UniqueAddressHeader, + 'resent-from': AddressHeader, + 'reply-to': UniqueAddressHeader, + } + +class HeaderRegistry: + + """A header_factory and header registry.""" + + def __init__(self, base_class=BaseHeader, default_class=UnstructuredHeader, + use_default_map=True): + """Create a header_factory that works with the Policy API. + + base_class is the class that will be the last class in the created + header class's __bases__ list. default_class is the class that will be + used if "name" (see __call__) does not appear in the registry. + use_default_map controls whether or not the default mapping of names to + specialized classes is copied in to the registry when the factory is + created. The default is True. + + """ + self.registry = {} + self.base_class = base_class + self.default_class = default_class + if use_default_map: + self.registry.update(_default_header_map) + + def map_to_type(self, name, cls): + """Register cls as the specialized class for handling "name" headers. + + """ + self.registry[name.lower()] = cls + + def __getitem__(self, name): + cls = self.registry.get(name.lower(), self.default_class) + return type('_'+cls.__name__, (cls, self.base_class), {}) + + def __call__(self, name, value): + """Create a header instance for header 'name' from 'value'. + + Creates a header instance by creating a specialized class for parsing + and representing the specified header by combining the factory + base_class with a specialized class from the registry or the + default_class, and passing the name and value to the constructed + class's constructor. + + """ + return self[name](name, value) diff --git a/Lib/email/_policybase.py b/Lib/email/_policybase.py index 05736d0..6bc298b 100644 --- a/Lib/email/_policybase.py +++ b/Lib/email/_policybase.py @@ -64,10 +64,16 @@ class _PolicyBase: except for the changes passed in as keyword arguments. """ + newpolicy = self.__class__.__new__(self.__class__) for attr, value in self.__dict__.items(): - if attr not in kw: - kw[attr] = value - return self.__class__(**kw) + object.__setattr__(newpolicy, attr, value) + for attr, value in kw.items(): + if not hasattr(self, attr): + raise TypeError( + "{!r} is an invalid keyword argument for {}".format( + attr, self.__class__.__name__)) + object.__setattr__(newpolicy, attr, value) + return newpolicy def __setattr__(self, name, value): if hasattr(self, name): diff --git a/Lib/email/errors.py b/Lib/email/errors.py index c04deb4..f916229 100644 --- a/Lib/email/errors.py +++ b/Lib/email/errors.py @@ -5,7 +5,6 @@ """email package exception classes.""" - class MessageError(Exception): """Base class for errors in the email package.""" @@ -30,9 +29,8 @@ class CharsetError(MessageError): """An illegal charset was given.""" - # These are parsing defects which the parser was able to work around. -class MessageDefect(Exception): +class MessageDefect(ValueError): """Base class for a message defect.""" def __init__(self, line=None): @@ -58,3 +56,42 @@ class MultipartInvariantViolationDefect(MessageDefect): class InvalidMultipartContentTransferEncodingDefect(MessageDefect): """An invalid content transfer encoding was set on the multipart itself.""" + +class UndecodableBytesDefect(MessageDefect): + """Header contained bytes that could not be decoded""" + +class InvalidBase64PaddingDefect(MessageDefect): + """base64 encoded sequence had an incorrect length""" + +class InvalidBase64CharactersDefect(MessageDefect): + """base64 encoded sequence had characters not in base64 alphabet""" + +# These errors are specific to header parsing. + +class HeaderDefect(MessageDefect): + """Base class for a header defect.""" + +class InvalidHeaderDefect(HeaderDefect): + """Header is not valid, message gives details.""" + +class HeaderMissingRequiredValue(HeaderDefect): + """A header that must have a value had none""" + +class NonPrintableDefect(HeaderDefect): + """ASCII characters outside the ascii-printable range found""" + + def __init__(self, non_printables): + super().__init__(non_printables) + self.non_printables = non_printables + + def __str__(self): + return ("the following ASCII non-printables found in header: " + "{}".format(self.non_printables)) + +class ObsoleteHeaderDefect(HeaderDefect): + """Header uses syntax declared obsolete by RFC 5322""" + +class NonASCIILocalPartDefect(HeaderDefect): + """local_part contains non-ASCII characters""" + # This defect only occurs during unicode parsing, not when + # parsing messages decoded from binary. diff --git a/Lib/email/generator.py b/Lib/email/generator.py index bfa288b..fcecf93 100644 --- a/Lib/email/generator.py +++ b/Lib/email/generator.py @@ -95,9 +95,15 @@ class Generator: self._encoded_NL = self._encode(self._NL) self._EMPTY = '' self._encoded_EMTPY = self._encode('') - p = self.policy + # Because we use clone (below) when we recursively process message + # subparts, and because clone uses the computed policy (not None), + # submessages will automatically get set to the computed policy when + # they are processed by this code. + old_gen_policy = self.policy + old_msg_policy = msg.policy try: self.policy = policy + msg.policy = policy if unixfrom: ufrom = msg.get_unixfrom() if not ufrom: @@ -105,7 +111,8 @@ class Generator: self.write(ufrom + self._NL) self._write(msg) finally: - self.policy = p + self.policy = old_gen_policy + msg.policy = old_msg_policy def clone(self, fp): """Clone this generator with the exact same options.""" diff --git a/Lib/email/policy.py b/Lib/email/policy.py index dae2dc7..ea90a8f 100644 --- a/Lib/email/policy.py +++ b/Lib/email/policy.py @@ -2,11 +2,178 @@ code that adds all the email6 features. """ -from email._policybase import Policy, compat32, Compat32 +from email._policybase import Policy, Compat32, compat32 +from email.utils import _has_surrogates +from email._headerregistry import HeaderRegistry as _HeaderRegistry -# XXX: temporarily derive everything from compat32. +__all__ = [ + 'Compat32', + 'compat32', + 'Policy', + 'EmailPolicy', + 'default', + 'strict', + 'SMTP', + 'HTTP', + ] -default = compat32 +class EmailPolicy(Policy): + + """+ + PROVISIONAL + + The API extensions enabled by this this policy are currently provisional. + Refer to the documentation for details. + + This policy adds new header parsing and folding algorithms. Instead of + simple strings, headers are custom objects with custom attributes + depending on the type of the field. The folding algorithm fully + implements RFCs 2047 and 5322. + + In addition to the settable attributes listed above that apply to + all Policies, this policy adds the following additional attributes: + + refold_source -- if the value for a header in the Message object + came from the parsing of some source, this attribute + indicates whether or not a generator should refold + that value when transforming the message back into + stream form. The possible values are: + + none -- all source values use original folding + long -- source values that have any line that is + longer than max_line_length will be + refolded + all -- all values are refolded. + + The default is 'long'. + + header_factory -- a callable that takes two arguments, 'name' and + 'value', where 'name' is a header field name and + 'value' is an unfolded header field value, and + returns a string-like object that represents that + header. A default header_factory is provided that + understands some of the RFC5322 header field types. + (Currently address fields and date fields have + special treatment, while all other fields are + treated as unstructured. This list will be + completed before the extension is marked stable.) + """ + + refold_source = 'long' + header_factory = _HeaderRegistry() + + def __init__(self, **kw): + # Ensure that each new instance gets a unique header factory + # (as opposed to clones, which share the factory). + if 'header_factory' not in kw: + object.__setattr__(self, 'header_factory', _HeaderRegistry()) + super().__init__(**kw) + + # The logic of the next three methods is chosen such that it is possible to + # switch a Message object between a Compat32 policy and a policy derived + # from this class and have the results stay consistent. This allows a + # Message object constructed with this policy to be passed to a library + # that only handles Compat32 objects, or to receive such an object and + # convert it to use the newer style by just changing its policy. It is + # also chosen because it postpones the relatively expensive full rfc5322 + # parse until as late as possible when parsing from source, since in many + # applications only a few headers will actually be inspected. + + def header_source_parse(self, sourcelines): + """+ + The name is parsed as everything up to the ':' and returned unmodified. + The value is determined by stripping leading whitespace off the + remainder of the first line, joining all subsequent lines together, and + stripping any trailing carriage return or linefeed characters. (This + is the same as Compat32). + + """ + name, value = sourcelines[0].split(':', 1) + value = value.lstrip(' \t') + ''.join(sourcelines[1:]) + return (name, value.rstrip('\r\n')) + + def header_store_parse(self, name, value): + """+ + The name is returned unchanged. If the input value has a 'name' + attribute and it matches the name ignoring case, the value is returned + unchanged. Otherwise the name and value are passed to header_factory + method, and the resulting custom header object is returned as the + value. In this case a ValueError is raised if the input value contains + CR or LF characters. + + """ + if hasattr(value, 'name') and value.name.lower() == name.lower(): + return (name, value) + if len(value.splitlines())>1: + raise ValueError("Header values may not contain linefeed " + "or carriage return characters") + return (name, self.header_factory(name, value)) + + def header_fetch_parse(self, name, value): + """+ + If the value has a 'name' attribute, it is returned to unmodified. + Otherwise the name and the value with any linesep characters removed + are passed to the header_factory method, and the resulting custom + header object is returned. Any surrogateescaped bytes get turned + into the unicode unknown-character glyph. + + """ + if hasattr(value, 'name'): + return value + return self.header_factory(name, ''.join(value.splitlines())) + + def fold(self, name, value): + """+ + Header folding is controlled by the refold_source policy setting. A + value is considered to be a 'source value' if and only if it does not + have a 'name' attribute (having a 'name' attribute means it is a header + object of some sort). If a source value needs to be refolded according + to the policy, it is converted into a custom header object by passing + the name and the value with any linesep characters removed to the + header_factory method. Folding of a custom header object is done by + calling its fold method with the current policy. + + Source values are split into lines using splitlines. If the value is + not to be refolded, the lines are rejoined using the linesep from the + policy and returned. The exception is lines containing non-ascii + binary data. In that case the value is refolded regardless of the + refold_source setting, which causes the binary data to be CTE encoded + using the unknown-8bit charset. + + """ + return self._fold(name, value, refold_binary=True) + + def fold_binary(self, name, value): + """+ + The same as fold if cte_type is 7bit, except that the returned value is + bytes. + + If cte_type is 8bit, non-ASCII binary data is converted back into + bytes. Headers with binary data are not refolded, regardless of the + refold_header setting, since there is no way to know whether the binary + data consists of single byte characters or multibyte characters. + + """ + folded = self._fold(name, value, refold_binary=self.cte_type=='7bit') + return folded.encode('ascii', 'surrogateescape') + + def _fold(self, name, value, refold_binary=False): + if hasattr(value, 'name'): + return value.fold(policy=self) + maxlen = self.max_line_length if self.max_line_length else float('inf') + lines = value.splitlines() + refold = (self.refold_source == 'all' or + self.refold_source == 'long' and + (len(lines[0])+len(name)+2 > maxlen or + any(len(x) > maxlen for x in lines[1:]))) + if refold or refold_binary and _has_surrogates(value): + return self.header_factory(name, ''.join(lines)).fold(policy=self) + return name + ': ' + self.linesep.join(lines) + self.linesep + + +default = EmailPolicy() +# Make the default policy use the class default header_factory +del default.header_factory strict = default.clone(raise_on_defect=True) SMTP = default.clone(linesep='\r\n') HTTP = default.clone(linesep='\r\n', max_line_length=None) diff --git a/Lib/email/utils.py b/Lib/email/utils.py index b82d5c5..b7e1bb9 100644 --- a/Lib/email/utils.py +++ b/Lib/email/utils.py @@ -62,6 +62,13 @@ escapesre = re.compile(r'[\\"]') _has_surrogates = re.compile( '([^\ud800-\udbff]|\A)[\udc00-\udfff]([^\udc00-\udfff]|\Z)').search +# How to deal with a string containing bytes before handing it to the +# application through the 'normal' interface. +def _sanitize(string): + # Turn any escaped bytes into unicode 'unknown' char. + original_bytes = string.encode('ascii', 'surrogateescape') + return original_bytes.decode('ascii', 'replace') + # Helpers diff --git a/Lib/test/test_email/__init__.py b/Lib/test/test_email/__init__.py index b05fb3c..75dc64d 100644 --- a/Lib/test/test_email/__init__.py +++ b/Lib/test/test_email/__init__.py @@ -65,3 +65,9 @@ class TestEmailBase(unittest.TestCase): def assertBytesEqual(self, first, second, msg): """Our byte strings are really encoded strings; improve diff output""" self.assertEqual(self._bytes_repr(first), self._bytes_repr(second)) + + def assertDefectsEqual(self, actual, expected): + self.assertEqual(len(actual), len(expected), actual) + for i in range(len(actual)): + self.assertIsInstance(actual[i], expected[i], + 'item {}'.format(i)) diff --git a/Lib/test/test_email/test__encoded_words.py b/Lib/test/test_email/test__encoded_words.py new file mode 100644 index 0000000..14395fe --- /dev/null +++ b/Lib/test/test_email/test__encoded_words.py @@ -0,0 +1,187 @@ +import unittest +from email import _encoded_words as _ew +from email import errors +from test.test_email import TestEmailBase + + +class TestDecodeQ(TestEmailBase): + + def _test(self, source, ex_result, ex_defects=[]): + result, defects = _ew.decode_q(source) + self.assertEqual(result, ex_result) + self.assertDefectsEqual(defects, ex_defects) + + def test_no_encoded(self): + self._test(b'foobar', b'foobar') + + def test_spaces(self): + self._test(b'foo=20bar=20', b'foo bar ') + self._test(b'foo_bar_', b'foo bar ') + + def test_run_of_encoded(self): + self._test(b'foo=20=20=21=2Cbar', b'foo !,bar') + + +class TestDecodeB(TestEmailBase): + + def _test(self, source, ex_result, ex_defects=[]): + result, defects = _ew.decode_b(source) + self.assertEqual(result, ex_result) + self.assertDefectsEqual(defects, ex_defects) + + def test_simple(self): + self._test(b'Zm9v', b'foo') + + def test_missing_padding(self): + self._test(b'dmk', b'vi', [errors.InvalidBase64PaddingDefect]) + + def test_invalid_character(self): + self._test(b'dm\x01k===', b'vi', [errors.InvalidBase64CharactersDefect]) + + def test_invalid_character_and_bad_padding(self): + self._test(b'dm\x01k', b'vi', [errors.InvalidBase64CharactersDefect, + errors.InvalidBase64PaddingDefect]) + + +class TestDecode(TestEmailBase): + + def test_wrong_format_input_raises(self): + with self.assertRaises(ValueError): + _ew.decode('=?badone?=') + with self.assertRaises(ValueError): + _ew.decode('=?') + with self.assertRaises(ValueError): + _ew.decode('') + + def _test(self, source, result, charset='us-ascii', lang='', defects=[]): + res, char, l, d = _ew.decode(source) + self.assertEqual(res, result) + self.assertEqual(char, charset) + self.assertEqual(l, lang) + self.assertDefectsEqual(d, defects) + + def test_simple_q(self): + self._test('=?us-ascii?q?foo?=', 'foo') + + def test_simple_b(self): + self._test('=?us-ascii?b?dmk=?=', 'vi') + + def test_q_case_ignored(self): + self._test('=?us-ascii?Q?foo?=', 'foo') + + def test_b_case_ignored(self): + self._test('=?us-ascii?B?dmk=?=', 'vi') + + def test_non_trivial_q(self): + self._test('=?latin-1?q?=20F=fcr=20Elise=20?=', ' Für Elise ', 'latin-1') + + def test_q_escpaed_bytes_preserved(self): + self._test(b'=?us-ascii?q?=20\xACfoo?='.decode('us-ascii', + 'surrogateescape'), + ' \uDCACfoo', + defects = [errors.UndecodableBytesDefect]) + + def test_b_undecodable_bytes_ignored_with_defect(self): + self._test(b'=?us-ascii?b?dm\xACk?='.decode('us-ascii', + 'surrogateescape'), + 'vi', + defects = [ + errors.InvalidBase64CharactersDefect, + errors.InvalidBase64PaddingDefect]) + + def test_b_invalid_bytes_ignored_with_defect(self): + self._test('=?us-ascii?b?dm\x01k===?=', + 'vi', + defects = [errors.InvalidBase64CharactersDefect]) + + def test_b_invalid_bytes_incorrect_padding(self): + self._test('=?us-ascii?b?dm\x01k?=', + 'vi', + defects = [ + errors.InvalidBase64CharactersDefect, + errors.InvalidBase64PaddingDefect]) + + def test_b_padding_defect(self): + self._test('=?us-ascii?b?dmk?=', + 'vi', + defects = [errors.InvalidBase64PaddingDefect]) + + def test_nonnull_lang(self): + self._test('=?us-ascii*jive?q?test?=', 'test', lang='jive') + + def test_unknown_8bit_charset(self): + self._test('=?unknown-8bit?q?foo=ACbar?=', + b'foo\xacbar'.decode('ascii', 'surrogateescape'), + charset = 'unknown-8bit', + defects = []) + + def test_unknown_charset(self): + self._test('=?foobar?q?foo=ACbar?=', + b'foo\xacbar'.decode('ascii', 'surrogateescape'), + charset = 'foobar', + # XXX Should this be a new Defect instead? + defects = [errors.CharsetError]) + + +class TestEncodeQ(TestEmailBase): + + def _test(self, src, expected): + self.assertEqual(_ew.encode_q(src), expected) + + def test_all_safe(self): + self._test(b'foobar', 'foobar') + + def test_spaces(self): + self._test(b'foo bar ', 'foo_bar_') + + def test_run_of_encodables(self): + self._test(b'foo ,,bar', 'foo__=2C=2Cbar') + + +class TestEncodeB(TestEmailBase): + + def test_simple(self): + self.assertEqual(_ew.encode_b(b'foo'), 'Zm9v') + + def test_padding(self): + self.assertEqual(_ew.encode_b(b'vi'), 'dmk=') + + +class TestEncode(TestEmailBase): + + def test_q(self): + self.assertEqual(_ew.encode('foo', 'utf-8', 'q'), '=?utf-8?q?foo?=') + + def test_b(self): + self.assertEqual(_ew.encode('foo', 'utf-8', 'b'), '=?utf-8?b?Zm9v?=') + + def test_auto_q(self): + self.assertEqual(_ew.encode('foo', 'utf-8'), '=?utf-8?q?foo?=') + + def test_auto_q_if_short_mostly_safe(self): + self.assertEqual(_ew.encode('vi.', 'utf-8'), '=?utf-8?q?vi=2E?=') + + def test_auto_b_if_enough_unsafe(self): + self.assertEqual(_ew.encode('.....', 'utf-8'), '=?utf-8?b?Li4uLi4=?=') + + def test_auto_b_if_long_unsafe(self): + self.assertEqual(_ew.encode('vi.vi.vi.vi.vi.', 'utf-8'), + '=?utf-8?b?dmkudmkudmkudmkudmku?=') + + def test_auto_q_if_long_mostly_safe(self): + self.assertEqual(_ew.encode('vi vi vi.vi ', 'utf-8'), + '=?utf-8?q?vi_vi_vi=2Evi_?=') + + def test_utf8_default(self): + self.assertEqual(_ew.encode('foo'), '=?utf-8?q?foo?=') + + def test_lang(self): + self.assertEqual(_ew.encode('foo', lang='jive'), '=?utf-8*jive?q?foo?=') + + def test_unknown_8bit(self): + self.assertEqual(_ew.encode('foo\uDCACbar', charset='unknown-8bit'), + '=?unknown-8bit?q?foo=ACbar?=') + + +if __name__ == '__main__': + unittest.main() diff --git a/Lib/test/test_email/test__header_value_parser.py b/Lib/test/test_email/test__header_value_parser.py new file mode 100644 index 0000000..75fe299 --- /dev/null +++ b/Lib/test/test_email/test__header_value_parser.py @@ -0,0 +1,2466 @@ +import string +import unittest +from email import _header_value_parser as parser +from email import errors +from email import policy +from test.test_email import TestEmailBase + +class TestTokens(TestEmailBase): + + # EWWhiteSpaceTerminal + + def test_EWWhiteSpaceTerminal(self): + x = parser.EWWhiteSpaceTerminal(' \t', 'fws') + self.assertEqual(x, ' \t') + self.assertEqual(str(x), '') + self.assertEqual(x.value, '') + self.assertEqual(x.encoded, ' \t') + + # UnstructuredTokenList + + def test_undecodable_bytes_error_preserved(self): + badstr = b"le pouf c\xaflebre".decode('ascii', 'surrogateescape') + unst = parser.get_unstructured(badstr) + self.assertDefectsEqual(unst.all_defects, [errors.UndecodableBytesDefect]) + parts = list(unst.parts) + self.assertDefectsEqual(parts[0].all_defects, []) + self.assertDefectsEqual(parts[1].all_defects, []) + self.assertDefectsEqual(parts[2].all_defects, [errors.UndecodableBytesDefect]) + + +class TestParser(TestEmailBase): + + # _wsp_splitter + + rfc_printable_ascii = bytes(range(33, 127)).decode('ascii') + rfc_atext_chars = (string.ascii_letters + string.digits + + "!#$%&\'*+-/=?^_`{}|~") + rfc_dtext_chars = rfc_printable_ascii.translate(str.maketrans('','',r'\[]')) + + def test__wsp_splitter_one_word(self): + self.assertEqual(parser._wsp_splitter('foo', 1), ['foo']) + + def test__wsp_splitter_two_words(self): + self.assertEqual(parser._wsp_splitter('foo def', 1), + ['foo', ' ', 'def']) + + def test__wsp_splitter_ws_runs(self): + self.assertEqual(parser._wsp_splitter('foo \t def jik', 1), + ['foo', ' \t ', 'def jik']) + + + # test harness + + def _test_get_x(self, method, input, string, value, defects, + remainder, comments=None): + token, rest = method(input) + self.assertEqual(str(token), string) + self.assertEqual(token.value, value) + self.assertDefectsEqual(token.all_defects, defects) + self.assertEqual(rest, remainder) + if comments is not None: + self.assertEqual(token.comments, comments) + return token + + # get_fws + + def test_get_fws_only(self): + fws = self._test_get_x(parser.get_fws, ' \t ', ' \t ', ' ', [], '') + self.assertEqual(fws.token_type, 'fws') + + def test_get_fws_space(self): + self._test_get_x(parser.get_fws, ' foo', ' ', ' ', [], 'foo') + + def test_get_fws_ws_run(self): + self._test_get_x(parser.get_fws, ' \t foo ', ' \t ', ' ', [], 'foo ') + + # get_encoded_word + + def test_get_encoded_word_missing_start_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_encoded_word('abc') + + def test_get_encoded_word_missing_end_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_encoded_word('=?abc') + + def test_get_encoded_word_missing_middle_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_encoded_word('=?abc?=') + + def test_get_encoded_word_valid_ew(self): + self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?this_is_a_test?= bird', + 'this is a test', + 'this is a test', + [], + ' bird') + + def test_get_encoded_word_internal_spaces(self): + self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?this is a test?= bird', + 'this is a test', + 'this is a test', + [errors.InvalidHeaderDefect], + ' bird') + + def test_get_encoded_word_gets_first(self): + self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?first?= =?utf-8?q?second?=', + 'first', + 'first', + [], + ' =?utf-8?q?second?=') + + def test_get_encoded_word_gets_first_even_if_no_space(self): + self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?first?==?utf-8?q?second?=', + 'first', + 'first', + [], + '=?utf-8?q?second?=') + + def test_get_encoded_word_sets_extra_attributes(self): + ew = self._test_get_x(parser.get_encoded_word, + '=?us-ascii*jive?q?first_second?=', + 'first second', + 'first second', + [], + '') + self.assertEqual(ew.encoded, '=?us-ascii*jive?q?first_second?=') + self.assertEqual(ew.charset, 'us-ascii') + self.assertEqual(ew.lang, 'jive') + + def test_get_encoded_word_lang_default_is_blank(self): + ew = self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?first_second?=', + 'first second', + 'first second', + [], + '') + self.assertEqual(ew.encoded, '=?us-ascii?q?first_second?=') + self.assertEqual(ew.charset, 'us-ascii') + self.assertEqual(ew.lang, '') + + def test_get_encoded_word_non_printable_defect(self): + self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?first\x02second?=', + 'first\x02second', + 'first\x02second', + [errors.NonPrintableDefect], + '') + + def test_get_encoded_word_leading_internal_space(self): + self._test_get_x(parser.get_encoded_word, + '=?us-ascii?q?=20foo?=', + ' foo', + ' foo', + [], + '') + + # get_unstructured + + def _get_unst(self, value): + token = parser.get_unstructured(value) + return token, '' + + def test_get_unstructured_null(self): + self._test_get_x(self._get_unst, '', '', '', [], '') + + def test_get_unstructured_one_word(self): + self._test_get_x(self._get_unst, 'foo', 'foo', 'foo', [], '') + + def test_get_unstructured_normal_phrase(self): + self._test_get_x(self._get_unst, 'foo bar bird', + 'foo bar bird', + 'foo bar bird', + [], + '') + + def test_get_unstructured_normal_phrase_with_whitespace(self): + self._test_get_x(self._get_unst, 'foo \t bar bird', + 'foo \t bar bird', + 'foo bar bird', + [], + '') + + def test_get_unstructured_leading_whitespace(self): + self._test_get_x(self._get_unst, ' foo bar', + ' foo bar', + ' foo bar', + [], + '') + + def test_get_unstructured_trailing_whitespace(self): + self._test_get_x(self._get_unst, 'foo bar ', + 'foo bar ', + 'foo bar ', + [], + '') + + def test_get_unstructured_leading_and_trailing_whitespace(self): + self._test_get_x(self._get_unst, ' foo bar ', + ' foo bar ', + ' foo bar ', + [], + '') + + def test_get_unstructured_one_valid_ew_no_ws(self): + self._test_get_x(self._get_unst, '=?us-ascii?q?bar?=', + 'bar', + 'bar', + [], + '') + + def test_get_unstructured_one_ew_trailing_ws(self): + self._test_get_x(self._get_unst, '=?us-ascii?q?bar?= ', + 'bar ', + 'bar ', + [], + '') + + def test_get_unstructured_one_valid_ew_trailing_text(self): + self._test_get_x(self._get_unst, '=?us-ascii?q?bar?= bird', + 'bar bird', + 'bar bird', + [], + '') + + def test_get_unstructured_phrase_with_ew_in_middle_of_text(self): + self._test_get_x(self._get_unst, 'foo =?us-ascii?q?bar?= bird', + 'foo bar bird', + 'foo bar bird', + [], + '') + + def test_get_unstructured_phrase_with_two_ew(self): + self._test_get_x(self._get_unst, + 'foo =?us-ascii?q?bar?= =?us-ascii?q?bird?=', + 'foo barbird', + 'foo barbird', + [], + '') + + def test_get_unstructured_phrase_with_two_ew_trailing_ws(self): + self._test_get_x(self._get_unst, + 'foo =?us-ascii?q?bar?= =?us-ascii?q?bird?= ', + 'foo barbird ', + 'foo barbird ', + [], + '') + + def test_get_unstructured_phrase_with_ew_with_leading_ws(self): + self._test_get_x(self._get_unst, + ' =?us-ascii?q?bar?=', + ' bar', + ' bar', + [], + '') + + def test_get_unstructured_phrase_with_two_ew_extra_ws(self): + self._test_get_x(self._get_unst, + 'foo =?us-ascii?q?bar?= \t =?us-ascii?q?bird?=', + 'foo barbird', + 'foo barbird', + [], + '') + + def test_get_unstructured_two_ew_extra_ws_trailing_text(self): + self._test_get_x(self._get_unst, + '=?us-ascii?q?test?= =?us-ascii?q?foo?= val', + 'testfoo val', + 'testfoo val', + [], + '') + + def test_get_unstructured_ew_with_internal_ws(self): + self._test_get_x(self._get_unst, + '=?iso-8859-1?q?hello=20world?=', + 'hello world', + 'hello world', + [], + '') + + def test_get_unstructured_ew_with_internal_leading_ws(self): + self._test_get_x(self._get_unst, + ' =?us-ascii?q?=20test?= =?us-ascii?q?=20foo?= val', + ' test foo val', + ' test foo val', + [], + '') + + def test_get_unstructured_invaild_ew(self): + self._test_get_x(self._get_unst, + '=?test val', + '=?test val', + '=?test val', + [], + '') + + def test_get_unstructured_undecodable_bytes(self): + self._test_get_x(self._get_unst, + b'test \xACfoo val'.decode('ascii', 'surrogateescape'), + 'test \uDCACfoo val', + 'test \uDCACfoo val', + [errors.UndecodableBytesDefect], + '') + + def test_get_unstructured_undecodable_bytes_in_EW(self): + self._test_get_x(self._get_unst, + (b'=?us-ascii?q?=20test?= =?us-ascii?q?=20\xACfoo?=' + b' val').decode('ascii', 'surrogateescape'), + ' test \uDCACfoo val', + ' test \uDCACfoo val', + [errors.UndecodableBytesDefect]*2, + '') + + def test_get_unstructured_missing_base64_padding(self): + self._test_get_x(self._get_unst, + '=?utf-8?b?dmk?=', + 'vi', + 'vi', + [errors.InvalidBase64PaddingDefect], + '') + + def test_get_unstructured_invalid_base64_character(self): + self._test_get_x(self._get_unst, + '=?utf-8?b?dm\x01k===?=', + 'vi', + 'vi', + [errors.InvalidBase64CharactersDefect], + '') + + def test_get_unstructured_invalid_base64_character_and_bad_padding(self): + self._test_get_x(self._get_unst, + '=?utf-8?b?dm\x01k?=', + 'vi', + 'vi', + [errors.InvalidBase64CharactersDefect, + errors.InvalidBase64PaddingDefect], + '') + + def test_get_unstructured_no_whitespace_between_ews(self): + self._test_get_x(self._get_unst, + '=?utf-8?q?foo?==?utf-8?q?bar?=', + 'foobar', + 'foobar', + [errors.InvalidHeaderDefect], + '') + + # get_qp_ctext + + def test_get_qp_ctext_only(self): + ptext = self._test_get_x(parser.get_qp_ctext, + 'foobar', 'foobar', ' ', [], '') + self.assertEqual(ptext.token_type, 'ptext') + + def test_get_qp_ctext_all_printables(self): + with_qp = self.rfc_printable_ascii.replace('\\', '\\\\') + with_qp = with_qp. replace('(', r'\(') + with_qp = with_qp.replace(')', r'\)') + ptext = self._test_get_x(parser.get_qp_ctext, + with_qp, self.rfc_printable_ascii, ' ', [], '') + + def test_get_qp_ctext_two_words_gets_first(self): + self._test_get_x(parser.get_qp_ctext, + 'foo de', 'foo', ' ', [], ' de') + + def test_get_qp_ctext_following_wsp_preserved(self): + self._test_get_x(parser.get_qp_ctext, + 'foo \t\tde', 'foo', ' ', [], ' \t\tde') + + def test_get_qp_ctext_up_to_close_paren_only(self): + self._test_get_x(parser.get_qp_ctext, + 'foo)', 'foo', ' ', [], ')') + + def test_get_qp_ctext_wsp_before_close_paren_preserved(self): + self._test_get_x(parser.get_qp_ctext, + 'foo )', 'foo', ' ', [], ' )') + + def test_get_qp_ctext_close_paren_mid_word(self): + self._test_get_x(parser.get_qp_ctext, + 'foo)bar', 'foo', ' ', [], ')bar') + + def test_get_qp_ctext_up_to_open_paren_only(self): + self._test_get_x(parser.get_qp_ctext, + 'foo(', 'foo', ' ', [], '(') + + def test_get_qp_ctext_wsp_before_open_paren_preserved(self): + self._test_get_x(parser.get_qp_ctext, + 'foo (', 'foo', ' ', [], ' (') + + def test_get_qp_ctext_open_paren_mid_word(self): + self._test_get_x(parser.get_qp_ctext, + 'foo(bar', 'foo', ' ', [], '(bar') + + def test_get_qp_ctext_non_printables(self): + ptext = self._test_get_x(parser.get_qp_ctext, + 'foo\x00bar)', 'foo\x00bar', ' ', + [errors.NonPrintableDefect], ')') + self.assertEqual(ptext.defects[0].non_printables[0], '\x00') + + # get_qcontent + + def test_get_qcontent_only(self): + ptext = self._test_get_x(parser.get_qcontent, + 'foobar', 'foobar', 'foobar', [], '') + self.assertEqual(ptext.token_type, 'ptext') + + def test_get_qcontent_all_printables(self): + with_qp = self.rfc_printable_ascii.replace('\\', '\\\\') + with_qp = with_qp. replace('"', r'\"') + ptext = self._test_get_x(parser.get_qcontent, with_qp, + self.rfc_printable_ascii, + self.rfc_printable_ascii, [], '') + + def test_get_qcontent_two_words_gets_first(self): + self._test_get_x(parser.get_qcontent, + 'foo de', 'foo', 'foo', [], ' de') + + def test_get_qcontent_following_wsp_preserved(self): + self._test_get_x(parser.get_qcontent, + 'foo \t\tde', 'foo', 'foo', [], ' \t\tde') + + def test_get_qcontent_up_to_dquote_only(self): + self._test_get_x(parser.get_qcontent, + 'foo"', 'foo', 'foo', [], '"') + + def test_get_qcontent_wsp_before_close_paren_preserved(self): + self._test_get_x(parser.get_qcontent, + 'foo "', 'foo', 'foo', [], ' "') + + def test_get_qcontent_close_paren_mid_word(self): + self._test_get_x(parser.get_qcontent, + 'foo"bar', 'foo', 'foo', [], '"bar') + + def test_get_qcontent_non_printables(self): + ptext = self._test_get_x(parser.get_qcontent, + 'foo\x00fg"', 'foo\x00fg', 'foo\x00fg', + [errors.NonPrintableDefect], '"') + self.assertEqual(ptext.defects[0].non_printables[0], '\x00') + + # get_atext + + def test_get_atext_only(self): + atext = self._test_get_x(parser.get_atext, + 'foobar', 'foobar', 'foobar', [], '') + self.assertEqual(atext.token_type, 'atext') + + def test_get_atext_all_atext(self): + atext = self._test_get_x(parser.get_atext, self.rfc_atext_chars, + self.rfc_atext_chars, + self.rfc_atext_chars, [], '') + + def test_get_atext_two_words_gets_first(self): + self._test_get_x(parser.get_atext, + 'foo bar', 'foo', 'foo', [], ' bar') + + def test_get_atext_following_wsp_preserved(self): + self._test_get_x(parser.get_atext, + 'foo \t\tbar', 'foo', 'foo', [], ' \t\tbar') + + def test_get_atext_up_to_special(self): + self._test_get_x(parser.get_atext, + 'foo@bar', 'foo', 'foo', [], '@bar') + + def test_get_atext_non_printables(self): + atext = self._test_get_x(parser.get_atext, + 'foo\x00bar(', 'foo\x00bar', 'foo\x00bar', + [errors.NonPrintableDefect], '(') + self.assertEqual(atext.defects[0].non_printables[0], '\x00') + + # get_bare_quoted_string + + def test_get_bare_quoted_string_only(self): + bqs = self._test_get_x(parser.get_bare_quoted_string, + '"foo"', '"foo"', 'foo', [], '') + self.assertEqual(bqs.token_type, 'bare-quoted-string') + + def test_get_bare_quoted_string_must_start_with_dquote(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_bare_quoted_string('foo"') + with self.assertRaises(errors.HeaderParseError): + parser.get_bare_quoted_string(' "foo"') + + def test_get_bare_quoted_string_following_wsp_preserved(self): + self._test_get_x(parser.get_bare_quoted_string, + '"foo"\t bar', '"foo"', 'foo', [], '\t bar') + + def test_get_bare_quoted_string_multiple_words(self): + self._test_get_x(parser.get_bare_quoted_string, + '"foo bar moo"', '"foo bar moo"', 'foo bar moo', [], '') + + def test_get_bare_quoted_string_multiple_words_wsp_preserved(self): + self._test_get_x(parser.get_bare_quoted_string, + '" foo moo\t"', '" foo moo\t"', ' foo moo\t', [], '') + + def test_get_bare_quoted_string_end_dquote_mid_word(self): + self._test_get_x(parser.get_bare_quoted_string, + '"foo"bar', '"foo"', 'foo', [], 'bar') + + def test_get_bare_quoted_string_quoted_dquote(self): + self._test_get_x(parser.get_bare_quoted_string, + r'"foo\"in"a', r'"foo\"in"', 'foo"in', [], 'a') + + def test_get_bare_quoted_string_non_printables(self): + self._test_get_x(parser.get_bare_quoted_string, + '"a\x01a"', '"a\x01a"', 'a\x01a', + [errors.NonPrintableDefect], '') + + def test_get_bare_quoted_string_no_end_dquote(self): + self._test_get_x(parser.get_bare_quoted_string, + '"foo', '"foo"', 'foo', + [errors.InvalidHeaderDefect], '') + self._test_get_x(parser.get_bare_quoted_string, + '"foo ', '"foo "', 'foo ', + [errors.InvalidHeaderDefect], '') + + def test_get_bare_quoted_string_empty_quotes(self): + self._test_get_x(parser.get_bare_quoted_string, + '""', '""', '', [], '') + + # get_comment + + def test_get_comment_only(self): + comment = self._test_get_x(parser.get_comment, + '(comment)', '(comment)', ' ', [], '', ['comment']) + self.assertEqual(comment.token_type, 'comment') + + def test_get_comment_must_start_with_paren(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_comment('foo"') + with self.assertRaises(errors.HeaderParseError): + parser.get_comment(' (foo"') + + def test_get_comment_following_wsp_preserved(self): + self._test_get_x(parser.get_comment, + '(comment) \t', '(comment)', ' ', [], ' \t', ['comment']) + + def test_get_comment_multiple_words(self): + self._test_get_x(parser.get_comment, + '(foo bar) \t', '(foo bar)', ' ', [], ' \t', ['foo bar']) + + def test_get_comment_multiple_words_wsp_preserved(self): + self._test_get_x(parser.get_comment, + '( foo bar\t ) \t', '( foo bar\t )', ' ', [], ' \t', + [' foo bar\t ']) + + def test_get_comment_end_paren_mid_word(self): + self._test_get_x(parser.get_comment, + '(foo)bar', '(foo)', ' ', [], 'bar', ['foo']) + + def test_get_comment_quoted_parens(self): + self._test_get_x(parser.get_comment, + '(foo\) \(\)bar)', '(foo\) \(\)bar)', ' ', [], '', ['foo) ()bar']) + + def test_get_comment_non_printable(self): + self._test_get_x(parser.get_comment, + '(foo\x7Fbar)', '(foo\x7Fbar)', ' ', + [errors.NonPrintableDefect], '', ['foo\x7Fbar']) + + def test_get_comment_no_end_paren(self): + self._test_get_x(parser.get_comment, + '(foo bar', '(foo bar)', ' ', + [errors.InvalidHeaderDefect], '', ['foo bar']) + self._test_get_x(parser.get_comment, + '(foo bar ', '(foo bar )', ' ', + [errors.InvalidHeaderDefect], '', ['foo bar ']) + + def test_get_comment_nested_comment(self): + comment = self._test_get_x(parser.get_comment, + '(foo(bar))', '(foo(bar))', ' ', [], '', ['foo(bar)']) + self.assertEqual(comment[1].content, 'bar') + + def test_get_comment_nested_comment_wsp(self): + comment = self._test_get_x(parser.get_comment, + '(foo ( bar ) )', '(foo ( bar ) )', ' ', [], '', ['foo ( bar ) ']) + self.assertEqual(comment[2].content, ' bar ') + + def test_get_comment_empty_comment(self): + self._test_get_x(parser.get_comment, + '()', '()', ' ', [], '', ['']) + + def test_get_comment_multiple_nesting(self): + comment = self._test_get_x(parser.get_comment, + '(((((foo)))))', '(((((foo)))))', ' ', [], '', ['((((foo))))']) + for i in range(4, 0, -1): + self.assertEqual(comment[0].content, '('*(i-1)+'foo'+')'*(i-1)) + comment = comment[0] + self.assertEqual(comment.content, 'foo') + + def test_get_comment_missing_end_of_nesting(self): + self._test_get_x(parser.get_comment, + '(((((foo)))', '(((((foo)))))', ' ', + [errors.InvalidHeaderDefect]*2, '', ['((((foo))))']) + + def test_get_comment_qs_in_nested_comment(self): + comment = self._test_get_x(parser.get_comment, + '(foo (b\)))', '(foo (b\)))', ' ', [], '', ['foo (b\))']) + self.assertEqual(comment[2].content, 'b)') + + # get_cfws + + def test_get_cfws_only_ws(self): + cfws = self._test_get_x(parser.get_cfws, + ' \t \t', ' \t \t', ' ', [], '', []) + self.assertEqual(cfws.token_type, 'cfws') + + def test_get_cfws_only_comment(self): + cfws = self._test_get_x(parser.get_cfws, + '(foo)', '(foo)', ' ', [], '', ['foo']) + self.assertEqual(cfws[0].content, 'foo') + + def test_get_cfws_only_mixed(self): + cfws = self._test_get_x(parser.get_cfws, + ' (foo ) ( bar) ', ' (foo ) ( bar) ', ' ', [], '', + ['foo ', ' bar']) + self.assertEqual(cfws[1].content, 'foo ') + self.assertEqual(cfws[3].content, ' bar') + + def test_get_cfws_ends_at_non_leader(self): + cfws = self._test_get_x(parser.get_cfws, + '(foo) bar', '(foo) ', ' ', [], 'bar', ['foo']) + self.assertEqual(cfws[0].content, 'foo') + + def test_get_cfws_ends_at_non_printable(self): + cfws = self._test_get_x(parser.get_cfws, + '(foo) \x07', '(foo) ', ' ', [], '\x07', ['foo']) + self.assertEqual(cfws[0].content, 'foo') + + def test_get_cfws_non_printable_in_comment(self): + cfws = self._test_get_x(parser.get_cfws, + '(foo \x07) "test"', '(foo \x07) ', ' ', + [errors.NonPrintableDefect], '"test"', ['foo \x07']) + self.assertEqual(cfws[0].content, 'foo \x07') + + def test_get_cfws_header_ends_in_comment(self): + cfws = self._test_get_x(parser.get_cfws, + ' (foo ', ' (foo )', ' ', + [errors.InvalidHeaderDefect], '', ['foo ']) + self.assertEqual(cfws[1].content, 'foo ') + + def test_get_cfws_multiple_nested_comments(self): + cfws = self._test_get_x(parser.get_cfws, + '(foo (bar)) ((a)(a))', '(foo (bar)) ((a)(a))', ' ', [], + '', ['foo (bar)', '(a)(a)']) + self.assertEqual(cfws[0].comments, ['foo (bar)']) + self.assertEqual(cfws[2].comments, ['(a)(a)']) + + # get_quoted_string + + def test_get_quoted_string_only(self): + qs = self._test_get_x(parser.get_quoted_string, + '"bob"', '"bob"', 'bob', [], '') + self.assertEqual(qs.token_type, 'quoted-string') + self.assertEqual(qs.quoted_value, '"bob"') + self.assertEqual(qs.content, 'bob') + + def test_get_quoted_string_with_wsp(self): + qs = self._test_get_x(parser.get_quoted_string, + '\t "bob" ', '\t "bob" ', ' bob ', [], '') + self.assertEqual(qs.quoted_value, ' "bob" ') + self.assertEqual(qs.content, 'bob') + + def test_get_quoted_string_with_comments_and_wsp(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (foo) "bob"(bar)', ' (foo) "bob"(bar)', ' bob ', [], '') + self.assertEqual(qs[0][1].content, 'foo') + self.assertEqual(qs[2][0].content, 'bar') + self.assertEqual(qs.content, 'bob') + self.assertEqual(qs.quoted_value, ' "bob" ') + + def test_get_quoted_string_with_multiple_comments(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (foo) (bar) "bob"(bird)', ' (foo) (bar) "bob"(bird)', ' bob ', + [], '') + self.assertEqual(qs[0].comments, ['foo', 'bar']) + self.assertEqual(qs[2].comments, ['bird']) + self.assertEqual(qs.content, 'bob') + self.assertEqual(qs.quoted_value, ' "bob" ') + + def test_get_quoted_string_non_printable_in_comment(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (\x0A) "bob"', ' (\x0A) "bob"', ' bob', + [errors.NonPrintableDefect], '') + self.assertEqual(qs[0].comments, ['\x0A']) + self.assertEqual(qs.content, 'bob') + self.assertEqual(qs.quoted_value, ' "bob"') + + def test_get_quoted_string_non_printable_in_qcontent(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (a) "a\x0B"', ' (a) "a\x0B"', ' a\x0B', + [errors.NonPrintableDefect], '') + self.assertEqual(qs[0].comments, ['a']) + self.assertEqual(qs.content, 'a\x0B') + self.assertEqual(qs.quoted_value, ' "a\x0B"') + + def test_get_quoted_string_internal_ws(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (a) "foo bar "', ' (a) "foo bar "', ' foo bar ', + [], '') + self.assertEqual(qs[0].comments, ['a']) + self.assertEqual(qs.content, 'foo bar ') + self.assertEqual(qs.quoted_value, ' "foo bar "') + + def test_get_quoted_string_header_ends_in_comment(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (a) "bob" (a', ' (a) "bob" (a)', ' bob ', + [errors.InvalidHeaderDefect], '') + self.assertEqual(qs[0].comments, ['a']) + self.assertEqual(qs[2].comments, ['a']) + self.assertEqual(qs.content, 'bob') + self.assertEqual(qs.quoted_value, ' "bob" ') + + def test_get_quoted_string_header_ends_in_qcontent(self): + qs = self._test_get_x(parser.get_quoted_string, + ' (a) "bob', ' (a) "bob"', ' bob', + [errors.InvalidHeaderDefect], '') + self.assertEqual(qs[0].comments, ['a']) + self.assertEqual(qs.content, 'bob') + self.assertEqual(qs.quoted_value, ' "bob"') + + def test_get_quoted_string_no_quoted_string(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_quoted_string(' (ab) xyz') + + def test_get_quoted_string_qs_ends_at_noncfws(self): + qs = self._test_get_x(parser.get_quoted_string, + '\t "bob" fee', '\t "bob" ', ' bob ', [], 'fee') + self.assertEqual(qs.content, 'bob') + self.assertEqual(qs.quoted_value, ' "bob" ') + + # get_atom + + def test_get_atom_only(self): + atom = self._test_get_x(parser.get_atom, + 'bob', 'bob', 'bob', [], '') + self.assertEqual(atom.token_type, 'atom') + + def test_get_atom_with_wsp(self): + self._test_get_x(parser.get_atom, + '\t bob ', '\t bob ', ' bob ', [], '') + + def test_get_atom_with_comments_and_wsp(self): + atom = self._test_get_x(parser.get_atom, + ' (foo) bob(bar)', ' (foo) bob(bar)', ' bob ', [], '') + self.assertEqual(atom[0][1].content, 'foo') + self.assertEqual(atom[2][0].content, 'bar') + + def test_get_atom_with_multiple_comments(self): + atom = self._test_get_x(parser.get_atom, + ' (foo) (bar) bob(bird)', ' (foo) (bar) bob(bird)', ' bob ', + [], '') + self.assertEqual(atom[0].comments, ['foo', 'bar']) + self.assertEqual(atom[2].comments, ['bird']) + + def test_get_atom_non_printable_in_comment(self): + atom = self._test_get_x(parser.get_atom, + ' (\x0A) bob', ' (\x0A) bob', ' bob', + [errors.NonPrintableDefect], '') + self.assertEqual(atom[0].comments, ['\x0A']) + + def test_get_atom_non_printable_in_atext(self): + atom = self._test_get_x(parser.get_atom, + ' (a) a\x0B', ' (a) a\x0B', ' a\x0B', + [errors.NonPrintableDefect], '') + self.assertEqual(atom[0].comments, ['a']) + + def test_get_atom_header_ends_in_comment(self): + atom = self._test_get_x(parser.get_atom, + ' (a) bob (a', ' (a) bob (a)', ' bob ', + [errors.InvalidHeaderDefect], '') + self.assertEqual(atom[0].comments, ['a']) + self.assertEqual(atom[2].comments, ['a']) + + def test_get_atom_no_atom(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_atom(' (ab) ') + + def test_get_atom_no_atom_before_special(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_atom(' (ab) @') + + def test_get_atom_atom_ends_at_special(self): + atom = self._test_get_x(parser.get_atom, + ' (foo) bob(bar) @bang', ' (foo) bob(bar) ', ' bob ', [], '@bang') + self.assertEqual(atom[0].comments, ['foo']) + self.assertEqual(atom[2].comments, ['bar']) + + def test_get_atom_atom_ends_at_noncfws(self): + atom = self._test_get_x(parser.get_atom, + 'bob fred', 'bob ', 'bob ', [], 'fred') + + # get_dot_atom_text + + def test_get_dot_atom_text(self): + dot_atom_text = self._test_get_x(parser.get_dot_atom_text, + 'foo.bar.bang', 'foo.bar.bang', 'foo.bar.bang', [], '') + self.assertEqual(dot_atom_text.token_type, 'dot-atom-text') + self.assertEqual(len(dot_atom_text), 5) + + def test_get_dot_atom_text_lone_atom_is_valid(self): + dot_atom_text = self._test_get_x(parser.get_dot_atom_text, + 'foo', 'foo', 'foo', [], '') + + def test_get_dot_atom_text_raises_on_leading_dot(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom_text('.foo.bar') + + def test_get_dot_atom_text_raises_on_trailing_dot(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom_text('foo.bar.') + + def test_get_dot_atom_text_raises_on_leading_non_atext(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom_text(' foo.bar') + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom_text('@foo.bar') + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom_text('"foo.bar"') + + def test_get_dot_atom_text_trailing_text_preserved(self): + dot_atom_text = self._test_get_x(parser.get_dot_atom_text, + 'foo@bar', 'foo', 'foo', [], '@bar') + + def test_get_dot_atom_text_trailing_ws_preserved(self): + dot_atom_text = self._test_get_x(parser.get_dot_atom_text, + 'foo .bar', 'foo', 'foo', [], ' .bar') + + # get_dot_atom + + def test_get_dot_atom_only(self): + dot_atom = self._test_get_x(parser.get_dot_atom, + 'foo.bar.bing', 'foo.bar.bing', 'foo.bar.bing', [], '') + self.assertEqual(dot_atom.token_type, 'dot-atom') + self.assertEqual(len(dot_atom), 1) + + def test_get_dot_atom_with_wsp(self): + self._test_get_x(parser.get_dot_atom, + '\t foo.bar.bing ', '\t foo.bar.bing ', ' foo.bar.bing ', [], '') + + def test_get_dot_atom_with_comments_and_wsp(self): + self._test_get_x(parser.get_dot_atom, + ' (sing) foo.bar.bing (here) ', ' (sing) foo.bar.bing (here) ', + ' foo.bar.bing ', [], '') + + def test_get_dot_atom_space_ends_dot_atom(self): + self._test_get_x(parser.get_dot_atom, + ' (sing) foo.bar .bing (here) ', ' (sing) foo.bar ', + ' foo.bar ', [], '.bing (here) ') + + def test_get_dot_atom_no_atom_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom(' (foo) ') + + def test_get_dot_atom_leading_dot_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom(' (foo) .bar') + + def test_get_dot_atom_two_dots_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom('bar..bang') + + def test_get_dot_atom_trailing_dot_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_dot_atom(' (foo) bar.bang. foo') + + # get_word (if this were black box we'd repeat all the qs/atom tests) + + def test_get_word_atom_yields_atom(self): + word = self._test_get_x(parser.get_word, + ' (foo) bar (bang) :ah', ' (foo) bar (bang) ', ' bar ', [], ':ah') + self.assertEqual(word.token_type, 'atom') + self.assertEqual(word[0].token_type, 'cfws') + + def test_get_word_qs_yields_qs(self): + word = self._test_get_x(parser.get_word, + '"bar " (bang) ah', '"bar " (bang) ', 'bar ', [], 'ah') + self.assertEqual(word.token_type, 'quoted-string') + self.assertEqual(word[0].token_type, 'bare-quoted-string') + self.assertEqual(word[0].value, 'bar ') + self.assertEqual(word.content, 'bar ') + + def test_get_word_ends_at_dot(self): + self._test_get_x(parser.get_word, + 'foo.', 'foo', 'foo', [], '.') + + # get_phrase + + def test_get_phrase_simple(self): + phrase = self._test_get_x(parser.get_phrase, + '"Fred A. Johnson" is his name, oh.', + '"Fred A. Johnson" is his name', + 'Fred A. Johnson is his name', + [], + ', oh.') + self.assertEqual(phrase.token_type, 'phrase') + + def test_get_phrase_complex(self): + phrase = self._test_get_x(parser.get_phrase, + ' (A) bird (in (my|your)) "hand " is messy\t<>\t', + ' (A) bird (in (my|your)) "hand " is messy\t', + ' bird hand is messy ', + [], + '<>\t') + self.assertEqual(phrase[0][0].comments, ['A']) + self.assertEqual(phrase[0][2].comments, ['in (my|your)']) + + def test_get_phrase_obsolete(self): + phrase = self._test_get_x(parser.get_phrase, + 'Fred A.(weird).O Johnson', + 'Fred A.(weird).O Johnson', + 'Fred A. .O Johnson', + [errors.ObsoleteHeaderDefect]*3, + '') + self.assertEqual(len(phrase), 7) + self.assertEqual(phrase[3].comments, ['weird']) + + def test_get_phrase_pharse_must_start_with_word(self): + phrase = self._test_get_x(parser.get_phrase, + '(even weirder).name', + '(even weirder).name', + ' .name', + [errors.InvalidHeaderDefect] + [errors.ObsoleteHeaderDefect]*2, + '') + self.assertEqual(len(phrase), 3) + self.assertEqual(phrase[0].comments, ['even weirder']) + + def test_get_phrase_ending_with_obsolete(self): + phrase = self._test_get_x(parser.get_phrase, + 'simple phrase.(with trailing comment):boo', + 'simple phrase.(with trailing comment)', + 'simple phrase. ', + [errors.ObsoleteHeaderDefect]*2, + ':boo') + self.assertEqual(len(phrase), 4) + self.assertEqual(phrase[3].comments, ['with trailing comment']) + + def get_phrase_cfws_only_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_phrase(' (foo) ') + + # get_local_part + + def test_get_local_part_simple(self): + local_part = self._test_get_x(parser.get_local_part, + 'dinsdale@python.org', 'dinsdale', 'dinsdale', [], '@python.org') + self.assertEqual(local_part.token_type, 'local-part') + self.assertEqual(local_part.local_part, 'dinsdale') + + def test_get_local_part_with_dot(self): + local_part = self._test_get_x(parser.get_local_part, + 'Fred.A.Johnson@python.org', + 'Fred.A.Johnson', + 'Fred.A.Johnson', + [], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson') + + def test_get_local_part_with_whitespace(self): + local_part = self._test_get_x(parser.get_local_part, + ' Fred.A.Johnson @python.org', + ' Fred.A.Johnson ', + ' Fred.A.Johnson ', + [], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson') + + def test_get_local_part_with_cfws(self): + local_part = self._test_get_x(parser.get_local_part, + ' (foo) Fred.A.Johnson (bar (bird)) @python.org', + ' (foo) Fred.A.Johnson (bar (bird)) ', + ' Fred.A.Johnson ', + [], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson') + self.assertEqual(local_part[0][0].comments, ['foo']) + self.assertEqual(local_part[0][2].comments, ['bar (bird)']) + + def test_get_local_part_simple_quoted(self): + local_part = self._test_get_x(parser.get_local_part, + '"dinsdale"@python.org', '"dinsdale"', '"dinsdale"', [], '@python.org') + self.assertEqual(local_part.token_type, 'local-part') + self.assertEqual(local_part.local_part, 'dinsdale') + + def test_get_local_part_with_quoted_dot(self): + local_part = self._test_get_x(parser.get_local_part, + '"Fred.A.Johnson"@python.org', + '"Fred.A.Johnson"', + '"Fred.A.Johnson"', + [], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson') + + def test_get_local_part_quoted_with_whitespace(self): + local_part = self._test_get_x(parser.get_local_part, + ' "Fred A. Johnson" @python.org', + ' "Fred A. Johnson" ', + ' "Fred A. Johnson" ', + [], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred A. Johnson') + + def test_get_local_part_quoted_with_cfws(self): + local_part = self._test_get_x(parser.get_local_part, + ' (foo) " Fred A. Johnson " (bar (bird)) @python.org', + ' (foo) " Fred A. Johnson " (bar (bird)) ', + ' " Fred A. Johnson " ', + [], + '@python.org') + self.assertEqual(local_part.local_part, ' Fred A. Johnson ') + self.assertEqual(local_part[0][0].comments, ['foo']) + self.assertEqual(local_part[0][2].comments, ['bar (bird)']) + + + def test_get_local_part_simple_obsolete(self): + local_part = self._test_get_x(parser.get_local_part, + 'Fred. A.Johnson@python.org', + 'Fred. A.Johnson', + 'Fred. A.Johnson', + [errors.ObsoleteHeaderDefect], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson') + + def test_get_local_part_complex_obsolete_1(self): + local_part = self._test_get_x(parser.get_local_part, + ' (foo )Fred (bar).(bird) A.(sheep)Johnson."and dogs "@python.org', + ' (foo )Fred (bar).(bird) A.(sheep)Johnson."and dogs "', + ' Fred . A. Johnson.and dogs ', + [errors.ObsoleteHeaderDefect], + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson.and dogs ') + + def test_get_local_part_complex_obsolete_invalid(self): + local_part = self._test_get_x(parser.get_local_part, + ' (foo )Fred (bar).(bird) A.(sheep)Johnson "and dogs"@python.org', + ' (foo )Fred (bar).(bird) A.(sheep)Johnson "and dogs"', + ' Fred . A. Johnson and dogs', + [errors.InvalidHeaderDefect]*2, + '@python.org') + self.assertEqual(local_part.local_part, 'Fred.A.Johnson and dogs') + + def test_get_local_part_no_part_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_local_part(' (foo) ') + + def test_get_local_part_special_instead_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_local_part(' (foo) @python.org') + + def test_get_local_part_trailing_dot(self): + local_part = self._test_get_x(parser.get_local_part, + ' borris.@python.org', + ' borris.', + ' borris.', + [errors.InvalidHeaderDefect]*2, + '@python.org') + self.assertEqual(local_part.local_part, 'borris.') + + def test_get_local_part_trailing_dot_with_ws(self): + local_part = self._test_get_x(parser.get_local_part, + ' borris. @python.org', + ' borris. ', + ' borris. ', + [errors.InvalidHeaderDefect]*2, + '@python.org') + self.assertEqual(local_part.local_part, 'borris.') + + def test_get_local_part_leading_dot(self): + local_part = self._test_get_x(parser.get_local_part, + '.borris@python.org', + '.borris', + '.borris', + [errors.InvalidHeaderDefect]*2, + '@python.org') + self.assertEqual(local_part.local_part, '.borris') + + def test_get_local_part_leading_dot_after_ws(self): + local_part = self._test_get_x(parser.get_local_part, + ' .borris@python.org', + ' .borris', + ' .borris', + [errors.InvalidHeaderDefect]*2, + '@python.org') + self.assertEqual(local_part.local_part, '.borris') + + def test_get_local_part_double_dot_raises(self): + local_part = self._test_get_x(parser.get_local_part, + ' borris.(foo).natasha@python.org', + ' borris.(foo).natasha', + ' borris. .natasha', + [errors.InvalidHeaderDefect]*2, + '@python.org') + self.assertEqual(local_part.local_part, 'borris..natasha') + + def test_get_local_part_quoted_strings_in_atom_list(self): + local_part = self._test_get_x(parser.get_local_part, + '""example" example"@example.com', + '""example" example"', + 'example example', + [errors.InvalidHeaderDefect]*3, + '@example.com') + self.assertEqual(local_part.local_part, 'example example') + + def test_get_local_part_valid_and_invalid_qp_in_atom_list(self): + local_part = self._test_get_x(parser.get_local_part, + r'"\\"example\\" example"@example.com', + r'"\\"example\\" example"', + r'\example\\ example', + [errors.InvalidHeaderDefect]*5, + '@example.com') + self.assertEqual(local_part.local_part, r'\example\\ example') + + def test_get_local_part_unicode_defect(self): + # Currently this only happens when parsing unicode, not when parsing + # stuff that was originally binary. + local_part = self._test_get_x(parser.get_local_part, + 'exámple@example.com', + 'exámple', + 'exámple', + [errors.NonASCIILocalPartDefect], + '@example.com') + self.assertEqual(local_part.local_part, 'exámple') + + # get_dtext + + def test_get_dtext_only(self): + dtext = self._test_get_x(parser.get_dtext, + 'foobar', 'foobar', 'foobar', [], '') + self.assertEqual(dtext.token_type, 'ptext') + + def test_get_dtext_all_dtext(self): + dtext = self._test_get_x(parser.get_dtext, self.rfc_dtext_chars, + self.rfc_dtext_chars, + self.rfc_dtext_chars, [], '') + + def test_get_dtext_two_words_gets_first(self): + self._test_get_x(parser.get_dtext, + 'foo bar', 'foo', 'foo', [], ' bar') + + def test_get_dtext_following_wsp_preserved(self): + self._test_get_x(parser.get_dtext, + 'foo \t\tbar', 'foo', 'foo', [], ' \t\tbar') + + def test_get_dtext_non_printables(self): + dtext = self._test_get_x(parser.get_dtext, + 'foo\x00bar]', 'foo\x00bar', 'foo\x00bar', + [errors.NonPrintableDefect], ']') + self.assertEqual(dtext.defects[0].non_printables[0], '\x00') + + def test_get_dtext_with_qp(self): + ptext = self._test_get_x(parser.get_dtext, + r'foo\]\[\\bar\b\e\l\l', + r'foo][\barbell', + r'foo][\barbell', + [errors.ObsoleteHeaderDefect], + '') + + def test_get_dtext_up_to_close_bracket_only(self): + self._test_get_x(parser.get_dtext, + 'foo]', 'foo', 'foo', [], ']') + + def test_get_dtext_wsp_before_close_bracket_preserved(self): + self._test_get_x(parser.get_dtext, + 'foo ]', 'foo', 'foo', [], ' ]') + + def test_get_dtext_close_bracket_mid_word(self): + self._test_get_x(parser.get_dtext, + 'foo]bar', 'foo', 'foo', [], ']bar') + + def test_get_dtext_up_to_open_bracket_only(self): + self._test_get_x(parser.get_dtext, + 'foo[', 'foo', 'foo', [], '[') + + def test_get_dtext_wsp_before_open_bracket_preserved(self): + self._test_get_x(parser.get_dtext, + 'foo [', 'foo', 'foo', [], ' [') + + def test_get_dtext_open_bracket_mid_word(self): + self._test_get_x(parser.get_dtext, + 'foo[bar', 'foo', 'foo', [], '[bar') + + # get_domain_literal + + def test_get_domain_literal_only(self): + domain_literal = domain_literal = self._test_get_x(parser.get_domain_literal, + '[127.0.0.1]', + '[127.0.0.1]', + '[127.0.0.1]', + [], + '') + self.assertEqual(domain_literal.token_type, 'domain-literal') + self.assertEqual(domain_literal.domain, '[127.0.0.1]') + self.assertEqual(domain_literal.ip, '127.0.0.1') + + def test_get_domain_literal_with_internal_ws(self): + domain_literal = self._test_get_x(parser.get_domain_literal, + '[ 127.0.0.1\t ]', + '[ 127.0.0.1\t ]', + '[ 127.0.0.1 ]', + [], + '') + self.assertEqual(domain_literal.domain, '[127.0.0.1]') + self.assertEqual(domain_literal.ip, '127.0.0.1') + + def test_get_domain_literal_with_surrounding_cfws(self): + domain_literal = self._test_get_x(parser.get_domain_literal, + '(foo)[ 127.0.0.1] (bar)', + '(foo)[ 127.0.0.1] (bar)', + ' [ 127.0.0.1] ', + [], + '') + self.assertEqual(domain_literal.domain, '[127.0.0.1]') + self.assertEqual(domain_literal.ip, '127.0.0.1') + + def test_get_domain_literal_no_start_char_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_domain_literal('(foo) ') + + def test_get_domain_literal_no_start_char_before_special_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_domain_literal('(foo) @') + + def test_get_domain_literal_bad_dtext_char_before_special_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_domain_literal('(foo) [abc[@') + + # get_domain + + def test_get_domain_regular_domain_only(self): + domain = self._test_get_x(parser.get_domain, + 'example.com', + 'example.com', + 'example.com', + [], + '') + self.assertEqual(domain.token_type, 'domain') + self.assertEqual(domain.domain, 'example.com') + + def test_get_domain_domain_literal_only(self): + domain = self._test_get_x(parser.get_domain, + '[127.0.0.1]', + '[127.0.0.1]', + '[127.0.0.1]', + [], + '') + self.assertEqual(domain.token_type, 'domain') + self.assertEqual(domain.domain, '[127.0.0.1]') + + def test_get_domain_with_cfws(self): + domain = self._test_get_x(parser.get_domain, + '(foo) example.com(bar)\t', + '(foo) example.com(bar)\t', + ' example.com ', + [], + '') + self.assertEqual(domain.domain, 'example.com') + + def test_get_domain_domain_literal_with_cfws(self): + domain = self._test_get_x(parser.get_domain, + '(foo)[127.0.0.1]\t(bar)', + '(foo)[127.0.0.1]\t(bar)', + ' [127.0.0.1] ', + [], + '') + self.assertEqual(domain.domain, '[127.0.0.1]') + + def test_get_domain_domain_with_cfws_ends_at_special(self): + domain = self._test_get_x(parser.get_domain, + '(foo)example.com\t(bar), next', + '(foo)example.com\t(bar)', + ' example.com ', + [], + ', next') + self.assertEqual(domain.domain, 'example.com') + + def test_get_domain_domain_literal_with_cfws_ends_at_special(self): + domain = self._test_get_x(parser.get_domain, + '(foo)[127.0.0.1]\t(bar), next', + '(foo)[127.0.0.1]\t(bar)', + ' [127.0.0.1] ', + [], + ', next') + self.assertEqual(domain.domain, '[127.0.0.1]') + + def test_get_domain_obsolete(self): + domain = self._test_get_x(parser.get_domain, + '(foo) example . (bird)com(bar)\t', + '(foo) example . (bird)com(bar)\t', + ' example . com ', + [errors.ObsoleteHeaderDefect], + '') + self.assertEqual(domain.domain, 'example.com') + + def test_get_domain_no_non_cfws_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_domain(" (foo)\t") + + def test_get_domain_no_atom_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_domain(" (foo)\t, broken") + + + # get_addr_spec + + def test_get_addr_spec_normal(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + 'dinsdale@example.com', + 'dinsdale@example.com', + 'dinsdale@example.com', + [], + '') + self.assertEqual(addr_spec.token_type, 'addr-spec') + self.assertEqual(addr_spec.local_part, 'dinsdale') + self.assertEqual(addr_spec.domain, 'example.com') + self.assertEqual(addr_spec.addr_spec, 'dinsdale@example.com') + + def test_get_addr_spec_with_doamin_literal(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + 'dinsdale@[127.0.0.1]', + 'dinsdale@[127.0.0.1]', + 'dinsdale@[127.0.0.1]', + [], + '') + self.assertEqual(addr_spec.local_part, 'dinsdale') + self.assertEqual(addr_spec.domain, '[127.0.0.1]') + self.assertEqual(addr_spec.addr_spec, 'dinsdale@[127.0.0.1]') + + def test_get_addr_spec_with_cfws(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + '(foo) dinsdale(bar)@ (bird) example.com (bog)', + '(foo) dinsdale(bar)@ (bird) example.com (bog)', + ' dinsdale@example.com ', + [], + '') + self.assertEqual(addr_spec.local_part, 'dinsdale') + self.assertEqual(addr_spec.domain, 'example.com') + self.assertEqual(addr_spec.addr_spec, 'dinsdale@example.com') + + def test_get_addr_spec_with_qouoted_string_and_cfws(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + '(foo) "roy a bug"(bar)@ (bird) example.com (bog)', + '(foo) "roy a bug"(bar)@ (bird) example.com (bog)', + ' "roy a bug"@example.com ', + [], + '') + self.assertEqual(addr_spec.local_part, 'roy a bug') + self.assertEqual(addr_spec.domain, 'example.com') + self.assertEqual(addr_spec.addr_spec, '"roy a bug"@example.com') + + def test_get_addr_spec_ends_at_special(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + '(foo) "roy a bug"(bar)@ (bird) example.com (bog) , next', + '(foo) "roy a bug"(bar)@ (bird) example.com (bog) ', + ' "roy a bug"@example.com ', + [], + ', next') + self.assertEqual(addr_spec.local_part, 'roy a bug') + self.assertEqual(addr_spec.domain, 'example.com') + self.assertEqual(addr_spec.addr_spec, '"roy a bug"@example.com') + + def test_get_addr_spec_quoted_strings_in_atom_list(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + '""example" example"@example.com', + '""example" example"@example.com', + 'example example@example.com', + [errors.InvalidHeaderDefect]*3, + '') + self.assertEqual(addr_spec.local_part, 'example example') + self.assertEqual(addr_spec.domain, 'example.com') + self.assertEqual(addr_spec.addr_spec, '"example example"@example.com') + + def test_get_addr_spec_dot_atom(self): + addr_spec = self._test_get_x(parser.get_addr_spec, + 'star.a.star@example.com', + 'star.a.star@example.com', + 'star.a.star@example.com', + [], + '') + self.assertEqual(addr_spec.local_part, 'star.a.star') + self.assertEqual(addr_spec.domain, 'example.com') + self.assertEqual(addr_spec.addr_spec, 'star.a.star@example.com') + + # get_obs_route + + def test_get_obs_route_simple(self): + obs_route = self._test_get_x(parser.get_obs_route, + '@example.com, @two.example.com:', + '@example.com, @two.example.com:', + '@example.com, @two.example.com:', + [], + '') + self.assertEqual(obs_route.token_type, 'obs-route') + self.assertEqual(obs_route.domains, ['example.com', 'two.example.com']) + + def test_get_obs_route_complex(self): + obs_route = self._test_get_x(parser.get_obs_route, + '(foo),, (blue)@example.com (bar),@two.(foo) example.com (bird):', + '(foo),, (blue)@example.com (bar),@two.(foo) example.com (bird):', + ' ,, @example.com ,@two. example.com :', + [errors.ObsoleteHeaderDefect], # This is the obs-domain + '') + self.assertEqual(obs_route.token_type, 'obs-route') + self.assertEqual(obs_route.domains, ['example.com', 'two.example.com']) + + def test_get_obs_route_no_route_before_end_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_obs_route('(foo) @example.com,') + + def test_get_obs_route_no_route_before_special_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_obs_route('(foo) [abc],') + + def test_get_obs_route_no_route_before_special_raises2(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_obs_route('(foo) @example.com [abc],') + + # get_angle_addr + + def test_get_angle_addr_simple(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '<dinsdale@example.com>', + '<dinsdale@example.com>', + '<dinsdale@example.com>', + [], + '') + self.assertEqual(angle_addr.token_type, 'angle-addr') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_with_cfws(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + ' (foo) <dinsdale@example.com>(bar)', + ' (foo) <dinsdale@example.com>(bar)', + ' <dinsdale@example.com> ', + [], + '') + self.assertEqual(angle_addr.token_type, 'angle-addr') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_qs_and_domain_literal(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '<"Fred Perfect"@[127.0.0.1]>', + '<"Fred Perfect"@[127.0.0.1]>', + '<"Fred Perfect"@[127.0.0.1]>', + [], + '') + self.assertEqual(angle_addr.local_part, 'Fred Perfect') + self.assertEqual(angle_addr.domain, '[127.0.0.1]') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, '"Fred Perfect"@[127.0.0.1]') + + def test_get_angle_addr_internal_cfws(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '<(foo) dinsdale@example.com(bar)>', + '<(foo) dinsdale@example.com(bar)>', + '< dinsdale@example.com >', + [], + '') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_obs_route(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '(foo)<@example.com, (bird) @two.example.com: dinsdale@example.com> (bar) ', + '(foo)<@example.com, (bird) @two.example.com: dinsdale@example.com> (bar) ', + ' <@example.com, @two.example.com: dinsdale@example.com> ', + [errors.ObsoleteHeaderDefect], + '') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertEqual(angle_addr.route, ['example.com', 'two.example.com']) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_missing_closing_angle(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '<dinsdale@example.com', + '<dinsdale@example.com>', + '<dinsdale@example.com>', + [errors.InvalidHeaderDefect], + '') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_missing_closing_angle_with_cfws(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '<dinsdale@example.com (foo)', + '<dinsdale@example.com (foo)>', + '<dinsdale@example.com >', + [errors.InvalidHeaderDefect], + '') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_ends_at_special(self): + angle_addr = self._test_get_x(parser.get_angle_addr, + '<dinsdale@example.com> (foo), next', + '<dinsdale@example.com> (foo)', + '<dinsdale@example.com> ', + [], + ', next') + self.assertEqual(angle_addr.local_part, 'dinsdale') + self.assertEqual(angle_addr.domain, 'example.com') + self.assertIsNone(angle_addr.route) + self.assertEqual(angle_addr.addr_spec, 'dinsdale@example.com') + + def test_get_angle_addr_no_angle_raise(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_angle_addr('(foo) ') + + def test_get_angle_addr_no_angle_before_special_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_angle_addr('(foo) , next') + + def test_get_angle_addr_no_angle_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_angle_addr('bar') + + def test_get_angle_addr_special_after_angle_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_angle_addr('(foo) <, bar') + + # get_display_name This is phrase but with a different value. + + def test_get_display_name_simple(self): + display_name = self._test_get_x(parser.get_display_name, + 'Fred A Johnson', + 'Fred A Johnson', + 'Fred A Johnson', + [], + '') + self.assertEqual(display_name.token_type, 'display-name') + self.assertEqual(display_name.display_name, 'Fred A Johnson') + + def test_get_display_name_complex1(self): + display_name = self._test_get_x(parser.get_display_name, + '"Fred A. Johnson" is his name, oh.', + '"Fred A. Johnson" is his name', + '"Fred A. Johnson is his name"', + [], + ', oh.') + self.assertEqual(display_name.token_type, 'display-name') + self.assertEqual(display_name.display_name, 'Fred A. Johnson is his name') + + def test_get_display_name_complex2(self): + display_name = self._test_get_x(parser.get_display_name, + ' (A) bird (in (my|your)) "hand " is messy\t<>\t', + ' (A) bird (in (my|your)) "hand " is messy\t', + ' "bird hand is messy" ', + [], + '<>\t') + self.assertEqual(display_name[0][0].comments, ['A']) + self.assertEqual(display_name[0][2].comments, ['in (my|your)']) + self.assertEqual(display_name.display_name, 'bird hand is messy') + + def test_get_display_name_obsolete(self): + display_name = self._test_get_x(parser.get_display_name, + 'Fred A.(weird).O Johnson', + 'Fred A.(weird).O Johnson', + '"Fred A. .O Johnson"', + [errors.ObsoleteHeaderDefect]*3, + '') + self.assertEqual(len(display_name), 7) + self.assertEqual(display_name[3].comments, ['weird']) + self.assertEqual(display_name.display_name, 'Fred A. .O Johnson') + + def test_get_display_name_pharse_must_start_with_word(self): + display_name = self._test_get_x(parser.get_display_name, + '(even weirder).name', + '(even weirder).name', + ' ".name"', + [errors.InvalidHeaderDefect] + [errors.ObsoleteHeaderDefect]*2, + '') + self.assertEqual(len(display_name), 3) + self.assertEqual(display_name[0].comments, ['even weirder']) + self.assertEqual(display_name.display_name, '.name') + + def test_get_display_name_ending_with_obsolete(self): + display_name = self._test_get_x(parser.get_display_name, + 'simple phrase.(with trailing comment):boo', + 'simple phrase.(with trailing comment)', + '"simple phrase." ', + [errors.ObsoleteHeaderDefect]*2, + ':boo') + self.assertEqual(len(display_name), 4) + self.assertEqual(display_name[3].comments, ['with trailing comment']) + self.assertEqual(display_name.display_name, 'simple phrase.') + + # get_name_addr + + def test_get_name_addr_angle_addr_only(self): + name_addr = self._test_get_x(parser.get_name_addr, + '<dinsdale@example.com>', + '<dinsdale@example.com>', + '<dinsdale@example.com>', + [], + '') + self.assertEqual(name_addr.token_type, 'name-addr') + self.assertIsNone(name_addr.display_name) + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertIsNone(name_addr.route) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_atom_name(self): + name_addr = self._test_get_x(parser.get_name_addr, + 'Dinsdale <dinsdale@example.com>', + 'Dinsdale <dinsdale@example.com>', + 'Dinsdale <dinsdale@example.com>', + [], + '') + self.assertEqual(name_addr.token_type, 'name-addr') + self.assertEqual(name_addr.display_name, 'Dinsdale') + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertIsNone(name_addr.route) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_atom_name_with_cfws(self): + name_addr = self._test_get_x(parser.get_name_addr, + '(foo) Dinsdale (bar) <dinsdale@example.com> (bird)', + '(foo) Dinsdale (bar) <dinsdale@example.com> (bird)', + ' Dinsdale <dinsdale@example.com> ', + [], + '') + self.assertEqual(name_addr.display_name, 'Dinsdale') + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertIsNone(name_addr.route) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_name_with_cfws_and_dots(self): + name_addr = self._test_get_x(parser.get_name_addr, + '(foo) Roy.A.Bear (bar) <dinsdale@example.com> (bird)', + '(foo) Roy.A.Bear (bar) <dinsdale@example.com> (bird)', + ' "Roy.A.Bear" <dinsdale@example.com> ', + [errors.ObsoleteHeaderDefect]*2, + '') + self.assertEqual(name_addr.display_name, 'Roy.A.Bear') + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertIsNone(name_addr.route) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_qs_name(self): + name_addr = self._test_get_x(parser.get_name_addr, + '"Roy.A.Bear" <dinsdale@example.com>', + '"Roy.A.Bear" <dinsdale@example.com>', + '"Roy.A.Bear" <dinsdale@example.com>', + [], + '') + self.assertEqual(name_addr.display_name, 'Roy.A.Bear') + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertIsNone(name_addr.route) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_with_route(self): + name_addr = self._test_get_x(parser.get_name_addr, + '"Roy.A.Bear" <@two.example.com: dinsdale@example.com>', + '"Roy.A.Bear" <@two.example.com: dinsdale@example.com>', + '"Roy.A.Bear" <@two.example.com: dinsdale@example.com>', + [errors.ObsoleteHeaderDefect], + '') + self.assertEqual(name_addr.display_name, 'Roy.A.Bear') + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertEqual(name_addr.route, ['two.example.com']) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_ends_at_special(self): + name_addr = self._test_get_x(parser.get_name_addr, + '"Roy.A.Bear" <dinsdale@example.com>, next', + '"Roy.A.Bear" <dinsdale@example.com>', + '"Roy.A.Bear" <dinsdale@example.com>', + [], + ', next') + self.assertEqual(name_addr.display_name, 'Roy.A.Bear') + self.assertEqual(name_addr.local_part, 'dinsdale') + self.assertEqual(name_addr.domain, 'example.com') + self.assertIsNone(name_addr.route) + self.assertEqual(name_addr.addr_spec, 'dinsdale@example.com') + + def test_get_name_addr_no_content_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_name_addr(' (foo) ') + + def test_get_name_addr_no_content_before_special_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_name_addr(' (foo) ,') + + def test_get_name_addr_no_angle_after_display_name_raises(self): + with self.assertRaises(errors.HeaderParseError): + parser.get_name_addr('foo bar') + + # get_mailbox + + def test_get_mailbox_addr_spec_only(self): + mailbox = self._test_get_x(parser.get_mailbox, + 'dinsdale@example.com', + 'dinsdale@example.com', + 'dinsdale@example.com', + [], + '') + self.assertEqual(mailbox.token_type, 'mailbox') + self.assertIsNone(mailbox.display_name) + self.assertEqual(mailbox.local_part, 'dinsdale') + self.assertEqual(mailbox.domain, 'example.com') + self.assertIsNone(mailbox.route) + self.assertEqual(mailbox.addr_spec, 'dinsdale@example.com') + + def test_get_mailbox_angle_addr_only(self): + mailbox = self._test_get_x(parser.get_mailbox, + '<dinsdale@example.com>', + '<dinsdale@example.com>', + '<dinsdale@example.com>', + [], + '') + self.assertEqual(mailbox.token_type, 'mailbox') + self.assertIsNone(mailbox.display_name) + self.assertEqual(mailbox.local_part, 'dinsdale') + self.assertEqual(mailbox.domain, 'example.com') + self.assertIsNone(mailbox.route) + self.assertEqual(mailbox.addr_spec, 'dinsdale@example.com') + + def test_get_mailbox_name_addr(self): + mailbox = self._test_get_x(parser.get_mailbox, + '"Roy A. Bear" <dinsdale@example.com>', + '"Roy A. Bear" <dinsdale@example.com>', + '"Roy A. Bear" <dinsdale@example.com>', + [], + '') + self.assertEqual(mailbox.token_type, 'mailbox') + self.assertEqual(mailbox.display_name, 'Roy A. Bear') + self.assertEqual(mailbox.local_part, 'dinsdale') + self.assertEqual(mailbox.domain, 'example.com') + self.assertIsNone(mailbox.route) + self.assertEqual(mailbox.addr_spec, 'dinsdale@example.com') + + def test_get_mailbox_ends_at_special(self): + mailbox = self._test_get_x(parser.get_mailbox, + '"Roy A. Bear" <dinsdale@example.com>, rest', + '"Roy A. Bear" <dinsdale@example.com>', + '"Roy A. Bear" <dinsdale@example.com>', + [], + ', rest') + self.assertEqual(mailbox.token_type, 'mailbox') + self.assertEqual(mailbox.display_name, 'Roy A. Bear') + self.assertEqual(mailbox.local_part, 'dinsdale') + self.assertEqual(mailbox.domain, 'example.com') + self.assertIsNone(mailbox.route) + self.assertEqual(mailbox.addr_spec, 'dinsdale@example.com') + + def test_get_mailbox_quoted_strings_in_atom_list(self): + mailbox = self._test_get_x(parser.get_mailbox, + '""example" example"@example.com', + '""example" example"@example.com', + 'example example@example.com', + [errors.InvalidHeaderDefect]*3, + '') + self.assertEqual(mailbox.local_part, 'example example') + self.assertEqual(mailbox.domain, 'example.com') + self.assertEqual(mailbox.addr_spec, '"example example"@example.com') + + # get_mailbox_list + + def test_get_mailbox_list_single_addr(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + 'dinsdale@example.com', + 'dinsdale@example.com', + 'dinsdale@example.com', + [], + '') + self.assertEqual(mailbox_list.token_type, 'mailbox-list') + self.assertEqual(len(mailbox_list.mailboxes), 1) + mailbox = mailbox_list.mailboxes[0] + self.assertIsNone(mailbox.display_name) + self.assertEqual(mailbox.local_part, 'dinsdale') + self.assertEqual(mailbox.domain, 'example.com') + self.assertIsNone(mailbox.route) + self.assertEqual(mailbox.addr_spec, 'dinsdale@example.com') + self.assertEqual(mailbox_list.mailboxes, + mailbox_list.all_mailboxes) + + def test_get_mailbox_list_two_simple_addr(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + 'dinsdale@example.com, dinsdale@test.example.com', + 'dinsdale@example.com, dinsdale@test.example.com', + 'dinsdale@example.com, dinsdale@test.example.com', + [], + '') + self.assertEqual(mailbox_list.token_type, 'mailbox-list') + self.assertEqual(len(mailbox_list.mailboxes), 2) + self.assertEqual(mailbox_list.mailboxes[0].addr_spec, + 'dinsdale@example.com') + self.assertEqual(mailbox_list.mailboxes[1].addr_spec, + 'dinsdale@test.example.com') + self.assertEqual(mailbox_list.mailboxes, + mailbox_list.all_mailboxes) + + def test_get_mailbox_list_two_name_addr(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + ('"Roy A. Bear" <dinsdale@example.com>,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + ('"Roy A. Bear" <dinsdale@example.com>,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + ('"Roy A. Bear" <dinsdale@example.com>,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + [], + '') + self.assertEqual(len(mailbox_list.mailboxes), 2) + self.assertEqual(mailbox_list.mailboxes[0].addr_spec, + 'dinsdale@example.com') + self.assertEqual(mailbox_list.mailboxes[0].display_name, + 'Roy A. Bear') + self.assertEqual(mailbox_list.mailboxes[1].addr_spec, + 'dinsdale@test.example.com') + self.assertEqual(mailbox_list.mailboxes[1].display_name, + 'Fred Flintstone') + self.assertEqual(mailbox_list.mailboxes, + mailbox_list.all_mailboxes) + + def test_get_mailbox_list_two_complex(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + ('(foo) "Roy A. Bear" <dinsdale@example.com>(bar),' + ' "Fred Flintstone" <dinsdale@test.(bird)example.com>'), + ('(foo) "Roy A. Bear" <dinsdale@example.com>(bar),' + ' "Fred Flintstone" <dinsdale@test.(bird)example.com>'), + (' "Roy A. Bear" <dinsdale@example.com> ,' + ' "Fred Flintstone" <dinsdale@test. example.com>'), + [errors.ObsoleteHeaderDefect], + '') + self.assertEqual(len(mailbox_list.mailboxes), 2) + self.assertEqual(mailbox_list.mailboxes[0].addr_spec, + 'dinsdale@example.com') + self.assertEqual(mailbox_list.mailboxes[0].display_name, + 'Roy A. Bear') + self.assertEqual(mailbox_list.mailboxes[1].addr_spec, + 'dinsdale@test.example.com') + self.assertEqual(mailbox_list.mailboxes[1].display_name, + 'Fred Flintstone') + self.assertEqual(mailbox_list.mailboxes, + mailbox_list.all_mailboxes) + + def test_get_mailbox_list_unparseable_mailbox_null(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + ('"Roy A. Bear"[] dinsdale@example.com,' + ' "Fred Flintstone" <dinsdale@test.(bird)example.com>'), + ('"Roy A. Bear"[] dinsdale@example.com,' + ' "Fred Flintstone" <dinsdale@test.(bird)example.com>'), + ('"Roy A. Bear"[] dinsdale@example.com,' + ' "Fred Flintstone" <dinsdale@test. example.com>'), + [errors.InvalidHeaderDefect, # the 'extra' text after the local part + errors.InvalidHeaderDefect, # the local part with no angle-addr + errors.ObsoleteHeaderDefect, # period in extra text (example.com) + errors.ObsoleteHeaderDefect], # (bird) in valid address. + '') + self.assertEqual(len(mailbox_list.mailboxes), 1) + self.assertEqual(len(mailbox_list.all_mailboxes), 2) + self.assertEqual(mailbox_list.all_mailboxes[0].token_type, + 'invalid-mailbox') + self.assertIsNone(mailbox_list.all_mailboxes[0].display_name) + self.assertEqual(mailbox_list.all_mailboxes[0].local_part, + 'Roy A. Bear') + self.assertIsNone(mailbox_list.all_mailboxes[0].domain) + self.assertEqual(mailbox_list.all_mailboxes[0].addr_spec, + '"Roy A. Bear"') + self.assertIs(mailbox_list.all_mailboxes[1], + mailbox_list.mailboxes[0]) + self.assertEqual(mailbox_list.mailboxes[0].addr_spec, + 'dinsdale@test.example.com') + self.assertEqual(mailbox_list.mailboxes[0].display_name, + 'Fred Flintstone') + + def test_get_mailbox_list_junk_after_valid_address(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + ('"Roy A. Bear" <dinsdale@example.com>@@,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + ('"Roy A. Bear" <dinsdale@example.com>@@,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + ('"Roy A. Bear" <dinsdale@example.com>@@,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + [errors.InvalidHeaderDefect], + '') + self.assertEqual(len(mailbox_list.mailboxes), 1) + self.assertEqual(len(mailbox_list.all_mailboxes), 2) + self.assertEqual(mailbox_list.all_mailboxes[0].addr_spec, + 'dinsdale@example.com') + self.assertEqual(mailbox_list.all_mailboxes[0].display_name, + 'Roy A. Bear') + self.assertEqual(mailbox_list.all_mailboxes[0].token_type, + 'invalid-mailbox') + self.assertIs(mailbox_list.all_mailboxes[1], + mailbox_list.mailboxes[0]) + self.assertEqual(mailbox_list.mailboxes[0].addr_spec, + 'dinsdale@test.example.com') + self.assertEqual(mailbox_list.mailboxes[0].display_name, + 'Fred Flintstone') + + def test_get_mailbox_list_empty_list_element(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + ('"Roy A. Bear" <dinsdale@example.com>, (bird),,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + ('"Roy A. Bear" <dinsdale@example.com>, (bird),,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + ('"Roy A. Bear" <dinsdale@example.com>, ,,' + ' "Fred Flintstone" <dinsdale@test.example.com>'), + [errors.ObsoleteHeaderDefect]*2, + '') + self.assertEqual(len(mailbox_list.mailboxes), 2) + self.assertEqual(mailbox_list.all_mailboxes, + mailbox_list.mailboxes) + self.assertEqual(mailbox_list.all_mailboxes[0].addr_spec, + 'dinsdale@example.com') + self.assertEqual(mailbox_list.all_mailboxes[0].display_name, + 'Roy A. Bear') + self.assertEqual(mailbox_list.mailboxes[1].addr_spec, + 'dinsdale@test.example.com') + self.assertEqual(mailbox_list.mailboxes[1].display_name, + 'Fred Flintstone') + + def test_get_mailbox_list_only_empty_elements(self): + mailbox_list = self._test_get_x(parser.get_mailbox_list, + '(foo),, (bar)', + '(foo),, (bar)', + ' ,, ', + [errors.ObsoleteHeaderDefect]*3, + '') + self.assertEqual(len(mailbox_list.mailboxes), 0) + self.assertEqual(mailbox_list.all_mailboxes, + mailbox_list.mailboxes) + + # get_group_list + + def test_get_group_list_cfws_only(self): + group_list = self._test_get_x(parser.get_group_list, + '(hidden);', + '(hidden)', + ' ', + [], + ';') + self.assertEqual(group_list.token_type, 'group-list') + self.assertEqual(len(group_list.mailboxes), 0) + self.assertEqual(group_list.mailboxes, + group_list.all_mailboxes) + + def test_get_group_list_mailbox_list(self): + group_list = self._test_get_x(parser.get_group_list, + 'dinsdale@example.org, "Fred A. Bear" <dinsdale@example.org>', + 'dinsdale@example.org, "Fred A. Bear" <dinsdale@example.org>', + 'dinsdale@example.org, "Fred A. Bear" <dinsdale@example.org>', + [], + '') + self.assertEqual(group_list.token_type, 'group-list') + self.assertEqual(len(group_list.mailboxes), 2) + self.assertEqual(group_list.mailboxes, + group_list.all_mailboxes) + self.assertEqual(group_list.mailboxes[1].display_name, + 'Fred A. Bear') + + def test_get_group_list_obs_group_list(self): + group_list = self._test_get_x(parser.get_group_list, + ', (foo),,(bar)', + ', (foo),,(bar)', + ', ,, ', + [errors.ObsoleteHeaderDefect], + '') + self.assertEqual(group_list.token_type, 'group-list') + self.assertEqual(len(group_list.mailboxes), 0) + self.assertEqual(group_list.mailboxes, + group_list.all_mailboxes) + + def test_get_group_list_comment_only_invalid(self): + group_list = self._test_get_x(parser.get_group_list, + '(bar)', + '(bar)', + ' ', + [errors.InvalidHeaderDefect], + '') + self.assertEqual(group_list.token_type, 'group-list') + self.assertEqual(len(group_list.mailboxes), 0) + self.assertEqual(group_list.mailboxes, + group_list.all_mailboxes) + + # get_group + + def test_get_group_empty(self): + group = self._test_get_x(parser.get_group, + 'Monty Python:;', + 'Monty Python:;', + 'Monty Python:;', + [], + '') + self.assertEqual(group.token_type, 'group') + self.assertEqual(group.display_name, 'Monty Python') + self.assertEqual(len(group.mailboxes), 0) + self.assertEqual(group.mailboxes, + group.all_mailboxes) + + def test_get_troup_null_addr_spec(self): + group = self._test_get_x(parser.get_group, + 'foo: <>;', + 'foo: <>;', + 'foo: <>;', + [errors.InvalidHeaderDefect], + '') + self.assertEqual(group.display_name, 'foo') + self.assertEqual(len(group.mailboxes), 0) + self.assertEqual(len(group.all_mailboxes), 1) + self.assertEqual(group.all_mailboxes[0].value, '<>') + + def test_get_group_cfws_only(self): + group = self._test_get_x(parser.get_group, + 'Monty Python: (hidden);', + 'Monty Python: (hidden);', + 'Monty Python: ;', + [], + '') + self.assertEqual(group.token_type, 'group') + self.assertEqual(group.display_name, 'Monty Python') + self.assertEqual(len(group.mailboxes), 0) + self.assertEqual(group.mailboxes, + group.all_mailboxes) + + def test_get_group_single_mailbox(self): + group = self._test_get_x(parser.get_group, + 'Monty Python: "Fred A. Bear" <dinsdale@example.com>;', + 'Monty Python: "Fred A. Bear" <dinsdale@example.com>;', + 'Monty Python: "Fred A. Bear" <dinsdale@example.com>;', + [], + '') + self.assertEqual(group.token_type, 'group') + self.assertEqual(group.display_name, 'Monty Python') + self.assertEqual(len(group.mailboxes), 1) + self.assertEqual(group.mailboxes, + group.all_mailboxes) + self.assertEqual(group.mailboxes[0].addr_spec, + 'dinsdale@example.com') + + def test_get_group_mixed_list(self): + group = self._test_get_x(parser.get_group, + ('Monty Python: "Fred A. Bear" <dinsdale@example.com>,' + '(foo) Roger <ping@exampele.com>, x@test.example.com;'), + ('Monty Python: "Fred A. Bear" <dinsdale@example.com>,' + '(foo) Roger <ping@exampele.com>, x@test.example.com;'), + ('Monty Python: "Fred A. Bear" <dinsdale@example.com>,' + ' Roger <ping@exampele.com>, x@test.example.com;'), + [], + '') + self.assertEqual(group.token_type, 'group') + self.assertEqual(group.display_name, 'Monty Python') + self.assertEqual(len(group.mailboxes), 3) + self.assertEqual(group.mailboxes, + group.all_mailboxes) + self.assertEqual(group.mailboxes[0].display_name, + 'Fred A. Bear') + self.assertEqual(group.mailboxes[1].display_name, + 'Roger') + self.assertEqual(group.mailboxes[2].local_part, 'x') + + def test_get_group_one_invalid(self): + group = self._test_get_x(parser.get_group, + ('Monty Python: "Fred A. Bear" <dinsdale@example.com>,' + '(foo) Roger ping@exampele.com, x@test.example.com;'), + ('Monty Python: "Fred A. Bear" <dinsdale@example.com>,' + '(foo) Roger ping@exampele.com, x@test.example.com;'), + ('Monty Python: "Fred A. Bear" <dinsdale@example.com>,' + ' Roger ping@exampele.com, x@test.example.com;'), + [errors.InvalidHeaderDefect, # non-angle addr makes local part invalid + errors.InvalidHeaderDefect], # and its not obs-local either: no dots. + '') + self.assertEqual(group.token_type, 'group') + self.assertEqual(group.display_name, 'Monty Python') + self.assertEqual(len(group.mailboxes), 2) + self.assertEqual(len(group.all_mailboxes), 3) + self.assertEqual(group.mailboxes[0].display_name, + 'Fred A. Bear') + self.assertEqual(group.mailboxes[1].local_part, 'x') + self.assertIsNone(group.all_mailboxes[1].display_name) + + # get_address + + def test_get_address_simple(self): + address = self._test_get_x(parser.get_address, + 'dinsdale@example.com', + 'dinsdale@example.com', + 'dinsdale@example.com', + [], + '') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 1) + self.assertEqual(address.mailboxes, + address.all_mailboxes) + self.assertEqual(address.mailboxes[0].domain, + 'example.com') + self.assertEqual(address[0].token_type, + 'mailbox') + + def test_get_address_complex(self): + address = self._test_get_x(parser.get_address, + '(foo) "Fred A. Bear" <(bird)dinsdale@example.com>', + '(foo) "Fred A. Bear" <(bird)dinsdale@example.com>', + ' "Fred A. Bear" < dinsdale@example.com>', + [], + '') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 1) + self.assertEqual(address.mailboxes, + address.all_mailboxes) + self.assertEqual(address.mailboxes[0].display_name, + 'Fred A. Bear') + self.assertEqual(address[0].token_type, + 'mailbox') + + def test_get_address_empty_group(self): + address = self._test_get_x(parser.get_address, + 'Monty Python:;', + 'Monty Python:;', + 'Monty Python:;', + [], + '') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 0) + self.assertEqual(address.mailboxes, + address.all_mailboxes) + self.assertEqual(address[0].token_type, + 'group') + self.assertEqual(address[0].display_name, + 'Monty Python') + + def test_get_address_group(self): + address = self._test_get_x(parser.get_address, + 'Monty Python: x@example.com, y@example.com;', + 'Monty Python: x@example.com, y@example.com;', + 'Monty Python: x@example.com, y@example.com;', + [], + '') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 2) + self.assertEqual(address.mailboxes, + address.all_mailboxes) + self.assertEqual(address[0].token_type, + 'group') + self.assertEqual(address[0].display_name, + 'Monty Python') + self.assertEqual(address.mailboxes[0].local_part, 'x') + + def test_get_address_quoted_local_part(self): + address = self._test_get_x(parser.get_address, + '"foo bar"@example.com', + '"foo bar"@example.com', + '"foo bar"@example.com', + [], + '') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 1) + self.assertEqual(address.mailboxes, + address.all_mailboxes) + self.assertEqual(address.mailboxes[0].domain, + 'example.com') + self.assertEqual(address.mailboxes[0].local_part, + 'foo bar') + self.assertEqual(address[0].token_type, 'mailbox') + + def test_get_address_ends_at_special(self): + address = self._test_get_x(parser.get_address, + 'dinsdale@example.com, next', + 'dinsdale@example.com', + 'dinsdale@example.com', + [], + ', next') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 1) + self.assertEqual(address.mailboxes, + address.all_mailboxes) + self.assertEqual(address.mailboxes[0].domain, + 'example.com') + self.assertEqual(address[0].token_type, 'mailbox') + + def test_get_address_invalid_mailbox_invalid(self): + address = self._test_get_x(parser.get_address, + 'ping example.com, next', + 'ping example.com', + 'ping example.com', + [errors.InvalidHeaderDefect, # addr-spec with no domain + errors.InvalidHeaderDefect, # invalid local-part + errors.InvalidHeaderDefect, # missing .s in local-part + ], + ', next') + self.assertEqual(address.token_type, 'address') + self.assertEqual(len(address.mailboxes), 0) + self.assertEqual(len(address.all_mailboxes), 1) + self.assertIsNone(address.all_mailboxes[0].domain) + self.assertEqual(address.all_mailboxes[0].local_part, 'ping example.com') + self.assertEqual(address[0].token_type, 'invalid-mailbox') + + def test_get_address_quoted_strings_in_atom_list(self): + address = self._test_get_x(parser.get_address, + '""example" example"@example.com', + '""example" example"@example.com', + 'example example@example.com', + [errors.InvalidHeaderDefect]*3, + '') + self.assertEqual(address.all_mailboxes[0].local_part, 'example example') + self.assertEqual(address.all_mailboxes[0].domain, 'example.com') + self.assertEqual(address.all_mailboxes[0].addr_spec, '"example example"@example.com') + + + # get_address_list + + def test_get_address_list_mailboxes_simple(self): + address_list = self._test_get_x(parser.get_address_list, + 'dinsdale@example.com', + 'dinsdale@example.com', + 'dinsdale@example.com', + [], + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 1) + self.assertEqual(address_list.mailboxes, + address_list.all_mailboxes) + self.assertEqual([str(x) for x in address_list.mailboxes], + [str(x) for x in address_list.addresses]) + self.assertEqual(address_list.mailboxes[0].domain, 'example.com') + self.assertEqual(address_list[0].token_type, 'address') + self.assertIsNone(address_list[0].display_name) + + def test_get_address_list_mailboxes_two_simple(self): + address_list = self._test_get_x(parser.get_address_list, + 'foo@example.com, "Fred A. Bar" <bar@example.com>', + 'foo@example.com, "Fred A. Bar" <bar@example.com>', + 'foo@example.com, "Fred A. Bar" <bar@example.com>', + [], + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 2) + self.assertEqual(address_list.mailboxes, + address_list.all_mailboxes) + self.assertEqual([str(x) for x in address_list.mailboxes], + [str(x) for x in address_list.addresses]) + self.assertEqual(address_list.mailboxes[0].local_part, 'foo') + self.assertEqual(address_list.mailboxes[1].display_name, "Fred A. Bar") + + def test_get_address_list_mailboxes_complex(self): + address_list = self._test_get_x(parser.get_address_list, + ('"Roy A. Bear" <dinsdale@example.com>, ' + '(ping) Foo <x@example.com>,' + 'Nobody Is. Special <y@(bird)example.(bad)com>'), + ('"Roy A. Bear" <dinsdale@example.com>, ' + '(ping) Foo <x@example.com>,' + 'Nobody Is. Special <y@(bird)example.(bad)com>'), + ('"Roy A. Bear" <dinsdale@example.com>, ' + 'Foo <x@example.com>,' + '"Nobody Is. Special" <y@example. com>'), + [errors.ObsoleteHeaderDefect, # period in Is. + errors.ObsoleteHeaderDefect], # cfws in domain + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 3) + self.assertEqual(address_list.mailboxes, + address_list.all_mailboxes) + self.assertEqual([str(x) for x in address_list.mailboxes], + [str(x) for x in address_list.addresses]) + self.assertEqual(address_list.mailboxes[0].domain, 'example.com') + self.assertEqual(address_list.mailboxes[0].token_type, 'mailbox') + self.assertEqual(address_list.addresses[0].token_type, 'address') + self.assertEqual(address_list.mailboxes[1].local_part, 'x') + self.assertEqual(address_list.mailboxes[2].display_name, + 'Nobody Is. Special') + + def test_get_address_list_mailboxes_invalid_addresses(self): + address_list = self._test_get_x(parser.get_address_list, + ('"Roy A. Bear" <dinsdale@example.com>, ' + '(ping) Foo x@example.com[],' + 'Nobody Is. Special <(bird)example.(bad)com>'), + ('"Roy A. Bear" <dinsdale@example.com>, ' + '(ping) Foo x@example.com[],' + 'Nobody Is. Special <(bird)example.(bad)com>'), + ('"Roy A. Bear" <dinsdale@example.com>, ' + 'Foo x@example.com[],' + '"Nobody Is. Special" < example. com>'), + [errors.InvalidHeaderDefect, # invalid address in list + errors.InvalidHeaderDefect, # 'Foo x' local part invalid. + errors.InvalidHeaderDefect, # Missing . in 'Foo x' local part + errors.ObsoleteHeaderDefect, # period in 'Is.' disp-name phrase + errors.InvalidHeaderDefect, # no domain part in addr-spec + errors.ObsoleteHeaderDefect], # addr-spec has comment in it + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 1) + self.assertEqual(len(address_list.all_mailboxes), 3) + self.assertEqual([str(x) for x in address_list.all_mailboxes], + [str(x) for x in address_list.addresses]) + self.assertEqual(address_list.mailboxes[0].domain, 'example.com') + self.assertEqual(address_list.mailboxes[0].token_type, 'mailbox') + self.assertEqual(address_list.addresses[0].token_type, 'address') + self.assertEqual(address_list.addresses[1].token_type, 'address') + self.assertEqual(len(address_list.addresses[0].mailboxes), 1) + self.assertEqual(len(address_list.addresses[1].mailboxes), 0) + self.assertEqual(len(address_list.addresses[1].mailboxes), 0) + self.assertEqual( + address_list.addresses[1].all_mailboxes[0].local_part, 'Foo x') + self.assertEqual( + address_list.addresses[2].all_mailboxes[0].display_name, + "Nobody Is. Special") + + def test_get_address_list_group_empty(self): + address_list = self._test_get_x(parser.get_address_list, + 'Monty Python: ;', + 'Monty Python: ;', + 'Monty Python: ;', + [], + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 0) + self.assertEqual(address_list.mailboxes, + address_list.all_mailboxes) + self.assertEqual(len(address_list.addresses), 1) + self.assertEqual(address_list.addresses[0].token_type, 'address') + self.assertEqual(address_list.addresses[0].display_name, 'Monty Python') + self.assertEqual(len(address_list.addresses[0].mailboxes), 0) + + def test_get_address_list_group_simple(self): + address_list = self._test_get_x(parser.get_address_list, + 'Monty Python: dinsdale@example.com;', + 'Monty Python: dinsdale@example.com;', + 'Monty Python: dinsdale@example.com;', + [], + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 1) + self.assertEqual(address_list.mailboxes, + address_list.all_mailboxes) + self.assertEqual(address_list.mailboxes[0].domain, 'example.com') + self.assertEqual(address_list.addresses[0].display_name, + 'Monty Python') + self.assertEqual(address_list.addresses[0].mailboxes[0].domain, + 'example.com') + + def test_get_address_list_group_and_mailboxes(self): + address_list = self._test_get_x(parser.get_address_list, + ('Monty Python: dinsdale@example.com, "Fred" <flint@example.com>;, ' + 'Abe <x@example.com>, Bee <y@example.com>'), + ('Monty Python: dinsdale@example.com, "Fred" <flint@example.com>;, ' + 'Abe <x@example.com>, Bee <y@example.com>'), + ('Monty Python: dinsdale@example.com, "Fred" <flint@example.com>;, ' + 'Abe <x@example.com>, Bee <y@example.com>'), + [], + '') + self.assertEqual(address_list.token_type, 'address-list') + self.assertEqual(len(address_list.mailboxes), 4) + self.assertEqual(address_list.mailboxes, + address_list.all_mailboxes) + self.assertEqual(len(address_list.addresses), 3) + self.assertEqual(address_list.mailboxes[0].local_part, 'dinsdale') + self.assertEqual(address_list.addresses[0].display_name, + 'Monty Python') + self.assertEqual(address_list.addresses[0].mailboxes[0].domain, + 'example.com') + self.assertEqual(address_list.addresses[0].mailboxes[1].local_part, + 'flint') + self.assertEqual(address_list.addresses[1].mailboxes[0].local_part, + 'x') + self.assertEqual(address_list.addresses[2].mailboxes[0].local_part, + 'y') + self.assertEqual(str(address_list.addresses[1]), + str(address_list.mailboxes[2])) + + +class TestFolding(TestEmailBase): + + policy = policy.default + + def _test(self, tl, folded, policy=policy): + self.assertEqual(tl.fold(policy=policy), folded, tl.ppstr()) + + def test_simple_unstructured_no_folds(self): + self._test(parser.get_unstructured("This is a test"), + "This is a test\n") + + def test_simple_unstructured_folded(self): + self._test(parser.get_unstructured("This is also a test, but this " + "time there are enough words (and even some " + "symbols) to make it wrap; at least in theory."), + "This is also a test, but this time there are enough " + "words (and even some\n" + " symbols) to make it wrap; at least in theory.\n") + + def test_unstructured_with_unicode_no_folds(self): + self._test(parser.get_unstructured("hübsch kleiner beißt"), + "=?utf-8?q?h=C3=BCbsch_kleiner_bei=C3=9Ft?=\n") + + def test_one_ew_on_each_of_two_wrapped_lines(self): + self._test(parser.get_unstructured("Mein kleiner Kaktus ist sehr " + "hübsch. Es hat viele Stacheln " + "und oft beißt mich."), + "Mein kleiner Kaktus ist sehr =?utf-8?q?h=C3=BCbsch=2E?= " + "Es hat viele Stacheln\n" + " und oft =?utf-8?q?bei=C3=9Ft?= mich.\n") + + def test_ews_combined_before_wrap(self): + self._test(parser.get_unstructured("Mein Kaktus ist hübsch. " + "Es beißt mich. " + "And that's all I'm sayin."), + "Mein Kaktus ist =?utf-8?q?h=C3=BCbsch=2E__Es_bei=C3=9Ft?= " + "mich. And that's\n" + " all I'm sayin.\n") + + # XXX Need test of an encoded word so long that it needs to be wrapped + + def test_simple_address(self): + self._test(parser.get_address_list("abc <xyz@example.com>")[0], + "abc <xyz@example.com>\n") + + def test_address_list_folding_at_commas(self): + self._test(parser.get_address_list('abc <xyz@example.com>, ' + '"Fred Blunt" <sharp@example.com>, ' + '"J.P.Cool" <hot@example.com>, ' + '"K<>y" <key@example.com>, ' + 'Firesale <cheap@example.com>, ' + '<end@example.com>')[0], + 'abc <xyz@example.com>, "Fred Blunt" <sharp@example.com>,\n' + ' "J.P.Cool" <hot@example.com>, "K<>y" <key@example.com>,\n' + ' Firesale <cheap@example.com>, <end@example.com>\n') + + def test_address_list_with_unicode_names(self): + self._test(parser.get_address_list( + 'Hübsch Kaktus <beautiful@example.com>, ' + 'beißt beißt <biter@example.com>')[0], + '=?utf-8?q?H=C3=BCbsch?= Kaktus <beautiful@example.com>,\n' + ' =?utf-8?q?bei=C3=9Ft_bei=C3=9Ft?= <biter@example.com>\n') + + def test_address_list_with_unicode_names_in_quotes(self): + self._test(parser.get_address_list( + '"Hübsch Kaktus" <beautiful@example.com>, ' + '"beißt" beißt <biter@example.com>')[0], + '=?utf-8?q?H=C3=BCbsch?= Kaktus <beautiful@example.com>,\n' + ' =?utf-8?q?bei=C3=9Ft_bei=C3=9Ft?= <biter@example.com>\n') + + # XXX Need tests with comments on various sides of a unicode token, + # and with unicode tokens in the comments. Spaces inside the quotes + # currently don't do the right thing. + + def test_initial_whitespace_splitting(self): + body = parser.get_unstructured(' ' + 'x'*77) + header = parser.Header([ + parser.HeaderLabel([parser.ValueTerminal('test:', 'atext')]), + parser.CFWSList([parser.WhiteSpaceTerminal(' ', 'fws')]), body]) + self._test(header, 'test: \n ' + 'x'*77 + '\n') + + def test_whitespace_splitting(self): + self._test(parser.get_unstructured('xxx ' + 'y'*77), + 'xxx \n ' + 'y'*77 + '\n') + +if __name__ == '__main__': + unittest.main() diff --git a/Lib/test/test_email/test__headerregistry.py b/Lib/test/test_email/test__headerregistry.py new file mode 100644 index 0000000..4398e29 --- /dev/null +++ b/Lib/test/test_email/test__headerregistry.py @@ -0,0 +1,717 @@ +import datetime +import textwrap +import unittest +from email import errors +from email import policy +from test.test_email import TestEmailBase +from email import _headerregistry +# Address and Group are public but I'm not sure where to put them yet. +from email._headerregistry import Address, Group + + +class TestHeaderRegistry(TestEmailBase): + + def test_arbitrary_name_unstructured(self): + factory = _headerregistry.HeaderRegistry() + h = factory('foobar', 'test') + self.assertIsInstance(h, _headerregistry.BaseHeader) + self.assertIsInstance(h, _headerregistry.UnstructuredHeader) + + def test_name_case_ignored(self): + factory = _headerregistry.HeaderRegistry() + # Whitebox check that test is valid + self.assertNotIn('Subject', factory.registry) + h = factory('Subject', 'test') + self.assertIsInstance(h, _headerregistry.BaseHeader) + self.assertIsInstance(h, _headerregistry.UniqueUnstructuredHeader) + + class FooBase: + def __init__(self, *args, **kw): + pass + + def test_override_default_base_class(self): + factory = _headerregistry.HeaderRegistry(base_class=self.FooBase) + h = factory('foobar', 'test') + self.assertIsInstance(h, self.FooBase) + self.assertIsInstance(h, _headerregistry.UnstructuredHeader) + + class FooDefault: + parse = _headerregistry.UnstructuredHeader.parse + + def test_override_default_class(self): + factory = _headerregistry.HeaderRegistry(default_class=self.FooDefault) + h = factory('foobar', 'test') + self.assertIsInstance(h, _headerregistry.BaseHeader) + self.assertIsInstance(h, self.FooDefault) + + def test_override_default_class_only_overrides_default(self): + factory = _headerregistry.HeaderRegistry(default_class=self.FooDefault) + h = factory('subject', 'test') + self.assertIsInstance(h, _headerregistry.BaseHeader) + self.assertIsInstance(h, _headerregistry.UniqueUnstructuredHeader) + + def test_dont_use_default_map(self): + factory = _headerregistry.HeaderRegistry(use_default_map=False) + h = factory('subject', 'test') + self.assertIsInstance(h, _headerregistry.BaseHeader) + self.assertIsInstance(h, _headerregistry.UnstructuredHeader) + + def test_map_to_type(self): + factory = _headerregistry.HeaderRegistry() + h1 = factory('foobar', 'test') + factory.map_to_type('foobar', _headerregistry.UniqueUnstructuredHeader) + h2 = factory('foobar', 'test') + self.assertIsInstance(h1, _headerregistry.BaseHeader) + self.assertIsInstance(h1, _headerregistry.UnstructuredHeader) + self.assertIsInstance(h2, _headerregistry.BaseHeader) + self.assertIsInstance(h2, _headerregistry.UniqueUnstructuredHeader) + + +class TestHeaderBase(TestEmailBase): + + factory = _headerregistry.HeaderRegistry() + + def make_header(self, name, value): + return self.factory(name, value) + + +class TestBaseHeaderFeatures(TestHeaderBase): + + def test_str(self): + h = self.make_header('subject', 'this is a test') + self.assertIsInstance(h, str) + self.assertEqual(h, 'this is a test') + self.assertEqual(str(h), 'this is a test') + + def test_substr(self): + h = self.make_header('subject', 'this is a test') + self.assertEqual(h[5:7], 'is') + + def test_has_name(self): + h = self.make_header('subject', 'this is a test') + self.assertEqual(h.name, 'subject') + + def _test_attr_ro(self, attr): + h = self.make_header('subject', 'this is a test') + with self.assertRaises(AttributeError): + setattr(h, attr, 'foo') + + def test_name_read_only(self): + self._test_attr_ro('name') + + def test_defects_read_only(self): + self._test_attr_ro('defects') + + def test_defects_is_tuple(self): + h = self.make_header('subject', 'this is a test') + self.assertEqual(len(h.defects), 0) + self.assertIsInstance(h.defects, tuple) + # Make sure it is still true when there are defects. + h = self.make_header('date', '') + self.assertEqual(len(h.defects), 1) + self.assertIsInstance(h.defects, tuple) + + # XXX: FIXME + #def test_CR_in_value(self): + # # XXX: this also re-raises the issue of embedded headers, + # # need test and solution for that. + # value = '\r'.join(['this is', ' a test']) + # h = self.make_header('subject', value) + # self.assertEqual(h, value) + # self.assertDefectsEqual(h.defects, [errors.ObsoleteHeaderDefect]) + + def test_RFC2047_value_decoded(self): + value = '=?utf-8?q?this_is_a_test?=' + h = self.make_header('subject', value) + self.assertEqual(h, 'this is a test') + + +class TestDateHeader(TestHeaderBase): + + datestring = 'Sun, 23 Sep 2001 20:10:55 -0700' + utcoffset = datetime.timedelta(hours=-7) + tz = datetime.timezone(utcoffset) + dt = datetime.datetime(2001, 9, 23, 20, 10, 55, tzinfo=tz) + + def test_parse_date(self): + h = self.make_header('date', self.datestring) + self.assertEqual(h, self.datestring) + self.assertEqual(h.datetime, self.dt) + self.assertEqual(h.datetime.utcoffset(), self.utcoffset) + self.assertEqual(h.defects, ()) + + def test_set_from_datetime(self): + h = self.make_header('date', self.dt) + self.assertEqual(h, self.datestring) + self.assertEqual(h.datetime, self.dt) + self.assertEqual(h.defects, ()) + + def test_date_header_properties(self): + h = self.make_header('date', self.datestring) + self.assertIsInstance(h, _headerregistry.UniqueDateHeader) + self.assertEqual(h.max_count, 1) + self.assertEqual(h.defects, ()) + + def test_resent_date_header_properties(self): + h = self.make_header('resent-date', self.datestring) + self.assertIsInstance(h, _headerregistry.DateHeader) + self.assertEqual(h.max_count, None) + self.assertEqual(h.defects, ()) + + def test_no_value_is_defect(self): + h = self.make_header('date', '') + self.assertEqual(len(h.defects), 1) + self.assertIsInstance(h.defects[0], errors.HeaderMissingRequiredValue) + + def test_datetime_read_only(self): + h = self.make_header('date', self.datestring) + with self.assertRaises(AttributeError): + h.datetime = 'foo' + + +class TestAddressHeader(TestHeaderBase): + + examples = { + + 'empty': + ('<>', + [errors.InvalidHeaderDefect], + '<>', + '', + '<>', + '', + '', + None), + + 'address_only': + ('zippy@pinhead.com', + [], + 'zippy@pinhead.com', + '', + 'zippy@pinhead.com', + 'zippy', + 'pinhead.com', + None), + + 'name_and_address': + ('Zaphrod Beblebrux <zippy@pinhead.com>', + [], + 'Zaphrod Beblebrux <zippy@pinhead.com>', + 'Zaphrod Beblebrux', + 'zippy@pinhead.com', + 'zippy', + 'pinhead.com', + None), + + 'quoted_local_part': + ('Zaphrod Beblebrux <"foo bar"@pinhead.com>', + [], + 'Zaphrod Beblebrux <"foo bar"@pinhead.com>', + 'Zaphrod Beblebrux', + '"foo bar"@pinhead.com', + 'foo bar', + 'pinhead.com', + None), + + 'quoted_parens_in_name': + (r'"A \(Special\) Person" <person@dom.ain>', + [], + '"A (Special) Person" <person@dom.ain>', + 'A (Special) Person', + 'person@dom.ain', + 'person', + 'dom.ain', + None), + + 'quoted_backslashes_in_name': + (r'"Arthur \\Backslash\\ Foobar" <person@dom.ain>', + [], + r'"Arthur \\Backslash\\ Foobar" <person@dom.ain>', + r'Arthur \Backslash\ Foobar', + 'person@dom.ain', + 'person', + 'dom.ain', + None), + + 'name_with_dot': + ('John X. Doe <jxd@example.com>', + [errors.ObsoleteHeaderDefect], + '"John X. Doe" <jxd@example.com>', + 'John X. Doe', + 'jxd@example.com', + 'jxd', + 'example.com', + None), + + 'quoted_strings_in_local_part': + ('""example" example"@example.com', + [errors.InvalidHeaderDefect]*3, + '"example example"@example.com', + '', + '"example example"@example.com', + 'example example', + 'example.com', + None), + + 'escaped_quoted_strings_in_local_part': + (r'"\"example\" example"@example.com', + [], + r'"\"example\" example"@example.com', + '', + r'"\"example\" example"@example.com', + r'"example" example', + 'example.com', + None), + + 'escaped_escapes_in_local_part': + (r'"\\"example\\" example"@example.com', + [errors.InvalidHeaderDefect]*5, + r'"\\example\\\\ example"@example.com', + '', + r'"\\example\\\\ example"@example.com', + r'\example\\ example', + 'example.com', + None), + + 'spaces_in_unquoted_local_part_collapsed': + ('merwok wok @example.com', + [errors.InvalidHeaderDefect]*2, + '"merwok wok"@example.com', + '', + '"merwok wok"@example.com', + 'merwok wok', + 'example.com', + None), + + 'spaces_around_dots_in_local_part_removed': + ('merwok. wok . wok@example.com', + [errors.ObsoleteHeaderDefect], + 'merwok.wok.wok@example.com', + '', + 'merwok.wok.wok@example.com', + 'merwok.wok.wok', + 'example.com', + None), + + } + + # XXX: Need many more examples, and in particular some with names in + # trailing comments, which aren't currently handled. comments in + # general are not handled yet. + + def _test_single_addr(self, source, defects, decoded, display_name, + addr_spec, username, domain, comment): + h = self.make_header('sender', source) + self.assertEqual(h, decoded) + self.assertDefectsEqual(h.defects, defects) + a = h.address + self.assertEqual(str(a), decoded) + self.assertEqual(len(h.groups), 1) + self.assertEqual([a], list(h.groups[0].addresses)) + self.assertEqual([a], list(h.addresses)) + self.assertEqual(a.display_name, display_name) + self.assertEqual(a.addr_spec, addr_spec) + self.assertEqual(a.username, username) + self.assertEqual(a.domain, domain) + # XXX: we have no comment support yet. + #self.assertEqual(a.comment, comment) + + for name in examples: + locals()['test_'+name] = ( + lambda self, name=name: + self._test_single_addr(*self.examples[name])) + + def _test_group_single_addr(self, source, defects, decoded, display_name, + addr_spec, username, domain, comment): + source = 'foo: {};'.format(source) + gdecoded = 'foo: {};'.format(decoded) if decoded else 'foo:;' + h = self.make_header('to', source) + self.assertEqual(h, gdecoded) + self.assertDefectsEqual(h.defects, defects) + self.assertEqual(h.groups[0].addresses, h.addresses) + self.assertEqual(len(h.groups), 1) + self.assertEqual(len(h.addresses), 1) + a = h.addresses[0] + self.assertEqual(str(a), decoded) + self.assertEqual(a.display_name, display_name) + self.assertEqual(a.addr_spec, addr_spec) + self.assertEqual(a.username, username) + self.assertEqual(a.domain, domain) + + for name in examples: + locals()['test_group_'+name] = ( + lambda self, name=name: + self._test_group_single_addr(*self.examples[name])) + + def test_simple_address_list(self): + value = ('Fred <dinsdale@python.org>, foo@example.com, ' + '"Harry W. Hastings" <hasty@example.com>') + h = self.make_header('to', value) + self.assertEqual(h, value) + self.assertEqual(len(h.groups), 3) + self.assertEqual(len(h.addresses), 3) + for i in range(3): + self.assertEqual(h.groups[i].addresses[0], h.addresses[i]) + self.assertEqual(str(h.addresses[0]), 'Fred <dinsdale@python.org>') + self.assertEqual(str(h.addresses[1]), 'foo@example.com') + self.assertEqual(str(h.addresses[2]), + '"Harry W. Hastings" <hasty@example.com>') + self.assertEqual(h.addresses[2].display_name, + 'Harry W. Hastings') + + def test_complex_address_list(self): + examples = list(self.examples.values()) + source = ('dummy list:;, another: (empty);,' + + ', '.join([x[0] for x in examples[:4]]) + ', ' + + r'"A \"list\"": ' + + ', '.join([x[0] for x in examples[4:6]]) + ';,' + + ', '.join([x[0] for x in examples[6:]]) + ) + # XXX: the fact that (empty) disappears here is a potential API design + # bug. We don't currently have a way to preserve comments. + expected = ('dummy list:;, another:;, ' + + ', '.join([x[2] for x in examples[:4]]) + ', ' + + r'"A \"list\"": ' + + ', '.join([x[2] for x in examples[4:6]]) + ';, ' + + ', '.join([x[2] for x in examples[6:]]) + ) + + h = self.make_header('to', source) + self.assertEqual(h.split(','), expected.split(',')) + self.assertEqual(h, expected) + self.assertEqual(len(h.groups), 7 + len(examples) - 6) + self.assertEqual(h.groups[0].display_name, 'dummy list') + self.assertEqual(h.groups[1].display_name, 'another') + self.assertEqual(h.groups[6].display_name, 'A "list"') + self.assertEqual(len(h.addresses), len(examples)) + for i in range(4): + self.assertIsNone(h.groups[i+2].display_name) + self.assertEqual(str(h.groups[i+2].addresses[0]), examples[i][2]) + for i in range(7, 7 + len(examples) - 6): + self.assertIsNone(h.groups[i].display_name) + self.assertEqual(str(h.groups[i].addresses[0]), examples[i-1][2]) + for i in range(len(examples)): + self.assertEqual(str(h.addresses[i]), examples[i][2]) + self.assertEqual(h.addresses[i].addr_spec, examples[i][4]) + + def test_address_read_only(self): + h = self.make_header('sender', 'abc@xyz.com') + with self.assertRaises(AttributeError): + h.address = 'foo' + + def test_addresses_read_only(self): + h = self.make_header('sender', 'abc@xyz.com') + with self.assertRaises(AttributeError): + h.addresses = 'foo' + + def test_groups_read_only(self): + h = self.make_header('sender', 'abc@xyz.com') + with self.assertRaises(AttributeError): + h.groups = 'foo' + + def test_addresses_types(self): + source = 'me <who@example.com>' + h = self.make_header('to', source) + self.assertIsInstance(h.addresses, tuple) + self.assertIsInstance(h.addresses[0], Address) + + def test_groups_types(self): + source = 'me <who@example.com>' + h = self.make_header('to', source) + self.assertIsInstance(h.groups, tuple) + self.assertIsInstance(h.groups[0], Group) + + def test_set_from_Address(self): + h = self.make_header('to', Address('me', 'foo', 'example.com')) + self.assertEqual(h, 'me <foo@example.com>') + + def test_set_from_Address_list(self): + h = self.make_header('to', [Address('me', 'foo', 'example.com'), + Address('you', 'bar', 'example.com')]) + self.assertEqual(h, 'me <foo@example.com>, you <bar@example.com>') + + def test_set_from_Address_and_Group_list(self): + h = self.make_header('to', [Address('me', 'foo', 'example.com'), + Group('bing', [Address('fiz', 'z', 'b.com'), + Address('zif', 'f', 'c.com')]), + Address('you', 'bar', 'example.com')]) + self.assertEqual(h, 'me <foo@example.com>, bing: fiz <z@b.com>, ' + 'zif <f@c.com>;, you <bar@example.com>') + self.assertEqual(h.fold(policy=policy.default.clone(max_line_length=40)), + 'to: me <foo@example.com>,\n' + ' bing: fiz <z@b.com>, zif <f@c.com>;,\n' + ' you <bar@example.com>\n') + + def test_set_from_Group_list(self): + h = self.make_header('to', [Group('bing', [Address('fiz', 'z', 'b.com'), + Address('zif', 'f', 'c.com')])]) + self.assertEqual(h, 'bing: fiz <z@b.com>, zif <f@c.com>;') + + +class TestAddressAndGroup(TestEmailBase): + + def _test_attr_ro(self, obj, attr): + with self.assertRaises(AttributeError): + setattr(obj, attr, 'foo') + + def test_address_display_name_ro(self): + self._test_attr_ro(Address('foo', 'bar', 'baz'), 'display_name') + + def test_address_username_ro(self): + self._test_attr_ro(Address('foo', 'bar', 'baz'), 'username') + + def test_address_domain_ro(self): + self._test_attr_ro(Address('foo', 'bar', 'baz'), 'domain') + + def test_group_display_name_ro(self): + self._test_attr_ro(Group('foo'), 'display_name') + + def test_group_addresses_ro(self): + self._test_attr_ro(Group('foo'), 'addresses') + + def test_address_from_username_domain(self): + a = Address('foo', 'bar', 'baz') + self.assertEqual(a.display_name, 'foo') + self.assertEqual(a.username, 'bar') + self.assertEqual(a.domain, 'baz') + self.assertEqual(a.addr_spec, 'bar@baz') + self.assertEqual(str(a), 'foo <bar@baz>') + + def test_address_from_addr_spec(self): + a = Address('foo', addr_spec='bar@baz') + self.assertEqual(a.display_name, 'foo') + self.assertEqual(a.username, 'bar') + self.assertEqual(a.domain, 'baz') + self.assertEqual(a.addr_spec, 'bar@baz') + self.assertEqual(str(a), 'foo <bar@baz>') + + def test_address_with_no_display_name(self): + a = Address(addr_spec='bar@baz') + self.assertEqual(a.display_name, '') + self.assertEqual(a.username, 'bar') + self.assertEqual(a.domain, 'baz') + self.assertEqual(a.addr_spec, 'bar@baz') + self.assertEqual(str(a), 'bar@baz') + + def test_null_address(self): + a = Address() + self.assertEqual(a.display_name, '') + self.assertEqual(a.username, '') + self.assertEqual(a.domain, '') + self.assertEqual(a.addr_spec, '<>') + self.assertEqual(str(a), '<>') + + def test_domain_only(self): + # This isn't really a valid address. + a = Address(domain='buzz') + self.assertEqual(a.display_name, '') + self.assertEqual(a.username, '') + self.assertEqual(a.domain, 'buzz') + self.assertEqual(a.addr_spec, '@buzz') + self.assertEqual(str(a), '@buzz') + + def test_username_only(self): + # This isn't really a valid address. + a = Address(username='buzz') + self.assertEqual(a.display_name, '') + self.assertEqual(a.username, 'buzz') + self.assertEqual(a.domain, '') + self.assertEqual(a.addr_spec, 'buzz') + self.assertEqual(str(a), 'buzz') + + def test_display_name_only(self): + a = Address('buzz') + self.assertEqual(a.display_name, 'buzz') + self.assertEqual(a.username, '') + self.assertEqual(a.domain, '') + self.assertEqual(a.addr_spec, '<>') + self.assertEqual(str(a), 'buzz <>') + + def test_quoting(self): + # Ideally we'd check every special individually, but I'm not up for + # writing that many tests. + a = Address('Sara J.', 'bad name', 'example.com') + self.assertEqual(a.display_name, 'Sara J.') + self.assertEqual(a.username, 'bad name') + self.assertEqual(a.domain, 'example.com') + self.assertEqual(a.addr_spec, '"bad name"@example.com') + self.assertEqual(str(a), '"Sara J." <"bad name"@example.com>') + + def test_il8n(self): + a = Address('Éric', 'wok', 'exàmple.com') + self.assertEqual(a.display_name, 'Éric') + self.assertEqual(a.username, 'wok') + self.assertEqual(a.domain, 'exàmple.com') + self.assertEqual(a.addr_spec, 'wok@exàmple.com') + self.assertEqual(str(a), 'Éric <wok@exàmple.com>') + + # XXX: there is an API design issue that needs to be solved here. + #def test_non_ascii_username_raises(self): + # with self.assertRaises(ValueError): + # Address('foo', 'wők', 'example.com') + + def test_non_ascii_username_in_addr_spec_raises(self): + with self.assertRaises(ValueError): + Address('foo', addr_spec='wők@example.com') + + def test_address_addr_spec_and_username_raises(self): + with self.assertRaises(TypeError): + Address('foo', username='bing', addr_spec='bar@baz') + + def test_address_addr_spec_and_domain_raises(self): + with self.assertRaises(TypeError): + Address('foo', domain='bing', addr_spec='bar@baz') + + def test_address_addr_spec_and_username_and_domain_raises(self): + with self.assertRaises(TypeError): + Address('foo', username='bong', domain='bing', addr_spec='bar@baz') + + def test_space_in_addr_spec_username_raises(self): + with self.assertRaises(ValueError): + Address('foo', addr_spec="bad name@example.com") + + def test_bad_addr_sepc_raises(self): + with self.assertRaises(ValueError): + Address('foo', addr_spec="name@ex[]ample.com") + + def test_empty_group(self): + g = Group('foo') + self.assertEqual(g.display_name, 'foo') + self.assertEqual(g.addresses, tuple()) + self.assertEqual(str(g), 'foo:;') + + def test_empty_group_list(self): + g = Group('foo', addresses=[]) + self.assertEqual(g.display_name, 'foo') + self.assertEqual(g.addresses, tuple()) + self.assertEqual(str(g), 'foo:;') + + def test_null_group(self): + g = Group() + self.assertIsNone(g.display_name) + self.assertEqual(g.addresses, tuple()) + self.assertEqual(str(g), 'None:;') + + def test_group_with_addresses(self): + addrs = [Address('b', 'b', 'c'), Address('a', 'b','c')] + g = Group('foo', addrs) + self.assertEqual(g.display_name, 'foo') + self.assertEqual(g.addresses, tuple(addrs)) + self.assertEqual(str(g), 'foo: b <b@c>, a <b@c>;') + + def test_group_with_addresses_no_display_name(self): + addrs = [Address('b', 'b', 'c'), Address('a', 'b','c')] + g = Group(addresses=addrs) + self.assertIsNone(g.display_name) + self.assertEqual(g.addresses, tuple(addrs)) + self.assertEqual(str(g), 'None: b <b@c>, a <b@c>;') + + def test_group_with_one_address_no_display_name(self): + addrs = [Address('b', 'b', 'c')] + g = Group(addresses=addrs) + self.assertIsNone(g.display_name) + self.assertEqual(g.addresses, tuple(addrs)) + self.assertEqual(str(g), 'b <b@c>') + + def test_display_name_quoting(self): + g = Group('foo.bar') + self.assertEqual(g.display_name, 'foo.bar') + self.assertEqual(g.addresses, tuple()) + self.assertEqual(str(g), '"foo.bar":;') + + def test_display_name_blanks_not_quoted(self): + g = Group('foo bar') + self.assertEqual(g.display_name, 'foo bar') + self.assertEqual(g.addresses, tuple()) + self.assertEqual(str(g), 'foo bar:;') + + +class TestFolding(TestHeaderBase): + + def test_short_unstructured(self): + h = self.make_header('subject', 'this is a test') + self.assertEqual(h.fold(policy=self.policy), + 'subject: this is a test\n') + + def test_long_unstructured(self): + h = self.make_header('Subject', 'This is a long header ' + 'line that will need to be folded into two lines ' + 'and will demonstrate basic folding') + self.assertEqual(h.fold(policy=self.policy), + 'Subject: This is a long header line that will ' + 'need to be folded into two lines\n' + ' and will demonstrate basic folding\n') + + def test_unstructured_short_max_line_length(self): + h = self.make_header('Subject', 'this is a short header ' + 'that will be folded anyway') + self.assertEqual( + h.fold(policy=policy.default.clone(max_line_length=20)), + textwrap.dedent("""\ + Subject: this is a + short header that + will be folded + anyway + """)) + + def test_fold_unstructured_single_word(self): + h = self.make_header('Subject', 'test') + self.assertEqual(h.fold(policy=self.policy), 'Subject: test\n') + + def test_fold_unstructured_short(self): + h = self.make_header('Subject', 'test test test') + self.assertEqual(h.fold(policy=self.policy), + 'Subject: test test test\n') + + def test_fold_unstructured_with_overlong_word(self): + h = self.make_header('Subject', 'thisisaverylonglineconsistingofa' + 'singlewordthatwontfit') + self.assertEqual( + h.fold(policy=policy.default.clone(max_line_length=20)), + 'Subject: thisisaverylonglineconsistingofasinglewordthatwontfit\n') + + def test_fold_unstructured_with_two_overlong_words(self): + h = self.make_header('Subject', 'thisisaverylonglineconsistingofa' + 'singlewordthatwontfit plusanotherverylongwordthatwontfit') + self.assertEqual( + h.fold(policy=policy.default.clone(max_line_length=20)), + 'Subject: thisisaverylonglineconsistingofasinglewordthatwontfit\n' + ' plusanotherverylongwordthatwontfit\n') + + def test_fold_unstructured_with_slightly_long_word(self): + h = self.make_header('Subject', 'thislongwordislessthanmaxlinelen') + self.assertEqual( + h.fold(policy=policy.default.clone(max_line_length=35)), + 'Subject:\n thislongwordislessthanmaxlinelen\n') + + def test_fold_unstructured_with_commas(self): + # The old wrapper would fold this at the commas. + h = self.make_header('Subject', "This header is intended to " + "demonstrate, in a fairly susinct way, that we now do " + "not give a , special treatment in unstructured headers.") + self.assertEqual( + h.fold(policy=policy.default.clone(max_line_length=60)), + textwrap.dedent("""\ + Subject: This header is intended to demonstrate, in a fairly + susinct way, that we now do not give a , special treatment + in unstructured headers. + """)) + + def test_fold_address_list(self): + h = self.make_header('To', '"Theodore H. Perfect" <yes@man.com>, ' + '"My address is very long because my name is long" <foo@bar.com>, ' + '"Only A. Friend" <no@yes.com>') + self.assertEqual(h.fold(policy=self.policy), textwrap.dedent("""\ + To: "Theodore H. Perfect" <yes@man.com>, + "My address is very long because my name is long" <foo@bar.com>, + "Only A. Friend" <no@yes.com> + """)) + + def test_fold_date_header(self): + h = self.make_header('Date', 'Sat, 2 Feb 2002 17:00:06 -0800') + self.assertEqual(h.fold(policy=self.policy), + 'Date: Sat, 02 Feb 2002 17:00:06 -0800\n') + + +if __name__ == '__main__': + unittest.main() diff --git a/Lib/test/test_email/test_generator.py b/Lib/test/test_email/test_generator.py index 8f5fde7..c1af024 100644 --- a/Lib/test/test_email/test_generator.py +++ b/Lib/test/test_email/test_generator.py @@ -6,14 +6,16 @@ from email.generator import Generator, BytesGenerator from email import policy from test.test_email import TestEmailBase -# XXX: move generator tests from test_email into here at some point. +class TestGeneratorBase: -class TestGeneratorBase(): + policy = policy.default - policy = policy.compat32 + def msgmaker(self, msg, policy=None): + policy = self.policy if policy is None else policy + return self.msgfunc(msg, policy=policy) - long_subject = { + refold_long_expected = { 0: textwrap.dedent("""\ To: whom_it_may_concern@example.com From: nobody_you_want_to_know@example.com @@ -23,33 +25,32 @@ class TestGeneratorBase(): None """), + # From is wrapped because wrapped it fits in 40. 40: textwrap.dedent("""\ To: whom_it_may_concern@example.com - From:\x20 + From: nobody_you_want_to_know@example.com Subject: We the willing led by the - unknowing are doing the - impossible for the ungrateful. We have - done so much for so long with so little - we are now qualified to do anything - with nothing. + unknowing are doing the impossible for + the ungrateful. We have done so much + for so long with so little we are now + qualified to do anything with nothing. None """), + # Neither to nor from fit even if put on a new line, + # so we leave them sticking out on the first line. 20: textwrap.dedent("""\ - To:\x20 - whom_it_may_concern@example.com - From:\x20 - nobody_you_want_to_know@example.com + To: whom_it_may_concern@example.com + From: nobody_you_want_to_know@example.com Subject: We the willing led by the unknowing are doing - the - impossible for the - ungrateful. We have - done so much for so - long with so little - we are now + the impossible for + the ungrateful. We + have done so much + for so long with so + little we are now qualified to do anything with nothing. @@ -57,65 +58,90 @@ class TestGeneratorBase(): None """), } - long_subject[100] = long_subject[0] - - def maxheaderlen_parameter_test(self, n): - msg = self.msgmaker(self.typ(self.long_subject[0])) + refold_long_expected[100] = refold_long_expected[0] + + refold_all_expected = refold_long_expected.copy() + refold_all_expected[0] = ( + "To: whom_it_may_concern@example.com\n" + "From: nobody_you_want_to_know@example.com\n" + "Subject: We the willing led by the unknowing are doing the " + "impossible for the ungrateful. We have done so much for " + "so long with so little we are now qualified to do anything " + "with nothing.\n" + "\n" + "None\n") + refold_all_expected[100] = ( + "To: whom_it_may_concern@example.com\n" + "From: nobody_you_want_to_know@example.com\n" + "Subject: We the willing led by the unknowing are doing the " + "impossible for the ungrateful. We have\n" + " done so much for so long with so little we are now qualified " + "to do anything with nothing.\n" + "\n" + "None\n") + + def _test_maxheaderlen_parameter(self, n): + msg = self.msgmaker(self.typ(self.refold_long_expected[0])) s = self.ioclass() - g = self.genclass(s, maxheaderlen=n) + g = self.genclass(s, maxheaderlen=n, policy=self.policy) g.flatten(msg) - self.assertEqual(s.getvalue(), self.typ(self.long_subject[n])) + self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[n])) - def test_maxheaderlen_parameter_0(self): - self.maxheaderlen_parameter_test(0) + for n in refold_long_expected: + locals()['test_maxheaderlen_parameter_' + str(n)] = ( + lambda self, n=n: + self._test_maxheaderlen_parameter(n)) - def test_maxheaderlen_parameter_100(self): - self.maxheaderlen_parameter_test(100) - - def test_maxheaderlen_parameter_40(self): - self.maxheaderlen_parameter_test(40) - - def test_maxheaderlen_parameter_20(self): - self.maxheaderlen_parameter_test(20) - - def maxheaderlen_policy_test(self, n): - msg = self.msgmaker(self.typ(self.long_subject[0])) + def _test_max_line_length_policy(self, n): + msg = self.msgmaker(self.typ(self.refold_long_expected[0])) s = self.ioclass() - g = self.genclass(s, policy=policy.default.clone(max_line_length=n)) + g = self.genclass(s, policy=self.policy.clone(max_line_length=n)) g.flatten(msg) - self.assertEqual(s.getvalue(), self.typ(self.long_subject[n])) - - def test_maxheaderlen_policy_0(self): - self.maxheaderlen_policy_test(0) - - def test_maxheaderlen_policy_100(self): - self.maxheaderlen_policy_test(100) - - def test_maxheaderlen_policy_40(self): - self.maxheaderlen_policy_test(40) + self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[n])) - def test_maxheaderlen_policy_20(self): - self.maxheaderlen_policy_test(20) + for n in refold_long_expected: + locals()['test_max_line_length_policy' + str(n)] = ( + lambda self, n=n: + self._test_max_line_length_policy(n)) - def maxheaderlen_parm_overrides_policy_test(self, n): - msg = self.msgmaker(self.typ(self.long_subject[0])) + def _test_maxheaderlen_parm_overrides_policy(self, n): + msg = self.msgmaker(self.typ(self.refold_long_expected[0])) s = self.ioclass() g = self.genclass(s, maxheaderlen=n, - policy=policy.default.clone(max_line_length=10)) + policy=self.policy.clone(max_line_length=10)) g.flatten(msg) - self.assertEqual(s.getvalue(), self.typ(self.long_subject[n])) + self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[n])) - def test_maxheaderlen_parm_overrides_policy_0(self): - self.maxheaderlen_parm_overrides_policy_test(0) + for n in refold_long_expected: + locals()['test_maxheaderlen_parm_overrides_policy' + str(n)] = ( + lambda self, n=n: + self._test_maxheaderlen_parm_overrides_policy(n)) - def test_maxheaderlen_parm_overrides_policy_100(self): - self.maxheaderlen_parm_overrides_policy_test(100) + def _test_refold_none_does_not_fold(self, n): + msg = self.msgmaker(self.typ(self.refold_long_expected[0])) + s = self.ioclass() + g = self.genclass(s, policy=self.policy.clone(refold_source='none', + max_line_length=n)) + g.flatten(msg) + self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[0])) + + for n in refold_long_expected: + locals()['test_refold_none_does_not_fold' + str(n)] = ( + lambda self, n=n: + self._test_refold_none_does_not_fold(n)) - def test_maxheaderlen_parm_overrides_policy_40(self): - self.maxheaderlen_parm_overrides_policy_test(40) + def _test_refold_all(self, n): + msg = self.msgmaker(self.typ(self.refold_long_expected[0])) + s = self.ioclass() + g = self.genclass(s, policy=self.policy.clone(refold_source='all', + max_line_length=n)) + g.flatten(msg) + self.assertEqual(s.getvalue(), self.typ(self.refold_all_expected[n])) - def test_maxheaderlen_parm_overrides_policy_20(self): - self.maxheaderlen_parm_overrides_policy_test(20) + for n in refold_long_expected: + locals()['test_refold_all' + str(n)] = ( + lambda self, n=n: + self._test_refold_all(n)) def test_crlf_control_via_policy(self): source = "Subject: test\r\n\r\ntest body\r\n" @@ -138,30 +164,24 @@ class TestGeneratorBase(): class TestGenerator(TestGeneratorBase, TestEmailBase): + msgfunc = staticmethod(message_from_string) genclass = Generator ioclass = io.StringIO typ = str - def msgmaker(self, msg, policy=None): - policy = self.policy if policy is None else policy - return message_from_string(msg, policy=policy) - class TestBytesGenerator(TestGeneratorBase, TestEmailBase): + msgfunc = staticmethod(message_from_bytes) genclass = BytesGenerator ioclass = io.BytesIO typ = lambda self, x: x.encode('ascii') - def msgmaker(self, msg, policy=None): - policy = self.policy if policy is None else policy - return message_from_bytes(msg, policy=policy) - def test_cte_type_7bit_handles_unknown_8bit(self): source = ("Subject: Maintenant je vous présente mon " "collègue\n\n").encode('utf-8') - expected = ('Subject: =?unknown-8bit?q?Maintenant_je_vous_pr=C3=A9sente_mon_' - 'coll=C3=A8gue?=\n\n').encode('ascii') + expected = ('Subject: Maintenant je vous =?unknown-8bit?q?' + 'pr=C3=A9sente_mon_coll=C3=A8gue?=\n\n').encode('ascii') msg = message_from_bytes(source) s = io.BytesIO() g = BytesGenerator(s, policy=self.policy.clone(cte_type='7bit')) diff --git a/Lib/test/test_email/test_pickleable.py b/Lib/test/test_email/test_pickleable.py new file mode 100644 index 0000000..e4c77ca --- /dev/null +++ b/Lib/test/test_email/test_pickleable.py @@ -0,0 +1,57 @@ +import unittest +import textwrap +import copy +import pickle +from email import policy +from email import message_from_string +from email._headerregistry import HeaderRegistry +from test.test_email import TestEmailBase + +class TestPickleCopyHeader(TestEmailBase): + + unstructured = HeaderRegistry()('subject', 'this is a test') + + def test_deepcopy_unstructured(self): + h = copy.deepcopy(self.unstructured) + self.assertEqual(str(h), str(self.unstructured)) + + def test_pickle_unstructured(self): + p = pickle.dumps(self.unstructured) + h = pickle.loads(p) + self.assertEqual(str(h), str(self.unstructured)) + + address = HeaderRegistry()('from', 'frodo@mordor.net') + + def test_deepcopy_address(self): + h = copy.deepcopy(self.address) + self.assertEqual(str(h), str(self.address)) + + def test_pickle_address(self): + p = pickle.dumps(self.address) + h = pickle.loads(p) + self.assertEqual(str(h), str(self.address)) + + +class TestPickleCopyMessage(TestEmailBase): + + testmsg = message_from_string(textwrap.dedent("""\ + From: frodo@mordor.net + To: bilbo@underhill.org + Subject: help + + I think I forgot the ring. + """), policy=policy.default) + + def test_deepcopy(self): + msg2 = copy.deepcopy(self.testmsg) + self.assertEqual(msg2.as_string(), self.testmsg.as_string()) + + def test_pickle(self): + p = pickle.dumps(self.testmsg) + msg2 = pickle.loads(p) + self.assertEqual(msg2.as_string(), self.testmsg.as_string()) + + + +if __name__ == '__main__': + unittest.main() diff --git a/Lib/test/test_email/test_policy.py b/Lib/test/test_email/test_policy.py index 07925a7..45a7666 100644 --- a/Lib/test/test_email/test_policy.py +++ b/Lib/test/test_email/test_policy.py @@ -5,49 +5,70 @@ import unittest import email.policy import email.parser import email.generator +from email import _headerregistry + +def make_defaults(base_defaults, differences): + defaults = base_defaults.copy() + defaults.update(differences) + return defaults class PolicyAPITests(unittest.TestCase): longMessage = True - # These default values are the ones set on email.policy.default. - # If any of these defaults change, the docs must be updated. - policy_defaults = { + # Base default values. + compat32_defaults = { 'max_line_length': 78, 'linesep': '\n', 'cte_type': '8bit', 'raise_on_defect': False, } - - # For each policy under test, we give here the values of the attributes - # that are different from the defaults for that policy. + # These default values are the ones set on email.policy.default. + # If any of these defaults change, the docs must be updated. + policy_defaults = compat32_defaults.copy() + policy_defaults.update({ + 'raise_on_defect': False, + 'header_factory': email.policy.EmailPolicy.header_factory, + 'refold_source': 'long', + }) + + # For each policy under test, we give here what we expect the defaults to + # be for that policy. The second argument to make defaults is the + # difference between the base defaults and that for the particular policy. + new_policy = email.policy.EmailPolicy() policies = { - email.policy.Compat32(): {}, - email.policy.compat32: {}, - email.policy.default: {}, - email.policy.SMTP: {'linesep': '\r\n'}, - email.policy.HTTP: {'linesep': '\r\n', 'max_line_length': None}, - email.policy.strict: {'raise_on_defect': True}, + email.policy.compat32: make_defaults(compat32_defaults, {}), + email.policy.default: make_defaults(policy_defaults, {}), + email.policy.SMTP: make_defaults(policy_defaults, + {'linesep': '\r\n'}), + email.policy.HTTP: make_defaults(policy_defaults, + {'linesep': '\r\n', + 'max_line_length': None}), + email.policy.strict: make_defaults(policy_defaults, + {'raise_on_defect': True}), + new_policy: make_defaults(policy_defaults, {}), } + # Creating a new policy creates a new header factory. There is a test + # later that proves this. + policies[new_policy]['header_factory'] = new_policy.header_factory def test_defaults(self): - for policy, changed_defaults in self.policies.items(): - expected = self.policy_defaults.copy() - expected.update(changed_defaults) + for policy, expected in self.policies.items(): for attr, value in expected.items(): self.assertEqual(getattr(policy, attr), value, ("change {} docs/docstrings if defaults have " "changed").format(policy)) def test_all_attributes_covered(self): - for attr in dir(email.policy.default): - if (attr.startswith('_') or - isinstance(getattr(email.policy.Policy, attr), - types.FunctionType)): - continue - else: - self.assertIn(attr, self.policy_defaults, - "{} is not fully tested".format(attr)) + for policy, expected in self.policies.items(): + for attr in dir(policy): + if (attr.startswith('_') or + isinstance(getattr(email.policy.EmailPolicy, attr), + types.FunctionType)): + continue + else: + self.assertIn(attr, expected, + "{} is not fully tested".format(attr)) def test_abc(self): with self.assertRaises(TypeError) as cm: @@ -62,18 +83,20 @@ class PolicyAPITests(unittest.TestCase): self.assertIn(method, msg) def test_policy_is_immutable(self): - for policy in self.policies: - for attr in self.policy_defaults: + for policy, defaults in self.policies.items(): + for attr in defaults: with self.assertRaisesRegex(AttributeError, attr+".*read-only"): setattr(policy, attr, None) with self.assertRaisesRegex(AttributeError, 'no attribute.*foo'): policy.foo = None - def test_set_policy_attrs_when_calledl(self): - testattrdict = { attr: None for attr in self.policy_defaults } - for policyclass in self.policies: + def test_set_policy_attrs_when_cloned(self): + # None of the attributes has a default value of None, so we set them + # all to None in the clone call and check that it worked. + for policyclass, defaults in self.policies.items(): + testattrdict = {attr: None for attr in defaults} policy = policyclass.clone(**testattrdict) - for attr in self.policy_defaults: + for attr in defaults: self.assertIsNone(getattr(policy, attr)) def test_reject_non_policy_keyword_when_called(self): @@ -105,7 +128,7 @@ class PolicyAPITests(unittest.TestCase): self.defects = [] obj = Dummy() defect = object() - policy = email.policy.Compat32() + policy = email.policy.EmailPolicy() policy.register_defect(obj, defect) self.assertEqual(obj.defects, [defect]) defect2 = object() @@ -134,7 +157,7 @@ class PolicyAPITests(unittest.TestCase): email.policy.default.handle_defect(foo, defect2) self.assertEqual(foo.defects, [defect1, defect2]) - class MyPolicy(email.policy.Compat32): + class MyPolicy(email.policy.EmailPolicy): defects = None def __init__(self, *args, **kw): super().__init__(*args, defects=[], **kw) @@ -159,6 +182,49 @@ class PolicyAPITests(unittest.TestCase): self.assertEqual(my_policy.defects, [defect1, defect2]) self.assertEqual(foo.defects, []) + def test_default_header_factory(self): + h = email.policy.default.header_factory('Test', 'test') + self.assertEqual(h.name, 'Test') + self.assertIsInstance(h, _headerregistry.UnstructuredHeader) + self.assertIsInstance(h, _headerregistry.BaseHeader) + + class Foo: + parse = _headerregistry.UnstructuredHeader.parse + + def test_each_Policy_gets_unique_factory(self): + policy1 = email.policy.EmailPolicy() + policy2 = email.policy.EmailPolicy() + policy1.header_factory.map_to_type('foo', self.Foo) + h = policy1.header_factory('foo', 'test') + self.assertIsInstance(h, self.Foo) + self.assertNotIsInstance(h, _headerregistry.UnstructuredHeader) + h = policy2.header_factory('foo', 'test') + self.assertNotIsInstance(h, self.Foo) + self.assertIsInstance(h, _headerregistry.UnstructuredHeader) + + def test_clone_copies_factory(self): + policy1 = email.policy.EmailPolicy() + policy2 = policy1.clone() + policy1.header_factory.map_to_type('foo', self.Foo) + h = policy1.header_factory('foo', 'test') + self.assertIsInstance(h, self.Foo) + h = policy2.header_factory('foo', 'test') + self.assertIsInstance(h, self.Foo) + + def test_new_factory_overrides_default(self): + mypolicy = email.policy.EmailPolicy() + myfactory = mypolicy.header_factory + newpolicy = mypolicy + email.policy.strict + self.assertEqual(newpolicy.header_factory, myfactory) + newpolicy = email.policy.strict + mypolicy + self.assertEqual(newpolicy.header_factory, myfactory) + + def test_adding_default_policies_preserves_default_factory(self): + newpolicy = email.policy.default + email.policy.strict + self.assertEqual(newpolicy.header_factory, + email.policy.EmailPolicy.header_factory) + self.assertEqual(newpolicy.__dict__, {'raise_on_defect': True}) + # XXX: Need subclassing tests. # For adding subclassed objects, make sure the usual rules apply (subclass # wins), but that the order still works (right overrides left). |