moilanik
b5faebcaae
CI Feature / Load example-gitea-env.conf to pipeline env (push) Successful in 20s
acc-tests Cucumber test report
CI Feature / Cucumber tests (push) Failing after 51s
CI Feature / Report Summary (push) Has been cancelled
CI Feature / Bats tests (push) Has been cancelled
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
import http.server, json, sys, os, threading
|
|
|
|
# Daemonize: detach from parent process group
|
|
if os.fork() > 0:
|
|
sys.exit(0)
|
|
os.setsid()
|
|
if os.fork() > 0:
|
|
sys.exit(0)
|
|
|
|
PORT = int(sys.argv[1])
|
|
CONFIG = sys.argv[2]
|
|
REQ_FILE = sys.argv[3]
|
|
|
|
idx_lock = threading.Lock()
|
|
mode = 'SINGLE'
|
|
default_code = 200
|
|
responses = []
|
|
idx_path = ''
|
|
|
|
with open(CONFIG) as f:
|
|
mode = f.readline().strip()
|
|
if mode == 'SEQUENCE':
|
|
seq_raw = f.readline().strip()
|
|
idx_path = f.readline().strip()
|
|
responses = json.loads(seq_raw)
|
|
else:
|
|
default_code = int(f.readline().strip())
|
|
|
|
def read_idx():
|
|
try:
|
|
with open(idx_path) as f2:
|
|
return int(f2.read().strip())
|
|
except:
|
|
return 0
|
|
|
|
def write_idx(v):
|
|
with open(idx_path, 'w') as f2:
|
|
f2.write(str(v))
|
|
|
|
class H(http.server.BaseHTTPRequestHandler):
|
|
def _get_response(self):
|
|
if mode == 'SEQUENCE':
|
|
with idx_lock:
|
|
i = read_idx()
|
|
r = responses[min(i, len(responses)-1)]
|
|
write_idx(i + 1)
|
|
code = r.get('code', 200)
|
|
body = r.get('body', {'id':1})
|
|
return code, json.dumps(body)
|
|
return default_code, json.dumps({'id':1})
|
|
|
|
def _log_request(self, method):
|
|
path = self.path
|
|
content_len = int(self.headers.get('Content-Length', 0))
|
|
body = self.rfile.read(content_len).decode(errors='replace') if content_len else ''
|
|
line = f'{method} {path}\n{body}\n'
|
|
with open(REQ_FILE, 'a') as f:
|
|
f.write(line)
|
|
|
|
def do_GET(self):
|
|
self._log_request('GET')
|
|
code, body = self._get_response()
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(body.encode())
|
|
|
|
def do_POST(self):
|
|
self._log_request('POST')
|
|
code, body = self._get_response()
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(body.encode())
|
|
|
|
def do_PATCH(self):
|
|
self._log_request('PATCH')
|
|
code, body = self._get_response()
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(body.encode())
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
s = http.server.HTTPServer(('', PORT), H)
|
|
s.allow_reuse_address = True
|
|
s.serve_forever()
|