1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
#!/usr/bin/env python3
import asyncio
import dummy_killer
import tornado.ioloop
import tornado.web
import tornado.httpserver
import ssl
import argparse
import os
import sys
import traceback
class MainHandler(tornado.web.RequestHandler):
@tornado.gen.coroutine
def get(self, path):
if path == '/empty':
# Return an empty reply
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/error_403':
# Return a 403 HTTP error
raise tornado.web.HTTPError(403)
elif path == '/timeout':
# Wait for 4 seconds before returning an empty reply
yield tornado.gen.sleep(4)
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/request':
# Return a string 'hello world'
self.set_header("Content-Type", "text/plain")
self.write("hello world")
elif path == '/map-simple':
# Return a string 'hello map'
self.set_header("Content-Type", "text/plain")
self.write("hello map")
elif path == '/map-query':
# Parse the 'key' argument from the HTTP request
key = self.get_query_argument("key", default=None)
if key == 'au':
# Return a string 'hit' if 'key' is equal to 'au'
self.set_header("Content-Type", "text/plain")
self.write("1.0")
else:
# Return a 404 HTTP error if 'key' is not equal to 'au'
raise tornado.web.HTTPError(404)
elif path == '/settings':
self.set_header("Content-Type", "application/json")
self.write("{\"actions\": { \"reject\": 1.0}, \"symbols\": { \"EXTERNAL_SETTINGS\": 1.0 }}")
else:
raise tornado.web.HTTPError(404)
@tornado.gen.coroutine
def post(self, path):
if path == '/empty':
# Return an empty reply
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/error_403':
# Return a 403 HTTP error
raise tornado.web.HTTPError(403)
elif path == '/request':
# Return a string 'hello post'
self.set_header("Content-Type", "text/plain")
self.write("hello post")
elif path == '/timeout':
# Wait for 4 seconds before returning an empty reply
yield tornado.gen.sleep(4)
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/map-simple':
# Return a string 'hello map'
self.set_header("Content-Type", "text/plain")
self.write("hello map")
elif path == '/map-query':
# Parse the 'key' argument from the HTTP request
key = self.get_query_argument("key", default="")
if key == 'au':
# Return a string 'hit' if 'key' is equal to 'au'
self.set_header("Content-Type", "text/plain")
self.write("hit")
else:
# Return a 404 HTTP error if 'key' is not equal to 'au'
raise tornado.web.HTTPError(404)
elif path == '/settings':
self.set_header("Content-Type", "application/json")
self.write("{\"actions\": { \"reject\": 1.0}, \"symbols\": { \"EXTERNAL_SETTINGS\": 1.0 }}")
else:
raise tornado.web.HTTPError(404)
def head(self, path):
self.set_header("Content-Type", "text/plain")
if path == "/redirect1":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/hello")
elif path == "/redirect2":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/redirect1")
elif self.path == "/redirect3":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/redirect4")
elif self.path == "/redirect4":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/redirect3")
else:
self.send_response(200)
self.set_header("Content-Type", "text/plain")
def make_app():
return tornado.web.Application([
(r"(/[^/]+)", MainHandler),
])
async def main():
parser = argparse.ArgumentParser()
parser.add_argument("--bind", "-b", default="127.0.0.1", help="bind address")
parser.add_argument("--port", "-p", type=int, default=18080, help="bind port")
parser.add_argument("--keyfile", "-k", help="server private key file")
parser.add_argument("--certfile", "-c", help="server certificate file")
parser.add_argument("--pidfile", "-pf", help="path to the PID file")
args = parser.parse_args()
print(f"dummy_http.py: Starting server on {args.bind}:{args.port}", file=sys.stderr)
# Create the Tornado application
app = make_app()
# If keyfile and certfile are provided, create an HTTPS server.
# Otherwise, create an HTTP server.
if args.keyfile and args.certfile:
print(f"dummy_http.py: Using SSL with cert={args.certfile}, key={args.keyfile}", file=sys.stderr)
ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_ctx.load_cert_chain(args.certfile, args.keyfile)
server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_ctx)
else:
server = tornado.httpserver.HTTPServer(app)
# Start the server
server.bind(args.port, args.bind)
server.start(1)
# Write the PID to the specified PID file, if provided
# Do this AFTER successful bind/start so we know the server is actually running
if args.pidfile:
dummy_killer.write_pid(args.pidfile)
print(f"dummy_http.py: PID file written to {args.pidfile}", file=sys.stderr)
print(f"dummy_http.py: Server started successfully, listening on {args.bind}:{args.port}", file=sys.stderr)
await asyncio.Event().wait()
if __name__ == "__main__":
try:
if hasattr(asyncio, 'run') and callable(getattr(asyncio, 'run')):
asyncio.run(main())
else:
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
except Exception as e:
print(f"dummy_http.py: FATAL ERROR: {type(e).__name__}: {e}", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
sys.exit(1)
|
