niko
bd6ed5c2c2
CI Gitea Reports Main / Config load (push) Successful in 17s
CI Main / Config load (push) Successful in 20s
CI Gitea Reports Main / Latest version (push) Successful in 17s
CI Main / Latest versio (push) Successful in 19s
ci-helm-build-push Helm push 0.2.0
CI Gitea Reports Main / Build & Push Helm chart (push) Successful in 33s
gitops/gitea-ci-library/gitea-reports GitOps: gitea-reports 0.2.0
CI Gitea Reports Main / GitOps (push) Successful in 33s
CI Gitea Reports Main / Report Summary (push) Successful in 3s
CI Main / Bats tests (push) Failing after 1m16s
CI Main / Cucumber tests (push) Failing after 1m22s
CI Main / Build & Push Docker (push) Has been skipped
CI Main / GitOps (push) Has been skipped
CI Main / Move provider version tag (push) Has been skipped
CI Main / Report Summary (push) Successful in 3s
Co-authored-by: moilanik <niko.moilanen@tietoevry.com> Reviewed-on: #48
92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
import http.server, json, sys, os, threading
|
|
|
|
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_PUT(self):
|
|
self._log_request('PUT')
|
|
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()
|