##// END OF EJS Templates
tests: force `dummysmtpd.py` to write its log file with '\n' on Windows...
Matt Harbison -
r53371:dbd2d562 default
parent child Browse files
Show More
@@ -1,161 +1,169 b''
1 1 #!/usr/bin/env python
2 2
3 3 """dummy SMTP server for use in tests"""
4 4
5 5
6 import io
6 7 import optparse
7 8 import os
8 9 import socket
9 10 import ssl
10 11 import sys
11 12
12 13 from mercurial import (
13 14 pycompat,
14 15 server,
15 16 sslutil,
16 17 ui as uimod,
17 18 )
18 19
20 if pycompat.iswindows:
21 sys.stdout = io.TextIOWrapper(
22 sys.stdout.buffer,
23 sys.stdout.encoding,
24 sys.stdout.errors,
25 newline="\n",
26 )
19 27
20 28 if os.environ.get('HGIPV6', '0') == '1':
21 29 family = socket.AF_INET6
22 30 else:
23 31 family = socket.AF_INET
24 32
25 33
26 34 def log(msg):
27 35 sys.stdout.write(msg)
28 36 sys.stdout.flush()
29 37
30 38
31 39 def mocksmtpserversession(conn, addr):
32 40 conn.send(b'220 smtp.example.com ESMTP\r\n')
33 41
34 42 try:
35 43 # Newer versions of OpenSSL raise on EOF
36 44 line = conn.recv(1024)
37 45 except ssl.SSLError:
38 46 log('no hello: EOF\n')
39 47 return
40 48
41 49 if not line.lower().startswith(b'ehlo '):
42 50 # Older versions of OpenSSl don't raise
43 51 log('no hello: %s\n' % line)
44 52 return
45 53
46 54 conn.send(b'250 Hello\r\n')
47 55
48 56 line = conn.recv(1024)
49 57 if not line.lower().startswith(b'mail from:'):
50 58 log('no mail from: %s\n' % line)
51 59 return
52 60 mailfrom = line[10:].decode().rstrip()
53 61 if mailfrom.startswith('<') and mailfrom.endswith('>'):
54 62 mailfrom = mailfrom[1:-1]
55 63
56 64 conn.send(b'250 Ok\r\n')
57 65
58 66 rcpttos = []
59 67 while True:
60 68 line = conn.recv(1024)
61 69 if not line.lower().startswith(b'rcpt to:'):
62 70 break
63 71 rcptto = line[8:].decode().rstrip()
64 72 if rcptto.startswith('<') and rcptto.endswith('>'):
65 73 rcptto = rcptto[1:-1]
66 74 rcpttos.append(rcptto)
67 75
68 76 conn.send(b'250 Ok\r\n')
69 77
70 78 if not line.lower().strip() == b'data':
71 79 log('no rcpt to or data: %s' % line)
72 80
73 81 conn.send(b'354 Go ahead\r\n')
74 82
75 83 data = b''
76 84 while True:
77 85 line = conn.recv(1024)
78 86 if not line:
79 87 log('connection closed before end of data')
80 88 break
81 89 data += line
82 90 if data.endswith(b'\r\n.\r\n'):
83 91 data = data[:-5]
84 92 break
85 93
86 94 conn.send(b'250 Ok\r\n')
87 95
88 96 log(
89 97 '%s from=%s to=%s\n%s\n'
90 98 % (addr[0], mailfrom, ', '.join(rcpttos), data.decode())
91 99 )
92 100
93 101
94 102 def run(host, port, certificate):
95 103 ui = uimod.ui.load()
96 104 with socket.socket(family, socket.SOCK_STREAM) as s:
97 105 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
98 106 s.bind((host, port))
99 107 # log('listening at %s:%d\n' % (host, port))
100 108 s.listen(1)
101 109 try:
102 110 while True:
103 111 conn, addr = s.accept()
104 112 if certificate:
105 113 try:
106 114 conn = sslutil.wrapserversocket(
107 115 conn, ui, certfile=certificate
108 116 )
109 117 except ssl.SSLError as e:
110 118 log('%s ssl error: %s\n' % (addr[0], e))
111 119 conn.close()
112 120 continue
113 121 log("connection from %s:%s\n" % addr)
114 122 mocksmtpserversession(conn, addr)
115 123 conn.close()
116 124 except KeyboardInterrupt:
117 125 pass
118 126
119 127
120 128 def _encodestrsonly(v):
121 129 if isinstance(v, type(u'')):
122 130 return v.encode('ascii')
123 131 return v
124 132
125 133
126 134 def bytesvars(obj):
127 135 unidict = vars(obj)
128 136 bd = {k.encode('ascii'): _encodestrsonly(v) for k, v in unidict.items()}
129 137 if bd[b'daemon_postexec'] is not None:
130 138 bd[b'daemon_postexec'] = [
131 139 _encodestrsonly(v) for v in bd[b'daemon_postexec']
132 140 ]
133 141 return bd
134 142
135 143
136 144 def main():
137 145 op = optparse.OptionParser()
138 146 op.add_option('-d', '--daemon', action='store_true')
139 147 op.add_option('--daemon-postexec', action='append')
140 148 op.add_option('-p', '--port', type=int, default=8025)
141 149 op.add_option('-a', '--address', default='localhost')
142 150 op.add_option('--pid-file', metavar='FILE')
143 151 op.add_option('--tls', choices=['none', 'smtps'], default='none')
144 152 op.add_option('--certificate', metavar='FILE')
145 153 op.add_option('--logfile', metavar='FILE')
146 154
147 155 opts, args = op.parse_args()
148 156 if (opts.tls == 'smtps') != bool(opts.certificate):
149 157 op.error('--certificate must be specified with --tls=smtps')
150 158
151 159 server.runservice(
152 160 bytesvars(opts),
153 161 runfn=lambda: run(opts.address, opts.port, opts.certificate),
154 162 runargs=[pycompat.sysexecutable, pycompat.fsencode(__file__)]
155 163 + pycompat.sysargv[1:],
156 164 logfile=opts.logfile,
157 165 )
158 166
159 167
160 168 if __name__ == '__main__':
161 169 main()
General Comments 0
You need to be logged in to leave comments. Login now