cpython: 6d12285e250b (original) (raw)

new file mode 100644 --- /dev/null +++ b/Doc/library/email.contentmanager.rst @@ -0,0 +1,427 @@ +:mod:email.contentmanager: Managing MIME Content +-------------------------------------------------- + +.. module:: email.contentmanager

:synopsis: Storing and Retrieving Content from MIME Parts +

+.. moduleauthor:: R. David Murray rdmurray@bitdance.com +.. sectionauthor:: R. David Murray rdmurray@bitdance.com + + +.. note:: +

The contentmanager module has been included in the standard library on a
:term:provisional basis <provisional package>. Backwards incompatible
changes (up to and including removal of the module) may occur if deemed
necessary by the core developers. +

+.. versionadded:: 3.4

as a :term:provisional module <provisional package>. +

+The :mod:~email.message module provides a class that can represent an +arbitrary email message. That basic message model has a useful and flexible +API, but it provides only a lower-level API for interacting with the generic +parts of a message (the headers, generic header parameters, and the payload, +which may be a list of sub-parts). This module provides classes and tools +that provide an enhanced and extensible API for dealing with various specific +types of content, including the ability to retrieve the content of the message +as a specialized object type rather than as a simple bytes object. The module +automatically takes care of the RFC-specified MIME details (required headers +and parameters, etc.) for the certain common content types content properties, +and support for additional types can be added by an application using the +extension mechanisms. + +This module defines the eponymous "Content Manager" classes. The base +:class:.ContentManager class defines an API for registering content +management functions which extract data from Message objects or insert data +and headers into Message objects, thus providing a way of converting +between Message objects containing data and other representations of that +data (Python data types, specialized Python objects, external files, etc). The +module also defines one concrete content manager: :data:raw_data_manager +converts between MIME content types and str or bytes data. It also +provides a convenient API for managing the MIME parameters when inserting +content into Message\ s. It also handles inserting and extracting +Message objects when dealing with the message/rfc822 content type. + +Another part of the enhanced interface is subclasses of +:class:~email.message.Message that provide new convenience API functions, +including convenience methods for calling the Content Managers derived from +this module. + +.. note:: +

Although :class:.EmailMessage and :class:.MIMEPart are currently
documented in this module because of the provisional nature of the code, the
implementation lives in the :mod:email.message module. +

+ +.. class:: EmailMessage(policy=default) +

If policy is specified (it must be an instance of a :mod:~email.policy
class) use the rules it specifies to udpate and serialize the representation
of the message. If policy is not set, use the
:class:~email.policy.default policy, which follows the rules of the email
RFCs except for line endings (instead of the RFC mandated \r\n, it uses
the Python standard \n line endings). For more information see the
:mod:~email.policy documentation. +
This class is a subclass of :class:~email.message.Message. It adds
the following methods: +

.. attribute:: is_attachment +

 Set to ``True`` if there is a :mailheader:`Content-Disposition` header[](#l1.78)

 and its (case insensitive) value is ``attachment``, ``False`` otherwise.[](#l1.79)

+ +

.. method:: get_body(preferencelist=('related', 'html', 'plain')) +

 Return the MIME part that is the best candidate to be the "body" of the[](#l1.84)

```
 message.[](#l1.85)
```

 *preferencelist* must be a sequence of strings from the set ``related``,[](#l1.87)

 ``html``, and ``plain``, and indicates the order of preference for the[](#l1.88)

 content type of the part returned.[](#l1.89)

 Start looking for candidate matches with the object on which the[](#l1.91)

 ``get_body`` method is called.[](#l1.92)

 If ``related`` is not included in *preferencelist*, consider the root[](#l1.94)

 part (or subpart of the root part) of any related encountered as a[](#l1.95)

 candidate if the (sub-)part matches a preference.[](#l1.96)

 When encountering a ``multipart/related``, check the ``start`` parameter[](#l1.98)

 and if a part with a matching :mailheader:`Content-ID` is found, consider[](#l1.99)

 only it when looking for candidate matches.  Otherwise consider only the[](#l1.100)

 first (default root) part of the ``multipart/related``.[](#l1.101)

 If a part has a :mailheader:``Content-Disposition`` header, only consider[](#l1.103)

 the part a candidate match if the value of the header is ``inline``.[](#l1.104)

 If none of the candidates matches any of the preferences in[](#l1.106)

 *preferneclist*, return ``None``.[](#l1.107)

 Notes: (1) For most applications the only *preferencelist* combinations[](#l1.109)

 that really make sense are ``('plain',)``, ``('html', 'plain')``, and the[](#l1.110)

 default, ``('related', 'html', 'plain')``.  (2) Because matching starts[](#l1.111)

 with the object on which ``get_body`` is called, calling ``get_body`` on[](#l1.112)

 a ``multipart/related`` will return the object itself unless[](#l1.113)

 *preferencelist* has a non-default value. (3) Messages (or message parts)[](#l1.114)

 that do not specify a :mailheader:`Content-Type` or whose[](#l1.115)

 :mailheader:`Content-Type` header is invalid will be treated as if they[](#l1.116)

 are of type ``text/plain``, which may occasionally cause ``get_body`` to[](#l1.117)

```
 return unexpected results.[](#l1.118)
```

+ +

.. method:: iter_attachments() +

 Return an iterator over all of the parts of the message that are not[](#l1.123)

 candidate "body" parts.  That is, skip the first occurrence of each of[](#l1.124)

 ``text/plain``, ``text/html``, ``multipart/related``, or[](#l1.125)

 ``multipart/alternative`` (unless they are explicitly marked as[](#l1.126)

 attachments via :mailheader:`Content-Disposition: attachment`), and[](#l1.127)

 return all remaining parts.  When applied directly to a[](#l1.128)

 ``multipart/related``, return an iterator over the all the related parts[](#l1.129)

 except the root part (ie: the part pointed to by the ``start`` parameter,[](#l1.130)

 or the first part if there is no ``start`` parameter or the ``start``[](#l1.131)

 parameter doesn't match the :mailheader:`Content-ID` of any of the[](#l1.132)

 parts).  When applied directly to a ``multipart/alternative`` or a[](#l1.133)

 non-``multipart``, return an empty iterator.[](#l1.134)

+ +

.. method:: iter_parts() +

 Return an iterator over all of the immediate sub-parts of the message,[](#l1.139)

 which will be empty for a non-``multipart``.  (See also[](#l1.140)

 :meth:``~email.message.walk``.)[](#l1.141)

+ +

.. method:: get_content(*args, content_manager=None, **kw) +

 Call the ``get_content`` method of the *content_manager*, passing self[](#l1.146)

 as the message object, and passing along any other arguments or keywords[](#l1.147)

 as additional arguments.  If *content_manager* is not specified, use[](#l1.148)

 the ``content_manager`` specified by the current :mod:`~email.policy`.[](#l1.149)

+ +

.. method:: set_content(*args, content_manager=None, **kw) +

 Call the ``set_content`` method of the *content_manager*, passing self[](#l1.154)

 as the message object, and passing along any other arguments or keywords[](#l1.155)

 as additional arguments.  If *content_manager* is not specified, use[](#l1.156)

 the ``content_manager`` specified by the current :mod:`~email.policy`.[](#l1.157)

+ +

.. method:: make_related(boundary=None) +

 Convert a non-``multipart`` message into a ``multipart/related`` message,[](#l1.162)

 moving any existing :mailheader:`Content-` headers and payload into a[](#l1.163)

 (new) first part of the ``multipart``.  If *boundary* is specified, use[](#l1.164)

 it as the boundary string in the multipart, otherwise leave the boundary[](#l1.165)

 to be automatically created when it is needed (for example, when the[](#l1.166)

```
 message is serialized).[](#l1.167)
```

+ +

.. method:: make_alternative(boundary=None) +

 Convert a non-``multipart`` or a ``multipart/related`` into a[](#l1.172)

 ``multipart/alternative``, moving any existing :mailheader:`Content-`[](#l1.173)

 headers and payload into a (new) first part of the ``multipart``.  If[](#l1.174)

 *boundary* is specified, use it as the boundary string in the multipart,[](#l1.175)

 otherwise leave the boundary to be automatically created when it is[](#l1.176)

 needed (for example, when the message is serialized).[](#l1.177)

+ +

.. method:: make_mixed(boundary=None) +

 Convert a non-``multipart``, a ``multipart/related``, or a[](#l1.182)

 ``multipart-alternative`` into a ``multipart/mixed``, moving any existing[](#l1.183)

 :mailheader:`Content-` headers and payload into a (new) first part of the[](#l1.184)

 ``multipart``.  If *boundary* is specified, use it as the boundary string[](#l1.185)

 in the multipart, otherwise leave the boundary to be automatically[](#l1.186)

 created when it is needed (for example, when the message is serialized).[](#l1.187)

+ +

.. method:: add_related(*args, content_manager=None, **kw) +

 If the message is a ``multipart/related``, create a new message[](#l1.192)

 object, pass all of the arguments to its :meth:`set_content` method,[](#l1.193)

 and :meth:`~email.message.Message.attach` it to the ``multipart``.  If[](#l1.194)

 the message is a non-``multipart``, call :meth:`make_related` and then[](#l1.195)

 proceed as above.  If the message is any other type of ``multipart``,[](#l1.196)

 raise a :exc:`TypeError`. If *content_manager* is not specified, use[](#l1.197)

 the ``content_manager`` specified by the current :mod:`~email.policy`.[](#l1.198)

 If the added part has no :mailheader:`Content-Disposition` header,[](#l1.199)

 add one with the value ``inline``.[](#l1.200)

+ +

.. method:: add_alternative(*args, content_manager=None, **kw) +

 If the message is a ``multipart/alternative``, create a new message[](#l1.205)

 object, pass all of the arguments to its :meth:`set_content` method, and[](#l1.206)

 :meth:`~email.message.Message.attach` it to the ``multipart``.  If the[](#l1.207)

 message is a non-``multipart`` or ``multipart/related``, call[](#l1.208)

 :meth:`make_alternative` and then proceed as above.  If the message is[](#l1.209)

 any other type of ``multipart``, raise a :exc:`TypeError`. If[](#l1.210)

 *content_manager* is not specified, use the ``content_manager`` specified[](#l1.211)

 by the current :mod:`~email.policy`.[](#l1.212)

+ +

.. method:: add_attachment(*args, content_manager=None, **kw) +

 If the message is a ``multipart/mixed``, create a new message object,[](#l1.217)

 pass all of the arguments to its :meth:`set_content` method, and[](#l1.218)

 :meth:`~email.message.Message.attach` it to the ``multipart``.  If the[](#l1.219)

 message is a non-``multipart``, ``multipart/related``, or[](#l1.220)

 ``multipart/alternative``, call :meth:`make_mixed` and then proceed as[](#l1.221)

 above. If *content_manager* is not specified, use the ``content_manager``[](#l1.222)

 specified by the current :mod:`~email.policy`.  If the added part[](#l1.223)

 has no :mailheader:`Content-Disposition` header, add one with the value[](#l1.224)

 ``attachment``.  This method can be used both for explicit attachments[](#l1.225)

 (:mailheader:`Content-Disposition: attachment` and ``inline`` attachments[](#l1.226)

 (:mailheader:`Content-Disposition: inline`), by passing appropriate[](#l1.227)

 options to the ``content_manager``.[](#l1.228)

+ +

.. method:: clear() +

 Remove the payload and all of the headers.[](#l1.233)

+ +

.. method:: clear_content() +

 Remove the payload and all of the :exc:`Content-` headers, leaving[](#l1.238)

 all other headers intact and in their original order.[](#l1.239)

+ + +.. class:: ContentManager() +

Base class for content managers. Provides the standard registry mechanisms
to register converters between MIME content and other representations, as
well as the get_content and set_content dispatch methods. +

.. method:: get_content(msg, *args, **kw) +

 Look up a handler function based on the ``mimetype`` of *msg* (see next[](#l1.251)

 paragraph), call it, passing through all arguments, and return the result[](#l1.252)

 of the call.  The expectation is that the handler will extract the[](#l1.253)

 payload from *msg* and return an object that encodes information about[](#l1.254)

```
 the extracted data.[](#l1.255)
```

 To find the handler, look for the following keys in the registry,[](#l1.257)

 stopping with the first one found:[](#l1.258)

       * the string representing the full MIME type (``maintype/subtype``)[](#l1.260)

       * the string representing the ``maintype``[](#l1.261)

```
       * the empty string[](#l1.262)
```

 If none of these keys produce a handler, raise a :exc:`KeyError` for the[](#l1.264)

```
 full MIME type.[](#l1.265)
```

+ +

.. method:: set_content(msg, obj, *args, **kw) +

 If the ``maintype`` is ``multipart``, raise a :exc:`TypeError`; otherwise[](#l1.270)

 look up a handler function based on the type of *obj* (see next[](#l1.271)

 paragraph), call :meth:`~email.message.EmailMessage.clear_content` on the[](#l1.272)

 *msg*, and call the handler function, passing through all arguments.  The[](#l1.273)

 expectation is that the handler will transform and store *obj* into[](#l1.274)

 *msg*, possibly making other changes to *msg* as well, such as adding[](#l1.275)

 various MIME headers to encode information needed to interpret the stored[](#l1.276)

```
 data.[](#l1.277)
```

 To find the handler, obtain the type of *obj* (``typ = type(obj)``), and[](#l1.279)

 look for the following keys in the registry, stopping with the first one[](#l1.280)

```
 found:[](#l1.281)
```

      * the type itself (``typ``)[](#l1.283)

      * the type's fully qualified name (``typ.__module__ + '.' +[](#l1.284)

```
        typ.__qualname__``).[](#l1.285)
```

      * the type's qualname (``typ.__qualname__``)[](#l1.286)

      * the type's name (``typ.__name__``).[](#l1.287)

 If none of the above match, repeat all of the checks above for each of[](#l1.289)

 the types in the :term:`MRO` (``typ.__mro__``).  Finally, if no other key[](#l1.290)

 yields a handler, check for a handler for the key ``None``.  If there is[](#l1.291)

 no handler for ``None``, raise a :exc:`KeyError` for the fully[](#l1.292)

```
 qualified name of the type.[](#l1.293)
```

 Also add a :mailheader:`MIME-Version` header if one is not present (see[](#l1.295)

```
 also :class:`.MIMEPart`).[](#l1.296)
```

+ +

.. method:: add_get_handler(key, handler) +

 Record the function *handler* as the handler for *key*.  For the possible[](#l1.301)

 values of *key*, see :meth:`get_content`.[](#l1.302)

+ +

.. method:: add_set_handler(typekey, handler) +

 Record *handler* as the function to call when an object of a type[](#l1.307)

 matching *typekey* is passed to :meth:`set_content`.  For the possible[](#l1.308)

 values of *typekey*, see :meth:`set_content`.[](#l1.309)

+ + +.. class:: MIMEPart(policy=default) +

This class represents a subpart of a MIME message. It is identical to
:class:EmailMessage, except that no :mailheader:MIME-Version headers are
added when :meth:~EmailMessage.set_content is called, since sub-parts do
not need their own :mailheader:MIME-Version headers.

+ + +Content Manager Instances +~~~~~~~~~~~~~~~~~~~~~~~~~ + +Currently the email package provides only one concrete content manager, +:data:raw_data_manager, although more may be added in the future. +:data:raw_data_manager is the +:attr:~email.policy.EmailPolicy.content_manager provided by +:attr:~email.policy.EmailPolicy and its derivatives. + + +.. data:: raw_data_manager +

This content manager provides only a minimum interface beyond that provided
by :class:~email.message.Message itself: it deals only with text, raw
byte strings, and :class:~email.message.Message objects. Nevertheless, it
provides significant advantages compared to the base API: get_content on
a text part will return a unicode string without the application needing to
manually decode it, set_content provides a rich set of options for
controlling the headers added to a part and controlling the content transfer
encoding, and it enables the use of the various add_ methods, thereby
simplifying the creation of multipart messages. +
.. method:: get_content(msg, errors='replace') +

 Return the payload of the part as either a string (for ``text`` parts), a[](#l1.344)

 :class:`~email.message.EmailMessage` object (for ``message/rfc822``[](#l1.345)

 parts), or a ``bytes`` object (for all other non-multipart types).  Raise[](#l1.346)

 a :exc:`KeyError` if called on a ``multipart``.  If the part is a[](#l1.347)

 ``text`` part and *errors* is specified, use it as the error handler when[](#l1.348)

 decoding the payload to unicode.  The default error handler is[](#l1.349)

```
 ``replace``.[](#l1.350)
```

.. method:: set_content(msg, <'str'>, subtype="plain", charset='utf-8' [](#l1.352)

                      cte=None, \[](#l1.353)

                      disposition=None, filename=None, cid=None, \[](#l1.354)

                      params=None, headers=None)[](#l1.355)

          set_content(msg, <'bytes'>, maintype, subtype, cte="base64", \[](#l1.356)

                      disposition=None, filename=None, cid=None, \[](#l1.357)

                      params=None, headers=None)[](#l1.358)

          set_content(msg, <'Message'>, cte=None, \[](#l1.359)

                      disposition=None, filename=None, cid=None, \[](#l1.360)

                      params=None, headers=None)[](#l1.361)

          set_content(msg, <'list'>, subtype='mixed', \[](#l1.362)

                      disposition=None, filename=None, cid=None, \[](#l1.363)

                      params=None, headers=None)[](#l1.364)

  Add headers and payload to *msg*:[](#l1.366)

  Add a :mailheader:`Content-Type` header with a ``maintype/subtype``[](#l1.368)

```
  value.[](#l1.369)
```

      * For ``str``, set the MIME ``maintype`` to ``text``, and set the[](#l1.371)

        subtype to *subtype* if it is specified, or ``plain`` if it is not.[](#l1.372)

      * For ``bytes``, use the specified *maintype* and *subtype*, or[](#l1.373)

        raise a :exc:`TypeError` if they are not specified.[](#l1.374)

      * For :class:`~email.message.Message` objects, set the maintype to[](#l1.375)

        ``message``, and set the subtype to *subtype* if it is specified[](#l1.376)

        or ``rfc822`` if it is not.  If *subtype* is ``partial``, raise an[](#l1.377)

        error (``bytes`` objects must be used to construct[](#l1.378)

        ``message/partial`` parts).[](#l1.379)

      * For *<'list'>*, which should be a list of[](#l1.380)

        :class:`~email.message.Message` objects, set the ``maintype`` to[](#l1.381)

        ``multipart``, and the ``subtype`` to *subtype* if it is[](#l1.382)

        specified, and ``mixed`` if it is not.  If the message parts in[](#l1.383)

        the *<'list'>* have :mailheader:`MIME-Version` headers, remove[](#l1.384)

```
        them.[](#l1.385)
```

  If *charset* is provided (which is valid only for ``str``), encode the[](#l1.387)

  string to bytes using the specified character set.  The default is[](#l1.388)

  ``utf-8``.  If the specified *charset* is a known alias for a standard[](#l1.389)

  MIME charset name, use the standard charset instead.[](#l1.390)

  If *cte* is set, encode the payload using the specified content transfer[](#l1.392)

  encoding, and set the :mailheader:`Content-Transfer-Endcoding` header to[](#l1.393)

  that value.  For ``str`` objects, if it is not set use heuristics to[](#l1.394)

  determine the most compact encoding.  Possible values for *cte* are[](#l1.395)

  ``quoted-printable``, ``base64``, ``7bit``, ``8bit``, and ``binary``.[](#l1.396)

  If the input cannot be encoded in the specified encoding (eg: ``7bit``),[](#l1.397)

  raise a :exc:`ValueError`.  For :class:`~email.message.Message`, per[](#l1.398)

  :rfc:`2046`, raise an error if a *cte* of ``quoted-printable`` or[](#l1.399)

  ``base64`` is requested for *subtype* ``rfc822``, and for any *cte*[](#l1.400)

  other than ``7bit`` for *subtype* ``external-body``.  For[](#l1.401)

  ``message/rfc822``, use ``8bit`` if *cte* is not specified.  For all[](#l1.402)

  other values of *subtype*, use ``7bit``.[](#l1.403)

  .. note:: A *cte* of ``binary`` does not actually work correctly yet.[](#l1.405)

     The ``Message`` object as modified by ``set_content`` is correct, but[](#l1.406)

     :class:`~email.generator.BytesGenerator` does not serialize it[](#l1.407)

```
     correctly.[](#l1.408)
```

  If *disposition* is set, use it as the value of the[](#l1.410)

  :mailheader:`Content-Disposition` header.  If not specified, and[](#l1.411)

  *filename* is specified, add the header with the value ``attachment``.[](#l1.412)

  If it is not specified and *filename* is also not specified, do not add[](#l1.413)

  the header.  The only valid values for *disposition* are ``attachment``[](#l1.414)

```
  and ``inline``.[](#l1.415)
```

  If *filename* is specified, use it as the value of the ``filename``[](#l1.417)

  parameter of the :mailheader:`Content-Disposition` header.  There is no[](#l1.418)

```
  default.[](#l1.419)
```

  If *cid* is specified, add a :mailheader:`Content-ID` header with[](#l1.421)

```
  *cid* as its value.[](#l1.422)
```

  If *params* is specified, iterate its ``items`` method and use the[](#l1.424)

  resulting ``(key, value)`` pairs to set additional paramters on the[](#l1.425)

  :mailheader:`Content-Type` header.[](#l1.426)

  If *headers* is specified and is a list of strings of the form[](#l1.428)

  ``headername: headervalue`` or a list of ``header`` objects[](#l1.429)

  (distinguised from strings by having a ``name`` attribute), add the[](#l1.430)

```
  headers to *msg*.[](#l1.431)
```

--- a/Doc/library/email.message.rst +++ b/Doc/library/email.message.rst @@ -33,10 +33,11 @@ Here are the methods of the :class:`Mess .. class:: Message(policy=compat32)

The policy argument determiens the :mod:~email.policy that will be used
to update the message model. The default value, :class:`compat32
<email.policy.Compat32>` maintains backward compatibility with the
Python 3.2 version of the email package. For more information see the

If policy is specified (it must be an instance of a :mod:~email.policy
class) use the rules it specifies to udpate and serialize the representation
of the message. If policy is not set, use the :class`compat32
<email.policy.Compat32>` policy, which maintains backward compatibility with
the Python 3.2 version of the email package. For more information see the :mod:~email.policy documentation. .. versionchanged:: 3.3 The policy keyword argument was added. @@ -465,7 +466,8 @@ Here are the methods of the :class:`Mess to False.

.. method:: set_param(param, value, header='Content-Type', requote=True, charset=None, language='')

.. method:: set_param(param, value, header='Content-Type', requote=True,

                    charset=None, language='', replace=False)[](#l2.25)

Set a parameter in the :mailheader:Content-Type header. If the parameter already exists in the header, its value will be replaced with @@ -482,6 +484,12 @@ Here are the methods of the :class:`Mess language, defaulting to the empty string. Both charset and language should be strings.

 If *replace* is ``False`` (the default) the header is moved to the[](#l2.33)

 end of the list of headers.  If *replace* is ``True``, the header[](#l2.34)

```
 will be updated in place.[](#l2.35)
```

 .. versionchanged: 3.4 ``replace`` keyword was added.[](#l2.37)

+ .. method:: del_param(param, header='content-type', requote=True)

--- a/Doc/library/email.policy.rst +++ b/Doc/library/email.policy.rst @@ -371,7 +371,7 @@ added matters. To illustrate:: 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

simple strings, headers are str subclasses with attributes that depend on the type of the field. The parsing and folding algorithm fully implement :rfc:2047 and :rfc:5322.

@@ -408,6 +408,20 @@ added matters. To illustrate:: fields are treated as unstructured. This list will be completed before the extension is marked stable.)

.. attribute:: content_manager +

 An object with at least two methods: get_content and set_content.  When[](#l3.18)

 the :meth:`~email.message.Message.get_content` or[](#l3.19)

 :meth:`~email.message.Message.set_content` method of a[](#l3.20)

 :class:`~email.message.Message` object is called, it calls the[](#l3.21)

 corresponding method of this object, passing it the message object as its[](#l3.22)

 first argument, and any arguments or keywords that were passed to it as[](#l3.23)

 additional arguments.  By default ``content_manager`` is set to[](#l3.24)

 :data:`~email.contentmanager.raw_data_manager`.[](#l3.25)

```
 .. versionadded 3.4[](#l3.27)
```

+ + The class provides the following concrete implementations of the abstract methods of :class:Policy: @@ -427,7 +441,7 @@ added matters. To illustrate:: 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[](#l3.37)

 ``header_factory``, and the resulting header object is returned as[](#l3.38)
 the value.  In this case a ``ValueError`` is raised if the input value[](#l3.39)
 contains CR or LF characters.[](#l3.40)

@@ -435,7 +449,7 @@ added matters. To illustrate:: 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[](#l3.46)

 removed, are passed to the ``header_factory``, and the resulting[](#l3.47)
 header object is returned.  Any surrogateescaped bytes get turned into[](#l3.48)
 the unicode unknown-character glyph.[](#l3.49)

@@ -445,9 +459,9 @@ added matters. To illustrate:: 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[](#l3.55)

 according to the policy, it is converted into a header object by[](#l3.56)
 passing the *name* and the *value* with any CR and LF characters removed[](#l3.57)

 to the ``header_factory``.  Folding of a custom header object is done by[](#l3.58)

 to the ``header_factory``.  Folding of a header object is done by[](#l3.59)
 calling its ``fold`` method with the current policy.[](#l3.60)

Source values are split into lines using :meth:~str.splitlines. If @@ -502,23 +516,23 @@ With all of these :class:EmailPolicies [](#l3.63) the email package is changed from the Python 3.2 API in the following ways:[](#l3.64) [](#l3.65) * Setting a header on a :class:~email.message.Message` results in that

header being parsed and a custom header object created.[](#l3.67)

header being parsed and a header object created.[](#l3.68)

* Fetching a header value from a :class:~email.message.Message results

in that header being parsed and a custom header object created and[](#l3.71)

in that header being parsed and a header object created and[](#l3.72)
returned.[](#l3.73)

- Any custom header object, or any header that is refolded due to the

- Any 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 +:class:~email.message.Message is a header object with extra 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 in +The header objects and their attributes are described in :mod:~email.headerregistry.

--- a/Doc/library/email.rst +++ b/Doc/library/email.rst @@ -53,6 +53,7 @@ Contents of the :mod:email package doc email.generator.rst email.policy.rst email.headerregistry.rst

email.contentmanager.rst email.mime.rst email.header.rst email.charset.rst

--- a/Doc/whatsnew/3.4.rst +++ b/Doc/whatsnew/3.4.rst @@ -280,6 +280,21 @@ result: a bytes object containing the f (Contributed by R. David Murray in :issue:18600.) +A pair of new subclasses of :class:~email.message.Message have been added, +along with a new sub-module, :mod:~email.contentmanager. All documentation +is currently in the new module, which is being added as part of the new +:term:provisional <provosional package> email API. These classes provide a +number of new methods that make extracting content from and inserting content +into email messages much easier. See the :mod:~email.contentmanager +documentation for details. + +These API additions complete the bulk of the work that was planned as part of +the email6 project. The currently provisional API is scheduled to become final +in Python 3.5 (possibly with a few minor additions in the area of error +handling). + +(Contributed by R. David Murray in :issue:18891.) + functools ---------

new file mode 100644 --- /dev/null +++ b/Lib/email/contentmanager.py @@ -0,0 +1,249 @@ +import binascii +import email.charset +import email.message +import email.errors +from email import quoprimime + +class ContentManager: +

def init(self):
```
   self.get_handlers = {}[](#l6.14)
```
```
   self.set_handlers = {}[](#l6.15)
```

def add_get_handler(self, key, handler):

   self.get_handlers[key] = handler[](#l6.18)

def get_content(self, msg, *args, **kw):

   content_type = msg.get_content_type()[](#l6.21)

   if content_type in self.get_handlers:[](#l6.22)

       return self.get_handlers[content_type](msg, *args, **kw)[](#l6.23)

   maintype = msg.get_content_maintype()[](#l6.24)

   if maintype in self.get_handlers:[](#l6.25)

       return self.get_handlers[maintype](msg, *args, **kw)[](#l6.26)

   if '' in self.get_handlers:[](#l6.27)

       return self.get_handlers[''](msg, *args, **kw)[](#l6.28)

   raise KeyError(content_type)[](#l6.29)

def add_set_handler(self, typekey, handler):

   self.set_handlers[typekey] = handler[](#l6.32)

def set_content(self, msg, obj, *args, **kw):

   if msg.get_content_maintype() == 'multipart':[](#l6.35)

       # XXX: is this error a good idea or not?  We can remove it later,[](#l6.36)

       # but we can't add it later, so do it for now.[](#l6.37)

       raise TypeError("set_content not valid on multipart")[](#l6.38)

   handler = self._find_set_handler(msg, obj)[](#l6.39)

```
   msg.clear_content()[](#l6.40)
```

   handler(msg, obj, *args, **kw)[](#l6.41)

def _find_set_handler(self, msg, obj):
```
   full_path_for_error = None[](#l6.44)
```

   for typ in type(obj).__mro__:[](#l6.45)

       if typ in self.set_handlers:[](#l6.46)

           return self.set_handlers[typ][](#l6.47)

       qname = typ.__qualname__[](#l6.48)

       modname = getattr(typ, '__module__', '')[](#l6.49)

       full_path = '.'.join((modname, qname)) if modname else qname[](#l6.50)

       if full_path_for_error is None:[](#l6.51)

           full_path_for_error = full_path[](#l6.52)

       if full_path in self.set_handlers:[](#l6.53)

           return self.set_handlers[full_path][](#l6.54)

       if qname in self.set_handlers:[](#l6.55)

           return self.set_handlers[qname][](#l6.56)

```
       name = typ.__name__[](#l6.57)
```

       if name in self.set_handlers:[](#l6.58)

           return self.set_handlers[name][](#l6.59)

   if None in self.set_handlers:[](#l6.60)

       return self.set_handlers[None][](#l6.61)

   raise KeyError(full_path_for_error)[](#l6.62)

+ + +raw_data_manager = ContentManager() + + +def get_text_content(msg, errors='replace'):

content = msg.get_payload(decode=True)
charset = msg.get_param('charset', 'ASCII')
return content.decode(charset, errors=errors)

+raw_data_manager.add_get_handler('text', get_text_content) + + +def get_non_text_content(msg):

return msg.get_payload(decode=True)

+for maintype in 'audio image video application'.split():

raw_data_manager.add_get_handler(maintype, get_non_text_content)

+ + +def get_message_content(msg):

return msg.get_payload(0)

+for subtype in 'rfc822 external-body'.split():

raw_data_manager.add_get_handler('message/'+subtype, get_message_content)

+ + +def get_and_fixup_unknown_message_content(msg):

If we don't understand a message subtype, we are supposed to treat it as
if it were application/octet-stream, per
tools.ietf.org/html/rfc2046#section-5.2.4. Feedparser doesn't do that,
so do our best to fix things up. Note that it is not appropriate to
model message/partial content as Message objects, so they are handled
here as well. (How to reassemble them is out of scope for this comment :)
return bytes(msg.get_payload(0))

+raw_data_manager.add_get_handler('message',

                            get_and_fixup_unknown_message_content)[](#l6.96)

+ + +def _prepare_set(msg, maintype, subtype, headers):

msg['Content-Type'] = '/'.join((maintype, subtype))
if headers:

   if not hasattr(headers[0], 'name'):[](#l6.102)

```
       mp = msg.policy[](#l6.103)
```

       headers = [mp.header_factory(*mp.header_source_parse([header]))[](#l6.104)

                  for header in headers][](#l6.105)

```
   try:[](#l6.106)
```

       for header in headers:[](#l6.107)

           if header.defects:[](#l6.108)

               raise header.defects[0][](#l6.109)

           msg[header.name] = header[](#l6.110)

   except email.errors.HeaderDefect as exc:[](#l6.111)

       raise ValueError("Invalid header: {}".format([](#l6.112)

                           header.fold(policy=msg.policy))) from exc[](#l6.113)

+ + +def _finalize_set(msg, disposition, filename, cid, params):

if disposition is None and filename is not None:

   disposition = 'attachment'[](#l6.118)

if disposition is not None:

   msg['Content-Disposition'] = disposition[](#l6.120)

if filename is not None:
```
   msg.set_param('filename',[](#l6.122)
```
```
                 filename,[](#l6.123)
```

                 header='Content-Disposition',[](#l6.124)

                 replace=True)[](#l6.125)

if cid is not None:
```
   msg['Content-ID'] = cid[](#l6.127)
```
if params is not None:

   for key, value in params.items():[](#l6.129)

       msg.set_param(key, value)[](#l6.130)

+ + +# XXX: This is a cleaned-up version of base64mime.body_encode. It would +# be nice to drop both this and quoprimime.body_encode in favor of +# enhanced binascii routines that accepted a max_line_length parameter. +def _encode_base64(data, max_line_length):

encoded_lines = []
unencoded_bytes_per_line = max_line_length * 3 // 4
for i in range(0, len(data), unencoded_bytes_per_line):

   thisline = data[i:i+unencoded_bytes_per_line][](#l6.140)

   encoded_lines.append(binascii.b2a_base64(thisline).decode('ascii'))[](#l6.141)

return ''.join(encoded_lines)

+ + +def _encode_text(string, charset, cte, policy):

lines = string.encode(charset).splitlines()
linesep = policy.linesep.encode('ascii')
def embeded_body(lines): return linesep.join(lines) + linesep
def normal_body(lines): return b'\n'.join(lines) + b'\n'
if cte==None:

   # Use heuristics to decide on the "best" encoding.[](#l6.151)

```
   try:[](#l6.152)
```

       return '7bit', normal_body(lines).decode('ascii')[](#l6.153)

   except UnicodeDecodeError:[](#l6.154)

```
       pass[](#l6.155)
```

   if (policy.cte_type == '8bit' and[](#l6.156)

           max(len(x) for x in lines) <= policy.max_line_length):[](#l6.157)

       return '8bit', normal_body(lines).decode('ascii', 'surrogateescape')[](#l6.158)

   sniff = embeded_body(lines[:10])[](#l6.159)

   sniff_qp = quoprimime.body_encode(sniff.decode('latin-1'),[](#l6.160)

                                     policy.max_line_length)[](#l6.161)

   sniff_base64 = binascii.b2a_base64(sniff)[](#l6.162)

   # This is a little unfair to qp; it includes lineseps, base64 doesn't.[](#l6.163)

   if len(sniff_qp) > len(sniff_base64):[](#l6.164)

```
       cte = 'base64'[](#l6.165)
```
```
   else:[](#l6.166)
```

       cte = 'quoted-printable'[](#l6.167)

```
       if len(lines) <= 10:[](#l6.168)
```

           return cte, sniff_qp[](#l6.169)

if cte == '7bit':

   data = normal_body(lines).decode('ascii')[](#l6.171)

elif cte == '8bit':

   data = normal_body(lines).decode('ascii', 'surrogateescape')[](#l6.173)

elif cte == 'quoted-printable':

   data = quoprimime.body_encode(normal_body(lines).decode('latin-1'),[](#l6.175)

                                 policy.max_line_length)[](#l6.176)

elif cte == 'base64':

   data = _encode_base64(embeded_body(lines), policy.max_line_length)[](#l6.178)

else:

   raise ValueError("Unknown content transfer encoding {}".format(cte))[](#l6.180)

return cte, data

+ + +def set_text_content(msg, string, subtype="plain", charset='utf-8', cte=None,

                disposition=None, filename=None, cid=None,[](#l6.185)

                params=None, headers=None):[](#l6.186)

_prepare_set(msg, 'text', subtype, headers)
cte, payload = _encode_text(string, charset, cte, msg.policy)
msg.set_payload(payload)
msg.set_param('charset',

             email.charset.ALIASES.get(charset, charset),[](#l6.191)

```
             replace=True)[](#l6.192)
```
msg['Content-Transfer-Encoding'] = cte
_finalize_set(msg, disposition, filename, cid, params)

+raw_data_manager.add_set_handler(str, set_text_content) + + +def set_message_content(msg, message, subtype="rfc822", cte=None,

                  disposition=None, filename=None, cid=None,[](#l6.199)

                  params=None, headers=None):[](#l6.200)

if subtype == 'partial':

   raise ValueError("message/partial is not supported for Message objects")[](#l6.202)

if subtype == 'rfc822':

   if cte not in (None, '7bit', '8bit', 'binary'):[](#l6.204)

       # http://tools.ietf.org/html/rfc2046#section-5.2.1 mandate.[](#l6.205)

```
       raise ValueError([](#l6.206)
```

           "message/rfc822 parts do not support cte={}".format(cte))[](#l6.207)

   # 8bit will get coerced on serialization if policy.cte_type='7bit'.  We[](#l6.208)

   # may end up claiming 8bit when it isn't needed, but the only negative[](#l6.209)

   # result of that should be a gateway that needs to coerce to 7bit[](#l6.210)

   # having to look through the whole embedded message to discover whether[](#l6.211)

   # or not it actually has to do anything.[](#l6.212)

   cte = '8bit' if cte is None else cte[](#l6.213)

elif subtype == 'external-body':

   if cte not in (None, '7bit'):[](#l6.215)

       # http://tools.ietf.org/html/rfc2046#section-5.2.3 mandate.[](#l6.216)

```
       raise ValueError([](#l6.217)
```

           "message/external-body parts do not support cte={}".format(cte))[](#l6.218)

```
   cte = '7bit'[](#l6.219)
```
elif cte is None:

   # http://tools.ietf.org/html/rfc2046#section-5.2.4 says all future[](#l6.221)

   # subtypes should be restricted to 7bit, so assume that.[](#l6.222)

```
   cte = '7bit'[](#l6.223)
```
_prepare_set(msg, 'message', subtype, headers)
msg.set_payload([message])
msg['Content-Transfer-Encoding'] = cte
_finalize_set(msg, disposition, filename, cid, params)

+raw_data_manager.add_set_handler(email.message.Message, set_message_content) + + +def set_bytes_content(msg, data, maintype, subtype, cte='base64',

                disposition=None, filename=None, cid=None,[](#l6.232)

                params=None, headers=None):[](#l6.233)

_prepare_set(msg, maintype, subtype, headers)
if cte == 'base64':

   data = _encode_base64(data, max_line_length=msg.policy.max_line_length)[](#l6.236)

elif cte == 'quoted-printable':

   # XXX: quoprimime.body_encode won't encode newline characters in data,[](#l6.238)

   # so we can't use it.  This means max_line_length is ignored.  Another[](#l6.239)

   # bug to fix later.  (Note: encoders.quopri is broken on line ends.)[](#l6.240)

   data = binascii.b2a_qp(data, istext=False, header=False, quotetabs=True)[](#l6.241)

   data = data.decode('ascii')[](#l6.242)

elif cte == '7bit':

   # Make sure it really is only ASCII.  The early warning here seems[](#l6.244)

   # worth the overhead...if you care write your own content manager :).[](#l6.245)

```
   data.encode('ascii')[](#l6.246)
```
elif cte in ('8bit', 'binary'):

   data = data.decode('ascii', 'surrogateescape')[](#l6.248)

msg.set_payload(data)
msg['Content-Transfer-Encoding'] = cte
_finalize_set(msg, disposition, filename, cid, params)

+for typ in (bytes, bytearray, memoryview):

raw_data_manager.add_set_handler(typ, set_bytes_content)

--- a/Lib/email/message.py +++ b/Lib/email/message.py @@ -8,8 +8,6 @@ import re import uu -import base64 -import binascii from io import BytesIO, StringIO

Intrapackage imports

@@ -679,7 +677,7 @@ class Message: return failobj def set_param(self, param, value, header='Content-Type', requote=True,

             charset=None, language=''):[](#l7.16)

             charset=None, language='', replace=False):[](#l7.17)
   """Set a parameter in the Content-Type header.[](#l7.18)

If the parameter already exists in the header, its value will be @@ -723,8 +721,11 @@ class Message: else: ctype = SEMISPACE.join([ctype, append_param]) if ctype != self.get(header):

```
       del self[header][](#l7.25)
```
```
       self[header] = ctype[](#l7.26)
```

```
       if replace:[](#l7.27)
```

           self.replace_header(header, ctype)[](#l7.28)

```
       else:[](#l7.29)
```
```
           del self[header][](#l7.30)
```

           self[header] = ctype[](#l7.31)

def del_param(self, param, header='content-type', requote=True): """Remove the given parameter completely from the Content-Type header. @@ -905,3 +906,208 @@ class Message: # I.e. def walk(self): ... from email.iterators import walk + + +class MIMEPart(Message): +

def init(self, policy=None):
```
   if policy is None:[](#l7.44)
```

       from email.policy import default[](#l7.45)

```
       policy = default[](#l7.46)
```

   Message.__init__(self, policy)[](#l7.47)

@property
def is_attachment(self):

   c_d = self.get('content-disposition')[](#l7.51)

```
   if c_d is None:[](#l7.52)
```
```
       return False[](#l7.53)
```

   return c_d.lower() == 'attachment'[](#l7.54)

def _find_body(self, part, preferencelist):
```
   if part.is_attachment:[](#l7.57)
```
```
       return[](#l7.58)
```

   maintype, subtype = part.get_content_type().split('/')[](#l7.59)

```
   if maintype == 'text':[](#l7.60)
```

       if subtype in preferencelist:[](#l7.61)

           yield (preferencelist.index(subtype), part)[](#l7.62)

```
       return[](#l7.63)
```

   if maintype != 'multipart':[](#l7.64)

```
       return[](#l7.65)
```
```
   if subtype != 'related':[](#l7.66)
```

       for subpart in part.iter_parts():[](#l7.67)

           yield from self._find_body(subpart, preferencelist)[](#l7.68)

```
       return[](#l7.69)
```

   if 'related' in preferencelist:[](#l7.70)

       yield (preferencelist.index('related'), part)[](#l7.71)

```
   candidate = None[](#l7.72)
```

   start = part.get_param('start')[](#l7.73)

```
   if start:[](#l7.74)
```

       for subpart in part.iter_parts():[](#l7.75)

           if subpart['content-id'] == start:[](#l7.76)

               candidate = subpart[](#l7.77)

```
               break[](#l7.78)
```
```
   if candidate is None:[](#l7.79)
```

       subparts = part.get_payload()[](#l7.80)

       candidate = subparts[0] if subparts else None[](#l7.81)

```
   if candidate is not None:[](#l7.82)
```

       yield from self._find_body(candidate, preferencelist)[](#l7.83)

def get_body(self, preferencelist=('related', 'html', 'plain')):

   """Return best candidate mime part for display as 'body' of message.[](#l7.86)

   Do a depth first search, starting with self, looking for the first part[](#l7.88)

   matching each of the items in preferencelist, and return the part[](#l7.89)

   corresponding to the first item that has a match, or None if no items[](#l7.90)

   have a match.  If 'related' is not included in preferencelist, consider[](#l7.91)

   the root part of any multipart/related encountered as a candidate[](#l7.92)

   match.  Ignore parts with 'Content-Disposition: attachment'.[](#l7.93)

```
   """[](#l7.94)
```

   best_prio = len(preferencelist)[](#l7.95)

```
   body = None[](#l7.96)
```

   for prio, part in self._find_body(self, preferencelist):[](#l7.97)

```
       if prio < best_prio:[](#l7.98)
```
```
           best_prio = prio[](#l7.99)
```
```
           body = part[](#l7.100)
```
```
           if prio == 0:[](#l7.101)
```
```
               break[](#l7.102)
```
```
   return body[](#l7.103)
```

_body_types = {('text', 'plain'),

              ('text', 'html'),[](#l7.106)

              ('multipart', 'related'),[](#l7.107)

              ('multipart', 'alternative')}[](#l7.108)

def iter_attachments(self):

   """Return an iterator over the non-main parts of a multipart.[](#l7.110)

   Skip the first of each occurrence of text/plain, text/html,[](#l7.112)

   multipart/related, or multipart/alternative in the multipart (unless[](#l7.113)

   they have a 'Content-Disposition: attachment' header) and include all[](#l7.114)

   remaining subparts in the returned iterator.  When applied to a[](#l7.115)

   multipart/related, return all parts except the root part.  Return an[](#l7.116)

   empty iterator when applied to a multipart/alternative or a[](#l7.117)

```
   non-multipart.[](#l7.118)
```
```
   """[](#l7.119)
```

   maintype, subtype = self.get_content_type().split('/')[](#l7.120)

   if maintype != 'multipart' or subtype == 'alternative':[](#l7.121)

```
       return[](#l7.122)
```

   parts = self.get_payload()[](#l7.123)

   if maintype == 'multipart' and subtype == 'related':[](#l7.124)

       # For related, we treat everything but the root as an attachment.[](#l7.125)

       # The root may be indicated by 'start'; if there's no start or we[](#l7.126)

       # can't find the named start, treat the first subpart as the root.[](#l7.127)

       start = self.get_param('start')[](#l7.128)

```
       if start:[](#l7.129)
```
```
           found = False[](#l7.130)
```
```
           attachments = [][](#l7.131)
```

           for part in parts:[](#l7.132)

               if part.get('content-id') == start:[](#l7.133)

                   found = True[](#l7.134)

```
               else:[](#l7.135)
```

                   attachments.append(part)[](#l7.136)

```
           if found:[](#l7.137)
```

               yield from attachments[](#l7.138)

```
               return[](#l7.139)
```
```
       parts.pop(0)[](#l7.140)
```
```
       yield from parts[](#l7.141)
```
```
       return[](#l7.142)
```

   # Otherwise we more or less invert the remaining logic in get_body.[](#l7.143)

   # This only really works in edge cases (ex: non-text relateds or[](#l7.144)

   # alternatives) if the sending agent sets content-disposition.[](#l7.145)

   seen = []   # Only skip the first example of each candidate type.[](#l7.146)

```
   for part in parts:[](#l7.147)
```

       maintype, subtype = part.get_content_type().split('/')[](#l7.148)

       if ((maintype, subtype) in self._body_types and[](#l7.149)

               not part.is_attachment and subtype not in seen):[](#l7.150)

           seen.append(subtype)[](#l7.151)

```
           continue[](#l7.152)
```
```
       yield part[](#l7.153)
```

def iter_parts(self):

   """Return an iterator over all immediate subparts of a multipart.[](#l7.156)

   Return an empty iterator for a non-multipart.[](#l7.158)

```
   """[](#l7.159)
```

   if self.get_content_maintype() == 'multipart':[](#l7.160)

       yield from self.get_payload()[](#l7.161)

def get_content(self, *args, content_manager=None, **kw):

   if content_manager is None:[](#l7.164)

       content_manager = self.policy.content_manager[](#l7.165)

   return content_manager.get_content(self, *args, **kw)[](#l7.166)

def set_content(self, *args, content_manager=None, **kw):

   if content_manager is None:[](#l7.169)

       content_manager = self.policy.content_manager[](#l7.170)

   content_manager.set_content(self, *args, **kw)[](#l7.171)

def _make_multipart(self, subtype, disallowed_subtypes, boundary):

   if self.get_content_maintype() == 'multipart':[](#l7.174)

       existing_subtype = self.get_content_subtype()[](#l7.175)

       disallowed_subtypes = disallowed_subtypes + (subtype,)[](#l7.176)

       if existing_subtype in disallowed_subtypes:[](#l7.177)

           raise ValueError("Cannot convert {} to {}".format([](#l7.178)

               existing_subtype, subtype))[](#l7.179)

```
   keep_headers = [][](#l7.180)
```
```
   part_headers = [][](#l7.181)
```

   for name, value in self._headers:[](#l7.182)

       if name.lower().startswith('content-'):[](#l7.183)

           part_headers.append((name, value))[](#l7.184)

```
       else:[](#l7.185)
```

           keep_headers.append((name, value))[](#l7.186)

```
   if part_headers:[](#l7.187)
```

       # There is existing content, move it to the first subpart.[](#l7.188)

       part = type(self)(policy=self.policy)[](#l7.189)

       part._headers = part_headers[](#l7.190)

       part._payload = self._payload[](#l7.191)

       self._payload = [part][](#l7.192)

```
   else:[](#l7.193)
```
```
       self._payload = [][](#l7.194)
```

   self._headers = keep_headers[](#l7.195)

   self['Content-Type'] = 'multipart/' + subtype[](#l7.196)

```
   if boundary is not None:[](#l7.197)
```

       self.set_param('boundary', boundary)[](#l7.198)

def make_related(self, boundary=None):

   self._make_multipart('related', ('alternative', 'mixed'), boundary)[](#l7.201)

def make_alternative(self, boundary=None):

   self._make_multipart('alternative', ('mixed',), boundary)[](#l7.204)

def make_mixed(self, boundary=None):

   self._make_multipart('mixed', (), boundary)[](#l7.207)

def _add_multipart(self, _subtype, *args, _disp=None, **kw):

   if (self.get_content_maintype() != 'multipart' or[](#l7.210)

           self.get_content_subtype() != _subtype):[](#l7.211)

       getattr(self, 'make_' + _subtype)()[](#l7.212)

   part = type(self)(policy=self.policy)[](#l7.213)

   part.set_content(*args, **kw)[](#l7.214)

   if _disp and 'content-disposition' not in part:[](#l7.215)

       part['Content-Disposition'] = _disp[](#l7.216)

```
   self.attach(part)[](#l7.217)
```

def add_related(self, *args, **kw):

   self._add_multipart('related', *args, _disp='inline', **kw)[](#l7.220)

def add_alternative(self, *args, **kw):

   self._add_multipart('alternative', *args, **kw)[](#l7.223)

def add_attachment(self, *args, **kw):

   self._add_multipart('mixed', *args, _disp='attachment', **kw)[](#l7.226)

def clear(self):
```
   self._headers = [][](#l7.229)
```
```
   self._payload = None[](#l7.230)
```

def clear_content(self):

   self._headers = [(n, v) for n, v in self._headers[](#l7.233)

                    if not n.lower().startswith('content-')][](#l7.234)

```
   self._payload = None[](#l7.235)
```

+ + +class EmailMessage(MIMEPart): +

def set_content(self, *args, **kw):

   super().set_content(*args, **kw)[](#l7.241)

   if 'MIME-Version' not in self:[](#l7.242)

       self['MIME-Version'] = '1.0'[](#l7.243)

--- a/Lib/email/policy.py +++ b/Lib/email/policy.py @@ -5,6 +5,7 @@ code that adds all the email6 features. from email._policybase import Policy, Compat32, compat32, _extend_docstrings from email.utils import _has_surrogates from email.headerregistry import HeaderRegistry as HeaderRegistry +from email.contentmanager import raw_data_manager all = [ 'Compat32', @@ -58,10 +59,22 @@ class EmailPolicy(Policy): special treatment, while all other fields are treated as unstructured. This list will be completed before the extension is marked stable.) +

content_manager -- an object with at least two methods: get_content

                      and set_content.  When the get_content or[](#l8.17)

                      set_content method of a Message object is called,[](#l8.18)

                      it calls the corresponding method of this object,[](#l8.19)

                      passing it the message object as its first argument,[](#l8.20)

                      and any arguments or keywords that were passed to[](#l8.21)

                      it as additional arguments.  The default[](#l8.22)

                      content_manager is[](#l8.23)

                      :data:`~email.contentmanager.raw_data_manager`.[](#l8.24)

+ """ refold_source = 'long' header_factory = HeaderRegistry()

content_manager = raw_data_manager

def init(self, **kw): # Ensure that each new instance gets a unique header factory

--- a/Lib/email/utils.py +++ b/Lib/email/utils.py @@ -68,9 +68,13 @@ def _has_surrogates(s):

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')

Turn any escaped bytes into unicode 'unknown' char. If the escaped
bytes happen to be utf-8 they will instead get decoded, even if they
were invalid in the charset the source was supposed to be in. This
seems like it is not a bad thing; a defect was still registered.
original_bytes = string.encode('utf-8', 'surrogateescape')
return original_bytes.decode('utf-8', 'replace')

Helpers

--- a/Lib/test/test_email/init.py +++ b/Lib/test/test_email/init.py @@ -2,6 +2,7 @@ import os import sys import unittest import test.support +import collections import email from email.message import Message from email._policybase import compat32 @@ -42,6 +43,8 @@ class TestEmailBase(unittest.TestCase): # here we make minimal changes in the test_email tests compared to their # pre-3.3 state. policy = compat32

Likewise, the default message object is Message.
message = Message

def init(self, *args, **kw): super().init(*args, **kw) @@ -54,11 +57,23 @@ class TestEmailBase(unittest.TestCase): with openfile(filename) as fp: return email.message_from_file(fp, policy=self.policy)

def _str_msg(self, string, message=Message, policy=None):

def _str_msg(self, string, message=None, policy=None): if policy is None: policy = self.policy
```
   if message is None:[](#l10.28)
```

       message = self.message[](#l10.29)
   return email.message_from_string(string, message, policy=policy)[](#l10.30)

def _bytes_msg(self, bytestring, message=None, policy=None):
```
   if policy is None:[](#l10.33)
```
```
       policy = self.policy[](#l10.34)
```
```
   if message is None:[](#l10.35)
```

       message = self.message[](#l10.36)

   return email.message_from_bytes(bytestring, message, policy=policy)[](#l10.37)

def _make_message(self):

   return self.message(policy=self.policy)[](#l10.40)

+ def _bytes_repr(self, b): return [repr(x) for x in b.splitlines(keepends=True)] @@ -123,6 +138,7 @@ def parameterize(cls): """ paramdicts = {}

testers = collections.defaultdict(list) for name, attr in cls.dict.items(): if name.endswith('_params'): if not hasattr(attr, 'keys'):

@@ -134,7 +150,15 @@ def parameterize(cls): d[n] = x attr = d paramdicts[name[:-7] + 'as'] = attr

```
   if '_as_' in name:[](#l10.57)
```

       testers[name.split('_as_')[0] + '_as_'].append(name)[](#l10.58)

testfuncs = {}

for name in paramdicts:
```
   if name not in testers:[](#l10.61)
```

       raise ValueError("No tester found for {}".format(name))[](#l10.62)

for name in testers:

   if name not in paramdicts:[](#l10.64)

```
       raise ValueError("No params found for {}".format(name))[](#l10.65)
```
for name, attr in cls.dict.items(): for paramsname, paramsdict in paramdicts.items(): if name.startswith(paramsname):

new file mode 100644 --- /dev/null +++ b/Lib/test/test_email/test_contentmanager.py @@ -0,0 +1,796 @@ +import unittest +from test.test_email import TestEmailBase, parameterize +import textwrap +from email import policy +from email.message import EmailMessage +from email.contentmanager import ContentManager, raw_data_manager + + +@parameterize +class TestContentManager(TestEmailBase): +

policy = policy.default
message = EmailMessage

get_key_params = {

   'full_type':        (1, 'text/plain',),[](#l11.20)

   'maintype_only':    (2, 'text',),[](#l11.21)

   'null_key':         (3, '',),[](#l11.22)

```
   }[](#l11.23)
```

def get_key_as_get_content_key(self, order, key):

   def foo_getter(msg, foo=None):[](#l11.26)

       bar = msg['X-Bar-Header'][](#l11.27)

```
       return foo, bar[](#l11.28)
```
```
   cm = ContentManager()[](#l11.29)
```

   cm.add_get_handler(key, foo_getter)[](#l11.30)

```
   m = self._make_message()[](#l11.31)
```

   m['Content-Type'] = 'text/plain'[](#l11.32)

```
   m['X-Bar-Header'] = 'foo'[](#l11.33)
```

   self.assertEqual(cm.get_content(m, foo='bar'), ('bar', 'foo'))[](#l11.34)

def get_key_as_get_content_key_order(self, order, key):
```
   def bar_getter(msg):[](#l11.37)
```

       return msg['X-Bar-Header'][](#l11.38)

```
   def foo_getter(msg):[](#l11.39)
```

       return msg['X-Foo-Header'][](#l11.40)

```
   cm = ContentManager()[](#l11.41)
```

   cm.add_get_handler(key, foo_getter)[](#l11.42)

   for precedence, key in self.get_key_params.values():[](#l11.43)

       if precedence > order:[](#l11.44)

           cm.add_get_handler(key, bar_getter)[](#l11.45)

```
   m = self._make_message()[](#l11.46)
```

   m['Content-Type'] = 'text/plain'[](#l11.47)

```
   m['X-Bar-Header'] = 'bar'[](#l11.48)
```
```
   m['X-Foo-Header'] = 'foo'[](#l11.49)
```

   self.assertEqual(cm.get_content(m), ('foo'))[](#l11.50)

def test_get_content_raises_if_unknown_mimetype_and_no_default(self):
```
   cm = ContentManager()[](#l11.53)
```
```
   m = self._make_message()[](#l11.54)
```

   m['Content-Type'] = 'text/plain'[](#l11.55)

   with self.assertRaisesRegex(KeyError, 'text/plain'):[](#l11.56)

```
       cm.get_content(m)[](#l11.57)
```

class BaseThing(str):
```
   pass[](#l11.60)
```
baseobject_full_path = name + '.' + 'TestContentManager.BaseThing'
class Thing(BaseThing):
```
   pass[](#l11.63)
```
testobject_full_path = name + '.' + 'TestContentManager.Thing'

set_key_params = {

   'type':             (0,  Thing,),[](#l11.67)

   'full_path':        (1,  testobject_full_path,),[](#l11.68)

   'qualname':         (2,  'TestContentManager.Thing',),[](#l11.69)

   'name':             (3,  'Thing',),[](#l11.70)

   'base_type':        (4,  BaseThing,),[](#l11.71)

   'base_full_path':   (5,  baseobject_full_path,),[](#l11.72)

   'base_qualname':    (6,  'TestContentManager.BaseThing',),[](#l11.73)

   'base_name':        (7,  'BaseThing',),[](#l11.74)

   'str_type':         (8,  str,),[](#l11.75)

   'str_full_path':    (9,  'builtins.str',),[](#l11.76)

   'str_name':         (10, 'str',),   # str name and qualname are the same[](#l11.77)

   'null_key':         (11, None,),[](#l11.78)

```
   }[](#l11.79)
```

def set_key_as_set_content_key(self, order, key):

   def foo_setter(msg, obj, foo=None):[](#l11.82)

       msg['X-Foo-Header'] = foo[](#l11.83)

```
       msg.set_payload(obj)[](#l11.84)
```
```
   cm = ContentManager()[](#l11.85)
```

   cm.add_set_handler(key, foo_setter)[](#l11.86)

```
   m = self._make_message()[](#l11.87)
```
```
   msg_obj = self.Thing()[](#l11.88)
```

   cm.set_content(m, msg_obj, foo='bar')[](#l11.89)

   self.assertEqual(m['X-Foo-Header'], 'bar')[](#l11.90)

   self.assertEqual(m.get_payload(), msg_obj)[](#l11.91)

def set_key_as_set_content_key_order(self, order, key):
```
   def foo_setter(msg, obj):[](#l11.94)
```

       msg['X-FooBar-Header'] = 'foo'[](#l11.95)

```
       msg.set_payload(obj)[](#l11.96)
```
```
   def bar_setter(msg, obj):[](#l11.97)
```

       msg['X-FooBar-Header'] = 'bar'[](#l11.98)

```
   cm = ContentManager()[](#l11.99)
```

   cm.add_set_handler(key, foo_setter)[](#l11.100)

   for precedence, key in self.get_key_params.values():[](#l11.101)

       if precedence > order:[](#l11.102)

           cm.add_set_handler(key, bar_setter)[](#l11.103)

```
   m = self._make_message()[](#l11.104)
```
```
   msg_obj = self.Thing()[](#l11.105)
```

   cm.set_content(m, msg_obj)[](#l11.106)

   self.assertEqual(m['X-FooBar-Header'], 'foo')[](#l11.107)

   self.assertEqual(m.get_payload(), msg_obj)[](#l11.108)

def test_set_content_raises_if_unknown_type_and_no_default(self):
```
   cm = ContentManager()[](#l11.111)
```
```
   m = self._make_message()[](#l11.112)
```
```
   msg_obj = self.Thing()[](#l11.113)
```

   with self.assertRaisesRegex(KeyError, self.testobject_full_path):[](#l11.114)

       cm.set_content(m, msg_obj)[](#l11.115)

def test_set_content_raises_if_called_on_multipart(self):
```
   cm = ContentManager()[](#l11.118)
```
```
   m = self._make_message()[](#l11.119)
```

   m['Content-Type'] = 'multipart/foo'[](#l11.120)

   with self.assertRaises(TypeError):[](#l11.121)

       cm.set_content(m, 'test')[](#l11.122)

def test_set_content_calls_clear_content(self):
```
   m = self._make_message()[](#l11.125)
```
```
   m['Content-Foo'] = 'bar'[](#l11.126)
```

   m['Content-Type'] = 'text/html'[](#l11.127)

```
   m['To'] = 'test'[](#l11.128)
```
```
   m.set_payload('abc')[](#l11.129)
```
```
   cm = ContentManager()[](#l11.130)
```

   cm.add_set_handler(str, lambda *args, **kw: None)[](#l11.131)

   m.set_content('xyz', content_manager=cm)[](#l11.132)

   self.assertIsNone(m['Content-Foo'])[](#l11.133)

   self.assertIsNone(m['Content-Type'])[](#l11.134)

   self.assertEqual(m['To'], 'test')[](#l11.135)

   self.assertIsNone(m.get_payload())[](#l11.136)

+ + +@parameterize +class TestRawDataManager(TestEmailBase):

Note: these tests are dependent on the order in which headers are added
to the message objects by the code. There's no defined ordering in
RFC5322/MIME, so this makes the tests more fragile than the standards
require. However, if the header order changes it is best to understand
why, and make sure it isn't a subtle bug in whatever change was
applied.

policy = policy.default.clone(max_line_length=60,

                             content_manager=raw_data_manager)[](#l11.149)

message = EmailMessage

def test_get_text_plain(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.153)

       Content-Type: text/plain[](#l11.154)

```
       Basic text.[](#l11.156)
```
```
       """))[](#l11.157)
```

   self.assertEqual(raw_data_manager.get_content(m), "Basic text.\n")[](#l11.158)

def test_get_text_html(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.161)

       Content-Type: text/html[](#l11.162)

```
       <p>Basic text.</p>[](#l11.164)
```
```
       """))[](#l11.165)
```

   self.assertEqual(raw_data_manager.get_content(m),[](#l11.166)

                    "<p>Basic text.</p>\n")[](#l11.167)

def test_get_text_plain_latin1(self):

   m = self._bytes_msg(textwrap.dedent("""\[](#l11.170)

       Content-Type: text/plain; charset=latin1[](#l11.171)

```
       Basìc tëxt.[](#l11.173)
```

       """).encode('latin1'))[](#l11.174)

   self.assertEqual(raw_data_manager.get_content(m), "Basìc tëxt.\n")[](#l11.175)

def test_get_text_plain_latin1_quoted_printable(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.178)

       Content-Type: text/plain; charset="latin-1"[](#l11.179)

       Content-Transfer-Encoding: quoted-printable[](#l11.180)

```
       Bas=ECc t=EBxt.[](#l11.182)
```
```
       """))[](#l11.183)
```

   self.assertEqual(raw_data_manager.get_content(m), "Basìc tëxt.\n")[](#l11.184)

def test_get_text_plain_utf8_base64(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.187)

       Content-Type: text/plain; charset="utf8"[](#l11.188)

       Content-Transfer-Encoding: base64[](#l11.189)

```
       QmFzw6xjIHTDq3h0Lgo=[](#l11.191)
```
```
       """))[](#l11.192)
```

   self.assertEqual(raw_data_manager.get_content(m), "Basìc tëxt.\n")[](#l11.193)

def test_get_text_plain_bad_utf8_quoted_printable(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.196)

       Content-Type: text/plain; charset="utf8"[](#l11.197)

       Content-Transfer-Encoding: quoted-printable[](#l11.198)

       Bas=c3=acc t=c3=abxt=fd.[](#l11.200)

```
       """))[](#l11.201)
```

   self.assertEqual(raw_data_manager.get_content(m), "Basìc tëxt�.\n")[](#l11.202)

def test_get_text_plain_bad_utf8_quoted_printable_ignore_errors(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.205)

       Content-Type: text/plain; charset="utf8"[](#l11.206)

       Content-Transfer-Encoding: quoted-printable[](#l11.207)

       Bas=c3=acc t=c3=abxt=fd.[](#l11.209)

```
       """))[](#l11.210)
```

   self.assertEqual(raw_data_manager.get_content(m, errors='ignore'),[](#l11.211)

                    "Basìc tëxt.\n")[](#l11.212)

def test_get_text_plain_utf8_base64_recoverable_bad_CTE_data(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.215)

       Content-Type: text/plain; charset="utf8"[](#l11.216)

       Content-Transfer-Encoding: base64[](#l11.217)

       QmFzw6xjIHTDq3h0Lgo\xFF=[](#l11.219)

```
       """))[](#l11.220)
```

   self.assertEqual(raw_data_manager.get_content(m, errors='ignore'),[](#l11.221)

                    "Basìc tëxt.\n")[](#l11.222)

def test_get_text_invalid_keyword(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.225)

       Content-Type: text/plain[](#l11.226)

```
       Basic text.[](#l11.228)
```
```
       """))[](#l11.229)
```

   with self.assertRaises(TypeError):[](#l11.230)

       raw_data_manager.get_content(m, foo='ignore')[](#l11.231)

def test_get_non_text(self):

   template = textwrap.dedent("""\[](#l11.234)

```
       Content-Type: {}[](#l11.235)
```

       Content-Transfer-Encoding: base64[](#l11.236)

```
       Ym9ndXMgZGF0YQ==[](#l11.238)
```
```
       """)[](#l11.239)
```

   for maintype in 'audio image video application'.split():[](#l11.240)

       with self.subTest(maintype=maintype):[](#l11.241)

           m = self._str_msg(template.format(maintype+'/foo'))[](#l11.242)

           self.assertEqual(raw_data_manager.get_content(m), b"bogus data")[](#l11.243)

def test_get_non_text_invalid_keyword(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.246)

       Content-Type: image/jpg[](#l11.247)

       Content-Transfer-Encoding: base64[](#l11.248)

```
       Ym9ndXMgZGF0YQ==[](#l11.250)
```
```
       """))[](#l11.251)
```

   with self.assertRaises(TypeError):[](#l11.252)

       raw_data_manager.get_content(m, errors='ignore')[](#l11.253)

def test_get_raises_on_multipart(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.256)

       Content-Type: multipart/mixed; boundary="==="[](#l11.257)

```
       --===[](#l11.259)
```
```
       --===--[](#l11.260)
```
```
       """))[](#l11.261)
```

   with self.assertRaises(KeyError):[](#l11.262)

       raw_data_manager.get_content(m)[](#l11.263)

def test_get_message_rfc822_and_external_body(self):

   template = textwrap.dedent("""\[](#l11.266)

       Content-Type: message/{}[](#l11.267)

```
       To: foo@example.com[](#l11.269)
```

       From: bar@example.com[](#l11.270)

```
       Subject: example[](#l11.271)
```

```
       an example message[](#l11.273)
```
```
       """)[](#l11.274)
```

   for subtype in 'rfc822 external-body'.split():[](#l11.275)

       with self.subTest(subtype=subtype):[](#l11.276)

           m = self._str_msg(template.format(subtype))[](#l11.277)

           sub_msg = raw_data_manager.get_content(m)[](#l11.278)

           self.assertIsInstance(sub_msg, self.message)[](#l11.279)

           self.assertEqual(raw_data_manager.get_content(sub_msg),[](#l11.280)

                            "an example message\n")[](#l11.281)

           self.assertEqual(sub_msg['to'], 'foo@example.com')[](#l11.282)

           self.assertEqual(sub_msg['from'].addresses[0].username, 'bar')[](#l11.283)

def test_get_message_non_rfc822_or_external_body_yields_bytes(self):

   m = self._str_msg(textwrap.dedent("""\[](#l11.286)

       Content-Type: message/partial[](#l11.287)

```
       To: foo@example.com[](#l11.289)
```

       From: bar@example.com[](#l11.290)

```
       Subject: example[](#l11.291)
```

       The real body is in another message.[](#l11.293)

```
       """))[](#l11.294)
```

   self.assertEqual(raw_data_manager.get_content(m)[:10], b'To: foo@ex')[](#l11.295)

def test_set_text_plain(self):
```
   m = self._make_message()[](#l11.298)
```

   content = "Simple message.\n"[](#l11.299)

   raw_data_manager.set_content(m, content)[](#l11.300)

   self.assertEqual(str(m), textwrap.dedent("""\[](#l11.301)

       Content-Type: text/plain; charset="utf-8"[](#l11.302)

       Content-Transfer-Encoding: 7bit[](#l11.303)

```
       Simple message.[](#l11.305)
```
```
       """))[](#l11.306)
```

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.307)

   self.assertEqual(m.get_content(), content)[](#l11.308)

def test_set_text_html(self):
```
   m = self._make_message()[](#l11.311)
```

   content = "<p>Simple message.</p>\n"[](#l11.312)

   raw_data_manager.set_content(m, content, subtype='html')[](#l11.313)

   self.assertEqual(str(m), textwrap.dedent("""\[](#l11.314)

       Content-Type: text/html; charset="utf-8"[](#l11.315)

       Content-Transfer-Encoding: 7bit[](#l11.316)

       <p>Simple message.</p>[](#l11.318)

```
       """))[](#l11.319)
```

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.320)

   self.assertEqual(m.get_content(), content)[](#l11.321)

def test_set_text_charset_latin_1(self):
```
   m = self._make_message()[](#l11.324)
```

   content = "Simple message.\n"[](#l11.325)

   raw_data_manager.set_content(m, content, charset='latin-1')[](#l11.326)

   self.assertEqual(str(m), textwrap.dedent("""\[](#l11.327)

       Content-Type: text/plain; charset="iso-8859-1"[](#l11.328)

       Content-Transfer-Encoding: 7bit[](#l11.329)

```
       Simple message.[](#l11.331)
```
```
       """))[](#l11.332)
```

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.333)

   self.assertEqual(m.get_content(), content)[](#l11.334)

def test_set_text_short_line_minimal_non_ascii_heuristics(self):
```
   m = self._make_message()[](#l11.337)
```

   content = "et là il est monté sur moi et il commence à m'éto.\n"[](#l11.338)

   raw_data_manager.set_content(m, content)[](#l11.339)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.340)

       Content-Type: text/plain; charset="utf-8"[](#l11.341)

       Content-Transfer-Encoding: 8bit[](#l11.342)

       et là il est monté sur moi et il commence à m'éto.[](#l11.344)

       """).encode('utf-8'))[](#l11.345)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.346)

   self.assertEqual(m.get_content(), content)[](#l11.347)

def test_set_text_long_line_minimal_non_ascii_heuristics(self):
```
   m = self._make_message()[](#l11.350)
```

   content = ("j'ai un problème de python. il est sorti de son"[](#l11.351)

              " vivarium.  et là il est monté sur moi et il commence"[](#l11.352)

              " à m'éto.\n")[](#l11.353)

   raw_data_manager.set_content(m, content)[](#l11.354)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.355)

       Content-Type: text/plain; charset="utf-8"[](#l11.356)

       Content-Transfer-Encoding: quoted-printable[](#l11.357)

       j'ai un probl=C3=A8me de python. il est sorti de son vivari=[](#l11.359)

       um.  et l=C3=A0 il est mont=C3=A9 sur moi et il commence =[](#l11.360)

```
       =C3=A0 m'=C3=A9to.[](#l11.361)
```

       """).encode('utf-8'))[](#l11.362)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.363)

   self.assertEqual(m.get_content(), content)[](#l11.364)

def test_set_text_11_lines_long_line_minimal_non_ascii_heuristics(self):
```
   m = self._make_message()[](#l11.367)
```
```
   content = '\n'*10 + ([](#l11.368)
```

             "j'ai un problème de python. il est sorti de son"[](#l11.369)

             " vivarium.  et là il est monté sur moi et il commence"[](#l11.370)

             " à m'éto.\n")[](#l11.371)

   raw_data_manager.set_content(m, content)[](#l11.372)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.373)

       Content-Type: text/plain; charset="utf-8"[](#l11.374)

       Content-Transfer-Encoding: quoted-printable[](#l11.375)

```
       """ + '\n'*10 + """[](#l11.376)
```

       j'ai un probl=C3=A8me de python. il est sorti de son vivari=[](#l11.377)

       um.  et l=C3=A0 il est mont=C3=A9 sur moi et il commence =[](#l11.378)

```
       =C3=A0 m'=C3=A9to.[](#l11.379)
```

       """).encode('utf-8'))[](#l11.380)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.381)

   self.assertEqual(m.get_content(), content)[](#l11.382)

def test_set_text_maximal_non_ascii_heuristics(self):
```
   m = self._make_message()[](#l11.385)
```

   content = "áàäéèęöő.\n"[](#l11.386)

   raw_data_manager.set_content(m, content)[](#l11.387)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.388)

       Content-Type: text/plain; charset="utf-8"[](#l11.389)

       Content-Transfer-Encoding: 8bit[](#l11.390)

```
       áàäéèęöő.[](#l11.392)
```

       """).encode('utf-8'))[](#l11.393)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.394)

   self.assertEqual(m.get_content(), content)[](#l11.395)

def test_set_text_11_lines_maximal_non_ascii_heuristics(self):
```
   m = self._make_message()[](#l11.398)
```

   content = '\n'*10 + "áàäéèęöő.\n"[](#l11.399)

   raw_data_manager.set_content(m, content)[](#l11.400)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.401)

       Content-Type: text/plain; charset="utf-8"[](#l11.402)

       Content-Transfer-Encoding: 8bit[](#l11.403)

```
       """ + '\n'*10 + """[](#l11.404)
```
```
       áàäéèęöő.[](#l11.405)
```

       """).encode('utf-8'))[](#l11.406)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.407)

   self.assertEqual(m.get_content(), content)[](#l11.408)

def test_set_text_long_line_maximal_non_ascii_heuristics(self):
```
   m = self._make_message()[](#l11.411)
```

   content = ("áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő"[](#l11.412)

              "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő"[](#l11.413)

              "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő.\n")[](#l11.414)

   raw_data_manager.set_content(m, content)[](#l11.415)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.416)

       Content-Type: text/plain; charset="utf-8"[](#l11.417)

       Content-Transfer-Encoding: base64[](#l11.418)

       w6HDoMOkw6nDqMSZw7bFkcOhw6DDpMOpw6jEmcO2xZHDocOgw6TDqcOoxJnD[](#l11.420)

       tsWRw6HDoMOkw6nDqMSZw7bFkcOhw6DDpMOpw6jEmcO2xZHDocOgw6TDqcOo[](#l11.421)

       xJnDtsWRw6HDoMOkw6nDqMSZw7bFkcOhw6DDpMOpw6jEmcO2xZHDocOgw6TD[](#l11.422)

       qcOoxJnDtsWRw6HDoMOkw6nDqMSZw7bFkcOhw6DDpMOpw6jEmcO2xZHDocOg[](#l11.423)

```
       w6TDqcOoxJnDtsWRLgo=[](#l11.424)
```

       """).encode('utf-8'))[](#l11.425)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.426)

   self.assertEqual(m.get_content(), content)[](#l11.427)

def test_set_text_11_lines_long_line_maximal_non_ascii_heuristics(self):

   # Yes, it chooses "wrong" here.  It's a heuristic.  So this result[](#l11.430)

   # could change if we come up with a better heuristic.[](#l11.431)

```
   m = self._make_message()[](#l11.432)
```
```
   content = ('\n'*10 +[](#l11.433)
```

              "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő"[](#l11.434)

              "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő"[](#l11.435)

              "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő.\n")[](#l11.436)

   raw_data_manager.set_content(m, "\n"*10 +[](#l11.437)

                                   "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő"[](#l11.438)

                                   "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő"[](#l11.439)

                                   "áàäéèęöőáàäéèęöőáàäéèęöőáàäéèęöő.\n")[](#l11.440)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.441)

       Content-Type: text/plain; charset="utf-8"[](#l11.442)

       Content-Transfer-Encoding: quoted-printable[](#l11.443)

```
       """ + '\n'*10 + """[](#l11.444)
```

       =C3=A1=C3=A0=C3=A4=C3=A9=C3=A8=C4=99=C3=B6=C5=91=C3=A1=C3=[](#l11.445)

       =A0=C3=A4=C3=A9=C3=A8=C4=99=C3=B6=C5=91=C3=A1=C3=A0=C3=A4=[](#l11.446)

       =C3=A9=C3=A8=C4=99=C3=B6=C5=91=C3=A1=C3=A0=C3=A4=C3=A9=C3=[](#l11.447)

       =A8=C4=99=C3=B6=C5=91=C3=A1=C3=A0=C3=A4=C3=A9=C3=A8=C4=99=[](#l11.448)

       =C3=B6=C5=91=C3=A1=C3=A0=C3=A4=C3=A9=C3=A8=C4=99=C3=B6=C5=[](#l11.449)

       =91=C3=A1=C3=A0=C3=A4=C3=A9=C3=A8=C4=99=C3=B6=C5=91=C3=A1=[](#l11.450)

       =C3=A0=C3=A4=C3=A9=C3=A8=C4=99=C3=B6=C5=91=C3=A1=C3=A0=C3=[](#l11.451)

       =A4=C3=A9=C3=A8=C4=99=C3=B6=C5=91=C3=A1=C3=A0=C3=A4=C3=A9=[](#l11.452)

       =C3=A8=C4=99=C3=B6=C5=91=C3=A1=C3=A0=C3=A4=C3=A9=C3=A8=C4=[](#l11.453)

       =99=C3=B6=C5=91=C3=A1=C3=A0=C3=A4=C3=A9=C3=A8=C4=99=C3=B6=[](#l11.454)

```
       =C5=91.[](#l11.455)
```

       """).encode('utf-8'))[](#l11.456)

   self.assertEqual(m.get_payload(decode=True).decode('utf-8'), content)[](#l11.457)

   self.assertEqual(m.get_content(), content)[](#l11.458)

def test_set_text_non_ascii_with_cte_7bit_raises(self):
```
   m = self._make_message()[](#l11.461)
```

   with self.assertRaises(UnicodeError):[](#l11.462)

       raw_data_manager.set_content(m,"áàäéèęöő.\n", cte='7bit')[](#l11.463)

def test_set_text_non_ascii_with_charset_ascii_raises(self):
```
   m = self._make_message()[](#l11.466)
```

   with self.assertRaises(UnicodeError):[](#l11.467)

       raw_data_manager.set_content(m,"áàäéèęöő.\n", charset='ascii')[](#l11.468)

def test_set_text_non_ascii_with_cte_7bit_and_charset_ascii_raises(self):
```
   m = self._make_message()[](#l11.471)
```

   with self.assertRaises(UnicodeError):[](#l11.472)

       raw_data_manager.set_content(m,"áàäéèęöő.\n", cte='7bit', charset='ascii')[](#l11.473)

def test_set_message(self):
```
   m = self._make_message()[](#l11.476)
```

   m['Subject'] = "Forwarded message"[](#l11.477)

   content = self._make_message()[](#l11.478)

   content['To'] = 'python@vivarium.org'[](#l11.479)

   content['From'] = 'police@monty.org'[](#l11.480)

   content['Subject'] = "get back in your box"[](#l11.481)

   content.set_content("Or face the comfy chair.")[](#l11.482)

   raw_data_manager.set_content(m, content)[](#l11.483)

   self.assertEqual(str(m), textwrap.dedent("""\[](#l11.484)

       Subject: Forwarded message[](#l11.485)

       Content-Type: message/rfc822[](#l11.486)

       Content-Transfer-Encoding: 8bit[](#l11.487)

       To: python@vivarium.org[](#l11.489)

       From: police@monty.org[](#l11.490)

       Subject: get back in your box[](#l11.491)

       Content-Type: text/plain; charset="utf-8"[](#l11.492)

       Content-Transfer-Encoding: 7bit[](#l11.493)

```
       MIME-Version: 1.0[](#l11.494)
```

       Or face the comfy chair.[](#l11.496)

```
       """))[](#l11.497)
```

   payload = m.get_payload(0)[](#l11.498)

   self.assertIsInstance(payload, self.message)[](#l11.499)

   self.assertEqual(str(payload), str(content))[](#l11.500)

   self.assertIsInstance(m.get_content(), self.message)[](#l11.501)

   self.assertEqual(str(m.get_content()), str(content))[](#l11.502)

def test_set_message_with_non_ascii_and_coercion_to_7bit(self):
```
   m = self._make_message()[](#l11.505)
```

   m['Subject'] = "Escape report"[](#l11.506)

   content = self._make_message()[](#l11.507)

   content['To'] = 'police@monty.org'[](#l11.508)

   content['From'] = 'victim@monty.org'[](#l11.509)

   content['Subject'] = "Help"[](#l11.510)

   content.set_content("j'ai un problème de python. il est sorti de son"[](#l11.511)

                       " vivarium.")[](#l11.512)

   raw_data_manager.set_content(m, content)[](#l11.513)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.514)

       Subject: Escape report[](#l11.515)

       Content-Type: message/rfc822[](#l11.516)

       Content-Transfer-Encoding: 8bit[](#l11.517)

```
       To: police@monty.org[](#l11.519)
```

       From: victim@monty.org[](#l11.520)

```
       Subject: Help[](#l11.521)
```

       Content-Type: text/plain; charset="utf-8"[](#l11.522)

       Content-Transfer-Encoding: 8bit[](#l11.523)

```
       MIME-Version: 1.0[](#l11.524)
```

       j'ai un problème de python. il est sorti de son vivarium.[](#l11.526)

       """).encode('utf-8'))[](#l11.527)

   # The choice of base64 for the body encoding is because generator[](#l11.528)

   # doesn't bother with heuristics and uses it unconditionally for utf-8[](#l11.529)

```
   # text.[](#l11.530)
```

   # XXX: the first cte should be 7bit, too...that's a generator bug.[](#l11.531)

   # XXX: the line length in the body also looks like a generator bug.[](#l11.532)

   self.assertEqual(m.as_string(maxheaderlen=self.policy.max_line_length),[](#l11.533)

                    textwrap.dedent("""\[](#l11.534)

       Subject: Escape report[](#l11.535)

       Content-Type: message/rfc822[](#l11.536)

       Content-Transfer-Encoding: 8bit[](#l11.537)

```
       To: police@monty.org[](#l11.539)
```

       From: victim@monty.org[](#l11.540)

```
       Subject: Help[](#l11.541)
```

       Content-Type: text/plain; charset="utf-8"[](#l11.542)

```
       MIME-Version: 1.0[](#l11.543)
```

       Content-Transfer-Encoding: base64[](#l11.544)

       aidhaSB1biBwcm9ibMOobWUgZGUgcHl0aG9uLiBpbCBlc3Qgc29ydGkgZGUgc29uIHZpdmFyaXVt[](#l11.546)

```
       Lgo=[](#l11.547)
```
```
       """))[](#l11.548)
```

   self.assertIsInstance(m.get_content(), self.message)[](#l11.549)

   self.assertEqual(str(m.get_content()), str(content))[](#l11.550)

def test_set_message_invalid_cte_raises(self):
```
   m = self._make_message()[](#l11.553)
```

   content = self._make_message()[](#l11.554)

   for cte in 'quoted-printable base64'.split():[](#l11.555)

       for subtype in 'rfc822 external-body'.split():[](#l11.556)

           with self.subTest(cte=cte, subtype=subtype):[](#l11.557)

               with self.assertRaises(ValueError) as ar:[](#l11.558)

                   m.set_content(content, subtype, cte=cte)[](#l11.559)

               exc = str(ar.exception)[](#l11.560)

               self.assertIn(cte, exc)[](#l11.561)

               self.assertIn(subtype, exc)[](#l11.562)

   subtype = 'external-body'[](#l11.563)

   for cte in '8bit binary'.split():[](#l11.564)

       with self.subTest(cte=cte, subtype=subtype):[](#l11.565)

           with self.assertRaises(ValueError) as ar:[](#l11.566)

               m.set_content(content, subtype, cte=cte)[](#l11.567)

           exc = str(ar.exception)[](#l11.568)

           self.assertIn(cte, exc)[](#l11.569)

           self.assertIn(subtype, exc)[](#l11.570)

def test_set_image_jpg(self):

   for content in (b"bogus content",[](#l11.573)

                   bytearray(b"bogus content"),[](#l11.574)

                   memoryview(b"bogus content")):[](#l11.575)

       with self.subTest(content=content):[](#l11.576)

           m = self._make_message()[](#l11.577)

           raw_data_manager.set_content(m, content, 'image', 'jpeg')[](#l11.578)

           self.assertEqual(str(m), textwrap.dedent("""\[](#l11.579)

               Content-Type: image/jpeg[](#l11.580)

               Content-Transfer-Encoding: base64[](#l11.581)

               Ym9ndXMgY29udGVudA==[](#l11.583)

```
               """))[](#l11.584)
```

           self.assertEqual(m.get_payload(decode=True), content)[](#l11.585)

           self.assertEqual(m.get_content(), content)[](#l11.586)

def test_set_audio_aif_with_quoted_printable_cte(self):

   # Why you would use qp, I don't know, but it is technically supported.[](#l11.589)

   # XXX: the incorrect line length is because binascii.b2a_qp doesn't[](#l11.590)

   # support a line length parameter, but we must use it to get newline[](#l11.591)

```
   # encoding.[](#l11.592)
```

   # XXX: what about that lack of tailing newline?  Do we actually handle[](#l11.593)

   # that correctly in all cases?  That is, if the *source* has an[](#l11.594)

   # unencoded newline, do we add an extra newline to the returned payload[](#l11.595)

   # or not?  And can that actually be disambiguated based on the RFC?[](#l11.596)

```
   m = self._make_message()[](#l11.597)
```

   content = b'b\xFFgus\tcon\nt\rent ' + b'z'*100[](#l11.598)

   m.set_content(content, 'audio', 'aif', cte='quoted-printable')[](#l11.599)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.600)

       Content-Type: audio/aif[](#l11.601)

       Content-Transfer-Encoding: quoted-printable[](#l11.602)

```
       MIME-Version: 1.0[](#l11.603)
```

       b=FFgus=09con=0At=0Dent=20zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz=[](#l11.605)

       zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz""").encode('latin-1'))[](#l11.606)

   self.assertEqual(m.get_payload(decode=True), content)[](#l11.607)

   self.assertEqual(m.get_content(), content)[](#l11.608)

def test_set_video_mpeg_with_binary_cte(self):
```
   m = self._make_message()[](#l11.611)
```

   content = b'b\xFFgus\tcon\nt\rent ' + b'z'*100[](#l11.612)

   m.set_content(content, 'video', 'mpeg', cte='binary')[](#l11.613)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.614)

       Content-Type: video/mpeg[](#l11.615)

       Content-Transfer-Encoding: binary[](#l11.616)

```
       MIME-Version: 1.0[](#l11.617)
```

       """).encode('ascii') +[](#l11.619)

       # XXX: the second \n ought to be a \r, but generator gets it wrong.[](#l11.620)

       # THIS MEANS WE DON'T ACTUALLY SUPPORT THE 'binary' CTE.[](#l11.621)

       b'b\xFFgus\tcon\nt\nent zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz' +[](#l11.622)

       b'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz')[](#l11.623)

   self.assertEqual(m.get_payload(decode=True), content)[](#l11.624)

   self.assertEqual(m.get_content(), content)[](#l11.625)

def test_set_application_octet_stream_with_8bit_cte(self):

   # In 8bit mode, univeral line end logic applies.  It is up to the[](#l11.628)

   # application to make sure the lines are short enough; we don't check.[](#l11.629)

```
   m = self._make_message()[](#l11.630)
```

   content = b'b\xFFgus\tcon\nt\rent\n' + b'z'*60 + b'\n'[](#l11.631)

   m.set_content(content, 'application', 'octet-stream', cte='8bit')[](#l11.632)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.633)

       Content-Type: application/octet-stream[](#l11.634)

       Content-Transfer-Encoding: 8bit[](#l11.635)

```
       MIME-Version: 1.0[](#l11.636)
```

       """).encode('ascii') +[](#l11.638)

       b'b\xFFgus\tcon\nt\nent\n' +[](#l11.639)

       b'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz\n')[](#l11.640)

   self.assertEqual(m.get_payload(decode=True), content)[](#l11.641)

   self.assertEqual(m.get_content(), content)[](#l11.642)

def test_set_headers_from_header_objects(self):
```
   m = self._make_message()[](#l11.645)
```

   content = "Simple message.\n"[](#l11.646)

   header_factory = self.policy.header_factory[](#l11.647)

   raw_data_manager.set_content(m, content, headers=([](#l11.648)

       header_factory("To", "foo@example.com"),[](#l11.649)

       header_factory("From", "foo@example.com"),[](#l11.650)

       header_factory("Subject", "I'm talking to myself.")))[](#l11.651)

   self.assertEqual(str(m), textwrap.dedent("""\[](#l11.652)

       Content-Type: text/plain; charset="utf-8"[](#l11.653)

```
       To: foo@example.com[](#l11.654)
```

       From: foo@example.com[](#l11.655)

       Subject: I'm talking to myself.[](#l11.656)

       Content-Transfer-Encoding: 7bit[](#l11.657)

```
       Simple message.[](#l11.659)
```
```
       """))[](#l11.660)
```

def test_set_headers_from_strings(self):
```
   m = self._make_message()[](#l11.663)
```

   content = "Simple message.\n"[](#l11.664)

   raw_data_manager.set_content(m, content, headers=([](#l11.665)

```
       "X-Foo-Header: foo",[](#l11.666)
```

       "X-Bar-Header: bar",))[](#l11.667)

   self.assertEqual(str(m), textwrap.dedent("""\[](#l11.668)

       Content-Type: text/plain; charset="utf-8"[](#l11.669)

```
       X-Foo-Header: foo[](#l11.670)
```
```
       X-Bar-Header: bar[](#l11.671)
```

       Content-Transfer-Encoding: 7bit[](#l11.672)

```
       Simple message.[](#l11.674)
```
```
       """))[](#l11.675)
```

def test_set_headers_with_invalid_duplicate_string_header_raises(self):
```
   m = self._make_message()[](#l11.678)
```

   content = "Simple message.\n"[](#l11.679)

   with self.assertRaisesRegex(ValueError, 'Content-Type'):[](#l11.680)

       raw_data_manager.set_content(m, content, headers=([](#l11.681)

           "Content-Type: foo/bar",)[](#l11.682)

```
           )[](#l11.683)
```

def test_set_headers_with_invalid_duplicate_header_header_raises(self):
```
   m = self._make_message()[](#l11.686)
```

   content = "Simple message.\n"[](#l11.687)

   header_factory = self.policy.header_factory[](#l11.688)

   with self.assertRaisesRegex(ValueError, 'Content-Type'):[](#l11.689)

       raw_data_manager.set_content(m, content, headers=([](#l11.690)

           header_factory("Content-Type", " foo/bar"),)[](#l11.691)

```
           )[](#l11.692)
```

def test_set_headers_with_defective_string_header_raises(self):
```
   m = self._make_message()[](#l11.695)
```

   content = "Simple message.\n"[](#l11.696)

   with self.assertRaisesRegex(ValueError, 'a@fairly@@invalid@address'):[](#l11.697)

       raw_data_manager.set_content(m, content, headers=([](#l11.698)

           'To: a@fairly@@invalid@address',)[](#l11.699)

```
           )[](#l11.700)
```

       print(m['To'].defects)[](#l11.701)

def test_set_headers_with_defective_header_header_raises(self):
```
   m = self._make_message()[](#l11.704)
```

   content = "Simple message.\n"[](#l11.705)

   header_factory = self.policy.header_factory[](#l11.706)

   with self.assertRaisesRegex(ValueError, 'a@fairly@@invalid@address'):[](#l11.707)

       raw_data_manager.set_content(m, content, headers=([](#l11.708)

           header_factory('To', 'a@fairly@@invalid@address'),)[](#l11.709)

```
           )[](#l11.710)
```

       print(m['To'].defects)[](#l11.711)

def test_set_disposition_inline(self):
```
   m = self._make_message()[](#l11.714)
```

   m.set_content('foo', disposition='inline')[](#l11.715)

   self.assertEqual(m['Content-Disposition'], 'inline')[](#l11.716)

def test_set_disposition_attachment(self):
```
   m = self._make_message()[](#l11.719)
```

   m.set_content('foo', disposition='attachment')[](#l11.720)

   self.assertEqual(m['Content-Disposition'], 'attachment')[](#l11.721)

def test_set_disposition_foo(self):
```
   m = self._make_message()[](#l11.724)
```

   m.set_content('foo', disposition='foo')[](#l11.725)

   self.assertEqual(m['Content-Disposition'], 'foo')[](#l11.726)

XXX: we should have a 'strict' policy mode (beyond raise_on_defect) that
would cause 'foo' above to raise.

def test_set_filename(self):
```
   m = self._make_message()[](#l11.732)
```

   m.set_content('foo', filename='bar.txt')[](#l11.733)

   self.assertEqual(m['Content-Disposition'],[](#l11.734)

                    'attachment; filename="bar.txt"')[](#l11.735)

def test_set_filename_and_disposition_inline(self):
```
   m = self._make_message()[](#l11.738)
```

   m.set_content('foo', disposition='inline', filename='bar.txt')[](#l11.739)

   self.assertEqual(m['Content-Disposition'], 'inline; filename="bar.txt"')[](#l11.740)

def test_set_non_ascii_filename(self):
```
   m = self._make_message()[](#l11.743)
```

   m.set_content('foo', filename='ábárî.txt')[](#l11.744)

   self.assertEqual(bytes(m), textwrap.dedent("""\[](#l11.745)

       Content-Type: text/plain; charset="utf-8"[](#l11.746)

       Content-Transfer-Encoding: 7bit[](#l11.747)

       Content-Disposition: attachment;[](#l11.748)

        filename*=utf-8''%C3%A1b%C3%A1r%C3%AE.txt[](#l11.749)

```
       MIME-Version: 1.0[](#l11.750)
```

```
       foo[](#l11.752)
```

       """).encode('ascii'))[](#l11.753)

content_object_params = {

   'text_plain': ('content', ()),[](#l11.756)

   'text_html': ('content', ('html',)),[](#l11.757)

   'application_octet_stream': (b'content',[](#l11.758)

                                ('application', 'octet_stream')),[](#l11.759)

   'image_jpeg': (b'content', ('image', 'jpeg')),[](#l11.760)

   'message_rfc822': (message(), ()),[](#l11.761)

   'message_external_body': (message(), ('external-body',)),[](#l11.762)

```
   }[](#l11.763)
```

def content_object_as_header_receiver(self, obj, mimetype):
```
   m = self._make_message()[](#l11.766)
```

   m.set_content(obj, *mimetype, headers=([](#l11.767)

       'To: foo@example.com',[](#l11.768)

       'From: bar@simple.net'))[](#l11.769)

   self.assertEqual(m['to'], 'foo@example.com')[](#l11.770)

   self.assertEqual(m['from'], 'bar@simple.net')[](#l11.771)

def content_object_as_disposition_inline_receiver(self, obj, mimetype):
```
   m = self._make_message()[](#l11.774)
```

   m.set_content(obj, *mimetype, disposition='inline')[](#l11.775)

   self.assertEqual(m['Content-Disposition'], 'inline')[](#l11.776)

def content_object_as_non_ascii_filename_receiver(self, obj, mimetype):
```
   m = self._make_message()[](#l11.779)
```

   m.set_content(obj, *mimetype, disposition='inline', filename='bár.txt')[](#l11.780)

   self.assertEqual(m['Content-Disposition'], 'inline; filename="bár.txt"')[](#l11.781)

   self.assertEqual(m.get_filename(), "bár.txt")[](#l11.782)

   self.assertEqual(m['Content-Disposition'].params['filename'], "bár.txt")[](#l11.783)

def content_object_as_cid_receiver(self, obj, mimetype):
```
   m = self._make_message()[](#l11.786)
```

   m.set_content(obj, *mimetype, cid='some_random_stuff')[](#l11.787)

   self.assertEqual(m['Content-ID'], 'some_random_stuff')[](#l11.788)

def content_object_as_params_receiver(self, obj, mimetype):
```
   m = self._make_message()[](#l11.791)
```

   params = {'foo': 'bár', 'abc': 'xyz'}[](#l11.792)

   m.set_content(obj, *mimetype, params=params)[](#l11.793)

```
   if isinstance(obj, str):[](#l11.794)
```

       params['charset'] = 'utf-8'[](#l11.795)

   self.assertEqual(m['Content-Type'].params, params)[](#l11.796)

+ + +if name == 'main':

unittest.main()

--- a/Lib/test/test_email/test_headerregistry.py +++ b/Lib/test/test_email/test_headerregistry.py @@ -661,7 +661,7 @@ class TestContentTypeHeader(TestHeaderBa 'text/plain; name="ascii_is_the_default"'), 'rfc2231_bad_character_in_charset_parameter_value': (

       "text/plain; charset*=ascii''utf-8%E2%80%9D",[](#l12.7)

       "text/plain; charset*=ascii''utf-8%F1%F2%F3",[](#l12.8)
       'text/plain',[](#l12.9)
       'text',[](#l12.10)
       'plain',[](#l12.11)

@@ -669,6 +669,18 @@ class TestContentTypeHeader(TestHeaderBa [errors.UndecodableBytesDefect], 'text/plain; charset="utf-8\uFFFD\uFFFD\uFFFD"'),

   'rfc2231_utf_8_in_supposedly_ascii_charset_parameter_value': ([](#l12.16)

       "text/plain; charset*=ascii''utf-8%E2%80%9D",[](#l12.17)

```
       'text/plain',[](#l12.18)
```
```
       'text',[](#l12.19)
```
```
       'plain',[](#l12.20)
```

       {'charset': 'utf-8”'},[](#l12.21)

       [errors.UndecodableBytesDefect],[](#l12.22)

       'text/plain; charset="utf-8”"',[](#l12.23)

```
       ),[](#l12.24)
```

       # XXX: if the above were *re*folded, it would get tagged as utf-8[](#l12.25)

       # instead of ascii in the param, since it now contains non-ASCII.[](#l12.26)

+ 'rfc2231_encoded_then_unencoded_segments': ( ('application/x-foo;' '\tname0="us-ascii'en-us'My";'

--- a/Lib/test/test_email/test_message.py +++ b/Lib/test/test_email/test_message.py @@ -1,6 +1,13 @@ import unittest +import textwrap from email import policy -from test.test_email import TestEmailBase +from email.message import EmailMessage, MIMEPart +from test.test_email import TestEmailBase, parameterize + + +# Helper. +def first(iterable):

return next(filter(lambda x: x is not None, iterable), None)

class Test(TestEmailBase): @@ -14,5 +21,738 @@ class Test(TestEmailBase): m['To'] = 'xyz@abc' +@parameterize +class TestEmailMessageBase: +

policy = policy.default

The first argument is a triple (related, html, plain) of indices into the
list returned by 'walk' called on a Message constructed from the third.
The indices indicate which part should match the corresponding part-type
when passed to get_body (ie: the "first" part of that type in the
message). The second argument is a list of indices into the 'walk' list
of the attachments that should be returned by a call to
'iter_attachments'. The third argument is a list of indices into 'walk'
that should be returned by a call to 'iter_parts'. Note that the first
item returned by 'walk' is the Message itself.

message_params = {

```
   'empty_message': ([](#l13.39)
```
```
       (None, None, 0),[](#l13.40)
```
```
       (),[](#l13.41)
```
```
       (),[](#l13.42)
```
```
       ""),[](#l13.43)
```

```
   'non_mime_plain': ([](#l13.45)
```
```
       (None, None, 0),[](#l13.46)
```
```
       (),[](#l13.47)
```
```
       (),[](#l13.48)
```
```
       textwrap.dedent("""\[](#l13.49)
```

           To: foo@example.com[](#l13.50)

```
           simple text body[](#l13.52)
```
```
           """)),[](#l13.53)
```

```
   'mime_non_text': ([](#l13.55)
```
```
       (None, None, None),[](#l13.56)
```
```
       (),[](#l13.57)
```
```
       (),[](#l13.58)
```
```
       textwrap.dedent("""\[](#l13.59)
```

           To: foo@example.com[](#l13.60)

```
           MIME-Version: 1.0[](#l13.61)
```

           Content-Type: image/jpg[](#l13.62)

```
           bogus body.[](#l13.64)
```
```
           """)),[](#l13.65)
```

   'plain_html_alternative': ([](#l13.67)

```
       (None, 2, 1),[](#l13.68)
```
```
       (),[](#l13.69)
```
```
       (1, 2),[](#l13.70)
```
```
       textwrap.dedent("""\[](#l13.71)
```

           To: foo@example.com[](#l13.72)

```
           MIME-Version: 1.0[](#l13.73)
```

           Content-Type: multipart/alternative; boundary="==="[](#l13.74)

```
           preamble[](#l13.76)
```

```
           --===[](#l13.78)
```

           Content-Type: text/plain[](#l13.79)

```
           simple body[](#l13.81)
```

```
           --===[](#l13.83)
```

           Content-Type: text/html[](#l13.84)

           <p>simple body</p>[](#l13.86)

```
           --===--[](#l13.87)
```
```
           """)),[](#l13.88)
```

```
   'plain_html_mixed': ([](#l13.90)
```
```
       (None, 2, 1),[](#l13.91)
```
```
       (),[](#l13.92)
```
```
       (1, 2),[](#l13.93)
```
```
       textwrap.dedent("""\[](#l13.94)
```

           To: foo@example.com[](#l13.95)

```
           MIME-Version: 1.0[](#l13.96)
```

           Content-Type: multipart/mixed; boundary="==="[](#l13.97)

```
           preamble[](#l13.99)
```

```
           --===[](#l13.101)
```

           Content-Type: text/plain[](#l13.102)

```
           simple body[](#l13.104)
```

```
           --===[](#l13.106)
```

           Content-Type: text/html[](#l13.107)

           <p>simple body</p>[](#l13.109)

```
           --===--[](#l13.111)
```
```
           """)),[](#l13.112)
```

   'plain_html_attachment_mixed': ([](#l13.114)

```
       (None, None, 1),[](#l13.115)
```
```
       (2,),[](#l13.116)
```
```
       (1, 2),[](#l13.117)
```
```
       textwrap.dedent("""\[](#l13.118)
```

           To: foo@example.com[](#l13.119)

           MIME-Version: 1.0[](#l13.120)

           Content-Type: multipart/mixed; boundary="==="[](#l13.121)

```
           --===[](#l13.123)
```

           Content-Type: text/plain[](#l13.124)

```
           simple body[](#l13.126)
```

```
           --===[](#l13.128)
```

           Content-Type: text/html[](#l13.129)

           Content-Disposition: attachment[](#l13.130)

           <p>simple body</p>[](#l13.132)

```
           --===--[](#l13.134)
```
```
           """)),[](#l13.135)
```

   'html_text_attachment_mixed': ([](#l13.137)

```
       (None, 2, None),[](#l13.138)
```
```
       (1,),[](#l13.139)
```
```
       (1, 2),[](#l13.140)
```
```
       textwrap.dedent("""\[](#l13.141)
```

           To: foo@example.com[](#l13.142)

           MIME-Version: 1.0[](#l13.143)

           Content-Type: multipart/mixed; boundary="==="[](#l13.144)

```
           --===[](#l13.146)
```

           Content-Type: text/plain[](#l13.147)

           Content-Disposition: AtTaChment[](#l13.148)

```
           simple body[](#l13.150)
```

```
           --===[](#l13.152)
```

           Content-Type: text/html[](#l13.153)

           <p>simple body</p>[](#l13.155)

```
           --===--[](#l13.157)
```
```
           """)),[](#l13.158)
```

   'html_text_attachment_inline_mixed': ([](#l13.160)

```
       (None, 2, 1),[](#l13.161)
```
```
       (),[](#l13.162)
```
```
       (1, 2),[](#l13.163)
```
```
       textwrap.dedent("""\[](#l13.164)
```

           To: foo@example.com[](#l13.165)

           MIME-Version: 1.0[](#l13.166)

           Content-Type: multipart/mixed; boundary="==="[](#l13.167)

```
           --===[](#l13.169)
```

           Content-Type: text/plain[](#l13.170)

           Content-Disposition: InLine[](#l13.171)

```
           simple body[](#l13.173)
```

```
           --===[](#l13.175)
```

           Content-Type: text/html[](#l13.176)

           Content-Disposition: inline[](#l13.177)

           <p>simple body</p>[](#l13.179)

```
           --===--[](#l13.181)
```
```
           """)),[](#l13.182)
```

```
   # RFC 2387[](#l13.184)
```
```
   'related': ([](#l13.185)
```
```
       (0, 1, None),[](#l13.186)
```
```
       (2,),[](#l13.187)
```
```
       (1, 2),[](#l13.188)
```
```
       textwrap.dedent("""\[](#l13.189)
```

           To: foo@example.com[](#l13.190)

           MIME-Version: 1.0[](#l13.191)

           Content-Type: multipart/related; boundary="==="; type=text/html[](#l13.192)

```
           --===[](#l13.194)
```

           Content-Type: text/html[](#l13.195)

           <p>simple body</p>[](#l13.197)

```
           --===[](#l13.199)
```

           Content-Type: image/jpg[](#l13.200)

           Content-ID: <image1>[](#l13.201)

```
           bogus data[](#l13.203)
```

```
           --===--[](#l13.205)
```
```
           """)),[](#l13.206)
```

   # This message structure will probably never be seen in the wild, but[](#l13.208)

   # it proves we distinguish between text parts based on 'start'.  The[](#l13.209)

   # content would not, of course, actually work :)[](#l13.210)

```
   'related_with_start': ([](#l13.211)
```
```
       (0, 2, None),[](#l13.212)
```
```
       (1,),[](#l13.213)
```
```
       (1, 2),[](#l13.214)
```
```
       textwrap.dedent("""\[](#l13.215)
```

           To: foo@example.com[](#l13.216)

           MIME-Version: 1.0[](#l13.217)

           Content-Type: multipart/related; boundary="==="; type=text/html;[](#l13.218)

```
            start="<body>"[](#l13.219)
```

```
           --===[](#l13.221)
```

           Content-Type: text/html[](#l13.222)

           Content-ID: <include>[](#l13.223)

```
           useless text[](#l13.225)
```

```
           --===[](#l13.227)
```

           Content-Type: text/html[](#l13.228)

           Content-ID: <body>[](#l13.229)

           <p>simple body</p>[](#l13.231)

           <!--#include file="<include>"-->[](#l13.232)

```
           --===--[](#l13.234)
```
```
           """)),[](#l13.235)
```

+ +

   'mixed_alternative_plain_related': ([](#l13.238)

```
       (3, 4, 2),[](#l13.239)
```
```
       (6, 7),[](#l13.240)
```
```
       (1, 6, 7),[](#l13.241)
```
```
       textwrap.dedent("""\[](#l13.242)
```

           To: foo@example.com[](#l13.243)

           MIME-Version: 1.0[](#l13.244)

           Content-Type: multipart/mixed; boundary="==="[](#l13.245)

```
           --===[](#l13.247)
```

           Content-Type: multipart/alternative; boundary="+++"[](#l13.248)

```
           --+++[](#l13.250)
```

           Content-Type: text/plain[](#l13.251)

```
           simple body[](#l13.253)
```

```
           --+++[](#l13.255)
```

           Content-Type: multipart/related; boundary="___"[](#l13.256)

```
           --___[](#l13.258)
```

           Content-Type: text/html[](#l13.259)

           <p>simple body</p>[](#l13.261)

```
           --___[](#l13.263)
```

           Content-Type: image/jpg[](#l13.264)

           Content-ID: <image1@cid>[](#l13.265)

```
           bogus jpg body[](#l13.267)
```

```
           --___--[](#l13.269)
```

```
           --+++--[](#l13.271)
```

```
           --===[](#l13.273)
```

           Content-Type: image/jpg[](#l13.274)

           Content-Disposition: attachment[](#l13.275)

```
           bogus jpg body[](#l13.277)
```

```
           --===[](#l13.279)
```

           Content-Type: image/jpg[](#l13.280)

           Content-Disposition: AttacHmenT[](#l13.281)

           another bogus jpg body[](#l13.283)

```
           --===--[](#l13.285)
```
```
           """)),[](#l13.286)
```

   # This structure suggested by Stephen J. Turnbull...may not exist/be[](#l13.288)

   # supported in the wild, but we want to support it.[](#l13.289)

   'mixed_related_alternative_plain_html': ([](#l13.290)

```
       (1, 4, 3),[](#l13.291)
```
```
       (6, 7),[](#l13.292)
```
```
       (1, 6, 7),[](#l13.293)
```
```
       textwrap.dedent("""\[](#l13.294)
```

           To: foo@example.com[](#l13.295)

           MIME-Version: 1.0[](#l13.296)

           Content-Type: multipart/mixed; boundary="==="[](#l13.297)

```
           --===[](#l13.299)
```

           Content-Type: multipart/related; boundary="+++"[](#l13.300)

```
           --+++[](#l13.302)
```

           Content-Type: multipart/alternative; boundary="___"[](#l13.303)

```
           --___[](#l13.305)
```

           Content-Type: text/plain[](#l13.306)

```
           simple body[](#l13.308)
```

```
           --___[](#l13.310)
```

           Content-Type: text/html[](#l13.311)

           <p>simple body</p>[](#l13.313)

```
           --___--[](#l13.315)
```

```
           --+++[](#l13.317)
```

           Content-Type: image/jpg[](#l13.318)

           Content-ID: <image1@cid>[](#l13.319)

```
           bogus jpg body[](#l13.321)
```

```
           --+++--[](#l13.323)
```

```
           --===[](#l13.325)
```

           Content-Type: image/jpg[](#l13.326)

           Content-Disposition: attachment[](#l13.327)

```
           bogus jpg body[](#l13.329)
```

```
           --===[](#l13.331)
```

           Content-Type: image/jpg[](#l13.332)

           Content-Disposition: attachment[](#l13.333)

           another bogus jpg body[](#l13.335)

```
           --===--[](#l13.337)
```
```
           """)),[](#l13.338)
```

   # Same thing, but proving we only look at the root part, which is the[](#l13.340)

   # first one if there isn't any start parameter.  That is, this is a[](#l13.341)

```
   # broken related.[](#l13.342)
```

   'mixed_related_alternative_plain_html_wrong_order': ([](#l13.343)

```
       (1, None, None),[](#l13.344)
```
```
       (6, 7),[](#l13.345)
```
```
       (1, 6, 7),[](#l13.346)
```
```
       textwrap.dedent("""\[](#l13.347)
```

           To: foo@example.com[](#l13.348)

           MIME-Version: 1.0[](#l13.349)

           Content-Type: multipart/mixed; boundary="==="[](#l13.350)

```
           --===[](#l13.352)
```

           Content-Type: multipart/related; boundary="+++"[](#l13.353)

```
           --+++[](#l13.355)
```

           Content-Type: image/jpg[](#l13.356)

           Content-ID: <image1@cid>[](#l13.357)

```
           bogus jpg body[](#l13.359)
```

```
           --+++[](#l13.361)
```

           Content-Type: multipart/alternative; boundary="___"[](#l13.362)

```
           --___[](#l13.364)
```

           Content-Type: text/plain[](#l13.365)

```
           simple body[](#l13.367)
```

```
           --___[](#l13.369)
```

           Content-Type: text/html[](#l13.370)

           <p>simple body</p>[](#l13.372)

```
           --___--[](#l13.374)
```

```
           --+++--[](#l13.376)
```

```
           --===[](#l13.378)
```

           Content-Type: image/jpg[](#l13.379)

           Content-Disposition: attachment[](#l13.380)

```
           bogus jpg body[](#l13.382)
```

```
           --===[](#l13.384)
```

           Content-Type: image/jpg[](#l13.385)

           Content-Disposition: attachment[](#l13.386)

           another bogus jpg body[](#l13.388)

```
           --===--[](#l13.390)
```
```
           """)),[](#l13.391)
```

```
   'message_rfc822': ([](#l13.393)
```
```
       (None, None, None),[](#l13.394)
```
```
       (),[](#l13.395)
```
```
       (),[](#l13.396)
```
```
       textwrap.dedent("""\[](#l13.397)
```

           To: foo@example.com[](#l13.398)

           MIME-Version: 1.0[](#l13.399)

           Content-Type: message/rfc822[](#l13.400)

           To: bar@example.com[](#l13.402)

           From: robot@examp.com[](#l13.403)

           this is a message body.[](#l13.405)

```
           """)),[](#l13.406)
```

   'mixed_text_message_rfc822': ([](#l13.408)

```
       (None, None, 1),[](#l13.409)
```
```
       (2,),[](#l13.410)
```
```
       (1, 2),[](#l13.411)
```
```
       textwrap.dedent("""\[](#l13.412)
```

           To: foo@example.com[](#l13.413)

           MIME-Version: 1.0[](#l13.414)

           Content-Type: multipart/mixed; boundary="==="[](#l13.415)

```
           --===[](#l13.417)
```

           Content-Type: text/plain[](#l13.418)

           Your message has bounced, ser.[](#l13.420)

```
           --===[](#l13.422)
```

           Content-Type: message/rfc822[](#l13.423)

           To: bar@example.com[](#l13.425)

           From: robot@examp.com[](#l13.426)

           this is a message body.[](#l13.428)

```
           --===--[](#l13.430)
```
```
           """)),[](#l13.431)
```

```
    }[](#l13.433)
```

def message_as_get_body(self, body_parts, attachments, parts, msg):
```
   m = self._str_msg(msg)[](#l13.436)
```

   allparts = list(m.walk())[](#l13.437)

   expected = [None if n is None else allparts[n] for n in body_parts][](#l13.438)

   related = 0; html = 1; plain = 2[](#l13.439)

   self.assertEqual(m.get_body(), first(expected))[](#l13.440)

   self.assertEqual(m.get_body(preferencelist=([](#l13.441)

                                   'related', 'html', 'plain')),[](#l13.442)

                    first(expected))[](#l13.443)

   self.assertEqual(m.get_body(preferencelist=('related', 'html')),[](#l13.444)

                    first(expected[related:html+1]))[](#l13.445)

   self.assertEqual(m.get_body(preferencelist=('related', 'plain')),[](#l13.446)

                    first([expected[related], expected[plain]]))[](#l13.447)

   self.assertEqual(m.get_body(preferencelist=('html', 'plain')),[](#l13.448)

                    first(expected[html:plain+1]))[](#l13.449)

   self.assertEqual(m.get_body(preferencelist=['related']),[](#l13.450)

                    expected[related])[](#l13.451)

   self.assertEqual(m.get_body(preferencelist=['html']), expected[html])[](#l13.452)

   self.assertEqual(m.get_body(preferencelist=['plain']), expected[plain])[](#l13.453)

   self.assertEqual(m.get_body(preferencelist=('plain', 'html')),[](#l13.454)

                    first(expected[plain:html-1:-1]))[](#l13.455)

   self.assertEqual(m.get_body(preferencelist=('plain', 'related')),[](#l13.456)

                    first([expected[plain], expected[related]]))[](#l13.457)

   self.assertEqual(m.get_body(preferencelist=('html', 'related')),[](#l13.458)

                    first(expected[html::-1]))[](#l13.459)

   self.assertEqual(m.get_body(preferencelist=('plain', 'html', 'related')),[](#l13.460)

                    first(expected[::-1]))[](#l13.461)

   self.assertEqual(m.get_body(preferencelist=('html', 'plain', 'related')),[](#l13.462)

                    first([expected[html],[](#l13.463)

                           expected[plain],[](#l13.464)

                           expected[related]]))[](#l13.465)

def message_as_iter_attachment(self, body_parts, attachments, parts, msg):
```
   m = self._str_msg(msg)[](#l13.468)
```

   allparts = list(m.walk())[](#l13.469)

   attachments = [allparts[n] for n in attachments][](#l13.470)

   self.assertEqual(list(m.iter_attachments()), attachments)[](#l13.471)

def message_as_iter_parts(self, body_parts, attachments, parts, msg):
```
   m = self._str_msg(msg)[](#l13.474)
```

   allparts = list(m.walk())[](#l13.475)

   parts = [allparts[n] for n in parts][](#l13.476)

   self.assertEqual(list(m.iter_parts()), parts)[](#l13.477)

class _TestContentManager:

   def get_content(self, msg, *args, **kw):[](#l13.480)

```
       return msg, args, kw[](#l13.481)
```

   def set_content(self, msg, *args, **kw):[](#l13.482)

```
       self.msg = msg[](#l13.483)
```
```
       self.args = args[](#l13.484)
```
```
       self.kw = kw[](#l13.485)
```

def test_get_content_with_cm(self):
```
   m = self._str_msg('')[](#l13.488)
```

   cm = self._TestContentManager()[](#l13.489)

   self.assertEqual(m.get_content(content_manager=cm), (m, (), {}))[](#l13.490)

   msg, args, kw = m.get_content('foo', content_manager=cm, bar=1, k=2)[](#l13.491)

```
   self.assertEqual(msg, m)[](#l13.492)
```

   self.assertEqual(args, ('foo',))[](#l13.493)

   self.assertEqual(kw, dict(bar=1, k=2))[](#l13.494)

def test_get_content_default_cm_comes_from_policy(self):

   p = policy.default.clone(content_manager=self._TestContentManager())[](#l13.497)

   m = self._str_msg('', policy=p)[](#l13.498)

   self.assertEqual(m.get_content(), (m, (), {}))[](#l13.499)

   msg, args, kw = m.get_content('foo', bar=1, k=2)[](#l13.500)

```
   self.assertEqual(msg, m)[](#l13.501)
```

   self.assertEqual(args, ('foo',))[](#l13.502)

   self.assertEqual(kw, dict(bar=1, k=2))[](#l13.503)

def test_set_content_with_cm(self):
```
   m = self._str_msg('')[](#l13.506)
```

   cm = self._TestContentManager()[](#l13.507)

   m.set_content(content_manager=cm)[](#l13.508)

   self.assertEqual(cm.msg, m)[](#l13.509)

   self.assertEqual(cm.args, ())[](#l13.510)

   self.assertEqual(cm.kw, {})[](#l13.511)

   m.set_content('foo', content_manager=cm, bar=1, k=2)[](#l13.512)

   self.assertEqual(cm.msg, m)[](#l13.513)

   self.assertEqual(cm.args, ('foo',))[](#l13.514)

   self.assertEqual(cm.kw, dict(bar=1, k=2))[](#l13.515)

def test_set_content_default_cm_comes_from_policy(self):

   cm = self._TestContentManager()[](#l13.518)

   p = policy.default.clone(content_manager=cm)[](#l13.519)

   m = self._str_msg('', policy=p)[](#l13.520)

```
   m.set_content()[](#l13.521)
```

   self.assertEqual(cm.msg, m)[](#l13.522)

   self.assertEqual(cm.args, ())[](#l13.523)

   self.assertEqual(cm.kw, {})[](#l13.524)

   m.set_content('foo', bar=1, k=2)[](#l13.525)

   self.assertEqual(cm.msg, m)[](#l13.526)

   self.assertEqual(cm.args, ('foo',))[](#l13.527)

   self.assertEqual(cm.kw, dict(bar=1, k=2))[](#l13.528)

outcome is whether xxx_method should raise ValueError error when called
on multipart/subtype. Blank outcome means it depends on xxx (add
succeeds, make raises). Note: 'none' means there are content-type
headers but payload is None...this happening in practice would be very
unusual, so treating it as if there were content seems reasonable.
method subtype outcome
subtype_params = (

   ('related',      'no_content',     'succeeds'),[](#l13.537)

   ('related',      'none',           'succeeds'),[](#l13.538)

   ('related',      'plain',          'succeeds'),[](#l13.539)

   ('related',      'related',        ''),[](#l13.540)

   ('related',      'alternative',    'raises'),[](#l13.541)

   ('related',      'mixed',          'raises'),[](#l13.542)

   ('alternative',  'no_content',     'succeeds'),[](#l13.543)

   ('alternative',  'none',           'succeeds'),[](#l13.544)

   ('alternative',  'plain',          'succeeds'),[](#l13.545)

   ('alternative',  'related',        'succeeds'),[](#l13.546)

   ('alternative',  'alternative',    ''),[](#l13.547)

   ('alternative',  'mixed',          'raises'),[](#l13.548)

   ('mixed',        'no_content',     'succeeds'),[](#l13.549)

   ('mixed',        'none',           'succeeds'),[](#l13.550)

   ('mixed',        'plain',          'succeeds'),[](#l13.551)

   ('mixed',        'related',        'succeeds'),[](#l13.552)

   ('mixed',        'alternative',    'succeeds'),[](#l13.553)

   ('mixed',        'mixed',          ''),[](#l13.554)

```
   )[](#l13.555)
```

def _make_subtype_test_message(self, subtype):
```
   m = self.message()[](#l13.558)
```
```
   payload = None[](#l13.559)
```
```
   msg_headers =  [[](#l13.560)
```

       ('To', 'foo@bar.com'),[](#l13.561)

       ('From', 'bar@foo.com'),[](#l13.562)

```
       ][](#l13.563)
```

   if subtype != 'no_content':[](#l13.564)

       ('content-shadow', 'Logrus'),[](#l13.565)

   msg_headers.append(('X-Random-Header', 'Corwin'))[](#l13.566)

```
   if subtype == 'text':[](#l13.567)
```
```
       payload = ''[](#l13.568)
```

       msg_headers.append(('Content-Type', 'text/plain'))[](#l13.569)

```
       m.set_payload('')[](#l13.570)
```

   elif subtype != 'no_content':[](#l13.571)

```
       payload = [][](#l13.572)
```

       msg_headers.append(('Content-Type', 'multipart/' + subtype))[](#l13.573)

   msg_headers.append(('X-Trump', 'Random'))[](#l13.574)

```
   m.set_payload(payload)[](#l13.575)
```

   for name, value in msg_headers:[](#l13.576)

```
       m[name] = value[](#l13.577)
```

   return m, msg_headers, payload[](#l13.578)

def _check_disallowed_subtype_raises(self, m, method_name, subtype, method):

   with self.assertRaises(ValueError) as ar:[](#l13.581)

```
       getattr(m, method)()[](#l13.582)
```

   exc_text = str(ar.exception)[](#l13.583)

   self.assertIn(subtype, exc_text)[](#l13.584)

   self.assertIn(method_name, exc_text)[](#l13.585)

def _check_make_multipart(self, m, msg_headers, payload):
```
   count = 0[](#l13.588)
```

   for name, value in msg_headers:[](#l13.589)

       if not name.lower().startswith('content-'):[](#l13.590)

           self.assertEqual(m[name], value)[](#l13.591)

```
           count += 1[](#l13.592)
```

   self.assertEqual(len(m), count+1) # +1 for new Content-Type[](#l13.593)

   part = next(m.iter_parts())[](#l13.594)

```
   count = 0[](#l13.595)
```

   for name, value in msg_headers:[](#l13.596)

       if name.lower().startswith('content-'):[](#l13.597)

           self.assertEqual(part[name], value)[](#l13.598)

```
           count += 1[](#l13.599)
```

   self.assertEqual(len(part), count)[](#l13.600)

   self.assertEqual(part.get_payload(), payload)[](#l13.601)

def subtype_as_make(self, method, subtype, outcome):

   m, msg_headers, payload = self._make_subtype_test_message(subtype)[](#l13.604)

   make_method = 'make_' + method[](#l13.605)

   if outcome in ('', 'raises'):[](#l13.606)

       self._check_disallowed_subtype_raises(m, method, subtype, make_method)[](#l13.607)

```
       return[](#l13.608)
```

   getattr(m, make_method)()[](#l13.609)

   self.assertEqual(m.get_content_maintype(), 'multipart')[](#l13.610)

   self.assertEqual(m.get_content_subtype(), method)[](#l13.611)

   if subtype == 'no_content':[](#l13.612)

       self.assertEqual(len(m.get_payload()), 0)[](#l13.613)

       self.assertEqual(m.items(),[](#l13.614)

                        msg_headers + [('Content-Type',[](#l13.615)

                                        'multipart/'+method)])[](#l13.616)

```
   else:[](#l13.617)
```

       self.assertEqual(len(m.get_payload()), 1)[](#l13.618)

       self._check_make_multipart(m, msg_headers, payload)[](#l13.619)

def subtype_as_make_with_boundary(self, method, subtype, outcome):

   # Doing all variation is a bit of overkill...[](#l13.622)

```
   m = self.message()[](#l13.623)
```

   if outcome in ('', 'raises'):[](#l13.624)

       m['Content-Type'] = 'multipart/' + subtype[](#l13.625)

       with self.assertRaises(ValueError) as cm:[](#l13.626)

           getattr(m, 'make_' + method)()[](#l13.627)

```
       return[](#l13.628)
```
```
   if subtype == 'plain':[](#l13.629)
```

       m['Content-Type'] = 'text/plain'[](#l13.630)

   elif subtype != 'no_content':[](#l13.631)

       m['Content-Type'] = 'multipart/' + subtype[](#l13.632)

   getattr(m, 'make_' + method)(boundary="abc")[](#l13.633)

   self.assertTrue(m.is_multipart())[](#l13.634)

   self.assertEqual(m.get_boundary(), 'abc')[](#l13.635)

def test_policy_on_part_made_by_make_comes_from_message(self):

   for method in ('make_related', 'make_alternative', 'make_mixed'):[](#l13.638)

       m = self.message(policy=self.policy.clone(content_manager='foo'))[](#l13.639)

       m['Content-Type'] = 'text/plain'[](#l13.640)

```
       getattr(m, method)()[](#l13.641)
```

       self.assertEqual(m.get_payload(0).policy.content_manager, 'foo')[](#l13.642)

class _TestSetContentManager:

   def set_content(self, msg, content, *args, **kw):[](#l13.645)

       msg['Content-Type'] = 'text/plain'[](#l13.646)

       msg.set_payload(content)[](#l13.647)

def subtype_as_add(self, method, subtype, outcome):

   m, msg_headers, payload = self._make_subtype_test_message(subtype)[](#l13.650)

   cm = self._TestSetContentManager()[](#l13.651)

   add_method = 'add_attachment' if method=='mixed' else 'add_' + method[](#l13.652)

```
   if outcome == 'raises':[](#l13.653)
```

       self._check_disallowed_subtype_raises(m, method, subtype, add_method)[](#l13.654)

```
       return[](#l13.655)
```

   getattr(m, add_method)('test', content_manager=cm)[](#l13.656)

   self.assertEqual(m.get_content_maintype(), 'multipart')[](#l13.657)

   self.assertEqual(m.get_content_subtype(), method)[](#l13.658)

   if method == subtype or subtype == 'no_content':[](#l13.659)

       self.assertEqual(len(m.get_payload()), 1)[](#l13.660)

       for name, value in msg_headers:[](#l13.661)

           self.assertEqual(m[name], value)[](#l13.662)

       part = m.get_payload()[0][](#l13.663)

```
   else:[](#l13.664)
```

       self.assertEqual(len(m.get_payload()), 2)[](#l13.665)

       self._check_make_multipart(m, msg_headers, payload)[](#l13.666)

       part = m.get_payload()[1][](#l13.667)

   self.assertEqual(part.get_content_type(), 'text/plain')[](#l13.668)

   self.assertEqual(part.get_payload(), 'test')[](#l13.669)

```
   if method=='mixed':[](#l13.670)
```

       self.assertEqual(part['Content-Disposition'], 'attachment')[](#l13.671)

```
   elif method=='related':[](#l13.672)
```

       self.assertEqual(part['Content-Disposition'], 'inline')[](#l13.673)

```
   else:[](#l13.674)
```

       # Otherwise we don't guess.[](#l13.675)

       self.assertIsNone(part['Content-Disposition'])[](#l13.676)

class _TestSetRaisingContentManager:

   def set_content(self, msg, content, *args, **kw):[](#l13.679)

       raise Exception('test')[](#l13.680)

def test_default_content_manager_for_add_comes_from_policy(self):

   cm = self._TestSetRaisingContentManager()[](#l13.683)

   m = self.message(policy=self.policy.clone(content_manager=cm))[](#l13.684)

   for method in ('add_related', 'add_alternative', 'add_attachment'):[](#l13.685)

       with self.assertRaises(Exception) as ar:[](#l13.686)

           getattr(m, method)('')[](#l13.687)

       self.assertEqual(str(ar.exception), 'test')[](#l13.688)

def message_as_clear(self, body_parts, attachments, parts, msg):
```
   m = self._str_msg(msg)[](#l13.691)
```
```
   m.clear()[](#l13.692)
```

   self.assertEqual(len(m), 0)[](#l13.693)

   self.assertEqual(list(m.items()), [])[](#l13.694)

   self.assertIsNone(m.get_payload())[](#l13.695)

   self.assertEqual(list(m.iter_parts()), [])[](#l13.696)

def message_as_clear_content(self, body_parts, attachments, parts, msg):
```
   m = self._str_msg(msg)[](#l13.699)
```

   expected_headers = [h for h in m.keys()[](#l13.700)

                       if not h.lower().startswith('content-')][](#l13.701)

```
   m.clear_content()[](#l13.702)
```

   self.assertEqual(list(m.keys()), expected_headers)[](#l13.703)

   self.assertIsNone(m.get_payload())[](#l13.704)

   self.assertEqual(list(m.iter_parts()), [])[](#l13.705)

def test_is_attachment(self):
```
   m = self._make_message()[](#l13.708)
```

   self.assertFalse(m.is_attachment)[](#l13.709)

   m['Content-Disposition'] = 'inline'[](#l13.710)

   self.assertFalse(m.is_attachment)[](#l13.711)

   m.replace_header('Content-Disposition', 'attachment')[](#l13.712)

   self.assertTrue(m.is_attachment)[](#l13.713)

   m.replace_header('Content-Disposition', 'AtTachMent')[](#l13.714)

   self.assertTrue(m.is_attachment)[](#l13.715)

+ + + +class TestEmailMessage(TestEmailMessageBase, TestEmailBase):

message = EmailMessage

def test_set_content_adds_MIME_Version(self):
```
   m = self._str_msg('')[](#l13.723)
```

   cm = self._TestContentManager()[](#l13.724)

   self.assertNotIn('MIME-Version', m)[](#l13.725)

   m.set_content(content_manager=cm)[](#l13.726)

   self.assertEqual(m['MIME-Version'], '1.0')[](#l13.727)

class _MIME_Version_adding_CM:

   def set_content(self, msg, *args, **kw):[](#l13.730)

       msg['MIME-Version'] = '1.0'[](#l13.731)

def test_set_content_does_not_duplicate_MIME_Version(self):
```
   m = self._str_msg('')[](#l13.734)
```

   cm = self._MIME_Version_adding_CM()[](#l13.735)

   self.assertNotIn('MIME-Version', m)[](#l13.736)

   m.set_content(content_manager=cm)[](#l13.737)

   self.assertEqual(m['MIME-Version'], '1.0')[](#l13.738)

+ + +class TestMIMEPart(TestEmailMessageBase, TestEmailBase):

Doing the full test run here may seem a bit redundant, since the two
classes are almost identical. But what if they drift apart? So we do
the full tests so that any future drift doesn't introduce bugs.
message = MIMEPart

def test_set_content_does_not_add_MIME_Version(self):
```
   m = self._str_msg('')[](#l13.748)
```

   cm = self._TestContentManager()[](#l13.749)

   self.assertNotIn('MIME-Version', m)[](#l13.750)

   m.set_content(content_manager=cm)[](#l13.751)

   self.assertNotIn('MIME-Version', m)[](#l13.752)

+ + if name == 'main': unittest.main()

--- a/Lib/test/test_email/test_policy.py +++ b/Lib/test/test_email/test_policy.py @@ -30,6 +30,7 @@ class PolicyAPITests(unittest.TestCase): 'raise_on_defect': False, 'header_factory': email.policy.EmailPolicy.header_factory, 'refold_source': 'long',

   'content_manager':          email.policy.EmailPolicy.content_manager,[](#l14.7)
   })[](#l14.8)

# For each policy under test, we give here what we expect the defaults to

--- a/Misc/NEWS +++ b/Misc/NEWS @@ -42,6 +42,9 @@ Core and Builtins Library ------- +- Issue #18891: Completed the new email package (provisional) API additions

by adding new classes EmailMessage, MIMEPart, and ContentManager. +

Issue #18468: The re.split, re.findall, and re.sub functions and the group() and groups() methods of match object now always return a string or a bytes object.

cpython: 6d12285e250b (original) (raw)

If we don't understand a message subtype, we are supposed to treat it as

if it were application/octet-stream, per

tools.ietf.org/html/rfc2046#section-5.2.4. Feedparser doesn't do that,

so do our best to fix things up. Note that it is not appropriate to

model message/partial content as Message objects, so they are handled

here as well. (How to reassemble them is out of scope for this comment :)

Intrapackage imports

How to deal with a string containing bytes before handing it to the

application through the 'normal' interface.

Turn any escaped bytes into unicode 'unknown' char.

Turn any escaped bytes into unicode 'unknown' char. If the escaped

bytes happen to be utf-8 they will instead get decoded, even if they

were invalid in the charset the source was supposed to be in. This

seems like it is not a bad thing; a defect was still registered.

Helpers

Likewise, the default message object is Message.

Note: these tests are dependent on the order in which headers are added

to the message objects by the code. There's no defined ordering in

RFC5322/MIME, so this makes the tests more fragile than the standards

require. However, if the header order changes it is best to understand

why, and make sure it isn't a subtle bug in whatever change was

applied.

XXX: we should have a 'strict' policy mode (beyond raise_on_defect) that

would cause 'foo' above to raise.

The first argument is a triple (related, html, plain) of indices into the

list returned by 'walk' called on a Message constructed from the third.

The indices indicate which part should match the corresponding part-type

when passed to get_body (ie: the "first" part of that type in the

message). The second argument is a list of indices into the 'walk' list

of the attachments that should be returned by a call to

'iter_attachments'. The third argument is a list of indices into 'walk'

that should be returned by a call to 'iter_parts'. Note that the first

item returned by 'walk' is the Message itself.

outcome is whether xxx_method should raise ValueError error when called

on multipart/subtype. Blank outcome means it depends on xxx (add

succeeds, make raises). Note: 'none' means there are content-type

headers but payload is None...this happening in practice would be very

unusual, so treating it as if there were content seems reasonable.

method subtype outcome

Doing the full test run here may seem a bit redundant, since the two

classes are almost identical. But what if they drift apart? So we do

the full tests so that any future drift doesn't introduce bugs.