Don't have write access for some reason, here's a script to run a mock central server when working with client DBs that won't start up without sync:
"""
Mock central server for testing sync integration without a real mSupply central.
Responds to all /sync/v5/* endpoints with minimal valid responses so that
manualSync can reach the integration step and process sync_buffer records.
Accepts any username/password — auth is not validated.
Usage:
python3 mock_central_server.py --site-id <SITE_ID> [--port 8080]
--site-id must match the siteId of the site row in your local DB (look it
up in the `site` table, or copy from the real central before swapping).
Then configure local.yaml (username/password can be anything):
sync:
url: "http://localhost:8080"
username: "anything"
password_sha256: "anything"
"""
import argparse
import json
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
class MockCentralHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/")
if path == "/sync/v5/site":
self.json_response(
{
"id": "B87865705D4D8A428EFF41B1A9226EC7",
"siteId": self.server.site_id,
"code": "mock",
"name": "Open mSupply Central Server",
"initialisationStatus": "completed",
"isOmSupplyCentralServer": True,
"omSupplyCentralServerUrl": "",
"mSupplyCentralSiteId": 1,
}
)
elif path == "/sync/v5/central_records":
# Return empty batch — no new central records to pull
params = parse_qs(parsed.query)
cursor = int(params.get("cursor", ["0"])[0])
self.json_response({"maxCursor": cursor, "data": []})
elif path == "/sync/v5/queued_records":
# Return empty batch — no new remote records to pull
self.json_response({"queueLength": 0, "data": []})
elif path == "/sync/v5/site_status":
self.json_response(
{"code": "idle", "message": "idle", "data": None}
)
else:
self.send_error(404, f"Unknown GET path: {path}")
def do_POST(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/")
content_length = int(self.headers.get("Content-Length", 0))
if path == "/sync/v5/queued_records":
self.rfile.read(content_length) # consume body
# Accept pushed records
self.json_response({"integrationStarted": True})
elif path == "/sync/v5/acknowledged_records":
self.rfile.read(content_length) # consume body
self.send_response(204)
self.end_headers()
elif path == "/sync/v5/initialise":
self.rfile.read(content_length) # consume body
self.json_response({"queueLength": 0, "data": []})
elif path == "/api/v4/login":
# Accept any credentials — return full v4 login response
raw = self.rfile.read(content_length)
body = json.loads(raw) if content_length else {}
username = body.get("username", "test")
permissions = [True] * 150 + [False] * 900
self.json_response({
"status": "success",
"authenticated": True,
"username": username,
"userFirstName": "Mock",
"userLastName": "User",
"userJobTitle": "",
"userType": "user",
"service": "mobile",
"storeName": "Mock Store",
"userInfo": {
"user": {
"ID": "MOCKUSER00000000000000000000ABCD",
"name": username,
"startup_method": "",
"nblogins": 1,
"group_id": "",
"mode": "",
"active": True,
"lasttime": 0,
"initials": "",
"first_name": "Mock",
"last_name": "User",
"address_1": "",
"address_2": "",
"e_mail": "",
"phone1": "",
"phone2": "",
"job_title": "",
"responsible_officer": False,
"Language": 0,
"use_ldap": False,
"ldap_login_string": "",
"receives_sms_errors": False,
"is_group": False,
"windows_user_name": "",
"license_category_id": "",
"isInactiveAuthoriser": False,
"spare_1": "",
},
"userStores": [
{
"ID": "MOCKUSERSTORE000000000000000ABCD",
"user_ID": "MOCKUSER00000000000000000000ABCD",
"store_ID": "store_a",
"can_login": True,
"store_default": True,
"can_action_replenishments": False,
"permissions": permissions,
}
],
},
})
else:
self.rfile.read(content_length) # consume body
self.send_error(404, f"Unknown POST path: {path}")
def json_response(self, data, status=200):
body = json.dumps(data).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, format, *args):
print(f"[mock-central] {self.command} {self.path} -> {args[1] if len(args) > 1 else args[0]}")
def main():
parser = argparse.ArgumentParser(description="Mock mSupply central server for sync testing")
parser.add_argument("--port", type=int, default=8080, help="Port to listen on (default: 8080)")
parser.add_argument(
"--site-id",
type=int,
required=True,
help="siteId to report on /sync/v5/site — must match your local DB's site row",
)
args = parser.parse_args()
server = HTTPServer(("127.0.0.1", args.port), MockCentralHandler)
server.site_id = args.site_id
print(f"Mock central server listening on http://127.0.0.1:{args.port} (siteId={args.site_id})")
print("Configure local.yaml with (username/password can be anything):")
print(f' sync:')
print(f' url: "http://localhost:{args.port}"')
print(f' username: "anything"')
print(f' password_sha256: "anything"')
server.serve_forever()
if __name__ == "__main__":
main()
Don't have write access for some reason, here's a script to run a mock central server when working with client DBs that won't start up without sync: