Connect to the LMTP daemon, on either a Unix or a TCP socket.
(self, host='localhost', port=0, source_address=None)
| 1088 | source_address=source_address, timeout=timeout) |
| 1089 | |
| 1090 | def connect(self, host='localhost', port=0, source_address=None): |
| 1091 | """Connect to the LMTP daemon, on either a Unix or a TCP socket.""" |
| 1092 | if host[0] != '/': |
| 1093 | return super().connect(host, port, source_address=source_address) |
| 1094 | |
| 1095 | if self.timeout is not None and not self.timeout: |
| 1096 | raise ValueError('Non-blocking socket (timeout=0) is not supported') |
| 1097 | |
| 1098 | # Handle Unix-domain sockets. |
| 1099 | try: |
| 1100 | self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 1101 | if self.timeout is not socket._GLOBAL_DEFAULT_TIMEOUT: |
| 1102 | self.sock.settimeout(self.timeout) |
| 1103 | self.file = None |
| 1104 | self.sock.connect(host) |
| 1105 | except OSError: |
| 1106 | if self.debuglevel > 0: |
| 1107 | self._print_debug('connect fail:', host) |
| 1108 | if self.sock: |
| 1109 | self.sock.close() |
| 1110 | self.sock = None |
| 1111 | raise |
| 1112 | (code, msg) = self.getreply() |
| 1113 | if self.debuglevel > 0: |
| 1114 | self._print_debug('connect:', msg) |
| 1115 | return (code, msg) |
| 1116 | |
| 1117 | |
| 1118 | # Test the sendmail method, which tests most of the others. |
nothing calls this directly
no test coverage detected