Connect to the LMTP daemon, on either a Unix or a TCP socket.
(self, host='localhost', port=0, source_address=None)
| 1072 | source_address=source_address, timeout=timeout) |
| 1073 | |
| 1074 | def connect(self, host='localhost', port=0, source_address=None): |
| 1075 | """Connect to the LMTP daemon, on either a Unix or a TCP socket.""" |
| 1076 | if host[0] != '/': |
| 1077 | return super().connect(host, port, source_address=source_address) |
| 1078 | |
| 1079 | if self.timeout is not None and not self.timeout: |
| 1080 | raise ValueError('Non-blocking socket (timeout=0) is not supported') |
| 1081 | |
| 1082 | # Handle Unix-domain sockets. |
| 1083 | try: |
| 1084 | self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 1085 | if self.timeout is not socket._GLOBAL_DEFAULT_TIMEOUT: |
| 1086 | self.sock.settimeout(self.timeout) |
| 1087 | self.file = None |
| 1088 | self.sock.connect(host) |
| 1089 | except OSError: |
| 1090 | if self.debuglevel > 0: |
| 1091 | self._print_debug('connect fail:', host) |
| 1092 | if self.sock: |
| 1093 | self.sock.close() |
| 1094 | self.sock = None |
| 1095 | raise |
| 1096 | (code, msg) = self.getreply() |
| 1097 | if self.debuglevel > 0: |
| 1098 | self._print_debug('connect:', msg) |
| 1099 | return (code, msg) |
| 1100 | |
| 1101 | |
| 1102 | # Test the sendmail method, which tests most of the others. |
nothing calls this directly
no test coverage detected