feat(dispatch): implement dispatch-workflow.sh with dispatch, poll, and timeout (#4)
Add dispatch-workflow.sh script that dispatches a Gitea workflow in another repository and polls synchronously for completion. Refactor mock-api.sh to use Python3 HTTP server with sequence support, enabling stateful poll-response simulation in tests. --------- Co-authored-by: moilanik <niko.moilanen@tietoevry.com> Reviewed-on: #4
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
#!/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() 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 log_message(self, format, *args):
|
||||
pass
|
||||
|
||||
s = http.server.HTTPServer(('', PORT), H)
|
||||
s.allow_reuse_address = True
|
||||
s.serve_forever()
|
||||
Reference in New Issue
Block a user