way/pigway-frame-flow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 29 Wiki Activity

Add manually gated first production bootstrap with fail-closed residue checks

Browse Source
This commit is contained in:
Pigway
2026-10-08 00:06:22 +08:00
parent 2e8251abef
commit 3e48801e8b
4 changed files with 704 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/verify.yml
+1 -1
View File
@@ -7,7 +7,7 @@ on:
deploy_action:
description: Production action (deploy requires explicit enabled Variable)
type: choice
options: [plan, deploy, rollback]
options: [plan, bootstrap, deploy, rollback]
default: plan
jobs:
verify:
docs/deployment/GITEA_CD.md
+106 -2
View File
@@ -89,8 +89,8 @@ Never upload that journal or a full Docker inspect as a public CI artifact.
Pushes to other branches and pull requests never deploy. A manual run defaults
to plan, even when automatic deployment is enabled.
The script intentionally supports updates to an existing healthy deployment,
not first-time bootstrap. No real production deployment or rollback drill has
The deploy action supports updates to an existing healthy deployment. For first-time
creation use the explicit manual bootstrap flow below. No real production deployment or rollback drill has
been performed during implementation; those are first-run acceptance steps.
## Deployment transaction and rollback
@@ -167,3 +167,107 @@ FRAME_FLOW_DEPLOY_OWNER=<confirmed failed owner> CD_ACTION=recover python script
recovery fails. If the very first attempt stopped before a journal was written,
confirm that the original image is still running and healthy before an operator
removes the stale lock. Do not delete a live lock or any persistence volume.
## First-time production bootstrap (explicit manual action)
The confirmed NAS residue is a `created`, never-started container with dead
`/workspace/way/...` bind paths. It is not an adoption/rollback baseline.
No migration from those paths is performed. Production Compose remains independently usable.
Set these repository Actions **Variables** (not Secrets):
```text
PROD_DEPLOY_ENABLED=true
PROD_PROJECT=pigway-frame-flow
PROD_CONTAINER=pigway-frame-flow
PROD_IMAGE_REPO=pigway-frame-flow
PROD_DATA_DIR=/volume2/docker/pigway-frame-flow/data
PROD_CONFIG_DIR=/volume2/docker/pigway-frame-flow/config
PROD_WEB_BIND=192.0.2.10
PROD_WEB_PORT=18787
PROD_STATE_VOLUME=pigway-frame-flow-cd-state
PROD_MOCK=true
PROD_WRITE_ENABLED=false
```
Leave `PROD_PHOTOS_DIR` unset/empty. Existing mirror Variables still apply.
Enabling CD does not enable real frame writes. Before the first successful bootstrap,
ordinary main pushes report `production not bootstrapped / bootstrap required`.
They cannot create production. `bootstrap` additionally requires the Actions event
`workflow_dispatch`; the default action remains `plan`. Every action is gated by
successful complete Verify on main.
Create repository Actions **Secret** `PRODUCTION_ENV` as multiline bare `KEY=value`:
```dotenv
FRAME_FLOW_TOKEN=replace-with-a-long-random-private-token
KOKONNA_API_KEY=
```
Do not add shell quotes, `export`, interpolation or spaces around `=`. An existing
KoKonna key may be supplied but first bootstrap forcibly uses Mock with writes off.
Optional application secrets/settings may be added in the same format; bootstrap
always overrides Mock/write flags, data/config paths, timezone and build identity.
This token remains useful for API/smoke checks even if Web UI uses account login.
Do not paste the real Secret into logs or repository files.
### Safety and cleanup
Bootstrap builds the verified main revision first, then acquires the same atomic
Docker daemon deployment lock used by deploy. Host path checks run in a short-lived,
network-disabled helper with **read-only** NAS root mounted, not CI workspace paths.
Only explicitly configured data/config directories are subsequently mounted writable
for initialization, owned `10333:10333`, mode `0750`. Nonempty directories, symlinks
(including ancestor symlinks), overlapping paths and unknown journal resources fail closed.
Docker may create missing ancestor directories while creating these exact bind sources;
there is no recursive chmod/chown/delete on their parents.
A same-name residue can be removed **without force** only after checking all of:
Compose project/service labels; `Status=created`; `Running=false`;
`StartedAt=0001-01-01T00:00:00Z`; exactly two bind mounts `/data` and `/config`;
and both old host sources absent. A running, previously started, unknown or data-bearing
container is refused. An existing default network must be owned by this Compose
project and contain no endpoints other than the validated residue.
The production service uses `create_host_path=false`, the explicit NAS LAN bind,
non-root UID, `unless-stopped`, normal bridge network and existing resource/security
limits. Acceptance runs ready, all read-only smoke endpoints, version/commit,
preflight mount/port/project/user/restart checks, one restart, repeated checks,
Docker health and exact built-image ID before `BOOTSTRAP VERIFIED`.
On failure after transaction preparation, cleanup removes only this transaction's
owner-labelled new container and its newly created, empty, project-labelled network.
The previously validated never-started residue has no rollback value and is not recreated.
**Data/config files, images/build cache and the private CD journal volume are retained.**
No `down -v`, volume prune or removal of unrelated/CI resources is used.
Cleanup failure retains the lock so the always recovery step can retry using the journal.
A failed bootstrap journal prevents automatic retry: review preserved data and diagnostics
before another bootstrap; do not automatically empty those directories or delete the journal.
Abrupt host/runner loss still requires operator review of the lock/journal.
The journal is private (0700/0600) and may contain prior/candidate secrets; never export it
as an Actions artifact. First bootstrap has no prior healthy version to roll back to.
### First run and acceptance
1. In Gitea repository **Actions → Verify Frame Flow → Run workflow**, choose branch
`main`, action **plan**. This is read-only and reports bootstrap required for the
absent/created instance; it is not a bootstrap safety acceptance. Detailed host
residue/directory checks happen under the lock in bootstrap.
2. Run the same workflow on `main` with action **bootstrap**. Verify runs in full first.
No NAS bootstrap is executed merely by committing this implementation.
3. Require `BOOTSTRAP VERIFIED` in the deploy log. In Container Manager confirm
container `pigway-frame-flow` is running/healthy, user 10333, restart unless-stopped,
exact `/volume2/docker/pigway-frame-flow/{data,config}` mounts and
`192.0.2.10:18787` binding (not loopback).
4. Open `http://192.0.2.10:18787/api/v1/version`; its `commit` must equal the full
workflow checkout commit. Check `/ready` returns HTTP 200. The immutable image tag
is `pigway-frame-flow:release-<commit>-<transaction>`; compare its image ID in
Container Manager with the running container.
5. Keep `PROD_DEPLOY_ENABLED=true`. Subsequent main pushes Verify then deploy using
this same runtime and journal mechanism. The first update records bootstrap as
its rollback baseline. Repeating bootstrap against this instance is refused.
Enable real-device read-only access and write access only through separate explicit
operator decisions; no automatic change is made by bootstrap.
Docker tests are mocked locally. The first manual NAS Runner bootstrap, including
real host mounts and restart, remains the required live acceptance.
scripts/production-deploy.py
+274 -1
View File
@@ -381,6 +381,232 @@ class Deploy:
)
return config
def container_optional(self):
result = self.run("container", "ls", "-a", "--format", "{{.Names}}")
if self.env["PROD_CONTAINER"] not in result.stdout.decode().splitlines():
return None
return self.inspect(self.env["PROD_CONTAINER"])
def host_paths(self, paths):
# Inspect the daemon host, never the job container workspace. Read-only root.
code = "import json,sys;from pathlib import Path;print(json.dumps([{'exists':p.exists() or p.is_symlink() or any(a.is_symlink() for a in p.parents),'empty':p.is_dir() and not p.is_symlink() and not any(a.is_symlink() for a in p.parents) and not any(p.iterdir())} for p in [Path('/host'+v) for v in json.load(sys.stdin)]]))"
return json.loads(
self.run(
"run",
"--rm",
"-i",
"--user",
"0:0",
"--network",
"none",
"--mount",
"type=bind,source=/,target=/host,readonly",
"--entrypoint",
"python",
self.old_image,
"-c",
code,
data=json.dumps(paths).encode(),
).stdout
)
def bootstrap_check(self):
self.settings()
if self.env["PROD_MOCK"] != "true" or self.env["PROD_WRITE_ENABLED"] != "false":
raise ValueError("Bootstrap requires mock=true and write=false")
if self.env.get("PROD_PHOTOS_DIR"):
raise ValueError("First bootstrap must not attach Photos")
paths = [self.env[k] for k in ("PROD_DATA_DIR", "PROD_CONFIG_DIR")]
if any(".." in Path(p).parts or p.endswith("/") for p in paths) or any(
Path(a) in Path(b).parents
for a, b in ((paths[0], paths[1]), (paths[1], paths[0]))
):
raise ValueError("Unsafe or overlapping bootstrap directories")
old = self.container_optional()
if old:
labels = old["Config"].get("Labels") or {}
state = old["State"]
if (
state.get("Status") != "created"
or state.get("Running")
or state.get("StartedAt") != "0001-01-01T00:00:00Z"
or labels.get("com.docker.compose.project") != self.env["PROD_PROJECT"]
or labels.get("com.docker.compose.service") != "pigway-frame-flow"
):
raise ValueError(
"Bootstrap refused: existing container is not a never-started owned residue"
)
mounts = old["Mounts"]
if (
len(mounts) != 2
or {m["Destination"] for m in mounts} != {"/data", "/config"}
or any(m["Type"] != "bind" for m in mounts)
):
raise ValueError("Bootstrap refused: unknown persistent mount")
if any(v["exists"] for v in self.host_paths([m["Source"] for m in mounts])):
raise ValueError("Bootstrap refused: old mount has existing data/path")
if any(v["exists"] and not v["empty"] for v in self.host_paths(paths)):
raise ValueError("Bootstrap refused: target directory nonempty or unsafe")
volume = self.run("volume", "ls", "--format", "{{.Name}}")
if self.env["PROD_STATE_VOLUME"] in volume.stdout.decode().splitlines():
raise ValueError(
"Bootstrap refused: existing deployment journal requires review"
)
return old
def bootstrap_config(self):
config = yaml.safe_load((ROOT / "docker-compose.yml").read_text())
config["name"] = self.env["PROD_PROJECT"]
service = config["services"]["pigway-frame-flow"]
service.pop("build")
service.pop("env_file")
service["container_name"] = self.env["PROD_CONTAINER"]
service["environment"] = {}
service["ports"] = [
f"{self.env['PROD_WEB_BIND']}:{self.env['PROD_WEB_PORT']}:18787"
]
service["labels"] = {"frameflow.bootstrap.owner": self.owner}
service["volumes"] = [
{
"type": "bind",
"source": self.env[key],
"target": target,
"bind": {"create_host_path": False},
}
for key, target in (
("PROD_DATA_DIR", "/data"),
("PROD_CONFIG_DIR", "/config"),
)
]
return config
def bootstrap_cleanup(self):
old = self.container_optional()
if old:
if (old["Config"].get("Labels") or {}).get(
"frameflow.bootstrap.owner"
) != self.owner:
raise ValueError("Refusing cleanup of another container")
self.run("rm", "-f", old["Id"])
if self.state and self.state.get("network_created"):
name = self.env["PROD_PROJECT"] + "_default"
result = self.run("network", "inspect", name, check=False)
if result.returncode == 0:
network = json.loads(result.stdout)[0]
if (
network.get("Containers")
or (network.get("Labels") or {}).get("com.docker.compose.project")
!= self.env["PROD_PROJECT"]
):
raise ValueError("Refusing cleanup of occupied/unknown network")
self.run("network", "rm", network["Id"])
def bootstrap(self):
self.settings()
if self.env["PROD_MOCK"] != "true" or self.env["PROD_WRITE_ENABLED"] != "false":
raise ValueError("Bootstrap requires mock=true and write=false")
self.secret_config()
# Build before host inspection so helpers use this verified repository image.
candidate = self.build(self.bootstrap_config())
self.old_image = candidate["services"]["pigway-frame-flow"]["image"]
self.lock()
safe_unlock = True
try:
residue = self.bootstrap_check()
network_name = self.env["PROD_PROJECT"] + "_default"
networks = (
self.run("network", "ls", "--format", "{{.Name}}")
.stdout.decode()
.splitlines()
)
if network_name in networks:
network = json.loads(
self.run("network", "inspect", network_name).stdout
)[0]
if (
set(network.get("Containers") or {})
- ({residue["Id"]} if residue else set())
or (network.get("Labels") or {}).get("com.docker.compose.project")
!= self.env["PROD_PROJECT"]
):
raise ValueError(
"Bootstrap network identity/occupancy requires review"
)
self.state = {
"owner": self.owner,
"kind": "bootstrap",
"status": "prepared",
"candidate": candidate,
"network_created": network_name not in networks,
}
self.run(
"volume",
"create",
"--label",
"frameflow.deploy.project=" + self.env["PROD_PROJECT"],
self.env["PROD_STATE_VOLUME"],
)
self.state_io(self.state)
if residue:
self.run("rm", residue["Id"])
# -v creates only the two explicit daemon-host paths. No parent is mounted writable.
self.run(
"run",
"--rm",
"--user",
"0:0",
"--network",
"none",
"-v",
self.env["PROD_DATA_DIR"] + ":/data",
"-v",
self.env["PROD_CONFIG_DIR"] + ":/config",
"--entrypoint",
"python",
self.old_image,
"-c",
"import os;from pathlib import Path;paths=[Path('/data'),Path('/config')];assert all(not any(p.iterdir()) for p in paths);[(os.chown(p,10333,10333),os.chmod(p,0o750)) for p in paths]",
)
self.activate(candidate)
self.verify(candidate)
self.preflight()
self.run("restart", self.env["PROD_CONTAINER"])
self.verify(candidate)
self.preflight()
info = self.inspect(self.env["PROD_CONTAINER"])
for _ in range(60):
if info["State"].get("Health", {}).get("Status") == "healthy":
break
time.sleep(2)
info = self.inspect(self.env["PROD_CONTAINER"])
else:
raise RuntimeError("Bootstrap health timeout")
expected_image = json.loads(
self.run(
"image",
"inspect",
candidate["services"]["pigway-frame-flow"]["image"],
).stdout
)[0]["Id"]
if info["Image"] != expected_image:
raise RuntimeError("Bootstrap image mismatch")
self.state["status"] = "success"
self.state_io(self.state)
print("BOOTSTRAP VERIFIED", flush=True)
except BaseException:
print("BOOTSTRAP FAILED", flush=True)
self.diagnostics()
if self.state:
safe_unlock = False
self.bootstrap_cleanup()
self.state["status"] = "failed"
self.state_io(self.state)
safe_unlock = True
raise
finally:
if safe_unlock:
self.unlock()
def diagnostics(self):
try:
self._diagnostics()
@@ -393,6 +619,19 @@ class Deploy:
result = self.run("inspect", self.env["PROD_CONTAINER"], check=False)
if result.returncode == 0:
info = json.loads(result.stdout)[0]
self.secret_values.extend(
value
for key, value in (
entry.split("=", 1)
for entry in info.get("Config", {}).get("Env", [])
if "=" in entry
)
if value
and any(
word in key
for word in ("KEY", "TOKEN", "PASSWORD", "SECRET", "SESSION")
)
)
state = info["State"]
print(
"Scoped production diagnostic:",
@@ -510,6 +749,16 @@ class Deploy:
]:
raise ValueError("Recovery state volume ownership mismatch")
state = json.loads(self.state_io())
if state.get("kind") == "bootstrap":
self.state = state
if state["owner"] != self.owner:
raise ValueError("Bootstrap journal owner mismatch")
if state["status"] == "prepared":
self.bootstrap_cleanup()
state["status"] = "failed"
self.state_io(state)
recovered = True
return
previous = state["previous"]
service = previous["services"]["pigway-frame-flow"]
if (
@@ -561,6 +810,15 @@ def main():
if any(not tool.env.get(key) for key in REQUIRED):
print("PLAN ONLY: configure required Variables:", ", ".join(REQUIRED))
return
if tool.container_optional() is None:
print("PLAN ONLY: production not bootstrapped / bootstrap required")
return
info = tool.inspect(tool.env["PROD_CONTAINER"])
if info["State"].get("Status") == "created":
print(
"PLAN ONLY: production not bootstrapped; created residue requires bootstrap safety checks"
)
return
baseline = tool.preflight()
tool.secret_config()
tool.verify(baseline)
@@ -576,7 +834,22 @@ def main():
raise SystemExit(143)
signal.signal(signal.SIGTERM, cancel)
if action == "deploy":
if action == "bootstrap":
if tool.env.get("GITHUB_EVENT_NAME") != "workflow_dispatch":
raise SystemExit("Bootstrap requires explicit workflow_dispatch")
try:
tool.bootstrap()
except BaseException:
print(
"BOOTSTRAP FAILED: inspect redacted diagnostics; data/config retained",
flush=True,
)
raise
elif action == "deploy":
info = tool.container_optional()
if info is None or info["State"].get("Status") == "created":
print("production not bootstrapped / bootstrap required")
return
tool.deploy()
elif action == "rollback":
tool.recover(manual=True)
tests/test_production_bootstrap.py
+323
View File
@@ -0,0 +1,323 @@
"""First production bootstrap safety and failure paths, mocked Docker only."""
import json
from types import SimpleNamespace
import pytest
from test_production_deploy import existing
from test_production_deploy import tool as deploy_fixture
@pytest.fixture
def tool():
return deploy_fixture.__wrapped__()
def residue(tool):
value = existing(tool)
value.update(Id="residue-id")
value["State"] = {
"Status": "created",
"Running": False,
"StartedAt": "0001-01-01T00:00:00Z",
}
for mount in value["Mounts"]:
mount["Source"] = "/workspace/way/pigway-frame-flow" + mount["Destination"]
return value
def prepare(tool, monkeypatch, old=None):
tool.env["PROD_MOCK"] = "true"
monkeypatch.setattr(tool, "container_optional", lambda: old)
monkeypatch.setattr(
tool,
"host_paths",
lambda paths: [{"exists": False, "empty": False} for _ in paths],
)
monkeypatch.setattr(
tool, "run", lambda *args, **kw: SimpleNamespace(stdout=b"", returncode=0)
)
@pytest.mark.parametrize("kind", ["absent", "residue"])
def test_bootstrap_accepts_only_empty_first_install(tool, monkeypatch, kind):
old = residue(tool) if kind == "residue" else None
prepare(tool, monkeypatch, old)
assert tool.bootstrap_check() == old
@pytest.mark.parametrize(
"kind",
[
"running",
"stopped",
"old-path",
"extra-mount",
"identity",
"target-nonempty",
"target-symlink",
"journal",
],
)
def test_bootstrap_fail_closed(tool, monkeypatch, kind):
old = residue(tool)
prepare(tool, monkeypatch, old)
if kind == "running":
old["State"].update(Status="running", Running=True)
elif kind == "stopped":
old["State"].update(Status="exited", StartedAt="2026-10-01T00:00:00Z")
elif kind == "extra-mount":
old["Mounts"].append({"Type": "volume", "Destination": "/unknown"})
elif kind == "identity":
old["Config"]["Labels"] = {}
elif kind == "journal":
monkeypatch.setattr(
tool,
"run",
lambda *a, **kw: SimpleNamespace(
stdout=tool.env["PROD_STATE_VOLUME"].encode()
),
)
else:
monkeypatch.setattr(
tool,
"host_paths",
lambda paths: [
{
"exists": kind == "old-path"
or not paths[0].startswith("/workspace"),
"empty": False,
}
for _ in paths
],
)
with pytest.raises(ValueError):
tool.bootstrap_check()
@pytest.mark.parametrize("mock,writes", [("false", "false"), ("true", "true")])
def test_bootstrap_requires_mock_and_no_writes(tool, monkeypatch, mock, writes):
prepare(tool, monkeypatch)
tool.env.update(PROD_MOCK=mock, PROD_WRITE_ENABLED=writes)
with pytest.raises(ValueError, match="mock=true"):
tool.bootstrap_check()
@pytest.mark.parametrize("failure", [None, "ready", "smoke", "commit"])
def test_bootstrap_verifies_restart_and_preserves_data_on_failure(
tool, monkeypatch, failure
):
prepare(tool, monkeypatch)
candidate = tool.bootstrap_config()
candidate["services"]["pigway-frame-flow"]["image"] = "release-image"
calls, saved, verifies = [], [], []
monkeypatch.setattr(tool, "secret_config", lambda: {})
monkeypatch.setattr(tool, "build", lambda config: candidate)
monkeypatch.setattr(tool, "bootstrap_check", lambda: None)
monkeypatch.setattr(tool, "lock", lambda: calls.append(("lock",)))
monkeypatch.setattr(tool, "unlock", lambda: calls.append(("unlock",)))
monkeypatch.setattr(tool, "state_io", lambda state: saved.append(dict(state)))
monkeypatch.setattr(
tool, "activate", lambda config: calls.append(("activate", config))
)
monkeypatch.setattr(tool, "preflight", lambda: calls.append(("preflight",)))
monkeypatch.setattr(tool, "diagnostics", lambda: None)
monkeypatch.setattr(
tool,
"inspect",
lambda _: {"Image": "sha256:new", "State": {"Health": {"Status": "healthy"}}},
)
def run(*args, **kw):
calls.append(args)
output = b""
if args[:2] == ("image", "inspect"):
output = b'[{"Id":"sha256:new"}]'
return SimpleNamespace(
stdout=output, returncode=1 if args[:2] == ("network", "inspect") else 0
)
monkeypatch.setattr(tool, "run", run)
def verify(config):
verifies.append(config)
if failure:
raise RuntimeError(failure)
monkeypatch.setattr(tool, "verify", verify)
if failure:
with pytest.raises(RuntimeError, match=failure):
tool.bootstrap()
assert saved[-1]["status"] == "failed"
else:
tool.bootstrap()
assert len(verifies) == 2
assert ("restart", tool.env["PROD_CONTAINER"]) in calls
assert saved[-1]["status"] == "success"
assert not any("down" in c or "prune" in c for c in calls)
assert not any(c[0] in ("rm", "volume") and "remove" in c for c in calls)
def test_bootstrap_config_is_adoptable_by_preflight(tool, monkeypatch):
prepare(tool, monkeypatch)
config = tool.bootstrap_config()
runtime = existing(tool)
monkeypatch.setattr(tool, "inspect", lambda _: runtime)
monkeypatch.setattr(tool, "run", lambda *a, **kw: SimpleNamespace(returncode=1))
adopted = tool.preflight()
service = config["services"]["pigway-frame-flow"]
other = adopted["services"]["pigway-frame-flow"]
for key in ("volumes", "ports", "container_name", "restart", "user"):
assert service[key] == other[key]
assert config["name"] == adopted["name"]
def test_cleanup_only_removes_owned_container_and_empty_owned_network(
tool, monkeypatch
):
prepare(tool, monkeypatch)
old = residue(tool)
old["Config"]["Labels"]["frameflow.bootstrap.owner"] = tool.owner
monkeypatch.setattr(tool, "container_optional", lambda: old)
tool.state = {"network_created": True}
calls = []
def run(*args, **kw):
calls.append(args)
return SimpleNamespace(
returncode=0,
stdout=json.dumps(
[
{
"Id": "network-id",
"Containers": {},
"Labels": {
"com.docker.compose.project": tool.env["PROD_PROJECT"]
},
}
]
).encode(),
)
monkeypatch.setattr(tool, "run", run)
tool.bootstrap_cleanup()
assert calls == [
("rm", "-f", "residue-id"),
("network", "inspect", "production-frame_default"),
("network", "rm", "network-id"),
]
old["Config"]["Labels"].pop("frameflow.bootstrap.owner")
with pytest.raises(ValueError):
tool.bootstrap_cleanup()
def test_diagnostics_masks_all_secrets(tool, monkeypatch, capsys):
tool.secret_values = ["test-secret"]
monkeypatch.setattr(
tool,
"run",
lambda *a, **kw: SimpleNamespace(
returncode=1, stdout=b"test-secret https://private.invalid/key", stderr=b""
),
)
tool.diagnostics()
output = capsys.readouterr().out
assert "test-secret" not in output and "private.invalid" not in output
@pytest.mark.parametrize("failure", ["ready", "smoke", "commit"])
def test_actual_verify_rejects_ready_smoke_and_commit_failure(
tool, monkeypatch, failure
):
monkeypatch.setattr(
tool.verify.__func__.__globals__["time"], "sleep", lambda _: None
)
config = {
"services": {"pigway-frame-flow": {"environment": {"BUILD_COMMIT": "expected"}}}
}
def run(*args, **kw):
if "-i" in args and failure == "smoke":
raise RuntimeError("smoke failed")
if "-i" not in args and failure == "ready":
return SimpleNamespace(returncode=1)
return SimpleNamespace(returncode=0, stdout=b'{"commit":"wrong"}')
monkeypatch.setattr(tool, "run", run)
with pytest.raises(RuntimeError):
tool.verify(config)
def test_workflow_bootstrap_manual_only_and_ci_namespace_separate(tool):
from pathlib import Path
import yaml
root = Path(__file__).parents[1]
workflow = yaml.safe_load((root / ".github/workflows/verify.yml").read_text())
inputs = workflow[True]["workflow_dispatch"]["inputs"]["deploy_action"]
assert inputs["options"] == ["plan", "bootstrap", "deploy", "rollback"]
assert inputs["default"] == "plan"
assert workflow["jobs"]["deploy"]["needs"] == "verify"
assert "success()" in workflow["jobs"]["deploy"]["if"]
tool.env["PROD_PROJECT"] = "ffci-invalid"
with pytest.raises(ValueError):
tool.settings()
def test_automatic_push_without_production_never_creates_it(tool, monkeypatch, capsys):
import importlib.util
from pathlib import Path
spec = importlib.util.spec_from_file_location(
"bootstrap_main", Path(__file__).parents[1] / "scripts/production-deploy.py"
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
tool.env.update(PROD_DEPLOY_ENABLED="true", GITHUB_EVENT_NAME="push")
monkeypatch.setattr(module, "Deploy", lambda: tool)
monkeypatch.setattr(tool, "container_optional", lambda: None)
monkeypatch.setattr(tool, "deploy", lambda: pytest.fail("Production touched"))
monkeypatch.setattr(tool, "bootstrap", lambda: pytest.fail("Production touched"))
module.main()
assert "bootstrap required" in capsys.readouterr().out
tool.env["CD_ACTION"] = "bootstrap"
with pytest.raises(SystemExit, match="workflow_dispatch"):
module.main()
def test_bootstrap_recovery_cleans_only_owned_prepared_transaction(tool, monkeypatch):
prepare(tool, monkeypatch)
calls = []
tool.state = {
"owner": tool.owner,
"kind": "bootstrap",
"status": "prepared",
"network_created": True,
}
def run(*args, **kw):
calls.append(args)
value = [{"Labels": {"frameflow.deploy.project": tool.env["PROD_PROJECT"]}}]
if args[0] == "inspect":
value = [
{
"Id": "lock-id",
"Image": "built-image",
"Config": {"Labels": {"frameflow.deploy.owner": tool.owner}},
}
]
return SimpleNamespace(returncode=0, stdout=json.dumps(value).encode())
monkeypatch.setattr(tool, "run", run)
monkeypatch.setattr(tool, "bootstrap_cleanup", lambda: calls.append(("cleanup",)))
def state_io(payload=None):
if payload is not None:
tool.state = payload
return json.dumps(tool.state).encode()
monkeypatch.setattr(tool, "state_io", state_io)
tool.recover()
assert tool.state["status"] == "failed"
assert ("cleanup",) in calls and ("rm", "lock-id") in calls