use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use futures::future::try_join_all;
use pipedash_plugin_actions::{
dispatch_body,
dispatch_inputs,
};
use pipedash_plugin_api::*;
use crate::log_dialect::GiteaLogDialect;
use crate::{
client,
config,
mapper,
metadata,
types::JobAttempt,
};
#[derive(serde::Serialize, serde::Deserialize)]
struct LogPosition {
job: String,
attempt: JobAttempt,
offset: u64,
}
pub struct GiteaPlugin {
metadata: PluginMetadata,
client: Option<client::GiteaClient>,
provider_id: Option<i64>,
config: HashMap<String, String>,
status_source: Option<StatusSource>,
}
impl Default for GiteaPlugin {
fn default() -> Self {
Self::new()
}
}
impl GiteaPlugin {
pub fn new() -> Self {
Self {
metadata: metadata::create_metadata(),
client: None,
provider_id: None,
config: HashMap::new(),
status_source: None,
}
}
fn client(&self) -> PluginResult<&client::GiteaClient> {
self.client
.as_ref()
.ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
}
fn provider_id(&self) -> PluginResult<i64> {
self.provider_id
.ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))
}
async fn fetch_repo_pipelines(
&self, provider_id: i64, repo_full_name: &str,
) -> PluginResult<Vec<Pipeline>> {
let (owner, repo) = config::parse_repo(repo_full_name)?;
let client = self.client()?;
let workflows = client.list_workflows(owner, repo).await?;
let workflow_ids: Vec<String> = workflows.iter().map(|w| w.id.clone()).collect();
let latest_runs = client.latest_runs(owner, repo, &workflow_ids).await?;
Ok(workflows
.iter()
.map(|workflow| {
let latest_run = latest_runs.get(&workflow.id);
mapper::map_pipeline(provider_id, owner, repo, workflow, latest_run)
})
.collect())
}
}
#[async_trait]
impl Plugin for GiteaPlugin {
fn metadata(&self) -> &PluginMetadata {
&self.metadata
}
fn status_source(&self) -> Option<&StatusSource> {
self.status_source.as_ref()
}
fn initialize(
&mut self, provider_id: i64, config: HashMap<String, String>,
http_client: Option<Arc<reqwest::Client>>,
) -> PluginResult<()> {
let base_url = config::get_base_url(&config)?;
let token = config
.get("token")
.map(|token| token.trim())
.filter(|token| !token.is_empty())
.ok_or_else(|| PluginError::InvalidConfig("Missing Gitea access token".to_string()))?
.to_string();
let status_source = custom_status_source(&config)?;
let http_client = http_client.unwrap_or_else(|| {
Arc::new(
reqwest::Client::builder()
.use_rustls_tls()
.https_only(cfg!(target_os = "ios"))
.pool_max_idle_per_host(10)
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.tcp_keepalive(Duration::from_secs(60))
.build()
.expect("Failed to build HTTP client"),
)
});
self.status_source = status_source;
self.client = Some(client::GiteaClient::new(http_client, base_url, token));
self.provider_id = Some(provider_id);
self.config = config;
Ok(())
}
async fn validate_credentials(&self) -> PluginResult<bool> {
self.client()?.get_user().await?;
Ok(true)
}
async fn fetch_available_pipelines(
&self, params: Option<PaginationParams>,
) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
self.fetch_available_pipelines_filtered(None, None, params)
.await
}
async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
self.client()?.list_organizations().await
}
async fn ids_with_ci(&self, ids: Vec<String>) -> PluginResult<Vec<String>> {
let client = self.client()?;
Ok(retain_with_ci(ids, |id| async move {
let (owner, repo) = config::parse_repo(&id)?;
Ok(!client.list_workflows(owner, repo).await?.is_empty())
})
.await)
}
async fn fetch_available_pipelines_filtered(
&self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
let client = self.client()?;
let params = params.unwrap_or_default();
let (uid, exclusive) = match org {
Some(org) => (
org.parse().map_err(|_| {
PluginError::InvalidConfig(format!("Invalid organization ID: {org}"))
})?,
true,
),
None => (client.get_user().await?.id, false),
};
let response = client
.search_repositories(uid, exclusive, search.as_deref(), ¶ms)
.await?;
Ok(PaginatedResponse {
items: response
.items
.iter()
.map(mapper::map_available_pipeline)
.collect(),
page: response.page,
page_size: response.page_size,
total_count: response.total_count,
total_pages: response.total_pages,
has_more: response.has_more,
})
}
async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
let provider_id = self.provider_id()?;
let repositories = config::get_repositories(&self.config);
if repositories.is_empty() {
return Err(PluginError::InvalidConfig(
"No repositories configured".to_string(),
));
}
let results = try_join_all(
repositories
.iter()
.map(|repo| self.fetch_repo_pipelines(provider_id, repo)),
)
.await?;
Ok(results.into_iter().flatten().collect())
}
async fn fetch_run_history(
&self, pipeline_id: &str, limit: usize,
) -> PluginResult<Vec<PipelineRun>> {
let id = config::parse_pipeline_id(pipeline_id)?;
let runs = self
.client()?
.workflow_runs(id.owner, id.repo, id.workflow, limit)
.await?;
Ok(runs
.iter()
.map(|run| mapper::map_run(run, pipeline_id, id.owner))
.collect())
}
async fn fetch_run_details(
&self, pipeline_id: &str, run_number: i64,
) -> PluginResult<PipelineRun> {
let id = config::parse_pipeline_id(pipeline_id)?;
let run = self
.client()?
.find_run(id.owner, id.repo, id.workflow, run_number)
.await?;
Ok(mapper::map_run(&run, pipeline_id, id.owner))
}
async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
let id = config::parse_pipeline_id(run.pipeline_id)?;
let run_id = run
.metadata
.get("run_id")
.and_then(serde_json::Value::as_i64)
.ok_or_else(|| {
PluginError::InvalidConfig(format!("Run #{} has no Gitea run ID", run.run_number))
})?;
let jobs = self
.client()?
.list_run_jobs(id.owner, id.repo, run_id)
.await?;
Ok(mapper::jobs_to_log_nodes(&jobs))
}
async fn read_log(
&self, run: RunRef<'_>, node_id: &str, from: Option<&LogCursor>, sink: &mut LogSink,
) -> PluginResult<LogRead> {
let id = config::parse_pipeline_id(run.pipeline_id)?;
if node_id.is_empty() || !node_id.bytes().all(|b| b.is_ascii_digit()) {
return Err(PluginError::InvalidConfig(format!(
"Invalid Gitea job ID: {node_id}"
)));
}
let client = self.client()?;
let attempt = client.job_attempt(id.owner, id.repo, node_id).await?;
let position = from
.and_then(|cursor| cursor.parse::<LogPosition>())
.filter(|position| position.job == node_id && position.attempt == attempt);
if from.is_some() && position.is_none() {
sink.restart()?;
}
let (mode, offset) = client
.stream_job_log(id.owner, id.repo, node_id, position.map(|p| p.offset), sink)
.await?;
let job = node_id.to_string();
Ok(LogRead {
mode,
cursor: Some(LogCursor::new(&LogPosition {
job,
attempt,
offset,
})?),
})
}
fn log_dialect(&self) -> &'static dyn LogDialect {
&GiteaLogDialect
}
async fn fetch_workflow_parameters(
&self, workflow_id: &str,
) -> PluginResult<Vec<WorkflowParameter>> {
let id = config::parse_pipeline_id(workflow_id)?;
let client = self.client()?;
let workflows = client.list_workflows(id.owner, id.repo).await?;
let path = workflows
.iter()
.find(|workflow| workflow.id == id.workflow)
.map(|workflow| workflow.path.as_str())
.ok_or_else(|| {
PluginError::PipelineNotFound(format!("Workflow {} not found", id.workflow))
})?;
let source = client.raw_file(id.owner, id.repo, path).await?;
let mut parameters = vec![WorkflowParameter {
name: "ref".to_string(),
label: Some("Ref".to_string()),
description: Some("Branch, tag, or commit SHA to run workflow on".to_string()),
param_type: WorkflowParameterType::String {
default: Some("main".to_string()),
},
required: true,
}];
parameters.extend(dispatch_inputs(&source)?);
Ok(parameters)
}
async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
let id = config::parse_pipeline_id(¶ms.workflow_id)?;
let client = self.client()?;
let body = dispatch_body(params.inputs);
client
.dispatch_workflow(id.owner, id.repo, id.workflow, &body)
.await?;
Ok(client.actions_url(id.owner, id.repo))
}
async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
let id = config::parse_pipeline_id(pipeline_id)?;
let client = self.client()?;
let run = client
.find_run(id.owner, id.repo, id.workflow, run_number)
.await?;
client.cancel_run(id.owner, id.repo, run.id).await
}
}
#[cfg(all(test, not(target_os = "ios")))]
mod tests {
use std::sync::atomic::{
AtomicUsize,
Ordering,
};
use tokio::io::{
AsyncReadExt,
AsyncWriteExt,
};
use super::*;
const STUB_REPOS: usize = 150;
const STUB_ORGS: usize = 70;
const MAX_RESPONSE_ITEMS: usize = 50;
const DEFAULT_PAGING_NUM: usize = 30;
fn stub_page(url: &reqwest::Url, total: usize) -> std::ops::Range<usize> {
let query = |key: &str| {
url.query_pairs()
.find(|(k, _)| k == key)
.map(|(_, v)| v.parse::<usize>().unwrap())
};
let limit = query("limit")
.unwrap_or(DEFAULT_PAGING_NUM)
.min(MAX_RESPONSE_ITEMS);
let page = query("page").unwrap_or(1);
((page - 1) * limit).min(total)..(page * limit).min(total)
}
const JOB_LOG: &str = "2026-10-28T10:00:01.1000000Z ::group::Set up job\n\
2026-10-28T10:00:02.2000000Z cloning into /workspace\n\
2026-10-28T10:00:03.3000000Z ::endgroup::\n";
fn stub_jobs(page: usize) -> serde_json::Value {
let step = |number: i64, name: &str, status: &str, conclusion: &str| {
let completed_at = if status == "completed" {
format!("2026-10-28T10:00:0{}Z", number + 2)
} else {
"1970-01-01T00:00:00Z".to_string()
};
serde_json::json!({
"name": name, "number": number, "status": status, "conclusion": conclusion,
"started_at": format!("2026-10-28T10:00:0{}Z", number + 1),
"completed_at": completed_at
})
};
let job = |id: i64, name: &str, status: &str, conclusion: &str, started: &str| {
serde_json::json!({
"id": id, "name": name, "status": status, "conclusion": conclusion,
"html_url": format!("http://g/claude/smoke__ci/actions/runs/7/jobs/{}", id - 101),
"started_at": started, "completed_at": "1970-01-01T00:00:00Z",
"steps": null
})
};
let unset = "1970-01-01T00:00:00Z";
let mut build = job(101, "build", "completed", "success", "2026-10-28T10:00:00Z");
build["steps"] = serde_json::json!([
step(0, "Set up job", "completed", "success"),
step(1, "Run tests", "completed", "success")
]);
let mut lint = job(102, "lint", "in_progress", "", "2026-10-28T10:00:00Z");
lint["steps"] = serde_json::json!([step(0, "Set up job", "in_progress", "")]);
let jobs = match page {
1 => vec![build, lint],
2 => vec![
job(103, "deploy", "queued", "", unset),
job(104, "publish", "completed", "skipped", unset),
],
_ => Vec::new(),
};
serde_json::json!({"jobs": jobs, "total_count": 4})
}
async fn stub_gitea() -> (GiteaPlugin, tokio::task::JoinHandle<()>) {
let (plugin, server, _) = stub_gitea_counting().await;
(plugin, server)
}
async fn stub_gitea_counting() -> (GiteaPlugin, tokio::task::JoinHandle<()>, Arc<AtomicUsize>) {
let _ = rustls::crypto::ring::default_provider().install_default();
let requests = Arc::new(AtomicUsize::new(0));
let counter = requests.clone();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.unwrap();
let mut request = Vec::new();
let mut buffer = [0; 4096];
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
let count = stream.read(&mut buffer).await.unwrap();
if count == 0 {
break;
}
request.extend_from_slice(&buffer[..count]);
}
let request = String::from_utf8(request).unwrap();
counter.fetch_add(1, Ordering::SeqCst);
let mut head = request.split_whitespace();
let method = head.next().unwrap();
let url =
reqwest::Url::parse(&format!("http://{addr}{}", head.next().unwrap())).unwrap();
let mut total_count = None;
let (status, body) = match (method, url.path()) {
("GET", "/api/v1/repos/search") => {
total_count = Some(STUB_REPOS);
let data: Vec<_> = stub_page(&url, STUB_REPOS)
.map(|i| {
serde_json::json!({
"name": format!("r{i}"),
"full_name": format!("org/r{i}"),
"owner": {"login": "org"}
})
})
.collect();
(200, serde_json::json!({"data": data}))
}
("GET", "/api/v1/user/orgs") => {
total_count = Some(STUB_ORGS);
let orgs: Vec<_> = stub_page(&url, STUB_ORGS)
.map(|i| serde_json::json!({"id": 100 + i, "username": format!("o{i}")}))
.collect();
(200, serde_json::json!(orgs))
}
("GET", "/api/v1/user") => {
(200, serde_json::json!({"id": 2, "login": "claude"}))
}
("GET", "/api/v1/repos/claude/smoke__ci/actions/workflows") => (
200,
serde_json::json!({"workflows": [
{"id": "build__test.yaml", "name": "Build", "path": ".gitea/workflows/build__test.yaml"},
{"id": "lint.yaml", "name": "Lint", "path": ".gitea/workflows/lint.yaml"}
]}),
),
("GET", "/api/v1/repos/claude/smoke__ci/actions/runs") => {
let page: usize = url
.query_pairs()
.find(|(key, _)| key == "page")
.map(|(_, value)| value.parse().unwrap())
.unwrap();
let runs: Vec<_> = if page == 1 {
(0..50)
.map(|i| {
serde_json::json!({
"id": 100 - i, "run_number": 100 - i, "status": "queued",
"path": "build__test.yaml@refs/heads/main", "head_branch": "main",
"html_url": format!("http://g/claude/smoke__ci/actions/runs/{}", 100 - i)
})
})
.collect()
} else {
vec![
serde_json::json!({"id": 3, "run_number": 3, "status": "completed", "conclusion": "failure",
"path": "lint.yaml@refs/heads/main", "head_branch": "main",
"html_url": "http://g/claude/smoke__ci/actions/runs/3"}),
serde_json::json!({"id": 2, "run_number": 2, "status": "queued",
"path": "build__test.yaml@refs/heads/main", "head_branch": "main",
"html_url": "http://g/claude/smoke__ci/actions/runs/2"}),
serde_json::json!({"id": 1, "run_number": 1, "status": "completed", "conclusion": "success",
"path": "build__test.yaml@refs/heads/main", "head_branch": "main",
"html_url": "http://g/claude/smoke__ci/actions/runs/1"}),
]
};
(200, serde_json::json!({"workflow_runs": runs}))
}
("POST", "/api/v1/repos/claude/smoke__ci/actions/runs/2/cancel") => {
(404, serde_json::json!({"message": "not found"}))
}
("GET", "/api/v1/repos/claude/missing/actions/workflows") => {
(404, serde_json::json!({"message": "not found"}))
}
("GET", "/api/v1/repos/claude/empty/actions/workflows") => {
(200, serde_json::json!({"workflows": []}))
}
("GET", "/api/v1/repos/claude/smoke__ci/raw/.gitea/workflows/build__test.yaml") => {
(200, serde_json::json!("on:\n workflow_dispatch:\n inputs:\n greeting:\n description: Greeting\n default: hi\n"))
}
("GET", "/api/v1/repos/claude/smoke__ci/actions/runs/7/jobs") => {
let page = url
.query_pairs()
.find(|(key, _)| key == "page")
.map(|(_, value)| value.parse().unwrap())
.unwrap();
(200, stub_jobs(page))
}
("GET", "/api/v1/repos/claude/smoke__ci/actions/jobs/101") => (
200,
serde_json::json!({"id": 101, "name": "build", "status": "completed", "run_attempt": 1, "started_at": "2026-10-28T10:00:00Z"}),
),
("GET", "/api/v1/repos/claude/smoke__ci/actions/jobs/104") => (
200,
serde_json::json!({"id": 104, "name": "deploy", "status": "waiting"}),
),
("GET", "/api/v1/repos/claude/smoke__ci/actions/jobs/101/logs") => {
(200, serde_json::json!(JOB_LOG))
}
("GET", "/api/v1/repos/claude/smoke__ci/actions/jobs/104/logs") => {
(404, serde_json::json!({"message": "job not started"}))
}
("GET", "/api/v1/repos/claude/old/actions/runs/7/jobs") => {
(404, serde_json::json!({"message": "not found"}))
}
(method, path) => panic!("Unexpected stub request: {method} {path}"),
};
let body = match body {
serde_json::Value::String(raw) => raw,
json => json.to_string(),
};
let total_header = total_count
.map(|total| format!("X-Total-Count: {total}\r\n"))
.unwrap_or_default();
let response = format!(
"HTTP/1.1 {status} X\r\nContent-Type: application/json\r\n{total_header}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).await.unwrap();
}
});
let mut plugin = GiteaPlugin::new();
plugin
.initialize(
1,
HashMap::from([
("base_url".to_string(), format!("http://{addr}/")),
("token".to_string(), "stub-token".to_string()),
("selected_items".to_string(), "claude/smoke__ci".to_string()),
]),
None,
)
.unwrap();
(plugin, server, requests)
}
#[tokio::test]
async fn repositories_without_workflows_are_dropped_from_a_selection() {
let (plugin, server) = stub_gitea().await;
let kept = plugin
.ids_with_ci(
["claude/smoke__ci", "claude/empty", "claude/missing"]
.map(String::from)
.to_vec(),
)
.await
.unwrap();
server.abort();
assert_eq!(kept, ["claude/smoke__ci", "claude/missing"]);
}
#[tokio::test]
async fn pipelines_find_latest_run_of_each_workflow_beyond_first_page() {
let (plugin, server) = stub_gitea().await;
let pipelines = plugin.fetch_pipelines().await.unwrap();
let summary: Vec<_> = pipelines
.iter()
.map(|p| (p.id.as_str(), p.status))
.collect();
assert_eq!(
summary,
[
(
"gitea__1__claude/smoke__ci/build__test.yaml",
PipelineStatus::Running
),
(
"gitea__1__claude/smoke__ci/lint.yaml",
PipelineStatus::Failed
),
]
);
server.abort();
}
#[tokio::test]
async fn one_failing_repository_fails_the_whole_fetch() {
let (mut plugin, server) = stub_gitea().await;
plugin.config.insert(
"selected_items".to_string(),
"claude/smoke__ci,claude/missing".to_string(),
);
let error = plugin.fetch_pipelines().await.unwrap_err();
assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
server.abort();
}
#[tokio::test]
async fn workflow_parameters_include_dispatch_inputs_after_ref() {
let (plugin, server) = stub_gitea().await;
let params = plugin
.fetch_workflow_parameters("gitea__1__claude/smoke__ci/build__test.yaml")
.await
.unwrap();
assert_eq!(
params.iter().map(|p| p.name.as_str()).collect::<Vec<_>>(),
["ref", "inputs.greeting"]
);
assert_eq!(params[1].label.as_deref(), Some("greeting"));
server.abort();
}
#[tokio::test]
async fn run_history_filters_by_workflow_and_respects_limit() {
let (plugin, server) = stub_gitea().await;
let runs = plugin
.fetch_run_history("gitea__1__claude/smoke__ci/build__test.yaml", 1)
.await
.unwrap();
assert_eq!(runs.iter().map(|r| r.run_number).collect::<Vec<_>>(), [100]);
let details = plugin
.fetch_run_details("gitea__1__claude/smoke__ci/build__test.yaml", 1)
.await
.unwrap();
assert_eq!(details.status, PipelineStatus::Success);
assert!(plugin
.fetch_run_details("gitea__1__claude/smoke__ci/build__test.yaml", 3)
.await
.is_err());
server.abort();
}
#[tokio::test]
async fn cancel_reports_not_supported_when_endpoint_is_missing() {
let (plugin, server) = stub_gitea().await;
let error = plugin
.cancel_run("gitea__1__claude/smoke__ci/build__test.yaml", 2)
.await
.unwrap_err();
assert!(matches!(error, PluginError::NotSupported(_)), "{error}");
server.abort();
}
#[tokio::test]
async fn dispatch_is_not_resent_after_a_server_error() {
let _ = rustls::crypto::ring::default_provider().install_default();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let counter = requests.clone();
let server = tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.unwrap();
let mut buffer = [0; 4096];
let _ = stream.read(&mut buffer).await.unwrap();
counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
stream
.write_all(b"HTTP/1.1 503 X\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
.await
.unwrap();
}
});
let mut plugin = GiteaPlugin::new();
plugin
.initialize(
1,
HashMap::from([
("base_url".to_string(), format!("http://{addr}/")),
("token".to_string(), "stub-token".to_string()),
]),
None,
)
.unwrap();
let error = plugin
.client()
.unwrap()
.dispatch_workflow("o", "r", "w.yaml", &serde_json::json!({}))
.await
.unwrap_err();
server.abort();
assert!(
matches!(error, PluginError::ServiceUnavailable { .. }),
"{error}"
);
assert_eq!(requests.load(std::sync::atomic::Ordering::SeqCst), 1);
}
#[tokio::test]
async fn repo_picker_pages_reach_every_repo_when_gitea_caps_the_limit() {
let (plugin, server) = stub_gitea().await;
let mut names = Vec::new();
let mut has_more = true;
let mut page = 1;
while has_more {
let response = plugin
.fetch_available_pipelines_filtered(
Some("7".into()),
None,
Some(PaginationParams {
page,
page_size: 100,
}),
)
.await
.unwrap();
names.extend(response.items.into_iter().map(|item| item.name));
has_more = response.has_more;
page += 1;
}
server.abort();
let expected: Vec<_> = (0..STUB_REPOS).map(|i| format!("org/r{i}")).collect();
assert_eq!(names, expected);
}
#[tokio::test]
async fn organizations_include_every_page() {
let (plugin, server) = stub_gitea().await;
let orgs = plugin.fetch_organizations().await.unwrap();
server.abort();
assert_eq!(orgs.len(), STUB_ORGS + 1);
assert_eq!(orgs.last().unwrap().name, format!("o{}", STUB_ORGS - 1));
}
fn run_metadata() -> HashMap<String, serde_json::Value> {
HashMap::from([("run_id".to_string(), serde_json::json!(7))])
}
fn run_7<'a>(
pipeline_id: &'a str, metadata: &'a HashMap<String, serde_json::Value>,
) -> RunRef<'a> {
RunRef {
pipeline_id,
run_number: 3,
run_id: "gitea-run-7",
metadata,
}
}
const SMOKE_PIPELINE: &str = "gitea__1__claude/smoke__ci/build__test.yaml";
async fn read(plugin: &GiteaPlugin, node_id: &str) -> PluginResult<Vec<u8>> {
let metadata = run_metadata();
let (tx, mut rx) = tokio::sync::mpsc::channel(16);
let mut sink = LogSink::new(tx);
let result = plugin
.read_log(run_7(SMOKE_PIPELINE, &metadata), node_id, None, &mut sink)
.await;
drop(sink);
let mut bytes = Vec::new();
while let Some(chunk) = rx.recv().await {
bytes.extend_from_slice(&chunk);
}
result.map(|_| bytes)
}
#[tokio::test]
async fn run_jobs_and_steps_become_log_nodes_across_pages() {
let (plugin, server, requests) = stub_gitea_counting().await;
let metadata = run_metadata();
let nodes = plugin
.fetch_log_nodes(run_7(SMOKE_PIPELINE, &metadata))
.await
.unwrap();
server.abort();
let shape: Vec<_> = nodes
.iter()
.map(|node| {
(
node.id.as_str(),
node.parent_id.as_deref(),
node.kind,
node.status,
node.log,
)
})
.collect();
assert_eq!(
shape,
[
(
"101",
None,
LogNodeKind::Job,
PipelineStatus::Success,
Some(LogAvailability::Ready)
),
(
"101:0",
Some("101"),
LogNodeKind::Step,
PipelineStatus::Success,
None
),
(
"101:1",
Some("101"),
LogNodeKind::Step,
PipelineStatus::Success,
None
),
(
"102",
None,
LogNodeKind::Job,
PipelineStatus::Running,
Some(LogAvailability::Partial)
),
(
"102:0",
Some("102"),
LogNodeKind::Step,
PipelineStatus::Running,
None
),
(
"103",
None,
LogNodeKind::Job,
PipelineStatus::Running,
Some(LogAvailability::Pending)
),
("104", None, LogNodeKind::Job, PipelineStatus::Skipped, None),
]
);
assert_eq!(nodes[0].name, "build");
assert_eq!(nodes[2].name, "Run tests");
assert_eq!(
nodes[0].web_url.as_deref(),
Some("http://g/claude/smoke__ci/actions/runs/7/jobs/0")
);
assert_eq!(requests.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn job_log_streams_the_text_gitea_stores() {
let (plugin, server) = stub_gitea().await;
let bytes = read(&plugin, "101").await.unwrap();
server.abort();
assert_eq!(bytes, JOB_LOG.as_bytes());
}
#[tokio::test]
async fn stored_log_lines_reach_the_dialect_as_sections() {
let (plugin, server) = stub_gitea().await;
let bytes = read(&plugin, "101").await.unwrap();
server.abort();
let markers: Vec<_> = bytes
.split(|b| *b == b'\n')
.filter(|line| !line.is_empty())
.map(|line| plugin.log_dialect().parse(line).marker)
.collect();
assert_eq!(
markers,
[
LineMarker::GroupStart,
LineMarker::Plain,
LineMarker::GroupEnd
]
);
}
#[tokio::test]
async fn job_that_never_started_has_no_log() {
let (plugin, server) = stub_gitea().await;
let error = read(&plugin, "104").await.unwrap_err();
server.abort();
assert!(
matches!(error, PluginError::PipelineNotFound(_)),
"{error:?}"
);
}
#[tokio::test]
async fn node_ids_that_are_not_job_numbers_are_rejected_without_a_request() {
let (plugin, server, requests) = stub_gitea_counting().await;
for node_id in [
"",
"101:1",
"../runs/7",
"101/logs",
"x?y=1",
"+101",
"-1",
"1 2",
] {
let error = read(&plugin, node_id).await.unwrap_err();
assert!(
matches!(error, PluginError::InvalidConfig(_)),
"{node_id}: {error:?}"
);
}
server.abort();
assert_eq!(requests.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn run_without_a_gitea_run_id_is_rejected_without_a_request() {
let (plugin, server, requests) = stub_gitea_counting().await;
let metadata = HashMap::new();
let error = plugin
.fetch_log_nodes(run_7(SMOKE_PIPELINE, &metadata))
.await
.unwrap_err();
server.abort();
assert!(matches!(error, PluginError::InvalidConfig(_)), "{error:?}");
assert_eq!(requests.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn gitea_without_the_jobs_route_reports_the_minimum_version() {
let (plugin, server) = stub_gitea().await;
let metadata = run_metadata();
let error = plugin
.fetch_log_nodes(run_7("gitea__1__claude/old/ci.yaml", &metadata))
.await
.unwrap_err();
server.abort();
let PluginError::NotSupported(message) = error else {
panic!("expected NotSupported, got {error:?}");
};
assert!(message.contains("Gitea 1.25 or newer"), "{message}");
}
#[derive(Clone, Copy)]
enum RangeSupport {
Honored,
Ignored,
Misaligned,
}
const STARTED_AT: &str = "2026-10-28T10:00:00Z";
struct Log {
body: Vec<u8>,
range_support: RangeSupport,
ranges: Vec<Option<String>>,
run_attempt: i64,
started_at: &'static str,
}
type SharedLog = Arc<std::sync::Mutex<Log>>;
async fn stub_log(
body: &[u8], range_support: RangeSupport,
) -> (GiteaPlugin, SharedLog, tokio::task::JoinHandle<()>) {
let _ = rustls::crypto::ring::default_provider().install_default();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let log = SharedLog::new(std::sync::Mutex::new(Log {
body: body.to_vec(),
range_support,
ranges: Vec::new(),
run_attempt: 1,
started_at: STARTED_AT,
}));
let state = log.clone();
let server = tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.unwrap();
let mut request = Vec::new();
let mut buffer = [0; 4096];
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
let count = stream.read(&mut buffer).await.unwrap();
if count == 0 {
break;
}
request.extend_from_slice(&buffer[..count]);
}
let request = String::from_utf8(request).unwrap();
let range = request.lines().find_map(|line| {
let (name, value) = line.split_once(':')?;
name.eq_ignore_ascii_case("range")
.then(|| value.trim().to_string())
});
let target = request.split_whitespace().nth(1).unwrap().to_string();
let (status, content_range, payload) = {
let mut log = state.lock().unwrap();
if target.ends_with("/logs") {
log.ranges.push(range.clone());
let total = log.body.len();
let start = range.as_deref().and_then(|value| {
value
.strip_prefix("bytes=")?
.strip_suffix('-')?
.parse()
.ok()
});
match (start, log.range_support) {
(Some(start), RangeSupport::Honored) if start >= total => (
"416 Range Not Satisfiable",
format!("Content-Range: bytes */{total}\r\n"),
Vec::new(),
),
(Some(start), RangeSupport::Honored) => (
"206 Partial Content",
format!("Content-Range: bytes {start}-{}/{total}\r\n", total - 1),
log.body[start..].to_vec(),
),
(Some(_), RangeSupport::Misaligned) => (
"206 Partial Content",
format!("Content-Range: bytes 0-{}/{total}\r\n", total - 1),
log.body.clone(),
),
_ => ("200 OK", String::new(), log.body.clone()),
}
} else {
let job = serde_json::json!({
"id": 101,
"name": "build",
"status": "in_progress",
"run_attempt": log.run_attempt,
"started_at": log.started_at
});
("200 OK", String::new(), job.to_string().into_bytes())
}
};
let head = format!(
"HTTP/1.1 {status}\r\nContent-Type: text/plain\r\n{content_range}Content-Length: {}\r\nConnection: close\r\n\r\n",
payload.len()
);
stream.write_all(head.as_bytes()).await.unwrap();
stream.write_all(&payload).await.unwrap();
}
});
let mut plugin = GiteaPlugin::new();
plugin
.initialize(
1,
HashMap::from([
("base_url".to_string(), format!("http://{addr}/")),
("token".to_string(), "stub-token".to_string()),
]),
None,
)
.unwrap();
(plugin, log, server)
}
fn cursor(job: &str, offset: u64) -> LogCursor {
cursor_of(job, 1, STARTED_AT, offset)
}
fn cursor_of(job: &str, run: i64, started_at: &str, offset: u64) -> LogCursor {
LogCursor::new(&serde_json::json!({
"job": job,
"attempt": {"run": run, "started_at": started_at},
"offset": offset
}))
.unwrap()
}
async fn read_log(
plugin: &GiteaPlugin, from: Option<&LogCursor>,
) -> (PluginResult<LogRead>, Vec<u8>, bool) {
let metadata = run_metadata();
let (tx, mut rx) = tokio::sync::mpsc::channel(16);
let mut sink = LogSink::new(tx);
let result = plugin
.read_log(run_7(SMOKE_PIPELINE, &metadata), "101", from, &mut sink)
.await;
let restarted = sink.restarted();
drop(sink);
let mut bytes = Vec::new();
while let Some(chunk) = rx.recv().await {
bytes.extend_from_slice(&chunk);
}
(result, bytes, restarted)
}
fn ranges(log: &SharedLog) -> Vec<Option<String>> {
log.lock().unwrap().ranges.clone()
}
#[tokio::test]
async fn resumed_log_appends_only_the_new_bytes() {
let (plugin, log, server) = stub_log(b"a\nb\n", RangeSupport::Honored).await;
let (first, bytes, restarted) = read_log(&plugin, None).await;
assert_eq!(first.unwrap(), LogRead::replace(Some(cursor("101", 4))));
assert_eq!(bytes, b"a\nb\n");
assert!(!restarted);
log.lock().unwrap().body.extend_from_slice(b"c\n");
let (second, bytes, restarted) = read_log(&plugin, Some(&cursor("101", 4))).await;
assert_eq!(second.unwrap(), LogRead::append(cursor("101", 6)));
assert_eq!(bytes, b"c\n");
assert!(!restarted);
let (idle, bytes, restarted) = read_log(&plugin, Some(&cursor("101", 6))).await;
server.abort();
assert_eq!(idle.unwrap(), LogRead::append(cursor("101", 6)));
assert!(bytes.is_empty() && !restarted);
assert_eq!(
ranges(&log),
[None, Some("bytes=4-".into()), Some("bytes=6-".into())]
);
}
#[tokio::test]
async fn server_ignoring_range_restarts_with_the_whole_log() {
let (plugin, log, server) = stub_log(b"a\nb\nc\n", RangeSupport::Ignored).await;
let (read, bytes, restarted) = read_log(&plugin, Some(&cursor("101", 4))).await;
server.abort();
assert_eq!(read.unwrap(), LogRead::replace(Some(cursor("101", 6))));
assert_eq!(bytes, b"a\nb\nc\n");
assert!(restarted);
assert_eq!(ranges(&log), [Some("bytes=4-".into())]);
}
#[tokio::test]
async fn partial_reply_starting_elsewhere_restarts_with_the_whole_log() {
let (plugin, log, server) = stub_log(b"a\nb\nc\n", RangeSupport::Misaligned).await;
let (read, bytes, restarted) = read_log(&plugin, Some(&cursor("101", 4))).await;
server.abort();
assert_eq!(read.unwrap(), LogRead::replace(Some(cursor("101", 6))));
assert_eq!(bytes, b"a\nb\nc\n");
assert!(restarted);
assert_eq!(ranges(&log), [Some("bytes=4-".into()), None]);
}
#[tokio::test]
async fn shrunk_log_restarts_with_the_whole_log() {
let (plugin, log, server) = stub_log(b"a\n", RangeSupport::Honored).await;
let (read, bytes, restarted) = read_log(&plugin, Some(&cursor("101", 6))).await;
server.abort();
assert_eq!(read.unwrap(), LogRead::replace(Some(cursor("101", 2))));
assert_eq!(bytes, b"a\n");
assert!(restarted);
assert_eq!(ranges(&log), [Some("bytes=6-".into()), None]);
}
#[tokio::test]
async fn unusable_cursors_restart_without_a_range_request() {
let (plugin, log, server) = stub_log(b"a\nb\n", RangeSupport::Honored).await;
for stale in [LogCursor::new(&"garbage").unwrap(), cursor("102", 4)] {
let (read, bytes, restarted) = read_log(&plugin, Some(&stale)).await;
assert_eq!(read.unwrap(), LogRead::replace(Some(cursor("101", 4))));
assert_eq!(bytes, b"a\nb\n");
assert!(restarted);
}
server.abort();
assert_eq!(ranges(&log), [None, None]);
}
#[tokio::test]
async fn rerun_keeping_the_job_id_restarts_even_when_its_log_is_longer_than_the_cursor() {
let (plugin, log, server) = stub_log(b"a\nb\n", RangeSupport::Honored).await;
let (first, _, _) = read_log(&plugin, None).await;
let first = first.unwrap();
assert_eq!(first, LogRead::replace(Some(cursor("101", 4))));
for (run, started_at) in [(2, STARTED_AT), (1, "2026-10-28T11:00:00Z")] {
{
let mut log = log.lock().unwrap();
log.run_attempt = run;
log.started_at = started_at;
log.body = b"x\ny\nz\nw\n".to_vec();
}
let (read, bytes, restarted) = read_log(&plugin, first.cursor.as_ref()).await;
assert_eq!(
read.unwrap(),
LogRead::replace(Some(cursor_of("101", run, started_at, 8)))
);
assert_eq!(bytes, b"x\ny\nz\nw\n");
assert!(restarted);
}
server.abort();
assert_eq!(ranges(&log), [None, None, None]);
}
}