Mock Your Own Tests
keploy mock lets you use Keploy as a mocking framework for your existing test
suite — pytest, go test, jest/playwright, or any command that makes network
calls. It is a language-agnostic VCR / WireMock that works at the network layer,
so your test code needs no SDK and no changes.
The recorded mocks act as virtual services for your dependencies, which makes this a lightweight form of service virtualization.
keploy mock record -c "<your test command>"runs your tests and captures the real outgoing dependency calls (HTTP, MySQL, …) into a named mock set.keploy mock replay -c "<your test command>"runs your tests again with those calls served from the mock set, so the real dependencies can be offline.
Keploy propagates your test runner's exit code, so it drops straight into CI.
Using VS Code? The Keploy VS Code extension runs these commands for you from a panel, and its troubleshooting table explains the messages the panel shows when a run fails.
Quick start
# 1. Record the dependency calls your tests make (real dependencies must be up)
keploy mock record -c "pytest"
# 2. Replay — the dependencies can now be down; your tests run against the mocks
keploy mock replay -c "pytest"
The mocks are written to keploy/default/mocks.yaml. Commit them like a VCR
cassette. Re-recording overwrites the set in place, so a "re-record on merge
to main" job produces a clean, reviewable diff.
# go test
keploy mock record -c "go test ./..."
keploy mock replay -c "go test ./..."
# a named set (e.g. per service)
keploy mock record -c "npm test" --name orders
keploy mock replay -c "npm test" --name orders
On-miss policy
When an outgoing call matches no recorded mock, --on-miss decides what happens:
--on-miss | Behaviour |
|---|---|
fail (default) | The call gets an error (deterministic); the run fails. |
passthrough | The call goes to the real dependency; nothing is persisted. |
record | The call goes to the real dependency and is appended to the set (VCR "new episodes") — incremental refresh. |
# A new test hit a new endpoint? Capture just that call and keep it:
keploy mock replay -c "pytest" --on-miss record
Add --strict to fail the run if any recorded mock was missed (a dependency
contract drifted), even when the tests themselves passed.
Per-test scoping (optional)
By default a set is recorded and replayed suite-wide. For per-test isolation — so
each test gets exactly its own mocks — your test runner can mark test boundaries
through a tiny HTTP API. Keploy exports two variables into your test process: the
agent's address as KEPLOY_MOCK_AGENT, and the credential for it as
KEPLOY_MOCK_AGENT_TOKEN. Call the API at the start and end of each test,
sending the token as a bearer credential:
POST {KEPLOY_MOCK_AGENT}/agent/scope/begin {"name": "<test name>"}
POST {KEPLOY_MOCK_AGENT}/agent/scope/end {"name": "<test name>"}
Authorization: Bearer {KEPLOY_MOCK_AGENT_TOKEN}
At record time this writes a per-test mappings.yaml; at replay time it restricts
the served pool to that test's mocks. No scope calls ⇒ suite-level, which is still
correct. Run the tests one at a time (see below).
The agent's API is authenticated, so a scope call without the header is rejected
with 401 and that test is not scoped. Send the header whenever
KEPLOY_MOCK_AGENT_TOKEN is set, and omit it when it is not, so the same glue
code keeps working against an older Keploy that does not export it — that is what
every example below does.
Watch out for the common shape try: ... except: pass. It hides the 401, and
per-test scoping silently falls back to suite-level while your suite still
passes. Keploy logs the rejection on its own output with the variable to set, so
check there if scoping stops taking effect after an upgrade.
Per-test scoping keeps tests apart only when they run one after another:
go test -p 1 -parallel 1, Playwright --workers=1, jest --runInBand, pytest
without -n. Sending the test process's PID as pid (Node process.pid, Python
os.getpid()) does not keep tests that run at the same time apart. At record, a
call made by any other process (a child the test starts, an app server the tests
share) goes to whichever test was open when it was made; at replay, a process the
test did not start, such as a shared app server, is served the whole recording.
Leave pid out of both calls, as the calls above and every example below do. A
begin without a pid narrows what the whole run is served to the test's mocks,
on every connection. A begin with a pid narrows only the connections that the
test process, or a process it started, opens while a test is open. Pooled
connections the app opened outside a test, such as a database pool it filled at
start-up or between two tests, would be served every test's mocks at replay, and
so would every connection of an app the tests share but did not start, such as
Playwright's webServer or a server started before the suite.
These calls are for runs without KEPLOY_PROVEN=1. Under it Keploy refuses them
(422 upgrade_integration): there a test marks its boundaries through Keploy's
scope protocol v1, which two integrations speak: go-sdk's
keploytest for
go test (go test -p 1), and
@keploy/playwright for
Playwright (--workers=1 --retries=0). Keploy runs one test process at a time
there and refuses a second. There is no integration for other frameworks: their
test process makes the calls by hand. It claims the run
(POST {KEPLOY_MOCK_AGENT}/agent/instance with {"key": "<package folder>"},
the folder relative to the project) before its first test, then sends
each test's begin with "canSkip": true, skips the test when the begin answers
"action": "skip", and sends its end, never with pid, at, dir or suite.
pytest (conftest.py):
import os, json, urllib.request, warnings, pytest
AGENT = os.environ.get("KEPLOY_MOCK_AGENT")
TOKEN = os.environ.get("KEPLOY_MOCK_AGENT_TOKEN")
def _post(path, name):
if not AGENT:
return
body = {"name": name} # run pytest without -n: see above
headers = {"Content-Type": "application/json"}
if TOKEN: # absent on an older Keploy; the call still works there
headers["Authorization"] = "Bearer " + TOKEN
req = urllib.request.Request(AGENT + path, data=json.dumps(body).encode(),
headers=headers, method="POST")
try:
urllib.request.urlopen(req, timeout=3).read()
except Exception as e:
# Surfaced, not swallowed: a 401 here means the test simply is not
# scoped, and silence would make that look like it worked.
warnings.warn(f"keploy: scope call {path} failed: {e}")
@pytest.fixture(autouse=True)
def keploy_scope(request):
_post("/agent/scope/begin", request.node.name)
yield
_post("/agent/scope/end", request.node.name)
go test (TestMain helper):
func scope(path, name string) {
agent := os.Getenv("KEPLOY_MOCK_AGENT")
if agent == "" {
return
}
// Run go test -p 1 -parallel 1: see above.
body, _ := json.Marshal(map[string]any{"name": name})
req, err := http.NewRequest(http.MethodPost, agent+path, bytes.NewReader(body))
if err != nil {
return
}
req.Header.Set("Content-Type", "application/json")
// Absent on an older Keploy, which does not require it.
if tok := os.Getenv("KEPLOY_MOCK_AGENT_TOKEN"); tok != "" {
req.Header.Set("Authorization", "Bearer "+tok)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Printf("keploy: scope call %s failed: %v", path, err)
return
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
// 401 here means this test is not scoped; do not let that pass quietly.
log.Printf("keploy: scope call %s returned %s", path, resp.Status)
}
}
// In each test: scope("/agent/scope/begin", t.Name()); defer scope("/agent/scope/end", t.Name())
jest / playwright (a beforeEach/afterEach or reporter hook) makes the same
two calls with the test's name. Run jest with --runInBand and Playwright with
--workers=1:
// Playwright: in a fixture or beforeEach/afterEach
const post = async (path, name) => {
const token = process.env.KEPLOY_MOCK_AGENT_TOKEN;
const headers = {"Content-Type": "application/json"};
// Absent on an older Keploy, which does not require it.
if (token) headers.Authorization = `Bearer ${token}`;
try {
const r = await fetch(`${process.env.KEPLOY_MOCK_AGENT}${path}`, {
method: "POST",
headers,
body: JSON.stringify({name}),
});
// A 401 means this test is not scoped — say so rather than swallow it.
if (!r.ok) console.warn(`keploy: scope call ${path} returned ${r.status}`);
} catch (e) {
console.warn(`keploy: scope call ${path} failed: ${e}`);
}
};
test.beforeEach(({}, testInfo) => post("/agent/scope/begin", testInfo.title));
test.afterEach(({}, testInfo) => post("/agent/scope/end", testInfo.title));
Platforms
| Platform | How to run |
|---|---|
| Linux | Native — keploy mock record -c "pytest" (uses eBPF; needs root). |
| Windows (x86-64) | Native — same command. Userspace interception, so no Administrator. |
| macOS (Apple Silicon) | Native — same command. Userspace interception, so no sudo. Running your tests in a container, e.g. -c "docker compose run tests", also works. |
Natively on macOS and Windows, Keploy understands HTTP/HTTPS, MySQL and MongoDB
calls; calls to other services — PostgreSQL, Redis, Kafka, gRPC and the like —
are captured only as raw bytes and usually don't replay. If your tests depend on
one of those, run them in a container (as above) or on Linux/WSL. On macOS, also
start the test runner itself rather than through a launcher such as npm test or
a wrapper script, and use a Homebrew or uv Python (or a virtualenv built on one),
not Apple's /usr/bin/python3 — see
running natively on macOS.
On an Intel Mac, use Lima;
on Windows on ARM, use WSL.
Refresh in CI
Because re-recording overwrites the set in place and the runner's exit code is
propagated, refreshing mocks on a merge to main is a normal CI step:
# bring up the real dependencies, then:
keploy mock record -c "pytest" --name default
keploy sanitize # scrub secrets before committing
git add keploy/ && git commit -m "chore: refresh mocks" || echo "no changes"
By default keploy mock is registry-first: the set is uploaded to Keploy
after record and downloaded before replay automatically (registry use depends on
your plan). Pass --local to keep everything on disk — the --local loop is
also the one that runs without signing in.