use std::{collections::BTreeMap, env};
use onetaskgraph_live::artifact::{Run, Sweep, now_micros};
use onetaskgraph_live::{Credential, Exclusivity, Session, missing, required};
use onetaskgraph_plugin_api::{
Capabilities, DependencyEdge, DependencyEndpoint, DependencyKind, DependencySupport, Direction,
Document, DocumentQuery, ItemKind, ItemWrite, Label, LabelFilter, Location, NativeId,
PageRequest, Project, ProjectFilter, ProjectQuery, SecretResolver, SourceName, SourcePlugin,
Status, StatusCategory, Support, Task, TaskQuery, TaskSource, TextFields, TextQuery,
};
use secrecy::SecretString;
use serde_json::{Value, json};
struct Environment;
impl SecretResolver for Environment {
fn get(&self, var: &str) -> Option<SecretString> {
env::var(var).ok().map(SecretString::from)
}
}
macro_rules! ensure {
($condition:expr, $($message:tt)+) => {
if !$condition {
return Err(format!($($message)+));
}
};
}
#[allow(dead_code)]
mod cleanup;
use cleanup::{
ISSUE_PAGE_PROBE, LABEL_CREATE, PROJECT_STATUSES, SESSION_NAME, TEAM_STATES, artifact_label,
artifact_title, is_this_runs, linear, remove_artifacts, run_then_cleanup, sweep_orphans,
};
mod settle;
use settle::{
LINEAR_DOCUMENT_LISTING, LINEAR_INDEX, settled, settled_document_absent, settled_documents,
settled_label, settled_tasks, settled_walk, task_titles, walked_task_titles,
};
async fn fixture_states(key: &str, team_key: &str) -> Result<(String, String), String> {
let data = linear(
key,
TEAM_STATES,
json!({"key":team_key}),
"live workflow state discovery",
)
.await?;
let states = data
.pointer("/teams/nodes/0/states/nodes")
.and_then(Value::as_array)
.ok_or_else(|| {
format!("LINEAR_WRITE_TEAM={team_key} names no team this credential can see")
})?;
let named = |wanted: &str| {
states
.iter()
.find(|state| state.get("type").and_then(Value::as_str) == Some(wanted))
.and_then(|state| state.get("name").and_then(Value::as_str))
.map(str::to_owned)
};
match (named("unstarted"), named("completed")) {
(Some(open), Some(done)) => Ok((open, done)),
_ => Err(format!(
"team {team_key} has no workflow state of type unstarted and one of type \
completed, which this lane needs to file two issues a status filter can \
separate; add them to that team or point LINEAR_WRITE_TEAM at a scratch team \
that has them"
)),
}
}
async fn fixture_project_status(key: &str) -> Result<String, String> {
let data = linear(
key,
PROJECT_STATUSES,
json!({}),
"live project status discovery",
)
.await?;
let statuses = data
.pointer("/projectStatuses/nodes")
.and_then(Value::as_array)
.ok_or_else(|| "live project status discovery returned no statuses".to_owned())?;
let names = statuses
.iter()
.filter_map(|status| status.get("name").and_then(Value::as_str))
.collect::<Vec<_>>();
names
.iter()
.find(|name| {
names
.iter()
.filter(|other| other.eq_ignore_ascii_case(name))
.count()
== 1
})
.map(|name| (*name).to_owned())
.ok_or_else(|| {
"this workspace has no project status name that resolves uniquely, and the \
source refuses one two statuses answer to"
.to_owned()
})
}
async fn create_label(key: &str, team: &str, name: &str) -> Result<(), String> {
let data = linear(
key,
LABEL_CREATE,
json!({"input":{"name":name,"teamId":team,"color":"#bec2c8"}}),
"live label creation",
)
.await?;
if data
.pointer("/issueLabelCreate/issueLabel/name")
.and_then(Value::as_str)
!= Some(name)
{
return Err(format!(
"Linear did not confirm creating the label {name:?}"
));
}
Ok(())
}
fn label(name: &str) -> Label {
Label {
id: NativeId("live-source-label".into()),
name: name.to_owned(),
color: None,
}
}
fn blocks(far: &NativeId, kind: ItemKind) -> DependencyEdge {
DependencyEdge {
from: DependencyEndpoint::from_native(NativeId("live-source-item".into()), kind),
to: DependencyEndpoint::from_native(far.clone(), kind),
kind: DependencyKind::Blocks,
}
}
fn page(limit: u32) -> PageRequest {
PageRequest {
cursor: None,
limit,
}
}
fn sorted(mut titles: Vec<String>) -> Vec<String> {
titles.sort();
titles
}
struct LiveRun {
key: String,
id: Run,
stamp_micros: u64,
open_state: String,
done_state: String,
project_status: String,
}
async fn drive_every_declared_capability(
run: &LiveRun,
source: &dyn TaskSource,
team_id: &str,
) -> Result<(), String> {
let title = |offset: u64| artifact_title(run.id, run.stamp_micros + offset);
let (alpha, beta) = (title(0), title(1));
let (first, second, orphan) = (title(2), title(3), title(4));
let run_label = artifact_label(run.id, run.stamp_micros);
let only_label = artifact_label(run.id, run.stamp_micros + 1);
create_label(&run.key, team_id, &run_label).await?;
create_label(&run.key, team_id, &only_label).await?;
for name in [&run_label, &only_label] {
settled_label(LINEAR_INDEX, name, |query, variables| {
linear(&run.key, query, variables, "live label lookup")
})
.await?;
}
let open = Status {
category: StatusCategory::Todo,
name: run.open_state.clone(),
};
let done = Status {
category: StatusCategory::Done,
name: run.done_state.clone(),
};
let project_status = Status {
category: StatusCategory::Todo,
name: run.project_status.clone(),
};
let scoped = || TaskQuery {
labels: LabelFilter {
any_of: vec![run_label.clone()],
..LabelFilter::default()
},
..TaskQuery::default()
};
let project = |name: &str| Project {
id: NativeId("live-source-item".into()),
title: name.to_owned(),
content: Some("temporary credentialed write; the live lane removes this".into()),
status: project_status.clone(),
labels: vec![],
url: None,
location: None,
created_at: None,
updated_at: None,
metadata: BTreeMap::new(),
repositories: vec![],
};
let task = |name: &str, status: &Status, under: Option<&NativeId>, labels: Vec<Label>| Task {
id: NativeId("live-source-item".into()),
title: name.to_owned(),
content: Some("temporary credentialed write; the live lane removes this".into()),
status: status.clone(),
labels,
project: under.cloned(),
url: None,
location: None,
created_at: None,
updated_at: None,
metadata: BTreeMap::new(),
repositories: vec![],
delivers: Vec::new(),
delivered_by: Vec::new(),
};
let alpha_id = source
.write_project(&ItemWrite {
target: None,
item: project(&alpha),
depends_on: vec![],
})
.await
.map_err(|error| format!("live project write of {alpha:?} failed: {error}"))?;
let beta_id = source
.write_project(&ItemWrite {
target: None,
item: project(&beta),
depends_on: vec![blocks(&alpha_id, ItemKind::Project)],
})
.await
.map_err(|error| format!("live project write of {beta:?} failed: {error}"))?;
let first_id = source
.write_task(&ItemWrite {
target: None,
item: task(
&first,
&open,
Some(&alpha_id),
vec![label(&run_label), label(&only_label)],
),
depends_on: vec![],
})
.await
.map_err(|error| format!("live task write of {first:?} failed: {error}"))?;
let second_id = source
.write_task(&ItemWrite {
target: None,
item: task(&second, &open, Some(&beta_id), vec![label(&run_label)]),
depends_on: vec![blocks(&first_id, ItemKind::Task)],
})
.await
.map_err(|error| format!("live task write of {second:?} failed: {error}"))?;
let orphan_id = source
.write_task(&ItemWrite {
target: None,
item: task(&orphan, &done, None, vec![label(&run_label)]),
depends_on: vec![],
})
.await
.map_err(|error| format!("live task write of {orphan:?} failed: {error}"))?;
let all_three = sorted(vec![first.clone(), second.clone(), orphan.clone()]);
settled_tasks(
LINEAR_INDEX,
source,
&scoped(),
&format!("the three issues this run created, labelled {run_label},"),
&all_three,
)
.await?;
let mut listed = Vec::new();
let mut cursor = None;
loop {
let step = source
.query_projects(
&ProjectQuery::default(),
&PageRequest {
cursor,
limit: onetaskgraph_linear::MAX_PAGE_SIZE,
},
)
.await
.map_err(|error| format!("live project listing failed: {error}"))?;
listed.extend(step.items.into_iter().map(|project| project.title));
cursor = step.next;
if cursor.is_none() {
break;
}
ensure!(listed.len() < 100_000, "the project walk must terminate");
}
ensure!(
listed.contains(&alpha) && listed.contains(&beta),
"the two projects this run created are not in Linear's own project listing"
);
let under = |id: &NativeId| TaskQuery {
project: ProjectFilter::Is(id.clone()),
..TaskQuery::default()
};
settled_tasks(
LINEAR_INDEX,
source,
&under(&alpha_id),
"the issues of one of this run's two projects",
std::slice::from_ref(&first),
)
.await?;
settled_tasks(
LINEAR_INDEX,
source,
&under(&beta_id),
"the issues of the other of this run's two projects",
std::slice::from_ref(&second),
)
.await?;
settled_tasks(
LINEAR_INDEX,
source,
&TaskQuery {
project: ProjectFilter::Orphans,
..scoped()
},
"this run's issues belonging to no project",
std::slice::from_ref(&orphan),
)
.await?;
settled_tasks(
LINEAR_INDEX,
source,
&TaskQuery {
labels: LabelFilter {
any_of: vec![only_label.clone()],
..LabelFilter::default()
},
..TaskQuery::default()
},
"this run's issues carrying its second label",
std::slice::from_ref(&first),
)
.await?;
settled_tasks(
LINEAR_INDEX,
source,
&TaskQuery {
labels: LabelFilter {
any_of: vec![run_label.clone()],
none_of: vec![only_label.clone()],
..LabelFilter::default()
},
..TaskQuery::default()
},
"this run's issues not carrying its second label",
&[second.clone(), orphan.clone()],
)
.await?;
settled_tasks(
LINEAR_INDEX,
source,
&TaskQuery {
statuses: vec![StatusCategory::Todo],
..scoped()
},
"this run's unstarted issues",
&[first.clone(), second.clone()],
)
.await?;
settled_tasks(
LINEAR_INDEX,
source,
&TaskQuery {
statuses: vec![StatusCategory::Done],
..scoped()
},
"this run's completed issue",
std::slice::from_ref(&orphan),
)
.await?;
for fields in [
TextFields::Title,
TextFields::Content,
TextFields::TitleOrContent,
] {
settled_tasks(
LINEAR_INDEX,
source,
&TaskQuery {
text: Some(TextQuery {
terms: first.clone(),
fields,
}),
..scoped()
},
&format!(
"a {fields:?} search this source declares unsupported, which must return the \
wider set,"
),
&all_three,
)
.await?;
}
let task_edge = DependencyEdge {
from: DependencyEndpoint::from_native(second_id.clone(), ItemKind::Task),
to: DependencyEndpoint::from_native(first_id.clone(), ItemKind::Task),
kind: DependencyKind::Blocks,
};
let project_edge = DependencyEdge {
from: DependencyEndpoint::from_native(beta_id.clone(), ItemKind::Project),
to: DependencyEndpoint::from_native(alpha_id.clone(), ItemKind::Project),
kind: DependencyKind::Blocks,
};
for (near, direction, expected, level) in [
(&second_id, Direction::DependsOn, &task_edge, "task"),
(&first_id, Direction::DependedOnBy, &task_edge, "task"),
(&beta_id, Direction::DependsOn, &project_edge, "project"),
(&alpha_id, Direction::DependedOnBy, &project_edge, "project"),
] {
let read = if level == "task" {
source.task_dependencies(near, direction, &page(50)).await
} else {
source
.project_dependencies(near, direction, &page(50))
.await
}
.map_err(|error| format!("live {level} {direction:?} dependency read failed: {error}"))?;
ensure!(
read.items == vec![expected.clone()],
"the {level} {direction:?} read of {} returned {:?}",
near.0,
read.items
);
}
let walk = "a walk in pages of one over this run's own three issues";
settled_walk(LINEAR_INDEX, source, &scoped(), 10, walk, &all_three).await?;
let whole = source
.query_tasks(&scoped(), &page(50))
.await
.map_err(|error| format!("live whole-page read failed: {error}"))?
.items
.into_iter()
.map(|task| task.title)
.collect::<Vec<_>>();
let walked = walked_task_titles(source, &scoped(), 10, walk).await?;
ensure!(
walked == whole,
"a walk in pages of one reached {walked:?} where one whole page reports {whole:?}"
);
let ceiling = source.capabilities().max_page_size;
let over_the_ceiling = scoped();
let what = "a read at a limit one above the declared ceiling, which this source clamps \
rather than refuses,";
settled(LINEAR_INDEX, what, &all_three, || {
task_titles(source, &over_the_ceiling, ceiling + 1, what)
})
.await?;
linear(
&run.key,
ISSUE_PAGE_PROBE,
json!({"first":ceiling}),
"page size probe at the declared maximum",
)
.await?;
let written = source
.get_task(&first_id)
.await
.map_err(|error| format!("live task read-back failed: {error}"))?
.ok_or_else(|| "the written issue was not readable by its own id".to_owned())?;
ensure!(
written.title == first && written.status.category == StatusCategory::Todo,
"the live write did not round-trip its title and status: {written:?}"
);
let closed_back = source
.get_task(&orphan_id)
.await
.map_err(|error| format!("live completed-issue read-back failed: {error}"))?
.ok_or_else(|| "the completed issue was not readable by its own id".to_owned())?;
ensure!(
closed_back.status.category == StatusCategory::Done,
"the live write filed under done read back as {:?}",
closed_back.status.category
);
drive_documents(run, source, &alpha_id, &title(5), &title(6)).await
}
async fn drive_documents(
run: &LiveRun,
source: &dyn TaskSource,
under: &NativeId,
filed: &str,
loose: &str,
) -> Result<(), String> {
let document = |title: &str, project: Option<&NativeId>| Document {
id: NativeId("live-source-item".into()),
title: title.to_owned(),
content: Some("temporary credentialed write; the live lane removes this".into()),
project: project.cloned(),
labels: vec![],
url: None,
location: None,
created_at: None,
updated_at: None,
metadata: [("caller.count".to_owned(), json!(3))]
.into_iter()
.collect(),
repositories: vec![],
};
let write = |item: Document| ItemWrite {
target: None,
item,
depends_on: vec![],
};
let filed_id = source
.write_document(&write(document(filed, Some(under))))
.await
.map_err(|error| format!("live document write of {filed:?} failed: {error}"))?;
let loose_id = source
.write_document(&write(document(loose, None)))
.await
.map_err(|error| format!("live document write of {loose:?} failed: {error}"))?;
let read = source
.get_document(&filed_id)
.await
.map_err(|error| format!("live document read-back failed: {error}"))?
.ok_or_else(|| "the written document was not readable by its own id".to_owned())?;
ensure!(
read.title == filed,
"the live document write did not round-trip its title: {read:?}"
);
ensure!(
read.content.as_deref() == Some("temporary credentialed write; the live lane removes this"),
"the visible body a read reports is the text a person wrote: {:?}",
read.content
);
ensure!(
read.metadata.get("caller.count") == Some(&json!(3)),
"a caller's key did not round-trip with its JSON type: {:?}",
read.metadata
);
ensure!(
read.labels.is_empty(),
"Linear's own document type has no labels, so a read reports none: {:?}",
read.labels
);
ensure!(
matches!(&read.location, Some(Location::Url(url)) if url.starts_with("https://")),
"a document says where it is, as a link: {:?}",
read.location
);
ensure!(
read.project.as_ref() == Some(under),
"the document filed under this run's project read back filed under {:?}",
read.project
);
let refusal = source
.write_document(&write(Document {
labels: vec![label(&artifact_label(run.id, run.stamp_micros))],
..document(filed, Some(under))
}))
.await;
ensure!(
refusal
.as_ref()
.err()
.is_some_and(|error| error.to_string().contains("labels")),
"a document write carrying a label must be refused by name, and was {refusal:?}"
);
settled_documents(
LINEAR_INDEX,
LINEAR_DOCUMENT_LISTING,
source,
&DocumentQuery::default(),
&[filed_id.clone(), loose_id.clone()],
"the two documents this run created",
&[
(filed_id.clone(), filed.to_owned()),
(loose_id.clone(), loose.to_owned()),
],
)
.await?;
settled_documents(
LINEAR_INDEX,
LINEAR_DOCUMENT_LISTING,
source,
&DocumentQuery {
project: ProjectFilter::Is(under.clone()),
..DocumentQuery::default()
},
&[filed_id.clone(), loose_id.clone()],
"a document listing narrowed to this run's project",
&[(filed_id.clone(), filed.to_owned())],
)
.await?;
settled_documents(
LINEAR_INDEX,
LINEAR_DOCUMENT_LISTING,
source,
&DocumentQuery {
project: ProjectFilter::Orphans,
..DocumentQuery::default()
},
&[filed_id.clone(), loose_id.clone()],
"a document listing narrowed to the orphans",
&[(loose_id.clone(), loose.to_owned())],
)
.await?;
settled_documents(
LINEAR_INDEX,
LINEAR_DOCUMENT_LISTING,
source,
&DocumentQuery {
labels: LabelFilter {
any_of: vec![artifact_label(run.id, run.stamp_micros)],
..LabelFilter::default()
},
..DocumentQuery::default()
},
&[filed_id.clone(), loose_id.clone()],
"a document listing demanding a label, which no Linear document carries,",
&[],
)
.await?;
for (id, title) in [(&filed_id, filed), (&loose_id, loose)] {
source
.delete_document(id)
.await
.map_err(|error| format!("live document removal failed: {error}"))?;
settled_document_absent(LINEAR_INDEX, source, id, title).await?;
}
Ok(())
}
#[tokio::test]
async fn real_linear_applies_every_declared_capability_and_leaves_no_residue() {
let live_required = required(
env::var(onetaskgraph_live::REQUIRED_VARIABLE)
.ok()
.as_deref(),
)
.unwrap_or_else(|error| panic!("the Linear live lane cannot run: {error}"));
let skip = |reason: &str| -> Option<String> {
match missing(live_required, SESSION_NAME, reason) {
Ok(reason) => {
eprintln!("skipped live Linear journey: {reason}");
None
}
Err(error) => panic!("the Linear live lane cannot run: {error}"),
}
};
let Some(key) = env::var("LINEAR_API_KEY").ok().and_then(Credential::new) else {
skip("LINEAR_API_KEY is not set");
return;
};
let Some(team) = env::var("LINEAR_WRITE_TEAM")
.ok()
.filter(|value| !value.trim().is_empty())
else {
skip(
"LINEAR_WRITE_TEAM is not set, and this lane writes only to the scratch team that \
name gives rather than discovering one; no mutation was sent. Set it to that \
team's key — in CI it is the LINEAR_WRITE_TEAM repository variable, and locally \
it is an environment variable",
);
return;
};
let session = Session::open(SESSION_NAME, key, Exclusivity::OneAtATime)
.unwrap_or_else(|declined| declined.refuse());
let key = session.credential().expose().to_owned();
let source = onetaskgraph_linear::Plugin
.build(
&SourceName::new("live").unwrap(),
&json!({"team":team}),
&Environment,
)
.unwrap_or_else(|error| panic!("the Linear live lane cannot use this team: {error}"));
assert!(source.health().await.unwrap().reachable);
assert_eq!(
source.capabilities(),
Capabilities {
projects: Support::Native,
documents: Support::Native,
comments: Support::Native,
orphan_tasks: Support::Native,
filter_by_label: Support::Native,
filter_by_status: Support::Native,
search_title: Support::Unsupported,
search_content: Support::Unsupported,
task_dependencies: DependencySupport::BothDirections,
project_dependencies: DependencySupport::BothDirections,
max_page_size: onetaskgraph_linear::MAX_PAGE_SIZE,
}
);
let team_id = linear(&key, TEAM_STATES, json!({"key":team}), "live team lookup")
.await
.and_then(|data| {
data.pointer("/teams/nodes/0/id")
.and_then(Value::as_str)
.map(str::to_owned)
.ok_or_else(|| {
format!("LINEAR_WRITE_TEAM={team} names no team this credential can see")
})
})
.unwrap_or_else(|error| {
panic!("the Linear live lane cannot reach its scratch team: {error}")
});
let (open_state, done_state) = fixture_states(&key, &team)
.await
.unwrap_or_else(|error| panic!("the Linear live lane cannot file its fixture: {error}"));
let project_status = fixture_project_status(&key)
.await
.unwrap_or_else(|error| panic!("the Linear live lane cannot file its projects: {error}"));
let run = LiveRun {
key: key.clone(),
id: Run::current(),
stamp_micros: now_micros(),
open_state,
done_state,
project_status,
};
let id = run.id;
run_then_cleanup(
|| drive_every_declared_capability(&run, source.as_ref(), &team_id),
|| async {
let mine = remove_artifacts(&key, &|prefix, name| is_this_runs(id, prefix, name)).await;
let orphans = sweep_orphans(&key, &Sweep::of(id, now_micros())).await;
match (mine, orphans) {
(Ok(()), Ok(())) => Ok(()),
(Err(mine), Ok(())) => Err(mine),
(Ok(()), Err(orphans)) => Err(format!(
"residue left by an earlier interrupted run could not be cleared: {orphans}"
)),
(Err(mine), Err(orphans)) => Err(format!(
"{mine}; additionally, residue left by an earlier interrupted run could \
not be cleared: {orphans}"
)),
}
},
)
.await
.unwrap_or_else(|error| panic!("Linear live capability journey failed: {error}"));
}