Ensure an open connection to the email server. Return whether or not a new connection was required (True or False) or None if an exception passed silently.
(self)
| 111 | return ssl.create_default_context() |
| 112 | |
| 113 | def open(self): |
| 114 | """ |
| 115 | Ensure an open connection to the email server. Return whether or not a |
| 116 | new connection was required (True or False) or None if an exception |
| 117 | passed silently. |
| 118 | """ |
| 119 | if self.connection: |
| 120 | # Nothing to do if the connection is already open. |
| 121 | return False |
| 122 | |
| 123 | # If local_hostname is not specified, socket.getfqdn() gets used. |
| 124 | # For performance, we use the cached FQDN for local_hostname. |
| 125 | connection_params = {"local_hostname": DNS_NAME.get_fqdn()} |
| 126 | if self.timeout is not None: |
| 127 | connection_params["timeout"] = self.timeout |
| 128 | if self.use_ssl: |
| 129 | connection_params["context"] = self.ssl_context |
| 130 | try: |
| 131 | self.connection = self.connection_class( |
| 132 | self.host, self.port, **connection_params |
| 133 | ) |
| 134 | |
| 135 | # TLS/SSL are mutually exclusive, so only attempt TLS over |
| 136 | # non-secure connections. |
| 137 | if not self.use_ssl and self.use_tls: |
| 138 | self.connection.starttls(context=self.ssl_context) |
| 139 | if self.username and self.password: |
| 140 | self.connection.login(self.username, self.password) |
| 141 | return True |
| 142 | except OSError: |
| 143 | if not self.fail_silently: |
| 144 | raise |
| 145 | |
| 146 | def close(self): |
| 147 | """Close the connection to the email server.""" |
no test coverage detected