134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""Stateful mock Firefly III API for skill evals only. Stdlib http.server.
|
|
|
|
Persists posted transactions to a JSON file so a SECOND import of the same
|
|
statement correctly reports duplicates (the dedup eval depends on this).
|
|
|
|
Pre-seeded expense/revenue accounts let us verify the skill consolidates onto
|
|
an EXISTING account ("Sheetz") instead of auto-creating "SHEETZ #432".
|
|
|
|
Run: python mock_firefly.py <port> <state_file>
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import urllib.parse
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
STATE_FILE = "mock_state.json"
|
|
|
|
SEED_ACCOUNTS = [
|
|
{"id": "10", "name": "Sheetz", "type": "Expense account"},
|
|
{"id": "11", "name": "Amazon", "type": "Expense account"},
|
|
{"id": "12", "name": "Costco", "type": "Expense account"},
|
|
{"id": "13", "name": "Local Cafe", "type": "Expense account"},
|
|
{"id": "20", "name": "Employer Payroll", "type": "Revenue account"},
|
|
]
|
|
|
|
|
|
def load_state():
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
return {"txns": [], "next_id": 1000}
|
|
|
|
|
|
def save_state(s):
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump(s, f)
|
|
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def _send(self, code, obj, ct="application/json"):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ct)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
u = urllib.parse.urlparse(self.path)
|
|
q = urllib.parse.parse_qs(u.query)
|
|
path = u.path.replace("/api/v1/", "")
|
|
|
|
if path == "about":
|
|
return self._send(200, {"data": {"version": "6.1.0-mock"}})
|
|
|
|
if path == "search/transactions":
|
|
query = q.get("query", [""])[0]
|
|
ext = ""
|
|
if "external_id:" in query:
|
|
ext = query.split('external_id:"')[1].rstrip('"')
|
|
st = load_state()
|
|
hits = [t for t in st["txns"] if t["external_id"] == ext]
|
|
return self._send(
|
|
200,
|
|
{"data": [{"id": str(h["id"])} for h in hits],
|
|
"meta": {"pagination": {"total_pages": 1}}},
|
|
ct="application/vnd.api+json",
|
|
)
|
|
|
|
if path == "autocomplete/accounts":
|
|
query = q.get("query", [""])[0].lower()
|
|
types = q.get("types", [""])[0]
|
|
res = [
|
|
a for a in SEED_ACCOUNTS
|
|
if (query in a["name"].lower() or not query)
|
|
and (a["type"] in types if types else True)
|
|
]
|
|
return self._send(200, res)
|
|
|
|
if path == "accounts":
|
|
t = q.get("type", ["all"])[0]
|
|
tmap = {"expense": "Expense account", "revenue": "Revenue account"}
|
|
res = [
|
|
{"id": a["id"], "attributes": {"name": a["name"]}}
|
|
for a in SEED_ACCOUNTS
|
|
if a["type"] == tmap.get(t)
|
|
]
|
|
return self._send(
|
|
200,
|
|
{"data": res, "meta": {"pagination": {"total_pages": 1}}},
|
|
ct="application/vnd.api+json",
|
|
)
|
|
|
|
self._send(404, {"message": f"no mock for GET {path}"})
|
|
|
|
def do_POST(self):
|
|
u = urllib.parse.urlparse(self.path)
|
|
path = u.path.replace("/api/v1/", "")
|
|
n = int(self.headers.get("Content-Length", 0))
|
|
body = json.loads(self.rfile.read(n) or "{}")
|
|
|
|
if path == "transactions":
|
|
split = body["transactions"][0]
|
|
ext = split.get("external_id", "")
|
|
st = load_state()
|
|
if body.get("error_if_duplicate_hash") and any(
|
|
t["external_id"] == ext for t in st["txns"]
|
|
):
|
|
return self._send(
|
|
422,
|
|
{"message": "Duplicate transaction.",
|
|
"errors": {"transactions.0": ["Duplicate of existing."]}},
|
|
)
|
|
tid = st["next_id"]
|
|
st["next_id"] += 1
|
|
st["txns"].append({"id": tid, "external_id": ext, "split": split})
|
|
save_state(st)
|
|
return self._send(200, {"data": {"id": str(tid),
|
|
"attributes": split}})
|
|
|
|
self._send(404, {"message": f"no mock for POST {path}"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8088
|
|
if len(sys.argv) > 2:
|
|
STATE_FILE = sys.argv[2]
|
|
HTTPServer(("127.0.0.1", port), H).serve_forever()
|