2 Commits

Author SHA1 Message Date
laily d7ebb6ee85 feat: update 2026-10-14 12:14:31 +08:00
laily 19aac17a0a feat(ship): clarify delivery stop semantics 2026-10-14 11:49:50 +08:00
22 changed files with 861 additions and 93 deletions
+1 -1
View File
@@ -223,7 +223,7 @@ Docker 只有你点名镜像时才走。token 不会写进输出、提交或回
| `scripts/verify_deb.sh` | 核对包元数据、内容与 SHA-256 |
| `scripts/copy_artifact.sh` | 按 `COPY_SCRIPT` 分发产物,未配置时拷到 `COPY_DEST` |
| `scripts/obtain.sh` | 对精确 SHA 拿到产物;`BUILD_VENUE=ci` 时等流水线再 inspect |
| `scripts/wait_ci.sh` | 只等该 SHA 的 GitLab/GitHub 流水线,不触发 |
| `scripts/wait_ci.sh` | 只等该 SHA 的 GitLab/GitHub/Gitea 流水线,不触发;Gitea 读取 workflow jobs |
| `templates/makefile.android.builder` | Android APK + copy 轨道骨架 |
| `templates/makefile.ios.builder` | iOS IPA + copy 轨道骨架 |
+1 -1
View File
@@ -90,6 +90,6 @@ COPY_AS=latest.apk
按 [obtain.md](references/obtain.md) 执行 `scripts/obtain.sh --sha <commit>`。
- `BUILD_VENUE=local`:当前 HEAD 必须是该 SHA,跑 makefile 的 docker/deb。
- `BUILD_VENUE=ci`:等该 SHA 的 GitLab/GitHub 流水线,再 inspect `version.sh --docker` 的镜像 tag。不触发流水线,不本机 push。
- `BUILD_VENUE=ci`:等该 SHA 的 GitLab/GitHub/Gitea 流水线,再 inspect `version.sh --docker` 的镜像 tag。不触发流水线,不本机 push。
- 失败(无流水线、job 失败、digest 不对)必须原样返回给调用方,不得标 `artifact_built`。
- 合进默认分支若项目不跑 CI,这是正确失败;预览用 `dev/*`,正式用 tag。
+2 -2
View File
@@ -82,7 +82,7 @@ make -f makefile.builder <target>
| 变量 | 默认 | 说明 |
|------|------|------|
| `BUILD_VENUE` | `local` | 产物在哪构建:`local` 本机 make;`ci` 由 GitLab/GitHub 流水线构建。这不是新轨道。写在 `makefile.builder` 并提交。`ci` 时 `push-docker` 委托 `obtain.sh`(等流水线 + inspect),不跑 `publish_docker.sh` |
| `BUILD_VENUE` | `local` | 产物在哪构建:`local` 本机 make;`ci` 由 GitLab/GitHub/Gitea 流水线构建。这不是新轨道。写在 `makefile.builder` 并提交。`ci` 时 `push-docker` 委托 `obtain.sh`(等流水线 + inspect),不跑 `publish_docker.sh` |
| `ARCHES` | `amd64` | 本项目要发布的架构清单,空格分隔,子集只能是 `amd64` `arm64`。写在 `makefile.builder` 里并提交;不要放进 `.env.builder`。缺省或无 `arches` 目标时视为单架构 `amd64`。不要默认写成双架构 |
| `ARCH` | `$(firstword $(ARCHES))` | 这一次构建的架构。必须落在 `ARCHES` 里;非法值必须 `$(error)` 并提示合法值。命令行/环境传入时只构建/上传这一项 |
| `VERSION` | `` (空) | 为空时按「版本号」节从 Git 祖先稳定 tag 推导;禁止 `sort -V` 取全局最新 |
@@ -189,7 +189,7 @@ include $(HOME)/.pouch/skills/builder/scripts/version.mk
`PLATFORMS` 优先级(高 → 低):用户这轮 `--platform` 或已 export 的 `PLATFORMS` → 命令行 `ARCH=` 映射的 `linux/$(ARCH)`(由 `push-docker` 传入)→ `.env.builder` 的 `PLATFORMS` → `makefile.builder` 的 `ARCHES` → `linux/amd64`。
`BUILD_VENUE=ci` 时 Docker 发布路径是 `obtain.sh`:只认 **该 commit SHA** 的流水线。没有流水线、job 失败、超时、或 Harbor inspect 不到 `version.sh --docker` 算出的 tag,一律非零退出。不触发流水线。合进默认分支若项目声明不构建,这是正确失败;调用方应打 tag 或推送到项目会构建的 ref(例如 `dev/*`)。CI job 若已经 `docker push`,obtain 成功即视为产物已在 registry,不要再本机 push 一次。
`BUILD_VENUE=ci` 时 Docker 发布路径是 `obtain.sh`:只认 **该 commit SHA** 的流水线。没有流水线、job 失败、超时、或 Harbor inspect 不到 `version.sh --docker` 算出的 tag,一律非零退出。不触发流水线。GitHub 用 `gh`、GitLab 用 `glab`、Gitea 用 `curl` 读取 Actions API;Gitea 会再次校验 `head_sha` 并读取 workflow jobs。合进默认分支若项目声明不构建,这是正确失败;调用方应打 tag 或推送到项目会构建的 ref(例如 `dev/*`)。CI job 若已经 `docker push`,obtain 成功即视为产物已在 registry,不要再本机 push 一次。
配置来源优先级:shell 已显式设置的值 > 项目根 `.env.builder` > 失败并询问用户。
`.env.builder` 由 builder 脚本加载(不回显任何值);当前 shell 已设置的值优先。
+2 -1
View File
@@ -36,12 +36,13 @@ TAG=1.4.2
| 0 | 产物已核验 | `artifact_built` 完成 |
| 1 | 该 SHA 无流水线、job 失败、超时、inspect 失败 | 标 `failed`,不得 publish/deploy |
| 2 | 用法或配置错误 | 标 `blocked` |
| 3 | glab/gh 缺失或读不到该项目 | 停;可用**已打开且已登录**的同 host Orca tab 查流水线。禁止新开未登录页 |
| 3 | glab/gh/curl 缺失或读不到该项目 | 停;可用**已打开且已登录**的同 host Orca tab 查流水线。禁止新开未登录页 |
## `ci` 规则
- 只认该 SHA 的 pipeline,不认默认分支「最新一次绿」。
- 不触发流水线。
- Gitea 使用 Actions REST API 的 `head_sha` 精确过滤,并读取每个 workflow run 的 jobs;私有仓库可在 `.env.builder` 配置 `GITEA_TOKEN`。自定义 Gitea host 可设置 `BUILDER_CI_FORGE=gitea`,API 前缀默认是 `https://<origin-host>/api/v1`,也可用 `GITEA_API_URL` 覆盖。
- job 变绿不够,docker 轨道必须 inspect 到 digest。
- 合进 `main` 若项目不构建,`no_pipeline` 是正确失败。预览走项目会构建的 ref(如 `dev/*`),正式走 tag。
+26
View File
@@ -612,8 +612,23 @@ def origin_forge(project: Path) -> str:
check=False,
)
url = (result.stdout or "").strip()
configured = env_file_values(project).get("BUILDER_CI_FORGE", "").strip().lower()
forced = (os.environ.get("BUILDER_CI_FORGE") or configured).strip().lower()
if forced in ("github", "gitlab", "gitea"):
return forced
if "github.com" in url:
return "github"
host_match = re.search(r"(?:https?://|ssh://git@|git@)([^/:]+)", url)
host = host_match.group(1).lower() if host_match else ""
builder_keys = env_file_keys(project)
if (
"gitea" in host
or "GITEA_TOKEN" in builder_keys
or "GITEA_API_URL" in builder_keys
or os.environ.get("GITEA_TOKEN")
or os.environ.get("GITEA_API_URL")
):
return "gitea"
if url:
return "gitlab"
return ""
@@ -646,6 +661,17 @@ def check_ready_toolchain(
lines.append(f"gh: {'found' if gh_ok else 'MISSING (blocks ci venue)'}")
if not gh_ok:
problems.append("install gh and authenticate for this GitHub repo")
elif forge == "gitlab":
glab_ok = shutil.which("glab") is not None
lines.append(f"glab: {'found' if glab_ok else 'MISSING (blocks ci venue)'}")
if not glab_ok:
problems.append("install glab and authenticate for this GitLab project")
elif forge == "gitea":
curl_ok = shutil.which("curl") is not None
lines.append(f"curl: {'found' if curl_ok else 'MISSING (blocks ci venue)'}")
lines.append("Gitea API: /api/v1; GITEA_TOKEN optional for public repositories")
if not curl_ok:
problems.append("install curl to read Gitea Actions for this repository")
else:
glab_ok = shutil.which("glab") is not None
lines.append(f"glab: {'found' if glab_ok else 'MISSING (blocks ci venue)'}")
+1
View File
@@ -11,6 +11,7 @@ set -euo pipefail
# BUILD_VENUE (makefile.builder, default local):
# local HEAD must be SHA; run `make -f makefile.builder docker` or `deb`
# ci wait for the forge pipeline of this SHA, then inspect the image
# (GitHub Actions, GitLab CI, or Gitea Actions)
#
# Exit:
# 0 artifact verified; prints ARTIFACT= / DIGEST= / SHA= / VENUE=
+256 -67
View File
@@ -7,6 +7,11 @@ set -euo pipefail
# Usage:
# wait_ci.sh --sha SHA [--root DIR] [--timeout SEC]
#
# Supported forges:
# GitHub Actions (gh), GitLab CI (glab), and Gitea Actions (curl + REST API).
# Set BUILDER_CI_FORGE=gitea for a custom Gitea host that does not contain
# "gitea" in its hostname. GITEA_API_URL may override the /api/v1 endpoint.
#
# Exit:
# 0 pipeline succeeded; prints "OK pipeline=<url> sha=<sha>"
# 1 no pipeline for this SHA, or a job failed/canceled
@@ -14,20 +19,20 @@ set -euo pipefail
# 3 forge CLI missing, or it cannot read this project
usage() {
sed -n '2,16p' "$0"
sed -n '2,18p' "$0"
}
sha=
root=
timeout_sec=1200
poll_sec=10
appear_sec=90
poll_sec=${BUILDER_CI_POLL_SEC:-10}
appear_sec=${BUILDER_CI_APPEAR_SEC:-90}
while (($#)); do
case "$1" in
--sha) sha=$2; shift 2 ;;
--root) root=$2; shift 2 ;;
--timeout) timeout_sec=$2; shift 2 ;;
--sha) [[ $# -ge 2 ]] || { echo "Error: --sha needs a value" >&2; exit 2; }; sha=$2; shift 2 ;;
--root) [[ $# -ge 2 ]] || { echo "Error: --root needs a value" >&2; exit 2; }; root=$2; shift 2 ;;
--timeout) [[ $# -ge 2 ]] || { echo "Error: --timeout needs a value" >&2; exit 2; }; timeout_sec=$2; shift 2 ;;
-h|--help) usage; exit 0 ;;
*) echo "Error: unknown argument: $1" >&2; usage >&2; exit 2 ;;
esac
@@ -46,41 +51,67 @@ if [[ -z "$sha" ]]; then
fi
sha=$(git -C "$root" rev-parse "$sha")
# Load only builder configuration. Shell values win and values are never echoed.
if [[ -f "$root/.env.builder" ]]; then
while IFS='=' read -r key value; do
key=${key%%[[:space:]]*}
[[ -z "$key" || "$key" == \#* ]] && continue
[[ "$key" =~ ^[A-Za-z_][A-Za-z0-9_]*$ ]] || continue
if [[ -n "${!key:-}" ]]; then
continue
fi
value=${value%\"}; value=${value#\"}; value=${value%\'}; value=${value#\'}
[[ -z "$value" ]] && continue
printf -v "$key" '%s' "$value"
export "$key"
done < <(grep -v '^[[:space:]]*$' "$root/.env.builder")
fi
python_json() {
python3 -c "$1" "${@:2}"
}
parse_origin() {
python_json '
import os, re
url = os.environ["WAIT_CI_ORIGIN"].strip()
url = re.sub(r"^git@([^:]+):", r"ssh://git@\1/", url)
url = re.sub(r"^ssh://git@([^/]+):", r"ssh://git@\1/", url)
if re.match(r"^https?://", url):
m = re.match(r"^(https?)://([^/]+)(?:/(.*))?$", url)
scheme, host, path = (m.group(1), m.group(2), m.group(3) or "") if m else ("", "", "")
else:
m = re.match(r"^ssh://git@([^/]+)/(.+)$", url)
scheme, host, path = ("https", m.group(1), m.group(2)) if m else ("", "", "")
path = re.sub(r"\.git$", "", path)
override = os.environ.get("BUILDER_CI_FORGE", "").strip().lower()
if override in ("github", "gitlab", "gitea"):
kind = override
elif host == "github.com" or host.endswith(".github.com"):
kind = "github"
elif ("gitea" in host.lower() or os.environ.get("GITEA_TOKEN") or
os.environ.get("GITEA_API_URL")):
kind = "gitea"
else:
kind = "gitlab"
print(kind, host, path, (scheme or "https") + "://" + host)
'
}
origin=$(git -C "$root" remote get-url origin 2>/dev/null || true)
if [[ -z "$origin" ]]; then
echo "Error: no origin remote" >&2
exit 2
fi
python_json() {
python3 -c "$1"
}
parse_origin() {
python_json '
import os, re, sys
url = os.environ["WAIT_CI_ORIGIN"]
url = url.strip()
url = re.sub(r"^git@", "ssh://git@", url)
url = re.sub(r"^ssh://git@([^/]+):", r"ssh://git@\1/", url)
url = re.sub(r"^https?://", "ssh://git@", url)
# ssh://git@host/group/repo.git
m = re.match(r"ssh://git@([^/]+)/(.+?)(?:\.git)?$", url)
if not m:
print("", "", "")
raise SystemExit(0)
host, path = m.group(1), m.group(2)
kind = "github" if host == "github.com" or host.endswith(".github.com") else "gitlab"
print(kind, host, path)
'
}
export WAIT_CI_ORIGIN=$origin
read -r forge_kind forge_host forge_path <<<"$(parse_origin)"
read -r forge_kind forge_host forge_path forge_base <<<"$(parse_origin)"
if [[ -z "$forge_kind" || -z "$forge_path" ]]; then
echo "Error: cannot parse origin: $origin" >&2
exit 2
fi
export WAIT_CI_PATH=$forge_path
export WAIT_CI_SHA=$sha
gitlab_pipelines() {
local encoded
@@ -94,30 +125,42 @@ gitlab_jobs() {
glab api "projects/${encoded}/pipelines/${pid}/jobs"
}
export WAIT_CI_PATH=$forge_path
github_runs() {
gh run list --repo "$forge_path" --commit "$sha" --limit 20 \
--json databaseId,status,conclusion,url,headSha
}
if [[ "$forge_kind" == github ]]; then
if ! command -v gh >/dev/null 2>&1; then
echo "Error: gh is required to watch GitHub Actions for $forge_host" >&2
exit 3
fi
else
if ! command -v glab >/dev/null 2>&1; then
echo "Error: glab is required to watch GitLab CI for $forge_host" >&2
exit 3
fi
gitea_api_base=${GITEA_API_URL:-$forge_base/api/v1}
gitea_api_base=${gitea_api_base%/}
if [[ "$gitea_api_base" != */api/v1 ]]; then
gitea_api_base+="/api/v1"
fi
gitea_repo_path=$(python_json 'import os,urllib.parse; print(urllib.parse.quote(os.environ["WAIT_CI_PATH"], safe="/"))')
started=$(date +%s)
first_seen=0
empty_since=$started
gitea_curl() {
local url=$1
if [[ -n "${GITEA_TOKEN:-}" ]]; then
# Process substitution keeps the token out of curl's argument list and logs.
curl -fsS --retry 2 --connect-timeout 10 --max-time 30 \
--config <(printf 'header = "Authorization: token %s"\n' "$GITEA_TOKEN") "$url"
else
curl -fsS --retry 2 --connect-timeout 10 --max-time 30 "$url"
fi
}
gitea_runs() {
gitea_curl "${gitea_api_base}/repos/${gitea_repo_path}/actions/runs?head_sha=${sha}&limit=20&page=1"
}
gitea_jobs() {
gitea_curl "${gitea_api_base}/repos/${gitea_repo_path}/actions/runs/$1/jobs?limit=100&page=1"
}
summarize_gitlab() {
python_json '
import json, os, sys
raw = sys.stdin.read()
import json, sys
try:
data = json.loads(raw)
data = json.loads(sys.stdin.read())
except json.JSONDecodeError:
print("error")
raise SystemExit(0)
@@ -146,11 +189,6 @@ print(",".join(parts))
'
}
github_runs() {
gh run list --repo "$forge_path" --commit "$sha" --limit 20 \
--json databaseId,status,conclusion,url,headSha
}
summarize_github() {
python_json '
import json, sys
@@ -171,9 +209,143 @@ print(r.get("databaseId"), st, r.get("url", ""))
'
}
summarize_gitea_runs() {
python_json '
import json, sys
try:
data = json.loads(sys.stdin.read())
except json.JSONDecodeError:
print("error")
raise SystemExit(0)
if isinstance(data, dict) and data.get("message") and not (data.get("workflow_runs") or data.get("runs")):
print("auth", data.get("message", "")[:80].replace(" ", "_"))
raise SystemExit(0)
if isinstance(data, dict):
data = data.get("workflow_runs", data.get("runs", []))
if not isinstance(data, list):
print("error")
raise SystemExit(0)
want = sys.argv[1] if len(sys.argv) > 1 else ""
for run in data:
head = run.get("head_sha") or run.get("headSha") or (run.get("head_commit") or {}).get("id")
if not head or head.lower() != want.lower():
continue
run_id = run.get("id") or run.get("database_id")
if run_id is None:
continue
status = str(run.get("status") or "").lower()
conclusion = str(run.get("conclusion") or run.get("result") or "").lower()
url = run.get("html_url") or run.get("web_url") or run.get("url") or "-"
print("%s\t%s\t%s\t%s" % (run_id, status, conclusion, url))
' "$sha"
}
summarize_gitea_jobs() {
python_json '
import json, sys
try:
data = json.loads(sys.stdin.read())
except json.JSONDecodeError:
print("error")
raise SystemExit(0)
if isinstance(data, dict) and data.get("message") and not data.get("jobs"):
print("auth", data.get("message", "")[:80].replace(" ", "_"))
raise SystemExit(0)
if isinstance(data, dict):
data = data.get("jobs", [])
if not isinstance(data, list):
print("error")
raise SystemExit(0)
want = sys.argv[1] if len(sys.argv) > 1 else ""
parts = []
state = "success" if data else "empty"
for job in data:
head = job.get("head_sha") or job.get("headSha")
if head and want and head.lower() != want.lower():
continue
name = str(job.get("name") or job.get("id") or "job").replace(" ", "_")
status = str(job.get("status") or "").lower()
result = str(job.get("conclusion") or job.get("result") or "").lower()
value = result or status or "unknown"
parts.append("%s=%s" % (name, value))
if value in ("failure", "failed", "cancelled", "canceled", "skipped", "timed_out", "action_required"):
state = "failed"
elif value not in ("success", "succeeded", "completed"):
state = "running"
if not parts:
state = "empty"
print("%s\t%s" % (state, ",".join(parts)))
' "$sha"
}
gitea_status_line() {
local raw runs
if ! raw=$(gitea_runs); then
echo "error"
return 0
fi
runs=$(printf '%s' "$raw" | summarize_gitea_runs)
if [[ "$runs" == auth* || "$runs" == error ]]; then
printf '%s\n' "$runs"
return 0
fi
if [[ -z "$runs" ]]; then
echo "empty"
return 0
fi
local overall=success ids= urls= jobs_all= first=1
while IFS=$'\t' read -r id run_status conclusion url; do
[[ -z "$id" ]] && continue
local jobs job_line job_state job_summary
if ! jobs=$(gitea_jobs "$id"); then
echo "error"
return 0
fi
job_line=$(printf '%s' "$jobs" | summarize_gitea_jobs "$sha")
job_state=${job_line%%$'\t'*}
job_summary=${job_line#*$'\t'}
if [[ "$job_state" == auth* || "$job_state" == error ]]; then
printf '%s\n' "$job_state"
return 0
fi
if [[ "$first" -eq 0 ]]; then
ids+=","; urls+=","; jobs_all+=","
fi
ids+="$id"; urls+="$url"; jobs_all+="$job_summary"; first=0
if [[ "$run_status" =~ ^(failure|failed|cancelled|canceled|skipped|timed_out|action_required)$ || "$conclusion" =~ ^(failure|failed|cancelled|canceled|skipped|timed_out|action_required)$ || "$job_state" == failed ]]; then
overall=failed
elif [[ "$overall" == success && ( "$job_state" == running || "$job_state" == empty || ( "$run_status" != completed && "$run_status" != success ) ) ]]; then
overall=running
fi
done <<<"$runs"
printf '%s %s %s %s\n' "$ids" "$overall" "$urls" "$jobs_all"
}
if [[ "$forge_kind" == github ]]; then
if ! command -v gh >/dev/null 2>&1; then
echo "Error: gh is required to watch GitHub Actions for $forge_host" >&2
exit 3
fi
elif [[ "$forge_kind" == gitlab ]]; then
if ! command -v glab >/dev/null 2>&1; then
echo "Error: glab is required to watch GitLab CI for $forge_host" >&2
exit 3
fi
else
if ! command -v curl >/dev/null 2>&1; then
echo "Error: curl is required to watch Gitea Actions for $forge_host" >&2
exit 3
fi
fi
started=$(date +%s)
first_seen=0
empty_since=$started
probe_auth() {
local out rc kind
if [[ "$forge_kind" == github ]]; then
local out rc
set +e
out=$(github_runs 2>&1)
rc=$?
@@ -185,20 +357,35 @@ probe_auth() {
fi
return
fi
local out rc
if [[ "$forge_kind" == gitlab ]]; then
set +e
out=$(gitlab_pipelines 2>&1)
rc=$?
set -e
if [[ $rc -ne 0 ]]; then
echo "Error: glab cannot list pipelines for $forge_path on $forge_host" >&2
echo "$out" | sed '/token/Id; /Bearer/d' >&2
exit 3
fi
kind=$(printf '%s' "$out" | summarize_gitlab | awk '{print $1}')
if [[ "$kind" == auth ]]; then
echo "Error: glab is authenticated but cannot read $forge_path" >&2
exit 3
fi
return
fi
set +e
out=$(gitlab_pipelines 2>&1)
out=$(gitea_runs 2>&1)
rc=$?
set -e
if [[ $rc -ne 0 ]]; then
echo "Error: glab cannot list pipelines for $forge_path on $forge_host" >&2
echo "$out" | sed '/token/Id; /Bearer/d' >&2
echo "Error: curl cannot read Gitea Actions for $forge_path on $forge_host" >&2
echo "$out" | sed '/token/Id; /Bearer/d; /Authorization/Id' >&2
exit 3
fi
local kind
kind=$(printf '%s' "$out" | summarize_gitlab | awk '{print $1}')
kind=$(printf '%s' "$out" | summarize_gitea_runs | awk '{print $1}')
if [[ "$kind" == auth ]]; then
echo "Error: glab is authenticated but cannot read $forge_path" >&2
echo "Error: Gitea API rejected access to $forge_path" >&2
exit 3
fi
}
@@ -208,8 +395,10 @@ probe_auth
status_line() {
if [[ "$forge_kind" == github ]]; then
github_runs | summarize_github
else
elif [[ "$forge_kind" == gitlab ]]; then
gitlab_pipelines | summarize_gitlab
else
gitea_status_line
fi
}
@@ -224,7 +413,7 @@ while true; do
kind=${1:-empty}
if [[ "$kind" == empty ]]; then
if (( now - empty_since > appear_sec )); then
if (( now - empty_since >= appear_sec )); then
echo "FAIL reason=no_pipeline sha=$sha origin=$forge_host/$forge_path" >&2
echo "This SHA has no CI pipeline. Tag or push a ref the project actually builds." >&2
exit 1
@@ -240,6 +429,7 @@ while true; do
pid=$1
st=$2
url=$3
jobs_s=${4:-}
empty_since=$now
if [[ "$first_seen" -eq 0 ]]; then
echo "Watching pipeline $pid status=$st $url" >&2
@@ -248,18 +438,17 @@ while true; do
case "$st" in
success)
echo "OK pipeline=$url sha=$sha id=$pid"
echo "OK pipeline=$url sha=$sha id=$pid jobs=$jobs_s"
exit 0
;;
failed|canceled|cancelled|skipped)
jobs_s=
if [[ "$forge_kind" != github ]]; then
failed|failure|canceled|cancelled|skipped)
if [[ "$forge_kind" == gitlab ]]; then
jobs_s=$(gitlab_jobs "$pid" | summarize_jobs || true)
fi
echo "FAIL reason=job_$st sha=$sha pipeline=$url jobs=$jobs_s" >&2
exit 1
;;
created|pending|running|waiting_for_resource|preparing|waiting|queued|in_progress|waiting_for_resource)
created|pending|running|waiting_for_resource|preparing|waiting|queued|in_progress|requested)
sleep "$poll_sec"
;;
*)
+3
View File
@@ -11,6 +11,9 @@
# DOCKER_REPOSITORY=
# IMAGE_TAG=
# PLATFORMS= # optional override; project ARCHES lives in makefile.builder
# BUILDER_CI_FORGE= # optional: github, gitlab, or gitea; useful for custom hosts
# GITEA_API_URL= # optional: defaults to https://<origin-host>/api/v1
# GITEA_TOKEN= # optional token for private Gitea repositories
# COPY_DEST= # fallback: local dir, or docker dest container:/absolute/dir
# COPY_SCRIPT= # external/temporary executable override; project-local paths belong in makefile.builder
# COPY_AS= # optional extra basenames for the first file
@@ -7,27 +7,31 @@ change_ready -> verified -> pr_opened -> merged -> regression_passed -> artifact
```
入口可以从已有状态开始,但不能跳过本次变更所需的验证。内部里程碑可以更细,但项目
ACK 配置必须使用以下 canonical `stopAt`:
Ship 配置使用以下 canonical `stopAt`:
- `verified`:只完成代码和验证。
- `validation_ready`:不可变产物已部署到开发/测试环境并通过健康检查,可交给用户验证。
- `review_ready`:PR/MR、预览产物和非生产验证证据已齐备,等待审核。
- `released`:经过本次 approval 后完成稳定发布或生产部署。
- `develop_verified`:开发阶段验证完成,包含单元测试、新增功能集成测试和静态检查;不等于完整回归。
- `review_ready`:PR/MR 已创建,等待审核。
- `release_ready`:PR 已合并且完整回归通过,映射内部 `regression_passed`;不表示已发布或部署。
- `released`:产物已构建并发布,映射内部 `published`。
- `deployed`:目标环境实际部署完成,映射内部 `deployed`。
旧记录可继续使用 `verified`(`develop_verified` 的别名)和 `validation_ready`
(`release_ready` 的别名);新记录应使用 canonical 名称。
入口 skill 的自然语言停止点映射如下:
| 用户目标 | ACK stopAt | 必要内部里程碑 |
| 用户目标 | Ship stopAt | 必要内部里程碑 |
| --- | --- | --- |
| 只验证 | `verified` | `verified` |
| 只做开发验证 | `develop_verified` | `verified` |
| 发 PR | `review_ready` | `pr_opened`,必要时预览构建/测试 |
| 部署测试环境 | `validation_ready` | `artifact_built`、`deployed`、健康检查 |
| 合并后回归但不发布 | `review_ready` 或任务级中间状态 | `merged`、`regression_passed` |
| 完整正式部署 | `released` | `merged`、`regression_passed`、`artifact_built`、`published`、`deployed` |
| 合并后完整回归 | `release_ready` | `merged`、`regression_passed` |
| 构建并发布产物 | `released` | `merged`、`regression_passed`、`artifact_built`、`published` |
| 实际部署 | `deployed` | `merged`、`regression_passed`、`artifact_built`、`published`、`deployed` |
## 每个阶段的要求
- `change_ready`:记录基线 commit、工作树、目标分支、任务范围和风险。
- `verified`:根因或需求验收明确;自动测试和必要的黑盒/真机验证通过。
- `verified`:开发阶段根因或需求验收明确;任务相关单元测试、新增功能集成测试和静态检查通过。
- `pr_opened`:提交说明、测试证据、风险和回滚信息已写入 PR;不能把草稿 PR 当 `review_ready`。
- `merged`:目标分支的合并 commit 已确认,不能只确认源分支 commit。
- `regression_passed`:以合并 commit 执行项目规定的完整回归;失败则回到修复或停止。
@@ -39,7 +43,7 @@ ACK 配置必须使用以下 canonical `stopAt`:
测试面缺少结果时都不能通过回归。
- `artifact_built`:对**合并后的精确 commit SHA** 调用 builder `obtain.sh`。`BUILD_VENUE=ci` 时监听该 SHA 的流水线并 inspect 产物,不在本机重新编译。无流水线、job 失败或 digest 对不上则本阶段失败,不得进入 `published`。项目若声明默认分支不构建,合进 `main` 会 `no_pipeline`:预览推 `dev/*`,正式打 tag。
- `published`:镜像/APK 等外部分发位置可复核,记录 digest 或文件校验和。`venue=ci` 且 CI 已经 push 到 registry 时,obtain 成功即同时满足本阶段,不要再本机 `publish_docker.sh`。
- `deployed`:目标服务运行在预期版本,健康检查和关键路径验证通过;生产部署还必须进入 `released`。
- `deployed`:目标服务运行在预期版本,健康检查和关键路径验证通过。
## 配置和授权
+1 -1
View File
@@ -51,7 +51,7 @@
3. 按 JSON 行动:
- `status: done`:写完成报告,结束。
- `status: blocked` / `ask_user`:告诉用户缺口或问题,等回复。生产部署(`stopAt:
released` 且下一步是 `deploy`)必须再问一次,除非当前这句话已经授权。
deployed` 且下一步是 `deploy`)必须再问一次,除非当前这句话已经授权。
- `actor: manager`:补账本或问用户,再跑 `next_step.py`。
- `actor: worker`:开新 agent,等待,把结果写入 `manager.lastStep` /
`manager.lastResult` / `manager.lastSummary`,按 `targetState` / `taskId` 更新账本,立刻
+14
View File
@@ -90,6 +90,20 @@ evidence:
`delivery.state` 使用 delivery-loop 内部里程碑:`change_ready`、`verified`、
`pr_opened`、`merged`、`regression_passed`、`artifact_built`、`published`、`deployed`。
用户可见的 canonical `stopAt` 与内部里程碑映射如下:
| stopAt | 停止含义 | 内部里程碑 |
| --- | --- | --- |
| `develop_verified` | 开发验证完成:单元测试、新增功能集成测试和静态检查通过;不等于完整回归 | `verified` |
| `review_ready` | PR/MR 已创建,等待审核 | `pr_opened` |
| `release_ready` | PR 已合并且完整回归通过;尚未表示发布或部署 | `regression_passed` |
| `released` | 产物已构建并发布 | `published` |
| `deployed` | 目标环境实际部署完成并通过健康检查 | `deployed` |
为兼容旧记录,`verified` 是 `develop_verified` 的别名,`validation_ready` 是
`release_ready` 的别名;新记录应使用 canonical 名称。`delivery.state` 内部里程碑
名称保持不变。
`source.kind`:`user` | `hunt` | `ack` | `pr` | `feishu`。修 ACK 缺陷时
`source.kind: ack` 且 `ref` 为缺陷任务 id。
+4
View File
@@ -21,6 +21,10 @@ scripts/append_evidence.py 追加后跑 scripts/validate_ship.py。
验收:见 spec.md 与 tasks.yaml 里该任务的 expected / verification。
授权:<本步允许的 git/PR/部署动作;未写的不要做>
测试边界:`verify` / `verify_task` 只负责开发阶段验证,执行任务相关的单元测试、
新增功能集成测试和静态检查,不执行合并后的完整 tester 回归;`regress` 只在合并后
负责完整回归,必须覆盖发现的每个 tester catalog 根和项目声明的原生测试套件。
若步骤是 `verify_task` 或 `regress` 且 skill 是 `tester`:先执行
`python3 <tester-skill>/scripts/catalog.py roots --root <project-root>`。`verify_task` 按
任务范围选择相关测试根;`regress` 必须对返回的每个独立测试根分别执行 `validate`、
+7
View File
@@ -95,6 +95,13 @@ def create_ship(
(spec_dir / "risks.yaml").write_text("[]\n", encoding="utf-8")
ensure_stage_layout(dest)
manifest = _replace(
(templates / "test-manifest.template.yaml").read_text(encoding="utf-8"),
mapping,
)
(dest / "2-develop" / "test-manifest.yaml").write_text(
manifest, encoding="utf-8"
)
issues = validate_ship_dir(dest)
if issues:
+16 -8
View File
@@ -10,7 +10,13 @@ import sys
from pathlib import Path
from typing import Any
from ship_lib import ensure_stage_layout, load_yaml, record_stage_event, stage_for_step
from ship_lib import (
STOP_AT_STATE,
ensure_stage_layout,
load_yaml,
record_stage_event,
stage_for_step,
)
SPEC_PLACEHOLDERS = (
"<用户可见的目标",
@@ -26,12 +32,6 @@ TASK_PLACEHOLDERS = (
FAULT_PLACEHOLDER = "<没有故障时删除"
DONE_TASK = frozenset({"verified", "merged", "done"})
READY_DEP = frozenset({"verified", "merged", "done"})
STOP_AT_STATE = {
"verified": "verified",
"review_ready": "pr_opened",
"validation_ready": "deployed",
"released": "deployed",
}
DELIVERY_STEPS: tuple[tuple[str, str, str], ...] = (
("verified", "verify", "tester"),
("pr_opened", "open_pr", "delivery-loop"),
@@ -173,6 +173,11 @@ def _next_delivery(
nxt = DELIVERY_STEPS[start]
state, step, skill = nxt
summary = f"派 worker 把 delivery.state 推进到 {state},加载 {skill}"
if step == "verify":
summary = (
"派 tester worker 做开发阶段验证:执行单元测试、新增功能集成测试和静态检查;"
"不执行合并后的完整 tester 回归"
)
if step == "regress":
summary = (
"派 tester worker 做合并后的完整回归:先发现父仓库和子项目的所有 tester "
@@ -323,7 +328,10 @@ def _decide_short(
actor="manager",
step="ask_user",
status="ask_user",
summary="stopAt 未配置,主会话向用户确认 verified / review_ready / validation_ready / released",
summary=(
"stopAt 未配置,主会话向用户确认 develop_verified / review_ready / "
"release_ready / released / deployed"
),
)
if state == "change_ready" and _needs_hunt(spec_md) and not _has_change_commit(current):
last = _manager(current).get("lastStep")
+12
View File
@@ -48,6 +48,18 @@ DELIVERY_STATE = frozenset(
"deployed",
}
)
# User-facing stop points. The two legacy names remain accepted so existing
# ship records continue to resolve to the new phase meanings.
STOP_AT_STATE = {
"develop_verified": "verified",
"review_ready": "pr_opened",
"release_ready": "regression_passed",
"released": "published",
"deployed": "deployed",
"verified": "verified", # legacy alias for develop_verified
"validation_ready": "regression_passed", # legacy alias for release_ready
}
STOP_AT = frozenset(STOP_AT_STATE)
SOURCE_KIND = frozenset({"user", "hunt", "ack", "pr", "feishu"})
COMMIT_ROLE = frozenset({"baseline", "merged", "head"})
+25
View File
@@ -13,6 +13,7 @@ from ship_lib import (
DELIVERY_STATE,
KIND,
SHIP_STATUS,
STOP_AT,
SOURCE_KIND,
TASK_STATUS,
check_ship_id,
@@ -77,6 +78,17 @@ def validate_ship_dir(ship_dir: Path) -> list[str]:
if not isinstance(data.get("title"), str) or not str(data.get("title")).strip():
_issue(issues, "title 不能为空")
for path, value in (
("stopAt", data.get("stopAt")),
("delivery.stopAt", (data.get("delivery") or {}).get("stopAt")
if isinstance(data.get("delivery"), dict) else None),
):
if value not in (None, "") and value not in STOP_AT:
_issue(
issues,
f"{path} 无效: {value!r}({', '.join(sorted(STOP_AT))})",
)
source = data.get("source")
if source is not None:
src = _mapping(source, "source", issues)
@@ -211,6 +223,19 @@ def _validate_tasks(path: Path, ship_id: Any, kind: Any) -> list[str]:
verification = row.get("verification")
if not isinstance(verification, list) or not verification:
_issue(issues, f"{prefix}.verification 必须是非空列表")
test_plan = row.get("testPlan")
if test_plan is not None:
plan = _mapping(test_plan, f"{prefix}.testPlan", issues)
if plan is not None:
policy = plan.get("policy")
if policy not in {"required", "not_applicable"}:
_issue(
issues,
f"{prefix}.testPlan.policy 无效: {policy!r}(required|not_applicable)",
)
manifest = plan.get("manifest")
if not isinstance(manifest, str) or not manifest.strip():
_issue(issues, f"{prefix}.testPlan.manifest 不能为空")
for i, item in enumerate(tasks):
if not isinstance(item, dict):
continue
@@ -0,0 +1,138 @@
#!/usr/bin/env python3
"""Validate and inspect a ship test manifest."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
from ship_lib import load_yaml
POLICIES = {"required", "not_applicable"}
CHANGES = {"added", "changed"}
KINDS = {"unit", "integration", "e2e", "static"}
MODES = {"schema", "verify", "regress"}
def _issue(issues: list[str], message: str) -> None:
issues.append(message)
def _mapping(value: Any, path: str, issues: list[str]) -> dict[str, Any] | None:
if not isinstance(value, dict):
_issue(issues, f"{path} 必须是 mapping")
return None
return value
def validate_manifest(
manifest_path: Path,
*,
expected_ship_id: str | None = None,
policy: str = "required",
mode: str = "schema",
) -> tuple[list[str], dict[str, Any] | None]:
issues: list[str] = []
if mode not in MODES:
return [f"mode 无效: {mode!r}"], None
try:
data = load_yaml(manifest_path)
except Exception as exc: # noqa: BLE001
return [f"无法解析 test manifest: {exc}"], None
root = _mapping(data, "test-manifest", issues)
if root is None:
return issues, None
ship_id = root.get("shipId")
if not isinstance(ship_id, str) or not ship_id.strip():
_issue(issues, "test-manifest.shipId 不能为空")
elif expected_ship_id and ship_id != expected_ship_id:
_issue(issues, f"test-manifest.shipId 不一致: {ship_id!r} != {expected_ship_id!r}")
tests = root.get("tests")
if not isinstance(tests, list):
_issue(issues, "test-manifest.tests 必须是列表")
tests = []
ids: set[str] = set()
for index, item in enumerate(tests):
prefix = f"tests[{index}]"
row = _mapping(item, prefix, issues)
if row is None:
continue
test_id = row.get("id")
if not isinstance(test_id, str) or not test_id.strip():
_issue(issues, f"{prefix}.id 不能为空")
elif test_id in ids:
_issue(issues, f"重复测试 id: {test_id}")
else:
ids.add(test_id)
for field in ("runner", "target", "selector"):
if not isinstance(row.get(field), str) or not row[field].strip():
_issue(issues, f"{prefix}.{field} 不能为空")
if row.get("kind") not in KINDS:
_issue(issues, f"{prefix}.kind 无效: {row.get('kind')!r}")
if row.get("change") not in CHANGES:
_issue(issues, f"{prefix}.change 无效: {row.get('change')!r}")
covers = row.get("covers")
if not isinstance(covers, list) or not covers or not all(
isinstance(value, str) and value.strip() for value in covers
):
_issue(issues, f"{prefix}.covers 必须是非空字符串列表")
if policy not in POLICIES:
_issue(issues, f"test policy 无效: {policy!r}")
gap = root.get("testGap")
if gap is not None:
gap_row = _mapping(gap, "testGap", issues)
if gap_row is not None:
if not isinstance(gap_row.get("reason"), str) or not gap_row["reason"].strip():
_issue(issues, "testGap.reason 不能为空")
if mode in {"verify", "regress"} and not (
isinstance(gap_row.get("acceptedBy"), str)
and gap_row["acceptedBy"].strip()
):
_issue(issues, "testGap 在 verify/regress 前必须有 acceptedBy")
if mode in {"verify", "regress"}:
if policy == "required" and not tests and gap is None:
_issue(issues, "policy=required 时必须登记至少一个测试,或填写已批准的 testGap")
if policy == "not_applicable" and tests:
_issue(issues, "policy=not_applicable 时 tests 必须为空")
if policy == "not_applicable" and gap is None:
_issue(issues, "policy=not_applicable 时必须说明 testGap.reason")
return issues, root
def main() -> int:
parser = argparse.ArgumentParser(description="校验并列出 ship 测试 manifest")
parser.add_argument("manifest", help="test-manifest.yaml 路径")
parser.add_argument("--ship-id")
parser.add_argument("--policy", default="required", choices=sorted(POLICIES))
parser.add_argument("--mode", default="schema", choices=sorted(MODES))
parser.add_argument("--json", action="store_true", dest="as_json")
args = parser.parse_args()
issues, data = validate_manifest(
Path(args.manifest),
expected_ship_id=args.ship_id,
policy=args.policy,
mode=args.mode,
)
if args.as_json:
print(json.dumps({"ok": not issues, "issues": issues, "tests": (data or {}).get("tests", [])}, ensure_ascii=False))
elif issues:
for issue in issues:
print(f"✗ {issue}", file=sys.stderr)
else:
print(f"✓ test manifest 通过: {args.manifest}")
for row in (data or {}).get("tests", []):
print(f"- {row['id']}: {row['runner']} {row['target']} {row['selector']}")
return 1 if issues else 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -7,5 +7,8 @@ tasks:
- <observable acceptance signal>
verification:
- <how to verify>
testPlan:
policy: required
manifest: 2-develop/test-manifest.yaml
status: open
stopAt: null
@@ -0,0 +1,3 @@
shipId: <ship-id>
tests: []
testGap: null
+23
View File
@@ -213,6 +213,29 @@ class BuilderCheckTests(unittest.TestCase):
self.assertIn("[FAIL] 10. 轨道工具链", text)
self.assertIn("docker: MISSING", text)
def test_ready_uses_curl_for_gitea_actions(self) -> None:
with tempfile.TemporaryDirectory() as temp:
project = Path(temp)
init_repo(project)
write_contract_makefile(project)
(project / "Dockerfile").write_text("FROM scratch\n", encoding="utf-8")
git(project, "remote", "add", "origin", "https://gitea.example.test/acme/widget.git")
code, text = run_check(
project,
"--ready",
env={"BUILD_VENUE": "ci"},
which={
"docker": "/usr/bin/docker",
"curl": "/usr/bin/curl",
"gh": None,
"glab": None,
},
)
self.assertEqual(code, 0, text)
self.assertIn("curl: found", text)
self.assertIn("Gitea API: /api/v1", text)
self.assertNotIn("glab: ", text)
def test_user_makefile_does_not_satisfy_contract(self) -> None:
with tempfile.TemporaryDirectory() as temp:
project = Path(temp)
+249
View File
@@ -0,0 +1,249 @@
from __future__ import annotations
import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT = REPO_ROOT / "skills" / "builder" / "scripts" / "wait_ci.sh"
def git(cwd: Path, *args: str) -> str:
result = subprocess.run(
["git", *args],
cwd=cwd,
text=True,
capture_output=True,
check=False,
env={
**os.environ,
"GIT_AUTHOR_NAME": "Test",
"GIT_AUTHOR_EMAIL": "test@example.com",
"GIT_COMMITTER_NAME": "Test",
"GIT_COMMITTER_EMAIL": "test@example.com",
},
)
if result.returncode != 0:
raise AssertionError(f"git {args} failed: {result.stderr}")
return result.stdout.strip()
class BuilderCiTests(unittest.TestCase):
def make_repo(self, root: Path, origin: str) -> str:
git(root, "init", "-b", "main")
git(root, "config", "user.email", "test@example.com")
git(root, "config", "user.name", "Test")
(root / "README").write_text("x\n", encoding="utf-8")
git(root, "add", "README")
git(root, "commit", "-m", "init")
git(root, "remote", "add", "origin", origin)
return git(root, "rev-parse", "HEAD")
def fake_command(self, bin_dir: Path, name: str, body: str) -> None:
command = bin_dir / name
command.write_text(f"#!/bin/sh\n{body}\n", encoding="utf-8")
command.chmod(0o755)
def run_wait(
self,
root: Path,
sha: str,
bin_dir: Path,
extra_env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
env = {
**os.environ,
"PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
"BUILDER_CI_POLL_SEC": "0",
"BUILDER_CI_APPEAR_SEC": "0",
"BUILDER_CI_FORGE": "gitea",
}
if extra_env:
env.update(extra_env)
return subprocess.run(
["bash", str(SCRIPT), "--sha", sha, "--root", str(root)],
cwd=root,
env=env,
text=True,
capture_output=True,
check=False,
)
def test_gitea_success_reads_exact_run_and_jobs(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
sha = self.make_repo(root, "https://gitea.example.test/acme/widget.git")
bin_dir = root / "bin"
bin_dir.mkdir()
self.fake_command(
bin_dir,
"curl",
"""
case "$*" in
*"/jobs?"*) cat "$BUILDER_CI_FIXTURE/jobs.json" ;;
*"/actions/runs?"*) cat "$BUILDER_CI_FIXTURE/runs.json" ;;
*) exit 9 ;;
esac
""",
)
fixture = root / "fixture"
fixture.mkdir()
(fixture / "runs.json").write_text(
json.dumps(
{
"workflow_runs": [
{
"id": 9,
"head_sha": "0" * 40,
"status": "completed",
"conclusion": "success",
"html_url": "https://gitea.example.test/wrong",
},
{
"id": 12,
"head_sha": sha,
"status": "completed",
"conclusion": "success",
"html_url": "https://gitea.example.test/run/12",
},
]
}
),
encoding="utf-8",
)
(fixture / "jobs.json").write_text(
json.dumps(
{"jobs": [{"id": 1, "name": "build", "head_sha": sha, "status": "completed", "conclusion": "success"}]}
),
encoding="utf-8",
)
result = self.run_wait(
root,
sha,
bin_dir,
{"BUILDER_CI_FIXTURE": str(fixture)},
)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("OK pipeline=https://gitea.example.test/run/12", result.stdout)
self.assertIn("id=12", result.stdout)
self.assertIn("jobs=build=success", result.stdout)
self.assertNotIn("wrong", result.stdout + result.stderr)
def test_gitea_failed_job_is_failure(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
sha = self.make_repo(root, "git@gitea.example.test:acme/widget.git")
bin_dir = root / "bin"
bin_dir.mkdir()
fixture = root / "fixture"
fixture.mkdir()
self.fake_command(
bin_dir,
"curl",
"""
case "$*" in
*"/jobs?"*) cat "$BUILDER_CI_FIXTURE/jobs.json" ;;
*"/actions/runs?"*) cat "$BUILDER_CI_FIXTURE/runs.json" ;;
*) exit 9 ;;
esac
""",
)
(fixture / "runs.json").write_text(
json.dumps(
{
"workflow_runs": [
{
"id": 21,
"head_sha": sha,
"status": "completed",
"conclusion": "failure",
"html_url": "https://gitea.example.test/run/21",
}
]
}
),
encoding="utf-8",
)
(fixture / "jobs.json").write_text(
json.dumps(
{"jobs": [{"id": 2, "name": "test", "head_sha": sha, "status": "completed", "conclusion": "failure"}]}
),
encoding="utf-8",
)
result = self.run_wait(
root,
sha,
bin_dir,
{"BUILDER_CI_FIXTURE": str(fixture)},
)
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
self.assertIn("reason=job_failed", result.stderr)
self.assertIn("test=failure", result.stderr)
def test_gitea_does_not_treat_other_sha_as_pipeline(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
sha = self.make_repo(root, "https://gitea.example.test/acme/widget.git")
bin_dir = root / "bin"
bin_dir.mkdir()
fixture = root / "fixture"
fixture.mkdir()
self.fake_command(
bin_dir,
"curl",
'cat "$BUILDER_CI_FIXTURE/runs.json"',
)
(fixture / "runs.json").write_text(
json.dumps(
{
"workflow_runs": [
{
"id": 99,
"head_sha": "f" * 40,
"status": "completed",
"conclusion": "success",
}
]
}
),
encoding="utf-8",
)
result = self.run_wait(
root,
sha,
bin_dir,
{"BUILDER_CI_FIXTURE": str(fixture)},
)
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
self.assertIn("reason=no_pipeline", result.stderr)
def test_github_and_gitlab_branches_still_use_their_cli(self) -> None:
for forge, command, output in (
("github", "gh", "printf '%s\\n' '[]'"),
("gitlab", "glab", "printf '%s\\n' '[]'"),
):
with self.subTest(forge=forge), tempfile.TemporaryDirectory() as temp:
root = Path(temp)
sha = self.make_repo(root, f"https://{forge}.example.test/acme/widget.git")
bin_dir = root / "bin"
bin_dir.mkdir()
self.fake_command(bin_dir, command, output)
result = self.run_wait(
root,
sha,
bin_dir,
{
"BUILDER_CI_FORGE": forge,
"BUILDER_CI_APPEAR_SEC": "0",
},
)
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
self.assertIn("reason=no_pipeline", result.stderr)
if __name__ == "__main__":
unittest.main()
+58
View File
@@ -244,6 +244,64 @@ def _fill_task(ship_dir: Path) -> None:
class NextStepTests(unittest.TestCase):
def _prepare_delivery_dispatch(
self, root: Path, ship_id: str, stop_at: str, state: str
) -> None:
ship_dir = _create(root, ship_id, "short", stop_at=stop_at)
(ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8")
_fill_task(ship_dir)
_patch_current(ship_dir, delivery_state=state)
data = safe_load((ship_dir / "current.yaml").read_text(encoding="utf-8"))
data["evidence"]["commits"] = [{"sha": "abc1234deadbeef", "role": "head"}]
(ship_dir / "current.yaml").write_text(safe_dump(data), encoding="utf-8")
def test_stop_at_mappings_cover_each_delivery_boundary(self) -> None:
cases = (
("develop_verified", "change_ready", "verify", "verified"),
("review_ready", "verified", "open_pr", "pr_opened"),
("release_ready", "merged", "regress", "regression_passed"),
("released", "artifact_built", "publish", "published"),
("deployed", "published", "deploy", "deployed"),
)
for stop_at, state, step, target in cases:
with self.subTest(stop_at=stop_at):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
ship_id = f"ship-{stop_at.replace('_', '-')}"
self._prepare_delivery_dispatch(root, ship_id, stop_at, state)
doc = _next(root, ship_id, "short")
self.assertEqual(doc["step"], step)
self.assertEqual(doc["targetState"], target)
def test_legacy_stop_at_names_resolve_to_new_boundaries(self) -> None:
cases = (
("verified", "change_ready", "verify", "verified"),
("validation_ready", "merged", "regress", "regression_passed"),
)
for stop_at, state, step, target in cases:
with self.subTest(stop_at=stop_at):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
ship_id = f"legacy-{stop_at.replace('_', '-')}"
self._prepare_delivery_dispatch(root, ship_id, stop_at, state)
doc = _next(root, ship_id, "short")
self.assertEqual(doc["step"], step)
self.assertEqual(doc["targetState"], target)
def test_invalid_stop_at_fails_ship_validation(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
result = _run(
CREATE,
"--project-root", str(root),
"--ship-id", "invalid-stop-at",
"--kind", "short",
"--title", "无效停止点",
"--stop-at", "not_a_stop_point",
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("stopAt 无效", result.stderr)
def test_new_short_ship_writes_spec(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)