122 lines
4.0 KiB
Python
Executable File
122 lines
4.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# Minimal mock of the WordPress REST API, for testing post.sh and terms.sh
|
|
# without a real WordPress install.
|
|
#
|
|
# Handles POST /wp-json/wp/v2/posts(/ID) same as before, plus
|
|
# GET/POST /wp-json/wp/v2/categories and /wp-json/wp/v2/tags.
|
|
#
|
|
# Behavior is controlled by files (paths given via env vars):
|
|
# MOCK_WP_CONTROL_FILE - optional JSON {"status": int, "body": {...}} that
|
|
# determines the response to the next POST /posts request. If absent,
|
|
# responds 201 with the submitted fields echoed back under an "id".
|
|
# MOCK_WP_REQUEST_FILE - overwritten on every request with
|
|
# {"path": ..., "payload": ..., "user_agent": ...} so a test can assert
|
|
# on what was sent.
|
|
# MOCK_WP_TERMS_FILE - JSON {"categories": [...], "tags": [...]}, used as
|
|
# the backing store for the categories/tags endpoints. Created terms
|
|
# are appended and persisted back to this file.
|
|
#
|
|
# Listens on 127.0.0.1:$MOCK_WP_PORT (default 8899).
|
|
|
|
import http.server
|
|
import json
|
|
import os
|
|
import re
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
PORT = int(os.environ.get("MOCK_WP_PORT", "8899"))
|
|
CONTROL_FILE = os.environ["MOCK_WP_CONTROL_FILE"]
|
|
REQUEST_FILE = os.environ["MOCK_WP_REQUEST_FILE"]
|
|
TERMS_FILE = os.environ["MOCK_WP_TERMS_FILE"]
|
|
|
|
TERM_TYPES = {
|
|
"/wp-json/wp/v2/categories": "categories",
|
|
"/wp-json/wp/v2/tags": "tags",
|
|
}
|
|
|
|
|
|
def slugify(name):
|
|
return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
|
|
|
|
|
|
def load_terms():
|
|
if os.path.exists(TERMS_FILE):
|
|
with open(TERMS_FILE) as f:
|
|
return json.load(f)
|
|
return {"categories": [], "tags": []}
|
|
|
|
|
|
def save_terms(terms):
|
|
with open(TERMS_FILE, "w") as f:
|
|
json.dump(terms, f)
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def _write_json(self, status, body):
|
|
payload = json.dumps(body).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def _record_request(self, path, data):
|
|
with open(REQUEST_FILE, "w") as f:
|
|
json.dump({
|
|
"path": path,
|
|
"payload": data,
|
|
"user_agent": self.headers.get("User-Agent", ""),
|
|
}, f)
|
|
|
|
def do_GET(self):
|
|
parsed = urlparse(self.path)
|
|
term_type = TERM_TYPES.get(parsed.path)
|
|
if term_type is None:
|
|
self._write_json(404, {"code": "not_found"})
|
|
return
|
|
|
|
qs = parse_qs(parsed.query)
|
|
per_page = int(qs.get("per_page", ["10"])[0])
|
|
page = int(qs.get("page", ["1"])[0])
|
|
|
|
terms = load_terms()[term_type]
|
|
start = (page - 1) * per_page
|
|
self._write_json(200, terms[start:start + per_page])
|
|
|
|
def do_POST(self):
|
|
parsed = urlparse(self.path)
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(length)
|
|
try:
|
|
data = json.loads(body) if body else {}
|
|
except json.JSONDecodeError:
|
|
data = {"_raw": body.decode(errors="replace")}
|
|
|
|
self._record_request(self.path, data)
|
|
|
|
term_type = TERM_TYPES.get(parsed.path)
|
|
if term_type is not None:
|
|
terms = load_terms()
|
|
next_id = max([t["id"] for t in terms[term_type]], default=0) + 1
|
|
name = data.get("name", "")
|
|
term = {"id": next_id, "name": name, "slug": slugify(name), "count": 0}
|
|
terms[term_type].append(term)
|
|
save_terms(terms)
|
|
self._write_json(201, term)
|
|
return
|
|
|
|
status, resp_body = 201, {"id": 1, **data}
|
|
if os.path.exists(CONTROL_FILE):
|
|
with open(CONTROL_FILE) as f:
|
|
control = json.load(f)
|
|
status = control.get("status", status)
|
|
resp_body = control.get("body", resp_body)
|
|
self._write_json(status, resp_body)
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
http.server.HTTPServer(("127.0.0.1", PORT), Handler).serve_forever()
|