266 lines
14 KiB
Python
266 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify a generated MySQL service against a disposable local Docker database."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
|
|
from verifyScaffold import MODULE, run
|
|
|
|
|
|
def waitFor(check, description, timeout=20):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if check():
|
|
return
|
|
time.sleep(0.1)
|
|
raise AssertionError(f"timed out: {description}")
|
|
|
|
|
|
class MySQL:
|
|
def __init__(self, directory, env, image):
|
|
self.directory, self.env, self.image = directory, dict(env), image
|
|
self.name = "go-omni-check-" + uuid.uuid4().hex[:12]
|
|
self.container = None
|
|
self.password = secrets.token_hex(18)
|
|
self.env.update(MYSQL_ROOT_PASSWORD=secrets.token_hex(18), MYSQL_PASSWORD=self.password)
|
|
self.env["MYSQL_PWD"] = self.env["MYSQL_ROOT_PASSWORD"]
|
|
|
|
def sql(self, statement, database=None):
|
|
args = ["docker", "exec", "-i", "--env", "MYSQL_PWD", self.container,
|
|
"mysql", "--protocol=tcp", "-h127.0.0.1", "-uroot", "--batch", "--skip-column-names"]
|
|
if database:
|
|
args.append(database)
|
|
result = subprocess.run(args, cwd=self.directory, env=self.env, input=statement,
|
|
text=True, capture_output=True, timeout=10)
|
|
if result.returncode:
|
|
raise RuntimeError(f"test database SQL failed: {result.stderr}")
|
|
return result.stdout.strip()
|
|
|
|
def wait(self):
|
|
def ready():
|
|
try:
|
|
return self.sql("SELECT 1") == "1"
|
|
except (RuntimeError, subprocess.TimeoutExpired):
|
|
return False
|
|
waitFor(ready, "MySQL ready", timeout=90)
|
|
|
|
def start(self):
|
|
endpoint = self.env.get("DOCKER_HOST") or run(
|
|
["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"],
|
|
self.directory, self.env).strip()
|
|
if not endpoint.startswith(("unix://", "npipe://")):
|
|
raise RuntimeError("database verification requires a local Docker context")
|
|
# Docker may assign a different random published port after restart.
|
|
# Choose the loopback port once so outage/recovery keeps the same DSN.
|
|
with socket.socket() as listener:
|
|
listener.bind(("127.0.0.1", 0))
|
|
hostPort = listener.getsockname()[1]
|
|
print("Starting isolated MySQL container", flush=True)
|
|
self.container = run(["docker", "run", "--detach", "--name", self.name,
|
|
"--label", "go-omni.test=database", "--publish", f"127.0.0.1:{hostPort}:3306",
|
|
"--env", "MYSQL_ROOT_PASSWORD", "--env", "MYSQL_PASSWORD",
|
|
"--env", "MYSQL_DATABASE=omni_test", "--env", "MYSQL_USER=omni_test",
|
|
self.image], self.directory, self.env).strip()
|
|
self.wait()
|
|
port = run(["docker", "port", self.container, "3306/tcp"], self.directory, self.env).strip().rsplit(":", 1)[1]
|
|
self.dsn = (f"omni_test:{self.password}@tcp(127.0.0.1:{port})/omni_test"
|
|
"?parseTime=true&timeout=2s&readTimeout=2s&writeTimeout=2s")
|
|
self.sql("CREATE DATABASE omni_empty; GRANT ALL ON omni_empty.* TO 'omni_test'@'%';")
|
|
|
|
def close(self):
|
|
if self.container:
|
|
# Remove only the container and anonymous volume created by this run.
|
|
run(["docker", "rm", "--force", "--volumes", self.container], self.directory, self.env)
|
|
|
|
|
|
class Server:
|
|
def __init__(self, binary, directory, env):
|
|
self.binary, self.directory, self.env = binary, directory, dict(env)
|
|
with socket.socket() as listener:
|
|
listener.bind(("127.0.0.1", 0))
|
|
self.address = f"127.0.0.1:{listener.getsockname()[1]}"
|
|
self.env["GO_OMNI_HTTP_ADDR"] = self.address
|
|
self.process = None
|
|
self.logPath = directory / f"server-{uuid.uuid4().hex[:8]}.log"
|
|
self.client = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
|
|
def request(self, method, path, body=None, expected=200):
|
|
req = urllib.request.Request(f"http://{self.address}{path}", method=method,
|
|
data=None if body is None else json.dumps(body).encode(),
|
|
headers={"Content-Type": "application/json", "X-Request-ID": "databaseCheck"})
|
|
try:
|
|
response = self.client.open(req, timeout=3)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
with response:
|
|
raw = response.read()
|
|
if response.status != expected:
|
|
raise AssertionError(f"{method} {path}: expected {expected}, got {response.status}: {raw!r}")
|
|
if response.headers.get("X-Request-ID") != "databaseCheck":
|
|
raise AssertionError("missing requestId header")
|
|
if not raw:
|
|
return None
|
|
result = json.loads(raw)
|
|
if result.get("requestId") != "databaseCheck":
|
|
raise AssertionError("missing camelCase requestId")
|
|
return result
|
|
|
|
def start(self):
|
|
with self.logPath.open("w") as logs:
|
|
self.process = subprocess.Popen([str(self.binary)], cwd=self.directory, env=self.env,
|
|
stdout=logs, stderr=subprocess.STDOUT)
|
|
def ready():
|
|
if self.process.poll() is not None:
|
|
raise RuntimeError(f"service exited: {self.logPath.read_text()}")
|
|
try:
|
|
return self.request("GET", "/health/live")["data"]["status"] == "ok"
|
|
except (OSError, urllib.error.URLError):
|
|
return False
|
|
waitFor(ready, "service startup")
|
|
|
|
def stop(self):
|
|
if self.process and self.process.poll() is None:
|
|
self.process.send_signal(signal.SIGTERM)
|
|
code = self.process.wait(timeout=10)
|
|
if code != 0:
|
|
raise AssertionError(f"service exit {code}: {self.logPath.read_text()}")
|
|
|
|
def close(self):
|
|
if self.process and self.process.poll() is None:
|
|
self.process.kill()
|
|
self.process.wait(timeout=5)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--go", default="go")
|
|
parser.add_argument("--version", default="v0.3.1")
|
|
parser.add_argument("--cli", help="Use a locally built CLI")
|
|
parser.add_argument("--framework-dir", help="Local framework source for pre-release checks")
|
|
parser.add_argument("--image", default="mysql:8.4")
|
|
args = parser.parse_args()
|
|
env = {key: value for key, value in os.environ.items() if not key.startswith("GO_OMNI_")}
|
|
env["GOWORK"] = "off"
|
|
env["GOPRIVATE"] = ",".join(filter(None, [env.get("GOPRIVATE"), "git.yuntuo.hk"]))
|
|
with tempfile.TemporaryDirectory(prefix="go omni database ") as temporary:
|
|
directory = Path(temporary).resolve()
|
|
if args.cli:
|
|
cli = Path(args.cli).resolve()
|
|
else:
|
|
env["GOBIN"] = str(directory / "bin")
|
|
run([args.go, "install", f"{MODULE}/cmd/go-omni@{args.version}"], directory, env)
|
|
cli = directory / "bin" / "go-omni"
|
|
command = [str(cli), "new", "--module", "example.com/databaseCheck", "--database", "mysql"]
|
|
if args.framework_dir:
|
|
command += ["--framework-dir", str(Path(args.framework_dir).resolve())]
|
|
project = directory / "billing"
|
|
run(command + [str(project)], directory, env)
|
|
module = json.loads(run([args.go, "mod", "edit", "-json"], project, env))
|
|
if not args.framework_dir and module.get("Replace"):
|
|
raise AssertionError("remote MySQL project contains a local replace")
|
|
if {entry["Path"]: entry["Version"] for entry in module["Require"]}.get(MODULE) != args.version:
|
|
raise AssertionError("MySQL project did not pin the CLI version")
|
|
run([args.go, "mod", "tidy"], project, env)
|
|
binary = directory / "server"
|
|
run([args.go, "build", "-o", str(binary), "./cmd/server"], project, env)
|
|
database = MySQL(directory, env, args.image)
|
|
servers = []
|
|
try:
|
|
database.start()
|
|
database.sql((project / "schema.sql").read_text(), "omni_test")
|
|
testEnv = dict(env, GO_OMNI_TEST_MYSQL_DSN=database.dsn)
|
|
output = run([args.go, "test", "-race", "-v", "./..."], project, testEnv)
|
|
for name in ("TestDatabaseTransactions", "TestDatabasePoolHealthAndClose"):
|
|
if f"--- PASS: {name}" not in output:
|
|
raise AssertionError(f"integration test did not pass: {name}\n{output}")
|
|
print("PASS real MySQL: commit, error/panic/cancel rollback, pool limits, health timeout, close", flush=True)
|
|
|
|
serverEnv = dict(env, GO_OMNI_DB_DSN=database.dsn, GO_OMNI_MODE="release",
|
|
GO_OMNI_DB_PING_TIMEOUT="500ms", GO_OMNI_ENABLE_LOGGER="true")
|
|
server = Server(binary, project, serverEnv)
|
|
servers.append(server)
|
|
server.start()
|
|
ready = server.request("GET", "/health/ready")
|
|
if ready["code"] != "READY" or ready["data"]["checks"][0]["name"] != "database":
|
|
raise AssertionError(f"database probe not registered: {ready}")
|
|
server.request("POST", "/users", {"name": "", "email": "invalid"}, expected=400)
|
|
created = server.request("POST", "/users", {"name": " Alice ", "email": "ALICE@example.com"}, expected=201)
|
|
user = created["data"]
|
|
if user["email"] != "alice@example.com" or not user.get("createdAt"):
|
|
raise AssertionError(f"normalization/time parsing failed: {user}")
|
|
path = f"/users/{user['id']}"
|
|
duplicate = server.request("POST", "/users", {"name": "Duplicate", "email": "alice@example.com"}, expected=409)
|
|
if duplicate["code"] != "EMAIL_TAKEN":
|
|
raise AssertionError("duplicate email error was not mapped")
|
|
if server.request("GET", path)["data"]["name"] != "Alice":
|
|
raise AssertionError("GET did not load created user")
|
|
page = server.request("GET", "/users?page=1&pageSize=1")["data"]
|
|
if page["total"] != 1 or page["pageSize"] != 1:
|
|
raise AssertionError(f"pagination mismatch: {page}")
|
|
server.request("PUT", path, {"name": "Updated", "email": "updated@example.com"})
|
|
server.stop()
|
|
waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE USER='omni_test'") == "0",
|
|
"connection pool closed on service exit")
|
|
server = Server(binary, project, serverEnv)
|
|
servers.append(server)
|
|
server.start()
|
|
if server.request("GET", path)["data"]["name"] != "Updated":
|
|
raise AssertionError("data did not survive service restart")
|
|
print("PASS HTTP validation, unique email, create/read/update/list, restart persistence and pool cleanup", flush=True)
|
|
|
|
failureEnv = dict(serverEnv, GO_OMNI_DB_DSN=database.dsn.replace("/omni_test?", "/omni_empty?"))
|
|
failed = subprocess.run([str(binary)], cwd=project, env=failureEnv, capture_output=True, text=True, timeout=10)
|
|
if failed.returncode == 0 or "apply schema.sql" not in failed.stderr:
|
|
raise AssertionError("missing schema did not fail startup before listening")
|
|
waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE DB='omni_empty'") == "0",
|
|
"connection pool closed on schema failure")
|
|
print("PASS missing schema fails startup and releases connections", flush=True)
|
|
|
|
run(["docker", "stop", "--time", "10", database.container], directory, database.env)
|
|
readiness = server.request("GET", "/health/ready", expected=503)
|
|
if readiness["code"] != "NOT_READY" or readiness["data"]["checks"][0]["status"] != "not_ready":
|
|
raise AssertionError("database outage did not fail readiness")
|
|
server.request("GET", "/health/live")
|
|
unavailable = server.request("GET", path, expected=500)
|
|
if unavailable["code"] != "INTERNAL_ERROR" or database.password in json.dumps(unavailable):
|
|
raise AssertionError("database outage leaked an internal error")
|
|
failed = subprocess.run([str(binary)], cwd=project, env=serverEnv, capture_output=True, text=True, timeout=10)
|
|
if failed.returncode == 0 or "ping database" not in failed.stderr:
|
|
raise AssertionError("offline database did not fail startup")
|
|
run(["docker", "start", database.container], directory, database.env)
|
|
database.wait()
|
|
def recovered():
|
|
try:
|
|
return server.request("GET", "/health/ready")["code"] == "READY"
|
|
except (AssertionError, OSError, urllib.error.URLError):
|
|
return False
|
|
waitFor(recovered, "application reconnects after MySQL restart")
|
|
if server.request("GET", path)["data"]["name"] != "Updated":
|
|
raise AssertionError("database restart lost persistent data")
|
|
server.request("DELETE", path, expected=204)
|
|
server.request("GET", path, expected=404)
|
|
server.stop()
|
|
waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE USER='omni_test'") == "0",
|
|
"final connection cleanup")
|
|
print("PASS database outage: readiness 503, liveness 200, startup failure; recovery 200; delete 204/404", flush=True)
|
|
finally:
|
|
for server in servers:
|
|
server.close()
|
|
database.close()
|
|
print("Removed test container, anonymous volume and temporary project", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|