forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
27.7 KiB
1//! Turns pushes, tags, pull requests and manual requests into CI runs.
2//!
3//! Every entry point reads `.forge/ci.toml` at the commit it runs (64 KiB cap). No file means no
4//! run. A file that does not parse gives an Error run with the message in `config_error`, plus a
5//! `ci/config` status. A valid file gives one run whose jobs are created in dependency order.
6
7use forge_ci::config::{CiConfig, Event, parse_config};
8use forge_core::authz::{Action, authorize};
9use forge_core::ci::{CiRun, NewJob};
10use forge_core::store::commit_status::StatusState;
11use forge_core::{PullRequest, Repo};
12use forge_git::browse::{read_blob, resolve_commit};
13
14use super::{publish, sync_run_statuses};
15use crate::internal::is_zero_oid;
16use crate::state::AppState;
17
18/// The workflow file, relative to the repo root.
19pub const CONFIG_PATH: &str = ".forge/ci.toml";
20/// Largest workflow file read, in bytes.
21const MAX_CONFIG_BYTES: usize = 64 * 1024;
22
23/// A job ready to store: name, `JobSpec` as JSON, and its needs.
24type PlannedJob = (String, String, Vec<String>);
25
26/// What `.forge/ci.toml` held at one commit.
27enum Loaded {
28 Missing,
29 Invalid(String),
30 Valid(CiConfig),
31}
32
33/// What a request should do with the loaded config.
34enum Plan {
35 /// No workflow file at this commit.
36 Missing,
37 /// The file is valid but this event does not trigger it.
38 NotTriggered,
39 /// Create a run with these jobs, or an Error run when `config_error` is set.
40 Run {
41 jobs: Vec<PlannedJob>,
42 config_error: Option<String>,
43 },
44}
45
46/// The inputs of one run to create.
47struct Request<'a> {
48 repo: &'a Repo,
49 trigger: &'static str,
50 git_ref: &'a str,
51 commit: &'a str,
52 pull_number: Option<i64>,
53 actor_id: &'a str,
54 needs_approval: bool,
55}
56
57/// A branch or tag update. Called from `post_update::after_ref_update` for every update of
58/// `refs/heads/*` or `refs/tags/*`, including pull mirror updates (with an empty actor). Deletions
59/// are ignored. Failures are logged; they never fail the push.
60pub async fn on_ref_update(
61 state: &AppState,
62 repo: &Repo,
63 actor_id: &str,
64 refname: &str,
65 new_oid: &str,
66) {
67 if !state.config.ci.enabled || is_zero_oid(new_oid) {
68 return;
69 }
70 let (trigger, event) = if let Some(branch) = refname.strip_prefix("refs/heads/") {
71 ("push", Event::Push { branch })
72 } else if let Some(tag) = refname.strip_prefix("refs/tags/") {
73 ("tag", Event::Tag { tag })
74 } else {
75 return;
76 };
77 let result = async {
78 let loaded = load(state, &repo.id, new_oid).await;
79 if let Plan::Run { jobs, config_error } = plan_for(loaded, &event)? {
80 create(
81 state,
82 Request {
83 repo,
84 trigger,
85 git_ref: refname,
86 commit: new_oid,
87 pull_number: None,
88 actor_id,
89 needs_approval: false,
90 },
91 jobs,
92 config_error.as_deref(),
93 )
94 .await?;
95 }
96 forge_core::Result::Ok(())
97 }
98 .await;
99 if let Err(e) = result {
100 tracing::error!(error = %e, repo_id = %repo.id, refname, "ci: scheduling a push failed");
101 }
102}
103
104/// A pull request was opened, synchronized (its head moved) or reopened. The config is read at
105/// the head commit. A fork head, or an author without Write on the base repo, creates the run in
106/// WaitingApproval: its jobs wait until someone approves it.
107pub async fn on_pull_event(state: &AppState, repo: &Repo, pull: &PullRequest, actor_id: &str) {
108 if !state.config.ci.enabled {
109 return;
110 }
111 if let Err(e) = pull_run(state, repo, pull, actor_id).await {
112 tracing::error!(error = %e, repo_id = %repo.id, number = pull.issue.number, "ci: scheduling a pull request failed");
113 }
114}
115
116async fn pull_run(
117 state: &AppState,
118 repo: &Repo,
119 pull: &PullRequest,
120 actor_id: &str,
121) -> forge_core::Result<()> {
122 let loaded = load(state, &repo.id, &pull.head_oid).await;
123 let Plan::Run { jobs, config_error } = plan_for(loaded, &Event::PullRequest)? else {
124 return Ok(());
125 };
126 let needs_approval = needs_approval(state, repo, pull).await?;
127 let git_ref = format!("refs/pull/{}/head", pull.issue.number);
128 create(
129 state,
130 Request {
131 repo,
132 trigger: "pull_request",
133 git_ref: &git_ref,
134 commit: &pull.head_oid,
135 pull_number: Some(pull.issue.number),
136 actor_id,
137 needs_approval,
138 },
139 jobs,
140 config_error.as_deref(),
141 )
142 .await?;
143 Ok(())
144}
145
146/// Starts a run for `branch` by hand. Fails when the branch does not exist, when there is no
147/// workflow file, or when the workflow does not allow manual runs. A file that does not parse
148/// gives an Error run, as a push does.
149pub async fn manual_run(
150 state: &AppState,
151 repo: &Repo,
152 branch: &str,
153 actor_id: &str,
154) -> forge_core::Result<CiRun> {
155 if !state.config.ci.enabled {
156 return Err(forge_core::Error::Invalid(
157 "CI is disabled on this instance".into(),
158 ));
159 }
160 let git_ref = format!("refs/heads/{branch}");
161 let repo_dir = forge_git::repo::repo_path(&state.git.data_dir, &repo.id)
162 .map_err(|e| forge_core::Error::Git(e.to_string()))?;
163 // `resolve_commit` takes the short branch name and looks under refs/heads/ first.
164 let short = branch.to_owned();
165 let Some(commit) = blocking(move || resolve_commit(&repo_dir, &short)).await? else {
166 return Err(forge_core::Error::NotFound);
167 };
168 let loaded = load(state, &repo.id, &commit).await;
169 match plan_for(loaded, &Event::Manual)? {
170 Plan::Run { jobs, config_error } => {
171 create(
172 state,
173 Request {
174 repo,
175 trigger: "manual",
176 git_ref: &git_ref,
177 commit: &commit,
178 pull_number: None,
179 actor_id,
180 needs_approval: false,
181 },
182 jobs,
183 config_error.as_deref(),
184 )
185 .await
186 }
187 Plan::NotTriggered => Err(forge_core::Error::Invalid(
188 "manual runs are not enabled in .forge/ci.toml".into(),
189 )),
190 Plan::Missing => Err(forge_core::Error::Invalid(
191 "this branch has no .forge/ci.toml".into(),
192 )),
193 }
194}
195
196/// Creates the run, sets the statuses of its jobs (and `ci/config` for an error run), and wakes
197/// the runner claims when a job is queued.
198async fn create(
199 state: &AppState,
200 req: Request<'_>,
201 jobs: Vec<PlannedJob>,
202 config_error: Option<&str>,
203) -> forge_core::Result<CiRun> {
204 let new_jobs: Vec<NewJob<'_>> = jobs
205 .iter()
206 .map(|(name, spec_json, needs)| NewJob {
207 name,
208 spec_json,
209 needs: needs.as_slice(),
210 })
211 .collect();
212 let run = state
213 .store
214 .create_run(
215 &req.repo.id,
216 req.trigger,
217 req.git_ref,
218 req.commit,
219 req.pull_number,
220 req.actor_id,
221 req.needs_approval,
222 &new_jobs,
223 config_error,
224 )
225 .await?;
226 if let Some(message) = config_error {
227 let url = format!(
228 "{}/{}/{}/ci/runs/{}",
229 state.config.base_url.trim_end_matches('/'),
230 req.repo.owner_name,
231 req.repo.name,
232 run.number
233 );
234 publish(
235 state,
236 req.repo,
237 &run,
238 "ci/config",
239 StatusState::Error,
240 message,
241 &url,
242 )
243 .await;
244 }
245 if sync_run_statuses(state, &run).await? {
246 state.ci_waker.notify_waiters();
247 }
248 Ok(run)
249}
250
251/// Decides what a request does, given the loaded config and the event.
252fn plan_for(loaded: Loaded, event: &Event<'_>) -> forge_core::Result<Plan> {
253 match loaded {
254 Loaded::Missing => Ok(Plan::Missing),
255 Loaded::Invalid(message) => Ok(Plan::Run {
256 jobs: Vec::new(),
257 config_error: Some(message),
258 }),
259 Loaded::Valid(cfg) => {
260 if !cfg.triggered_by(event) {
261 return Ok(Plan::NotTriggered);
262 }
263 Ok(Plan::Run {
264 jobs: plan_jobs(&cfg)?,
265 config_error: None,
266 })
267 }
268 }
269}
270
271/// The jobs of a valid config in dependency order, each with its spec as JSON.
272fn plan_jobs(cfg: &CiConfig) -> forge_core::Result<Vec<PlannedJob>> {
273 let mut jobs = Vec::new();
274 for name in cfg.job_order() {
275 let Some(spec) = cfg.jobs.get(name) else {
276 continue;
277 };
278 let spec_json =
279 serde_json::to_string(spec).map_err(|e| forge_core::Error::Invalid(e.to_string()))?;
280 jobs.push((name.to_owned(), spec_json, spec.needs.clone()));
281 }
282 Ok(jobs)
283}
284
285/// Reads and parses `.forge/ci.toml` at `commit` from the repo `repo_id`.
286async fn load(state: &AppState, repo_id: &str, commit: &str) -> Loaded {
287 let repo_dir = match forge_git::repo::repo_path(&state.git.data_dir, repo_id) {
288 Ok(path) => path,
289 Err(e) => {
290 tracing::error!(error = %e, repo_id, "ci: bad repo path");
291 return Loaded::Invalid(format!("could not read {CONFIG_PATH}"));
292 }
293 };
294 let (dir, rev) = (repo_dir, commit.to_owned());
295 let blob = match blocking(move || read_blob(&dir, &rev, CONFIG_PATH, MAX_CONFIG_BYTES)).await {
296 Ok(blob) => blob,
297 Err(e) => {
298 tracing::error!(error = %e, repo_id, commit, "ci: reading the workflow failed");
299 return Loaded::Invalid(format!("could not read {CONFIG_PATH}"));
300 }
301 };
302 let Some(blob) = blob else {
303 return Loaded::Missing;
304 };
305 if blob.truncated {
306 return Loaded::Invalid(format!("{CONFIG_PATH} is larger than 64 KiB"));
307 }
308 if blob.is_binary {
309 return Loaded::Invalid(format!("{CONFIG_PATH} is not text"));
310 }
311 let Ok(text) = String::from_utf8(blob.data) else {
312 return Loaded::Invalid(format!("{CONFIG_PATH} is not valid UTF-8"));
313 };
314 match parse_config(&text) {
315 Ok(cfg) => Loaded::Valid(cfg),
316 Err(e) => Loaded::Invalid(e.to_string()),
317 }
318}
319
320/// Whether a pull request needs approval before its jobs run: a fork head, or an author who lacks
321/// Write on the base repo (or no longer exists).
322async fn needs_approval(
323 state: &AppState,
324 repo: &Repo,
325 pull: &PullRequest,
326) -> forge_core::Result<bool> {
327 if pull.head_repo_id.as_deref() != Some(repo.id.as_str()) {
328 return Ok(true);
329 }
330 let Some(author) = state.store.user_by_id(&pull.issue.author_id).await? else {
331 return Ok(true);
332 };
333 if author.is_org() {
334 return Ok(true);
335 }
336 let role = state.store.collaborator_role(&repo.id, &author.id).await?;
337 let org_role = state
338 .store
339 .org_repo_role(&repo.owner_id, &author.id, &repo.id)
340 .await?;
341 Ok(authorize(Some(&author), repo, role, org_role, Action::Write).is_err())
342}
343
344/// Runs blocking git work off the async runtime, mapping git and join errors to `forge_core`.
345async fn blocking<T, F>(work: F) -> forge_core::Result<T>
346where
347 F: FnOnce() -> forge_git::Result<T> + Send + 'static,
348 T: Send + 'static,
349{
350 match tokio::task::spawn_blocking(work).await {
351 Ok(Ok(value)) => Ok(value),
352 Ok(Err(e)) => Err(forge_core::Error::Git(e.to_string())),
353 Err(e) => Err(forge_core::Error::Git(format!("git task failed: {e}"))),
354 }
355}
356
357#[cfg(test)]
358mod tests {
359 use std::path::{Path, PathBuf};
360 use std::sync::Arc;
361 use std::time::Duration;
362
363 use forge_core::ci::{CiJob, CiRunner, JobState, RunState};
364 use forge_core::store::commit_status::{CommitStatus, StatusState};
365 use forge_core::time::stamp_after;
366 use forge_core::{User, Visibility};
367 use forge_git::repo::{init_bare, repo_path};
368 use tempfile::TempDir;
369
370 use super::*;
371 use crate::ci::log_path;
372 use crate::ci::reaper::{purge_once, reap_before};
373 use crate::post_update::after_ref_update;
374 use crate::test_support::test_state;
375
376 const ZERO: &str = "0000000000000000000000000000000000000000";
377
378 const CONFIG: &str = r#"
379[on]
380push = ["main"]
381pull_request = true
382
383[jobs.build]
384image = "alpine:3"
385run = "make"
386
387[jobs.test]
388image = "alpine:3"
389run = "make test"
390needs = ["build"]
391"#;
392
393 const CONFIG_GLOBS: &str = r#"
394[on]
395push = ["release/*"]
396
397[jobs.build]
398image = "alpine:3"
399run = "make"
400"#;
401
402 const CONFIG_TAGS: &str = r#"
403[on]
404push = []
405tags = ["v*"]
406
407[jobs.build]
408image = "alpine:3"
409run = "make"
410"#;
411
412 const CONFIG_ONE_JOB: &str = r#"
413[on]
414push = ["main"]
415
416[jobs.build]
417image = "alpine:3"
418run = "make"
419"#;
420
421 const CONFIG_NO_MANUAL: &str = r#"
422[on]
423push = ["main"]
424manual = false
425
426[jobs.build]
427image = "alpine:3"
428run = "make"
429"#;
430
431 fn git(dir: &Path, args: &[&str]) -> String {
432 let out = std::process::Command::new("git")
433 .arg("-C")
434 .arg(dir)
435 .args(args)
436 .env("GIT_AUTHOR_NAME", "test")
437 .env("GIT_AUTHOR_EMAIL", "test@example.com")
438 .env("GIT_COMMITTER_NAME", "test")
439 .env("GIT_COMMITTER_EMAIL", "test@example.com")
440 .output()
441 .expect("git runs");
442 assert!(
443 out.status.success(),
444 "git {args:?} failed: {}",
445 String::from_utf8_lossy(&out.stderr)
446 );
447 String::from_utf8_lossy(&out.stdout).trim().to_string()
448 }
449
450 struct Fixture {
451 _tmp: TempDir,
452 state: AppState,
453 repo: Repo,
454 owner: User,
455 bare: PathBuf,
456 work: PathBuf,
457 }
458
459 async fn fixture() -> Fixture {
460 let tmp = tempfile::tempdir().expect("tempdir");
461 let state = test_state(tmp.path()).await;
462 let owner = state
463 .store
464 .create_user("alice", None, "unused", false)
465 .await
466 .expect("user");
467 let repo = state
468 .store
469 .create_repo(&owner.id, "proj", "", Visibility::Private, "main")
470 .await
471 .expect("repo");
472 let bare = repo_path(&state.git.data_dir, &repo.id).expect("repo path");
473 init_bare(&state.git, &bare, "main")
474 .await
475 .expect("init bare");
476 let work = tmp.path().join("work");
477 git(tmp.path(), &["init", "-b", "main", "work"]);
478 Fixture {
479 _tmp: tmp,
480 state,
481 repo,
482 owner,
483 bare,
484 work,
485 }
486 }
487
488 /// Commits `files` in the work tree and returns the new commit.
489 fn commit(work: &Path, files: &[(&str, &str)]) -> String {
490 for (path, text) in files {
491 let full = work.join(path);
492 if let Some(parent) = full.parent() {
493 std::fs::create_dir_all(parent).expect("dirs");
494 }
495 std::fs::write(&full, text).expect("write");
496 git(work, &["add", path]);
497 }
498 git(work, &["commit", "--allow-empty", "-m", "change"]);
499 git(work, &["rev-parse", "HEAD"])
500 }
501
502 /// Makes `oid` the tip of `refname` in the bare repo. Fetch runs no hooks, unlike a push.
503 fn land(fx: &Fixture, oid: &str, refname: &str) {
504 let work = fx.work.to_str().expect("utf8 path");
505 git(
506 &fx.bare,
507 &["fetch", "--no-tags", work, &format!("+{oid}:{refname}")],
508 );
509 }
510
511 /// Commits `files`, lands them on `refname`, and runs the post-update hook for the update.
512 async fn push(fx: &Fixture, files: &[(&str, &str)], refname: &str) -> String {
513 let oid = commit(&fx.work, files);
514 land(fx, &oid, refname);
515 after_ref_update(&fx.state, &fx.repo, &fx.owner.id, refname, ZERO, &oid)
516 .await
517 .expect("post update");
518 oid
519 }
520
521 fn status<'a>(statuses: &'a [CommitStatus], context: &str) -> Option<&'a CommitStatus> {
522 statuses.iter().find(|s| s.context == context)
523 }
524
525 fn state_of(jobs: &[CiJob], name: &str) -> Option<JobState> {
526 jobs.iter().find(|j| j.name == name).map(|j| j.state)
527 }
528
529 async fn runs_total(fx: &Fixture) -> u64 {
530 fx.state
531 .store
532 .list_runs(&fx.repo.id, 1, 50)
533 .await
534 .expect("list runs")
535 .1
536 }
537
538 async fn runner(fx: &Fixture) -> CiRunner {
539 let token = fx
540 .state
541 .store
542 .create_registration_token(&fx.owner.id, 24)
543 .await
544 .expect("token");
545 let (runner, _secret) = fx
546 .state
547 .store
548 .register_runner(&token, "runner-1", &[])
549 .await
550 .expect("register");
551 runner
552 }
553
554 #[tokio::test]
555 async fn push_with_config_creates_run_and_pending_statuses() {
556 let fx = fixture().await;
557 let oid = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/main").await;
558
559 let run = fx
560 .state
561 .store
562 .run_by_number(&fx.repo.id, 1)
563 .await
564 .expect("query")
565 .expect("run exists");
566 assert_eq!(run.trigger, "push");
567 assert_eq!(run.git_ref, "refs/heads/main");
568 assert_eq!(run.commit_oid, oid);
569 assert!(run.config_error.is_none());
570
571 let jobs = fx.state.store.jobs_for_run(&run.id).await.expect("jobs");
572 assert_eq!(state_of(&jobs, "build"), Some(JobState::Queued));
573 assert_eq!(state_of(&jobs, "test"), Some(JobState::Blocked));
574
575 let statuses = fx
576 .state
577 .store
578 .statuses_for(&fx.repo.id, &oid)
579 .await
580 .expect("statuses");
581 let build = status(&statuses, "ci/build").expect("ci/build status");
582 assert_eq!(build.state, StatusState::Pending);
583 let url = build.target_url.clone().unwrap_or_default();
584 let job = jobs.iter().find(|j| j.name == "build").expect("build job");
585 assert_eq!(
586 url,
587 format!("https://forge.test/alice/proj/ci/jobs/{}", job.id)
588 );
589 let test = status(&statuses, "ci/test").expect("ci/test status");
590 assert_eq!(test.state, StatusState::Pending);
591 }
592
593 #[tokio::test]
594 async fn no_file_means_no_run() {
595 let fx = fixture().await;
596 let oid = push(&fx, &[("README.md", "hello")], "refs/heads/main").await;
597 assert_eq!(runs_total(&fx).await, 0);
598 let statuses = fx
599 .state
600 .store
601 .statuses_for(&fx.repo.id, &oid)
602 .await
603 .expect("statuses");
604 assert!(statuses.is_empty());
605 }
606
607 #[tokio::test]
608 async fn bad_file_gives_error_run_and_config_status() {
609 let fx = fixture().await;
610 let oid = push(
611 &fx,
612 &[(CONFIG_PATH, "[jobs.build\nimage = ")],
613 "refs/heads/main",
614 )
615 .await;
616 let run = fx
617 .state
618 .store
619 .run_by_number(&fx.repo.id, 1)
620 .await
621 .expect("query")
622 .expect("run exists");
623 assert_eq!(run.state, RunState::Error);
624 assert!(run.config_error.is_some());
625 let jobs = fx.state.store.jobs_for_run(&run.id).await.expect("jobs");
626 assert!(jobs.is_empty());
627
628 let statuses = fx
629 .state
630 .store
631 .statuses_for(&fx.repo.id, &oid)
632 .await
633 .expect("statuses");
634 let config = status(&statuses, "ci/config").expect("ci/config status");
635 assert_eq!(config.state, StatusState::Error);
636 assert!(!config.description.as_deref().unwrap_or("").is_empty());
637 }
638
639 #[tokio::test]
640 async fn branch_globs_select_runs() {
641 let fx = fixture().await;
642 push(&fx, &[(CONFIG_PATH, CONFIG_GLOBS)], "refs/heads/main").await;
643 assert_eq!(runs_total(&fx).await, 0, "main is not in the globs");
644
645 push(
646 &fx,
647 &[(CONFIG_PATH, CONFIG_GLOBS)],
648 "refs/heads/release/1.0",
649 )
650 .await;
651 assert_eq!(runs_total(&fx).await, 1);
652 let run = fx
653 .state
654 .store
655 .run_by_number(&fx.repo.id, 1)
656 .await
657 .expect("query")
658 .expect("run");
659 assert_eq!(run.git_ref, "refs/heads/release/1.0");
660 }
661
662 #[tokio::test]
663 async fn tag_push_creates_tag_run() {
664 let fx = fixture().await;
665 push(&fx, &[(CONFIG_PATH, CONFIG_TAGS)], "refs/tags/nightly").await;
666 assert_eq!(runs_total(&fx).await, 0, "nightly does not match v*");
667
668 push(&fx, &[(CONFIG_PATH, CONFIG_TAGS)], "refs/tags/v1.0").await;
669 assert_eq!(runs_total(&fx).await, 1);
670 let run = fx
671 .state
672 .store
673 .run_by_number(&fx.repo.id, 1)
674 .await
675 .expect("query")
676 .expect("run");
677 assert_eq!(run.trigger, "tag");
678 assert_eq!(run.git_ref, "refs/tags/v1.0");
679 }
680
681 /// Opens a pull request `author` -> `branch` whose head repo is `head_repo_id`, and runs the
682 /// PR hook for it.
683 async fn open_pull(
684 fx: &Fixture,
685 author_id: &str,
686 head_repo_id: &str,
687 branch: &str,
688 head_oid: &str,
689 ) -> PullRequest {
690 let pr = fx
691 .state
692 .store
693 .create_pull(
694 &fx.repo.id,
695 author_id,
696 "Change",
697 "",
698 "main",
699 head_repo_id,
700 branch,
701 head_oid,
702 head_oid,
703 )
704 .await
705 .expect("create pull");
706 on_pull_event(&fx.state, &fx.repo, &pr, author_id).await;
707 pr
708 }
709
710 #[tokio::test]
711 async fn fork_pull_request_waits_for_approval() {
712 let fx = fixture().await;
713 let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/feature").await;
714 let bob = fx
715 .state
716 .store
717 .create_user("bob", None, "unused", false)
718 .await
719 .expect("user");
720 let fork = fx
721 .state
722 .store
723 .create_repo(&bob.id, "proj", "", Visibility::Public, "main")
724 .await
725 .expect("fork");
726 let pr = open_pull(&fx, &fx.owner.id, &fork.id, "feature", &head).await;
727
728 let run = fx
729 .state
730 .store
731 .run_by_number(&fx.repo.id, 1)
732 .await
733 .expect("query")
734 .expect("run");
735 assert_eq!(run.trigger, "pull_request");
736 assert_eq!(run.state, RunState::WaitingApproval);
737 assert_eq!(run.pull_number, Some(pr.issue.number));
738
739 let statuses = fx
740 .state
741 .store
742 .statuses_for(&fx.repo.id, &head)
743 .await
744 .expect("statuses");
745 let build = status(&statuses, "ci/build").expect("ci/build");
746 assert_eq!(build.state, StatusState::Pending);
747 assert_eq!(build.description.as_deref(), Some("Waiting for approval"));
748 }
749
750 #[tokio::test]
751 async fn non_writer_pull_request_waits_for_approval() {
752 let fx = fixture().await;
753 let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/feature").await;
754 let bob = fx
755 .state
756 .store
757 .create_user("bob", None, "unused", false)
758 .await
759 .expect("user");
760 open_pull(&fx, &bob.id, &fx.repo.id, "feature", &head).await;
761 let run = fx
762 .state
763 .store
764 .run_by_number(&fx.repo.id, 1)
765 .await
766 .expect("query")
767 .expect("run");
768 assert_eq!(run.state, RunState::WaitingApproval);
769 }
770
771 #[tokio::test]
772 async fn writer_pull_request_runs_at_once() {
773 let fx = fixture().await;
774 let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/feature").await;
775 open_pull(&fx, &fx.owner.id, &fx.repo.id, "feature", &head).await;
776 let run = fx
777 .state
778 .store
779 .run_by_number(&fx.repo.id, 1)
780 .await
781 .expect("query")
782 .expect("run");
783 assert_ne!(run.state, RunState::WaitingApproval);
784 }
785
786 #[tokio::test]
787 async fn manual_run_uses_branch_head() {
788 let fx = fixture().await;
789 let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/main").await;
790 let before = runs_total(&fx).await;
791 let run = manual_run(&fx.state, &fx.repo, "main", &fx.owner.id)
792 .await
793 .expect("manual run");
794 assert_eq!(run.trigger, "manual");
795 assert_eq!(run.git_ref, "refs/heads/main");
796 assert_eq!(run.commit_oid, head);
797 assert_eq!(runs_total(&fx).await, before + 1);
798 }
799
800 #[tokio::test]
801 async fn manual_run_refuses_missing_branch_and_disabled_manual() {
802 let fx = fixture().await;
803 push(&fx, &[(CONFIG_PATH, CONFIG_NO_MANUAL)], "refs/heads/main").await;
804 let before = runs_total(&fx).await;
805 assert!(
806 manual_run(&fx.state, &fx.repo, "nope", &fx.owner.id)
807 .await
808 .is_err()
809 );
810 let refused = manual_run(&fx.state, &fx.repo, "main", &fx.owner.id).await;
811 assert!(matches!(refused, Err(forge_core::Error::Invalid(_))));
812 assert_eq!(runs_total(&fx).await, before);
813 }
814
815 #[tokio::test]
816 async fn reaper_marks_lost_jobs_and_syncs_statuses() {
817 let fx = fixture().await;
818 let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/main").await;
819 let runner = runner(&fx).await;
820 let (job, _run, _token) = fx
821 .state
822 .store
823 .claim_job(&runner)
824 .await
825 .expect("claim")
826 .expect("a queued job");
827 assert_eq!(job.name, "build");
828
829 // A cutoff in the future makes every running job look silent.
830 let reaped = reap_before(&fx.state, &stamp_after(Duration::from_secs(3600))).await;
831 assert_eq!(reaped, 1);
832
833 let job = fx
834 .state
835 .store
836 .job_by_id(&job.id)
837 .await
838 .expect("query")
839 .expect("job");
840 assert_eq!(job.state, JobState::Error);
841 assert_eq!(job.message.as_deref(), Some("runner lost"));
842
843 let statuses = fx
844 .state
845 .store
846 .statuses_for(&fx.repo.id, &head)
847 .await
848 .expect("statuses");
849 let build = status(&statuses, "ci/build").expect("ci/build");
850 assert_eq!(build.state, StatusState::Error);
851 }
852
853 #[tokio::test]
854 async fn purge_deletes_log_files() {
855 let fx = fixture().await;
856 push(&fx, &[(CONFIG_PATH, CONFIG_ONE_JOB)], "refs/heads/main").await;
857 let runner = runner(&fx).await;
858 let (job, _run, _token) = fx
859 .state
860 .store
861 .claim_job(&runner)
862 .await
863 .expect("claim")
864 .expect("a queued job");
865 // Purge only takes jobs that wrote a log and whose run has finished.
866 fx.state
867 .store
868 .add_log_bytes(&job.id, 12)
869 .await
870 .expect("log bytes");
871 fx.state
872 .store
873 .finish_job(&job.id, JobState::Success, Some(0), None)
874 .await
875 .expect("finish");
876
877 let path = log_path(&fx.state.config.data_dir, &job.id);
878 std::fs::create_dir_all(path.parent().expect("parent")).expect("log dir");
879 std::fs::write(&path, "build output\n").expect("write log");
880
881 // Zero days of retention: everything finished before now is purged.
882 let mut cfg = (*fx.state.config).clone();
883 cfg.ci.log_retention_days = 0;
884 let mut state = fx.state.clone();
885 state.config = Arc::new(cfg);
886 tokio::time::sleep(Duration::from_millis(1100)).await;
887
888 let (logs, _artifacts) = purge_once(&state).await;
889 assert_eq!(logs, 1);
890 assert!(!path.exists(), "log file is deleted");
891 }
892}