d58a911
use std::{collections::BTreeMap, path::PathBuf};
use anyhow::{Context, ensure};
use ci::{CiService, LoadedWorkflow, actions_environments::EnvironmentRules, runs::RunSeed};
use models::{
ci::{runner as runners, workflow_run},
repo::repos,
space::spaces,
};
use sea_orm::{ActiveModelTrait, ColumnTrait, Database, EntityTrait, QueryFilter, Set};
use serde_json::json;
use time::OffsetDateTime;
use uuid::Uuid;
const WORKFLOW: &str = r#"name: Fixture
on: [push, workflow_dispatch]
jobs:
plan:
runs-on: self-hosted
outputs:
matrix: ${{ steps.plan.outputs.matrix }}
steps:
- id: plan
run: echo 'matrix={"value":["a","b"]}' >> "$GITHUB_OUTPUT"
verify:
needs: plan
runs-on: self-hosted
name: verify-${{ matrix.value }}
strategy:
max-parallel: 1
matrix: ${{ fromJSON(needs.plan.outputs.matrix) }}
outputs:
value: ${{ steps.result.outputs.value }}
steps:
- id: result
run: echo 'value=${{ matrix.value }}' >> "$GITHUB_OUTPUT"
cleanup:
needs: verify
if: always()
runs-on: self-hosted
steps:
- run: echo 'fixture completed'
"#;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let directory = std::env::current_dir()?.join("target/actions-fixture");
tokio::fs::create_dir_all(&directory).await?;
let mut config = config::AppConfig::default();
config.database.url =
Some("postgres://postgres:fixture-password@127.0.0.1:45432/gitdata_ci_fixture".into());
config.database.max_connections = Some(20);
config.database.sqlx_logging = Some(false);
config.redis.url = Some("redis://127.0.0.1:46379/9".into());
config.s3.bucket = Some("ci-fixture".into());
config.s3.region = Some("us-east-1".into());
config.s3.endpoint = Some("http://127.0.0.1:49000".into());
config.s3.access_key_id = Some("fixture-access".into());
config.s3.secret_access_key = Some("fixture-secret-password".into());
config.s3.path_style = Some(true);
config.email.enabled = Some(false);
config.email.from_address = Some("fixture@localhost".into());
config.ci.enabled = true;
config.ci.secret_encryption_key = "QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE=".into();
// Built by the git-data-actions repository now; override with ACTIONS_ENGINE_BINARY.
config.ci.actions_engine_binary = std::env::var("ACTIONS_ENGINE_BINARY").unwrap_or_else(|_| {
std::env::current_dir()
.expect("current dir")
.join("target/gitdata-actions-engine")
.to_string_lossy()
.into_owned()
});
config.ci.runtime_base_url = "http://127.0.0.1:43120".into();
config.ci.event_queue_topic = "gitdata:ci-fixture:events".into();
config.ci.lease_secs = 120;
config.git.repo_root = directory.join("repos").to_string_lossy().into_owned();
config.git.base_url = "http://127.0.0.1:43121".into();
let database = Database::connect(config.database.url.clone().unwrap()).await?;
models::migrate::sync_schema(&database)
.await
.context("fixture schema sync")?;
let state = export::AppState::with_config(config).await?;
let service = CiService::new(state);
let (repo, space, owner, reviewer) = seed(&service).await?;
let http = axum::Router::new()
.nest(
"/api/v1",
ci::routes::runner_router().with_state(service.clone()),
)
.merge(ci::actions_runtime::router().with_state(service.clone()));
let listener = tokio::net::TcpListener::bind("127.0.0.1:43120").await?;
let server = tokio::spawn(async move { axum::serve(listener, http).await });
let git_service = git::service::GitService::new(service.state.clone());
let smart = git::smart::service::SmartService::new(git_service);
let git_listener = tokio::net::TcpListener::bind("127.0.0.1:43121").await?;
let git_server =
tokio::spawn(
async move { axum::serve(git_listener, git::smart::handler::router(smart)).await },
);
let runner_id = Uuid::now_v7();
let now = OffsetDateTime::now_utc();
runners::ActiveModel {
id: Set(runner_id),
name: Set("actions-fixture".into()),
labels: Set(json!(["self-hosted"])),
capacity: Set(4),
version: Set(Some("fixture".into())),
state: Set(runners::RunnerState::Active),
space_id: Set(Some(space.id)),
last_seen_at: Set(Some(now)),
created_at: Set(now),
updated_at: Set(now),
protocol_version: Set(Some(2)),
capabilities: Set(Some(json!(["actions-v2", "docker"]))),
}
.insert(service.db_write())
.await?;
let token_id = Uuid::now_v7();
let (token, hash, salt, last_eight) = ci::runners::generate_credential(token_id);
models::ci::runner_token::ActiveModel {
id: Set(token_id),
runner_id: Set(None),
name: Set("fixture-client".into()),
token_hash: Set(hash),
token_salt: Set(salt),
token_last_eight: Set(last_eight),
space_id: Set(Some(space.id)),
single_use: Set(true),
used_at: Set(None),
revoked_at: Set(None),
expires_at: Set(Some(now + time::Duration::days(1))),
created_at: Set(now),
updated_at: Set(now),
}
.insert(service.db_write())
.await?;
let runner_config = config::runner::RunnerConfig {
server_url: "http://127.0.0.1:43120".into(),
registration_token: token,
name: "fixture-client".into(),
actions_engine_binary: service.config().ci.actions_engine_binary.clone(),
workspace_root: directory.join("workspaces").to_string_lossy().into_owned(),
actions_node_runtimes: BTreeMap::from([(
"node24".into(),
std::env::current_dir()?
.join("target/node24/bin/node")
.to_string_lossy()
.into_owned(),
)]),
actions_cache_dir: directory
.join("action-cache")
.to_string_lossy()
.into_owned(),
..Default::default()
};
let client = runner::client::RunnerClient::new(&runner_config)?;
let registration = client
.register(&ci_protocol::runner::RegisterRequest {
token: runner_config.registration_token.clone(),
name: runner_config.name.clone(),
labels: vec!["self-hosted".into()],
capacity: 4,
version: Some("fixture".into()),
protocol_versions: vec![2, 1],
capabilities: vec!["actions-v2".into(), "docker".into()],
})
.await?;
let client = client.with_credential(registration.credential);
let run_id = create(&service, &repo, &space, owner, WORKFLOW).await?;
let count = drive(&service, &client, &runner_config, run_id).await?;
let result = workflow_run::Entity::find_by_id(run_id)
.one(service.db_write())
.await?
.unwrap();
ensure!(
result.state == models::ci::workflow_run::WorkflowRunState::Success,
"matrix run failed: {:?}",
service.run_detail(run_id).await?
);
ensure!(count == 4, "expected 4 physical jobs, got {count}");
let jobs = models::ci::workflow_job::Entity::find()
.filter(models::ci::workflow_job::Column::RunId.eq(run_id))
.all(service.db_write())
.await?;
ensure!(
jobs.iter()
.find(|job| job.job_key == "verify")
.and_then(|job| job.outputs.as_ref())
.is_some_and(|value| value["value"] == "b")
);
println!(
"PASS: protocol v2, persisted DAG, dynamic matrix, cross-job outputs, terminal step events"
);
let rerun = service.rerun(run_id, owner).await?;
drive(&service, &client, &runner_config, rerun).await?;
println!("PASS: snapshot rerun");
let rules = EnvironmentRules {
reviewers: vec![reviewer],
wait_minutes: 0,
..Default::default()
};
service
.upsert_actions_environment(space.id, repo.id, "production", &rules)
.await?;
let approval_workflow = "on: push\njobs:\n deploy:\n runs-on: self-hosted\n environment: production\n steps:\n - run: echo approved\n";
let approved = create(&service, &repo, &space, owner, approval_workflow).await?;
service.advance_jobs().await?;
let claim = service
.claim_job(registration.runner_id, &["self-hosted".into()])
.await?;
ensure!(claim.is_none(), "unapproved environment job was claimed");
let approvals = service.actions_approvals(repo.id, approved).await?;
ensure!(approvals.len() == 1);
service
.review_actions_environment(repo.id, approved, approvals[0].id, reviewer, true, None)
.await?;
drive(&service, &client, &runner_config, approved).await?;
println!("PASS: environment approval gate");
server.abort();
git_server.abort();
Ok(())
}
async fn create(
service: &CiService,
repo: &repos::Model,
space: &spaces::Model,
owner: Uuid,
source: &str,
) -> anyhow::Result<Uuid> {
let actions = service
.parse_actions_workflow(".github/workflows/fixture.yml", source)
.await?;
let workflow = LoadedWorkflow {
file: actions.file.clone(),
name: actions.name().unwrap_or("fixture".into()),
definition: actions.projection(),
actions: Some(actions),
};
let seed = RunSeed {
event: models::ci::workflow_run::WorkflowEventKind::Push,
event_payload: json!({"event_id":Uuid::now_v7()}),
commit_sha: "a".repeat(40),
git_ref: "refs/heads/main".into(),
branch: Some("main".into()),
tag: None,
base_sha: None,
base_ref: None,
pr_number: None,
trigger_user_id: Some(owner),
};
Ok(service.create_run(repo, space, &seed, &workflow).await?)
}
async fn drive(
service: &CiService,
client: &runner::client::RunnerClient,
config: &config::runner::RunnerConfig,
run_id: Uuid,
) -> anyhow::Result<usize> {
let mut count = 0;
for _ in 0..100 {
service.advance_jobs().await?;
service.enforce_concurrency().await?;
let result = client
.claim(&ci_protocol::runner::ClaimRequest { wait_secs: Some(0) })
.await?;
if let Some(payload) = result.job {
let finish = runner::actions::execute(client, config, &payload).await;
client.finish(&payload.job_id, &finish).await?;
count += 1;
}
service.advance_jobs().await?;
service.finalize_runs().await?;
let run = workflow_run::Entity::find_by_id(run_id)
.one(service.db_write())
.await?
.unwrap();
if matches!(
run.state,
models::ci::workflow_run::WorkflowRunState::Success
| models::ci::workflow_run::WorkflowRunState::Failure
| models::ci::workflow_run::WorkflowRunState::Cancelled
) {
return Ok(count);
}
}
anyhow::bail!("fixture run did not reach a terminal state")
}
async fn seed(service: &CiService) -> anyhow::Result<(repos::Model, spaces::Model, Uuid, Uuid)> {
let owner = Uuid::now_v7();
let reviewer = Uuid::now_v7();
let now = OffsetDateTime::now_utc();
for (id, name) in [
(owner, format!("fixture-{}", owner.simple())),
(reviewer, format!("fixture-{}", reviewer.simple())),
] {
models::users::user::ActiveModel {
uid: Set(id),
username: Set(name.clone()),
lower_name: Set(name.clone()),
display_name: Set(name),
avatar_url: Set(String::new()),
is_active: Set(true),
is_restricted: Set(false),
prohibit_login: Set(false),
visibility: Set(Default::default()),
version: Set(1),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
}
.insert(service.db_write())
.await?;
}
let space_id = Uuid::now_v7();
let name = format!("fixture-{}", space_id.simple());
let space = spaces::ActiveModel {
id: Set(space_id),
lower_name: Set(name.clone()),
name: Set(name),
owner_id: Set(owner),
is_active: Set(true),
allow_join: Set(false),
allow_invite: Set(false),
nums_members: Set(1),
allow_join_level: Set(Default::default()),
member_must_enable_2fa: Set(false),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
}
.insert(service.db_write())
.await?;
let dir = Uuid::now_v7();
models::dir::dirs::ActiveModel {
id: Set(dir),
space_id: Set(Some(space.id)),
name: Set("repos".into()),
kind: Set(Default::default()),
position: Set(0),
created_by_uid: Set(owner),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
}
.insert(service.db_write())
.await?;
let id = Uuid::now_v7();
let name = format!("fixture-{}", id.simple());
let repo = repos::ActiveModel {
id: Set(id),
dir_id: Set(dir),
lower_name: Set(name.clone()),
name: Set(name),
default_branch: Set("main".into()),
nums_watch: Set(0),
nums_star: Set(0),
nums_pr: Set(0),
nums_commits: Set(0),
nums_branches: Set(0),
nums_tags: Set(0),
enable_lfs: Set(false),
size: Set(0),
git_size: Set(0),
lfs_size: Set(0),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
}
.insert(service.db_write())
.await?;
let directory = PathBuf::from(&service.config().git.repo_root).join(repo.id.to_string());
tokio::fs::create_dir_all(&directory).await?;
let status = std::process::Command::new("git")
.args(["init", "--bare"])
.arg(directory)
.status()?;
ensure!(status.success());
Ok((repo, space, owner, reviewer))
}