forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
31.6 KiB
1//! Route tests for the CI run and job pages. The fixture: alice owns the private repo `proj`, bob
2//! has Write, dave has Read, and carol is outside the repo.
3
4use std::io::Write;
5use std::path::{Path, PathBuf};
6use std::sync::Arc;
7use std::time::Duration;
8
9use forge_core::ci::{CiJob, CiRun, JobState, NewJob, RunState};
10use forge_core::{Role, Store, Visibility};
11use forge_git::repo::{init_bare, repo_path};
12use topcoat::router::request::Request;
13use topcoat::router::{Body, HeaderMap, Router, to_bytes};
14
15use crate::auth::SESSION_COOKIE;
16use crate::blobs::BlobStore;
17use crate::ci::{CiLive, artifact_key, log_path};
18use crate::test_support::{app_router, create_session, test_state};
19
20const CSRF: &str = "ci-test-csrf";
21const SHA: &str = "0123456789abcdef0123456789abcdef01234567";
22const SPEC: &str = r#"{"runs_on":[]}"#;
23const RUN_BASE: &str = "/alice/proj/ci";
24
25struct Fixture {
26 store: Store,
27 app: Router,
28 live: Arc<CiLive>,
29 repo: forge_core::Repo,
30 alice: String,
31 data_dir: PathBuf,
32 blobs: BlobStore,
33}
34
35async fn fixture(dir: &Path) -> Fixture {
36 let state = test_state(dir).await;
37 let store = state.store.clone();
38 let alice = store
39 .create_user("alice", None, "unused", false)
40 .await
41 .expect("alice");
42 let bob = store
43 .create_user("bob", None, "unused", false)
44 .await
45 .expect("bob");
46 let dave = store
47 .create_user("dave", None, "unused", false)
48 .await
49 .expect("dave");
50 let carol = store
51 .create_user("carol", None, "unused", false)
52 .await
53 .expect("carol");
54 let repo = store
55 .create_repo(&alice.id, "proj", "", Visibility::Private, "main")
56 .await
57 .expect("repo");
58 let bare = repo_path(&state.git.data_dir, &repo.id).expect("repo path");
59 init_bare(&state.git, &bare, "main")
60 .await
61 .expect("init bare");
62 store
63 .set_collaborator(&repo.id, &bob.id, Role::Write)
64 .await
65 .expect("bob writes");
66 store
67 .set_collaborator(&repo.id, &dave.id, Role::Read)
68 .await
69 .expect("dave reads");
70 create_session(&store, &alice.id, "alice-token", CSRF).await;
71 create_session(&store, &bob.id, "bob-token", CSRF).await;
72 create_session(&store, &dave.id, "dave-token", CSRF).await;
73 create_session(&store, &carol.id, "carol-token", CSRF).await;
74 let data_dir = state.config.data_dir.clone();
75 let blobs = state.blobs.clone();
76 let live = Arc::clone(&state.ci_live);
77 Fixture {
78 store,
79 app: app_router(state),
80 live,
81 repo,
82 alice: alice.id,
83 data_dir,
84 blobs,
85 }
86}
87
88fn cookie_for(token: &str) -> String {
89 format!("__Host-{SESSION_COOKIE}={token}")
90}
91
92/// A response: status, headers and the whole body.
93struct Reply {
94 status: u16,
95 headers: HeaderMap,
96 body: Vec<u8>,
97}
98
99impl Reply {
100 fn text(&self) -> String {
101 String::from_utf8_lossy(&self.body).into_owned()
102 }
103
104 fn location(&self) -> Option<String> {
105 self.headers
106 .get("location")
107 .and_then(|value| value.to_str().ok())
108 .map(str::to_owned)
109 }
110}
111
112async fn get(app: &Router, uri: &str, cookie: Option<&str>) -> Reply {
113 let mut builder = Request::builder().method("GET").uri(uri);
114 if let Some(cookie) = cookie {
115 builder = builder.header("cookie", cookie);
116 }
117 let req = builder.body(Body::from(String::new())).expect("request");
118 collect(app.handle(req).await).await
119}
120
121/// A POST with the CSRF token and the given form fields.
122async fn post(app: &Router, uri: &str, cookie: &str, fields: &[(&str, &str)]) -> Reply {
123 let mut form = form_urlencoded::Serializer::new(String::new());
124 form.append_pair("csrf", CSRF);
125 for (name, value) in fields {
126 form.append_pair(name, value);
127 }
128 let req = Request::builder()
129 .method("POST")
130 .uri(uri)
131 .header("cookie", cookie)
132 .header("content-type", "application/x-www-form-urlencoded")
133 .body(Body::from(form.finish()))
134 .expect("request");
135 collect(app.handle(req).await).await
136}
137
138async fn collect(res: topcoat::router::response::Response) -> Reply {
139 let status = res.status().as_u16();
140 let headers = res.headers().clone();
141 let body = to_bytes(res.into_body(), usize::MAX).await.expect("body");
142 Reply {
143 status,
144 headers,
145 body: body.to_vec(),
146 }
147}
148
149/// Creates a run on `main` with one job per name. Jobs wait for approval when `needs_approval`.
150async fn run_with(fx: &Fixture, jobs: &[&str], needs_approval: bool) -> CiRun {
151 let needs: &[String] = &[];
152 let specs: Vec<NewJob<'_>> = jobs
153 .iter()
154 .map(|name| NewJob {
155 name,
156 spec_json: SPEC,
157 needs,
158 })
159 .collect();
160 fx.store
161 .create_run(
162 &fx.repo.id,
163 "push",
164 "refs/heads/main",
165 SHA,
166 None,
167 &fx.alice,
168 needs_approval,
169 &specs,
170 None,
171 )
172 .await
173 .expect("run")
174}
175
176/// Claims the one queued job with a fresh runner and finishes it as a success.
177async fn finish_job(fx: &Fixture) -> CiJob {
178 let token = fx
179 .store
180 .create_registration_token(&fx.alice, 24)
181 .await
182 .expect("token");
183 let (runner, _) = fx
184 .store
185 .register_runner(&token, "runner-1", &["linux".to_owned()])
186 .await
187 .expect("runner");
188 let (job, _, _) = fx
189 .store
190 .claim_job(&runner)
191 .await
192 .expect("claim")
193 .expect("a queued job");
194 fx.store
195 .finish_job(&job.id, JobState::Success, Some(0), None)
196 .await
197 .expect("finish");
198 job
199}
200
201async fn job_of(fx: &Fixture, run: &CiRun) -> CiJob {
202 fx.store
203 .jobs_for_run(&run.id)
204 .await
205 .expect("jobs")
206 .into_iter()
207 .next()
208 .expect("a job")
209}
210
211#[tokio::test]
212async fn private_repo_is_not_found_to_outsiders_and_anonymous() {
213 let dir = tempfile::tempdir().expect("tempdir");
214 let fx = fixture(dir.path()).await;
215 let run = run_with(&fx, &["build"], false).await;
216 let job = job_of(&fx, &run).await;
217 let urls = [
218 RUN_BASE.to_owned(),
219 format!("{RUN_BASE}/runs/{}", run.number),
220 format!("/alice/proj/ci/jobs/{}", job.id),
221 format!("/alice/proj/ci/jobs/{}/log.txt", job.id),
222 format!("/alice/proj/ci/jobs/{}/artifacts/x.bin", job.id),
223 ];
224 for url in &urls {
225 let anon = get(&fx.app, url, None).await;
226 assert_eq!(anon.status, 404, "anonymous {url}");
227 let outsider = get(&fx.app, url, Some(&cookie_for("carol-token"))).await;
228 assert_eq!(outsider.status, 404, "outsider {url}");
229 }
230 let reader = get(&fx.app, RUN_BASE, Some(&cookie_for("dave-token"))).await;
231 assert_eq!(reader.status, 200);
232}
233
234#[tokio::test]
235async fn buttons_and_manual_form_need_write() {
236 let dir = tempfile::tempdir().expect("tempdir");
237 let fx = fixture(dir.path()).await;
238 let run = run_with(&fx, &["build"], false).await;
239 let run_url = format!("{RUN_BASE}/runs/{}", run.number);
240 let writer = get(&fx.app, &run_url, Some(&cookie_for("bob-token"))).await;
241 assert!(writer.text().contains(">Re-run<"), "writer sees re-run");
242 assert!(writer.text().contains(">Cancel<"), "writer sees cancel");
243 let reader = get(&fx.app, &run_url, Some(&cookie_for("dave-token"))).await;
244 assert_eq!(reader.status, 200);
245 assert!(!reader.text().contains(">Re-run<"), "reader sees no re-run");
246 assert!(!reader.text().contains(">Cancel<"), "reader sees no cancel");
247 let list = get(&fx.app, RUN_BASE, Some(&cookie_for("dave-token"))).await;
248 assert!(
249 !list.text().contains("Run workflow"),
250 "reader has no manual form"
251 );
252 let list = get(&fx.app, RUN_BASE, Some(&cookie_for("bob-token"))).await;
253 assert!(
254 list.text().contains("Run workflow"),
255 "writer has the manual form"
256 );
257 let refused = post(
258 &fx.app,
259 &format!("{run_url}/cancel"),
260 &cookie_for("dave-token"),
261 &[],
262 )
263 .await;
264 assert_eq!(refused.status, 403, "reader cannot cancel");
265 let still = fx
266 .store
267 .run_by_id(&run.id)
268 .await
269 .expect("run")
270 .expect("run");
271 assert_eq!(still.state, RunState::Queued);
272}
273
274#[tokio::test]
275async fn run_list_pages_25_at_a_time() {
276 let dir = tempfile::tempdir().expect("tempdir");
277 let fx = fixture(dir.path()).await;
278 for _ in 0..30 {
279 run_with(&fx, &[], false).await;
280 }
281 let first = get(&fx.app, RUN_BASE, Some(&cookie_for("alice-token"))).await;
282 let text = first.text();
283 assert_eq!(text.matches("href=\"/alice/proj/ci/runs/").count(), 25);
284 assert!(text.contains("?page=2"), "older link");
285 assert!(!text.contains(">Newer<"));
286 let second = get(
287 &fx.app,
288 &format!("{RUN_BASE}?page=2"),
289 Some(&cookie_for("alice-token")),
290 )
291 .await;
292 let text = second.text();
293 assert_eq!(text.matches("href=\"/alice/proj/ci/runs/").count(), 5);
294 assert!(text.contains(">Newer<"), "newer link");
295 assert!(!text.contains("?page=3"), "no older link on the last page");
296 let bogus = get(
297 &fx.app,
298 &format!("{RUN_BASE}?page=abc"),
299 Some(&cookie_for("alice-token")),
300 )
301 .await;
302 assert_eq!(bogus.status, 200);
303 assert_eq!(
304 bogus.text().matches("href=\"/alice/proj/ci/runs/").count(),
305 25
306 );
307}
308
309/// The pages no longer reload themselves. A page that is not finished carries its live marker, and
310/// a finished one carries none.
311#[tokio::test]
312async fn pages_are_live_while_not_finished_and_never_meta_refresh() {
313 let dir = tempfile::tempdir().expect("tempdir");
314 let fx = fixture(dir.path()).await;
315 let run = run_with(&fx, &["build"], false).await;
316 let job = job_of(&fx, &run).await;
317 let url = format!("/alice/proj/ci/jobs/{}", job.id);
318 let run_url = format!("{RUN_BASE}/runs/{}", run.number);
319 let job_marker = format!("data-live=\"{url}/events\"");
320 let run_marker = format!("data-live=\"{run_url}/events\"");
321
322 let queued = get(&fx.app, &url, Some(&cookie_for("alice-token"))).await;
323 assert!(queued.text().contains(&job_marker), "job live while queued");
324 let run_page = get(&fx.app, &run_url, Some(&cookie_for("alice-token"))).await;
325 assert!(
326 run_page.text().contains(&run_marker),
327 "run live while queued"
328 );
329 for page in [&queued, &run_page] {
330 assert!(
331 !page.text().contains("http-equiv=\"refresh\""),
332 "no meta refresh"
333 );
334 }
335
336 finish_job(&fx).await;
337 let done = get(&fx.app, &url, Some(&cookie_for("alice-token"))).await;
338 assert!(
339 !done.text().contains("data-live="),
340 "job not live when finished"
341 );
342 assert!(
343 !done.text().contains("http-equiv=\"refresh\""),
344 "no meta refresh"
345 );
346}
347
348/// Opens an event stream without reading it. The response head carries the status.
349async fn open(
350 app: &Router,
351 uri: &str,
352 cookie: Option<&str>,
353) -> topcoat::router::response::Response {
354 let mut builder = Request::builder().method("GET").uri(uri);
355 if let Some(cookie) = cookie {
356 builder = builder.header("cookie", cookie);
357 }
358 app.handle(builder.body(Body::from(String::new())).expect("request"))
359 .await
360}
361
362/// The next frame of an event stream, or None once the stream has ended. Fails after five seconds.
363async fn next_event<S, E>(frames: &mut S) -> Option<Vec<u8>>
364where
365 S: futures_util::Stream<Item = Result<bytes::Bytes, E>> + Unpin,
366 E: std::fmt::Debug,
367{
368 use futures_util::StreamExt;
369
370 let frame = tokio::time::timeout(Duration::from_secs(5), frames.next())
371 .await
372 .expect("an event within five seconds")?;
373 Some(frame.expect("data").to_vec())
374}
375
376#[tokio::test]
377async fn live_events_need_the_same_access_as_the_page() {
378 let dir = tempfile::tempdir().expect("tempdir");
379 let fx = fixture(dir.path()).await;
380 let run = run_with(&fx, &["build"], false).await;
381 let job = job_of(&fx, &run).await;
382 let urls = [
383 format!("{RUN_BASE}/runs/{}/events", run.number),
384 format!("/alice/proj/ci/jobs/{}/events", job.id),
385 ];
386 for url in &urls {
387 let anon = open(&fx.app, url, None).await;
388 assert_eq!(anon.status().as_u16(), 404, "anonymous {url}");
389 let outsider = open(&fx.app, url, Some(&cookie_for("carol-token"))).await;
390 assert_eq!(outsider.status().as_u16(), 404, "outsider {url}");
391 let reader = open(&fx.app, url, Some(&cookie_for("dave-token"))).await;
392 assert_eq!(reader.status().as_u16(), 200, "reader {url}");
393 }
394}
395
396/// Each stream opens with its status. The run stream follows every bump; the job stream sends only
397/// status changes, so a log append is silent, and it ends once the job is terminal.
398#[tokio::test]
399async fn live_events_open_with_the_status_and_follow_changes() {
400 use futures_util::StreamExt;
401
402 let dir = tempfile::tempdir().expect("tempdir");
403 let fx = fixture(dir.path()).await;
404 let run = run_with(&fx, &["build"], false).await;
405 let job = job_of(&fx, &run).await;
406 let alice = cookie_for("alice-token");
407
408 let mut run_frames = open(
409 &fx.app,
410 &format!("{RUN_BASE}/runs/{}/events", run.number),
411 Some(&alice),
412 )
413 .await
414 .into_body()
415 .into_data_stream();
416 let first = next_event(&mut run_frames).await.expect("the run status");
417 assert_eq!(
418 first,
419 format!("event: changed\ndata: {}\n\n", run.state.as_str()).into_bytes()
420 );
421
422 let mut job_frames = open(
423 &fx.app,
424 &format!("/alice/proj/ci/jobs/{}/events", job.id),
425 Some(&alice),
426 )
427 .await
428 .into_body()
429 .into_data_stream();
430 let first = next_event(&mut job_frames).await.expect("the job status");
431 assert_eq!(
432 first,
433 format!("event: changed\ndata: {}\n\n", job.state.as_str()).into_bytes()
434 );
435
436 // A bump with no new log bytes and no status change sends nothing.
437 fx.live.bump_job(&job.id);
438 let quiet = tokio::time::timeout(Duration::from_millis(200), job_frames.next()).await;
439 assert!(quiet.is_err(), "a log append sends no status event");
440
441 // The run stream sends its status on every bump of the run.
442 fx.live.bump_run(&run.id);
443 let bumped = next_event(&mut run_frames).await.expect("the run status");
444 let now = fx
445 .store
446 .run_by_id(&run.id)
447 .await
448 .expect("run")
449 .expect("run");
450 assert_eq!(
451 bumped,
452 format!("event: changed\ndata: {}\n\n", now.state.as_str()).into_bytes()
453 );
454
455 // The job finishes: its final status is sent, then the stream ends.
456 finish_job(&fx).await;
457 fx.live.close_job(&job.id);
458 let last = next_event(&mut job_frames).await.expect("the final status");
459 assert_eq!(
460 last,
461 format!("event: changed\ndata: {}\n\n", JobState::Success.as_str()).into_bytes()
462 );
463 assert!(next_event(&mut job_frames).await.is_none(), "stream ends");
464}
465
466/// The job's log events: each append is sent as a `log` event with its bytes, the id is the offset
467/// past them, and the bytes before `?from=` are not sent again.
468#[tokio::test]
469async fn job_events_send_each_log_append_as_a_log_event() {
470 let dir = tempfile::tempdir().expect("tempdir");
471 let fx = fixture(dir.path()).await;
472 let run = run_with(&fx, &["build"], false).await;
473 let job = job_of(&fx, &run).await;
474 let path = log_path(&fx.data_dir, &job.id);
475 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
476 std::fs::write(&path, b"one\n").expect("log");
477
478 let mut frames = open(
479 &fx.app,
480 &format!("/alice/proj/ci/jobs/{}/events?from=4", job.id),
481 Some(&cookie_for("alice-token")),
482 )
483 .await
484 .into_body()
485 .into_data_stream();
486 let first = next_event(&mut frames).await.expect("the status");
487 assert_eq!(
488 first,
489 format!("event: changed\ndata: {}\n\n", job.state.as_str()).into_bytes()
490 );
491
492 let mut file = std::fs::OpenOptions::new()
493 .append(true)
494 .open(&path)
495 .expect("log");
496 file.write_all(b"two\nthree\n").expect("append");
497 fx.live.bump_job(&job.id);
498 let log = next_event(&mut frames).await.expect("the log event");
499 assert_eq!(
500 log,
501 b"event: log\ndata: two\ndata: three\ndata: \nid: 14\n\n".to_vec()
502 );
503}
504
505/// An append longer than 64 KiB is sent in several `log` events, in order.
506#[tokio::test]
507async fn job_events_split_a_long_append_into_64_kib_events() {
508 let dir = tempfile::tempdir().expect("tempdir");
509 let fx = fixture(dir.path()).await;
510 let run = run_with(&fx, &["build"], false).await;
511 let job = job_of(&fx, &run).await;
512 let path = log_path(&fx.data_dir, &job.id);
513 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
514 std::fs::write(&path, b"").expect("log");
515
516 let mut frames = open(
517 &fx.app,
518 &format!("/alice/proj/ci/jobs/{}/events", job.id),
519 Some(&cookie_for("alice-token")),
520 )
521 .await
522 .into_body()
523 .into_data_stream();
524 next_event(&mut frames).await.expect("the status");
525
526 let long = "a".repeat(150 * 1024);
527 let mut file = std::fs::OpenOptions::new()
528 .append(true)
529 .open(&path)
530 .expect("log");
531 file.write_all(long.as_bytes()).expect("append");
532 fx.live.bump_job(&job.id);
533
534 let mut sizes = Vec::new();
535 for _ in 0..3 {
536 let frame = next_event(&mut frames).await.expect("a log event");
537 let text = String::from_utf8(frame).expect("text");
538 assert!(text.starts_with("event: log\n"), "{text}");
539 let data = text
540 .lines()
541 .find_map(|line| line.strip_prefix("data: "))
542 .expect("data line");
543 sizes.push(data.len());
544 if sizes.len() == 3 {
545 assert!(
546 text.contains("id: 153600\n"),
547 "the last id is the end offset"
548 );
549 }
550 }
551 assert_eq!(sizes, vec![64 * 1024, 64 * 1024, 150 * 1024 - 128 * 1024]);
552}
553
554/// A reconnect sends `Last-Event-ID`, and the stream resumes after that offset, not at `?from=`.
555#[tokio::test]
556async fn job_events_resume_after_the_last_event_id() {
557 let dir = tempfile::tempdir().expect("tempdir");
558 let fx = fixture(dir.path()).await;
559 let run = run_with(&fx, &["build"], false).await;
560 let job = job_of(&fx, &run).await;
561 let path = log_path(&fx.data_dir, &job.id);
562 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
563 std::fs::write(&path, b"one\ntwo\n").expect("log");
564
565 let request = Request::builder()
566 .method("GET")
567 .uri(format!("/alice/proj/ci/jobs/{}/events?from=0", job.id))
568 .header("cookie", cookie_for("alice-token"))
569 .header("last-event-id", "4")
570 .body(Body::from(String::new()))
571 .expect("request");
572 let mut frames = fx.app.handle(request).await.into_body().into_data_stream();
573 next_event(&mut frames).await.expect("the status");
574 let log = next_event(&mut frames).await.expect("the log event");
575 assert_eq!(log, b"event: log\ndata: two\ndata: \nid: 8\n\n".to_vec());
576}
577
578/// The job page's log tail says where its text ends, so the events stream starts from there.
579#[tokio::test]
580async fn job_page_tail_carries_its_offset() {
581 let dir = tempfile::tempdir().expect("tempdir");
582 let fx = fixture(dir.path()).await;
583 let run = run_with(&fx, &["build"], false).await;
584 let job = job_of(&fx, &run).await;
585 let path = log_path(&fx.data_dir, &job.id);
586 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
587 std::fs::write(&path, b"one\ntwo\n").expect("log");
588
589 let page = get(
590 &fx.app,
591 &format!("/alice/proj/ci/jobs/{}", job.id),
592 Some(&cookie_for("alice-token")),
593 )
594 .await;
595 let text = page.text();
596 assert!(
597 text.contains("data-live-log=\"true\" data-offset=\"8\""),
598 "tail offset"
599 );
600 assert!(
601 text.contains(&format!("data-live-state=\"{}\"", job.state.as_str())),
602 "rendered status"
603 );
604}
605
606#[tokio::test]
607async fn job_page_tails_the_log_at_512_kib() {
608 let dir = tempfile::tempdir().expect("tempdir");
609 let fx = fixture(dir.path()).await;
610 let run = run_with(&fx, &["build"], false).await;
611 let job = job_of(&fx, &run).await;
612 let path = log_path(&fx.data_dir, &job.id);
613 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
614 let mut body = b"HEAD-MARK\n".to_vec();
615 let line = format!("{}\n", "x".repeat(99));
616 for _ in 0..6_000 {
617 body.extend_from_slice(line.as_bytes());
618 }
619 body.extend_from_slice(b"TAIL-MARK\n");
620 std::fs::write(&path, &body).expect("log");
621 let page = get(
622 &fx.app,
623 &format!("/alice/proj/ci/jobs/{}", job.id),
624 Some(&cookie_for("alice-token")),
625 )
626 .await;
627 let text = page.text();
628 assert!(text.contains("TAIL-MARK"));
629 assert!(!text.contains("HEAD-MARK"));
630 assert!(text.contains("Showing the last 512 KiB"));
631 assert!(
632 page.body.len() < 530 * 1024,
633 "page is {} bytes",
634 page.body.len()
635 );
636}
637
638#[tokio::test]
639async fn log_txt_is_the_full_file_for_a_finished_job() {
640 let dir = tempfile::tempdir().expect("tempdir");
641 let fx = fixture(dir.path()).await;
642 let run = run_with(&fx, &["build"], false).await;
643 let job = job_of(&fx, &run).await;
644 let path = log_path(&fx.data_dir, &job.id);
645 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
646 std::fs::write(&path, b"line one\nline two\n").expect("log");
647 finish_job(&fx).await;
648 let res = get(
649 &fx.app,
650 &format!("/alice/proj/ci/jobs/{}/log.txt", job.id),
651 Some(&cookie_for("alice-token")),
652 )
653 .await;
654 assert_eq!(res.status, 200);
655 assert_eq!(res.body, b"line one\nline two\n");
656 assert_eq!(
657 res.headers
658 .get("content-type")
659 .and_then(|v| v.to_str().ok()),
660 Some("text/plain; charset=utf-8")
661 );
662 assert_eq!(
663 res.headers
664 .get("x-content-type-options")
665 .and_then(|v| v.to_str().ok()),
666 Some("nosniff")
667 );
668}
669
670/// The next chunk a follower sends. Fails the test when none comes within five seconds.
671async fn next_chunk(
672 chunks: &mut tokio::sync::mpsc::UnboundedReceiver<bytes::Bytes>,
673) -> Option<bytes::Bytes> {
674 tokio::time::timeout(Duration::from_secs(5), chunks.recv())
675 .await
676 .expect("a chunk within five seconds")
677}
678
679/// A follower sends each append as soon as the job's signal is bumped. The wait is far longer
680/// than the timeouts here, so a chunk that arrives in time was woken by the bump, not by a timer.
681#[tokio::test]
682async fn log_follower_is_woken_by_bumps_and_ends_on_close() {
683 use futures_util::StreamExt;
684
685 let dir = tempfile::tempdir().expect("tempdir");
686 let fx = fixture(dir.path()).await;
687 let run = run_with(&fx, &["build"], false).await;
688 let job = job_of(&fx, &run).await;
689 let path = log_path(&fx.data_dir, &job.id);
690 std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
691 std::fs::write(&path, b"one\n").expect("log");
692 let live = Arc::new(CiLive::default());
693 let mut stream = Box::pin(super::ci_job::follow_log(
694 fx.store.clone(),
695 Arc::clone(&live),
696 job.id.clone(),
697 path.clone(),
698 ));
699 let (tx, mut chunks) = tokio::sync::mpsc::unbounded_channel();
700 tokio::spawn(async move {
701 while let Some(Ok(chunk)) = stream.next().await {
702 if tx.send(chunk).is_err() {
703 break;
704 }
705 }
706 });
707 assert_eq!(
708 next_chunk(&mut chunks).await.as_deref(),
709 Some(&b"one\n"[..])
710 );
711
712 let mut file = std::fs::OpenOptions::new()
713 .append(true)
714 .open(&path)
715 .expect("log");
716 file.write_all(b"two\n").expect("append");
717 live.bump_job(&job.id);
718 assert_eq!(
719 next_chunk(&mut chunks).await.as_deref(),
720 Some(&b"two\n"[..])
721 );
722
723 finish_job(&fx).await;
724 live.close_job(&job.id);
725 let end = tokio::time::timeout(Duration::from_secs(5), chunks.recv())
726 .await
727 .expect("the stream ends once the job is terminal");
728 assert!(end.is_none());
729}
730
731#[tokio::test]
732async fn artifact_download_sends_attachment_headers() {
733 let dir = tempfile::tempdir().expect("tempdir");
734 let fx = fixture(dir.path()).await;
735 let run = run_with(&fx, &["build"], false).await;
736 let job = job_of(&fx, &run).await;
737 let payload = b"artifact bytes";
738 let artifact = fx
739 .store
740 .add_artifact(&job.id, "report.txt", payload.len() as i64, SHA)
741 .await
742 .expect("artifact row");
743 let mut writer = fx
744 .blobs
745 .begin(&artifact_key(&artifact.id), 1024)
746 .await
747 .expect("begin");
748 writer.push(payload).await.expect("push");
749 writer.finish().await.expect("finish");
750 let res = get(
751 &fx.app,
752 &format!("/alice/proj/ci/jobs/{}/artifacts/report.txt", job.id),
753 Some(&cookie_for("alice-token")),
754 )
755 .await;
756 assert_eq!(res.status, 200);
757 assert_eq!(res.body, payload);
758 let header = |name: &str| {
759 res.headers
760 .get(name)
761 .and_then(|v| v.to_str().ok())
762 .map(str::to_owned)
763 };
764 assert_eq!(
765 header("content-disposition").as_deref(),
766 Some("attachment; filename=\"report.txt\"")
767 );
768 assert_eq!(header("content-length").as_deref(), Some("14"));
769 assert_eq!(header("x-content-type-options").as_deref(), Some("nosniff"));
770 assert_eq!(
771 header("content-type").as_deref(),
772 Some("application/octet-stream")
773 );
774}
775
776#[tokio::test]
777async fn approve_only_works_while_waiting_for_approval() {
778 let dir = tempfile::tempdir().expect("tempdir");
779 let fx = fixture(dir.path()).await;
780 let run = run_with(&fx, &["build"], true).await;
781 let run_url = format!("{RUN_BASE}/runs/{}", run.number);
782 let waiting = get(&fx.app, &run_url, Some(&cookie_for("bob-token"))).await;
783 assert!(
784 waiting.text().contains(">Approve and run<"),
785 "approve button while waiting"
786 );
787 let reader = get(&fx.app, &run_url, Some(&cookie_for("dave-token"))).await;
788 assert!(
789 !reader.text().contains(">Approve and run<"),
790 "no approve button for readers"
791 );
792
793 let approved = post(
794 &fx.app,
795 &format!("{run_url}/approve"),
796 &cookie_for("bob-token"),
797 &[],
798 )
799 .await;
800 assert!(
801 approved
802 .location()
803 .unwrap_or_default()
804 .ends_with("?notice=approved")
805 );
806 let now = fx
807 .store
808 .run_by_id(&run.id)
809 .await
810 .expect("run")
811 .expect("run");
812 assert_eq!(now.state, RunState::Queued);
813 let page = get(&fx.app, &run_url, Some(&cookie_for("bob-token"))).await;
814 assert!(
815 !page.text().contains(">Approve and run<"),
816 "no approve button after approval"
817 );
818
819 let again = post(
820 &fx.app,
821 &format!("{run_url}/approve"),
822 &cookie_for("bob-token"),
823 &[],
824 )
825 .await;
826 assert!(
827 again
828 .location()
829 .unwrap_or_default()
830 .ends_with("?notice=not-waiting")
831 );
832 let still = fx
833 .store
834 .run_by_id(&run.id)
835 .await
836 .expect("run")
837 .expect("run");
838 assert_eq!(still.state, RunState::Queued);
839}
840
841#[tokio::test]
842async fn cancel_and_rerun_post_and_redirect() {
843 let dir = tempfile::tempdir().expect("tempdir");
844 let fx = fixture(dir.path()).await;
845 let run = run_with(&fx, &["build"], false).await;
846 let run_url = format!("{RUN_BASE}/runs/{}", run.number);
847 let cancelled = post(
848 &fx.app,
849 &format!("{run_url}/cancel"),
850 &cookie_for("bob-token"),
851 &[],
852 )
853 .await;
854 assert!(
855 cancelled
856 .location()
857 .unwrap_or_default()
858 .ends_with("?notice=cancelled")
859 );
860 let now = fx
861 .store
862 .run_by_id(&run.id)
863 .await
864 .expect("run")
865 .expect("run");
866 assert_eq!(now.state, RunState::Cancelled);
867 let finished = post(
868 &fx.app,
869 &format!("{run_url}/cancel"),
870 &cookie_for("bob-token"),
871 &[],
872 )
873 .await;
874 assert!(
875 finished
876 .location()
877 .unwrap_or_default()
878 .ends_with("?notice=finished")
879 );
880
881 let rerun = post(
882 &fx.app,
883 &format!("{run_url}/rerun"),
884 &cookie_for("bob-token"),
885 &[],
886 )
887 .await;
888 assert_eq!(
889 rerun.location().as_deref(),
890 Some("/alice/proj/ci/runs/2?notice=rerun")
891 );
892 assert!(
893 fx.store
894 .run_by_number(&fx.repo.id, 2)
895 .await
896 .expect("lookup")
897 .is_some()
898 );
899}
900
901#[tokio::test]
902async fn manual_run_refuses_an_unknown_branch() {
903 let dir = tempfile::tempdir().expect("tempdir");
904 let fx = fixture(dir.path()).await;
905 let res = post(
906 &fx.app,
907 RUN_BASE,
908 &cookie_for("bob-token"),
909 &[("branch", "no-such-branch")],
910 )
911 .await;
912 assert!(
913 res.location()
914 .unwrap_or_default()
915 .ends_with("?notice=no-branch")
916 );
917 let reader = post(
918 &fx.app,
919 RUN_BASE,
920 &cookie_for("dave-token"),
921 &[("branch", "main")],
922 )
923 .await;
924 assert_eq!(reader.status, 403, "reader cannot start a manual run");
925}
926
927#[tokio::test]
928async fn badge_follows_read_access_and_reports_the_latest_run() {
929 let dir = tempfile::tempdir().expect("tempdir");
930 let fx = fixture(dir.path()).await;
931 let open = fx
932 .store
933 .create_repo(&fx.alice, "open", "", Visibility::Public, "main")
934 .await
935 .expect("public repo");
936 let svg_path = "/alice/proj/badge.svg";
937
938 // The private repo: a run exists, but only readers see its badge.
939 run_with(&fx, &["build"], false).await;
940 finish_job(&fx).await;
941 assert_eq!(get(&fx.app, svg_path, None).await.status, 404, "anonymous");
942 let outsider = get(&fx.app, svg_path, Some(&cookie_for("carol-token"))).await;
943 assert_eq!(outsider.status, 404, "outsider");
944 let reader = get(&fx.app, svg_path, Some(&cookie_for("dave-token"))).await;
945 assert_eq!(reader.status, 200, "reader");
946 assert!(reader.text().contains("passing"));
947 assert_eq!(
948 reader
949 .headers
950 .get("content-type")
951 .and_then(|v| v.to_str().ok()),
952 Some("image/svg+xml")
953 );
954 assert_eq!(
955 reader
956 .headers
957 .get("cache-control")
958 .and_then(|v| v.to_str().ok()),
959 Some("no-cache, max-age=0")
960 );
961
962 // The public repo: anyone sees the badge. Unknown branches and jobs have no status.
963 let needs: &[String] = &[];
964 let specs = [NewJob {
965 name: "build",
966 spec_json: SPEC,
967 needs,
968 }];
969 fx.store
970 .create_run(
971 &open.id,
972 "push",
973 "refs/heads/main",
974 SHA,
975 None,
976 &fx.alice,
977 false,
978 &specs,
979 None,
980 )
981 .await
982 .expect("run");
983 finish_job(&fx).await;
984 let anon = get(&fx.app, "/alice/open/badge.svg", None).await;
985 assert_eq!(anon.status, 200);
986 assert!(anon.text().contains(">passing</text>"));
987 let unknown = get(&fx.app, "/alice/open/badge.svg?branch=nope", None).await;
988 assert_eq!(unknown.status, 200);
989 assert!(unknown.text().contains(">no status</text>"));
990 let job = get(&fx.app, "/alice/open/badge.svg?job=build", None).await;
991 assert!(job.text().contains(">build</text>"));
992 assert!(job.text().contains(">passing</text>"));
993 let missing_job = get(&fx.app, "/alice/open/badge.svg?job=lint", None).await;
994 assert!(missing_job.text().contains(">no status</text>"));
995 let unknown_repo = get(&fx.app, "/alice/nothere/badge.svg", None).await;
996 assert_eq!(unknown_repo.status, 404);
997}