use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use chrono::TimeZone;
use pipedash_plugin_api::{
ComponentFilter,
ComponentHealth,
FeedFormat,
RateLimit,
StatusSource,
};
use reqwest::Url;
use serde_json::{
json,
Value,
};
use super::super::feed_subject;
use super::scan::scan;
use super::*;
use crate::application::services::provider_service::ProviderFactory;
use crate::application::test_support::{
pipeline,
statuspage_body,
stub_run,
FeedStub,
Gate,
StubFactory,
StubProvider,
};
use crate::domain::{
DomainError,
HealthLevel,
PipelineRun,
ProviderConfig,
};
use crate::infrastructure::HttpClientManager;
const BODY: &str = "acme/app ยท Provider 1";
fn at(minutes: i64) -> DateTime<Utc> {
Utc.with_ymd_and_hms(2026, 10, 26, 12, 0, 0).unwrap() + chrono::Duration::minutes(minutes)
}
fn run(provider: i64, name: &str, status: PipelineStatus, minute: i64) -> Pipeline {
let mut pipeline = pipeline(provider, name, status);
pipeline.last_run = Some(at(minute));
pipeline
}
fn summary(id: i64) -> ProviderSummary {
serde_json::from_value(json!({
"id": id, "name": format!("Provider {id}"), "provider_type": "github", "icon": null,
"pipeline_count": 0, "last_updated": null, "refresh_interval": 30,
"configured_repositories": [], "last_fetch_status": "success",
"last_fetch_error": null, "last_fetch_at": null, "version": 1
}))
.unwrap()
}
fn updated(provider_id: i64, pipelines: Vec<Pipeline>) -> CoreEvent {
CoreEvent::PipelinesUpdated {
pipelines,
provider_id,
timestamp: 0,
}
}
fn failed_event(pipelines: Vec<Pipeline>) -> CoreEvent {
CoreEvent::PipelinesFailed { pipelines }
}
fn alert_event(key: &str, alert_key: &str, title: &str, recovered: bool) -> CoreEvent {
CoreEvent::ProviderHealthAlert {
alert_id: "alert".to_owned(),
alert_key: alert_key.to_owned(),
notification_key: key.to_owned(),
provider_ids: vec![1],
level: HealthLevel::MajorOutage,
title: title.to_owned(),
body: String::new(),
link: None,
recovered,
}
}
fn titles(found: &[LocalNotification]) -> Vec<&str> {
found.iter().map(|notice| notice.title.as_str()).collect()
}
fn reload(state: &NotificationState) -> NotificationState {
serde_json::from_str(&serde_json::to_string(state).unwrap()).unwrap()
}
fn running_run(pipeline_id: &str, hours: i64) -> PipelineRun {
serde_json::from_value(json!({
"id": format!("{pipeline_id}__9"), "pipeline_id": pipeline_id, "run_number": 9,
"status": "running", "started_at": Utc::now() - chrono::Duration::hours(hours),
"logs_url": ""
}))
.unwrap()
}
fn summary_with(id: i64, signals: Vec<Value>, status_page: bool) -> ProviderSummary {
let page = status_page.then(|| {
json!({
"name": "Example", "url": "https://status.example.com/",
"checked_at": null, "error": null
})
});
let mut value = serde_json::to_value(summary(id)).unwrap();
value["health"] = json!({
"level": "major_outage", "signals": signals, "next_attempt_at": null,
"in_flight": false, "status_page": page
});
serde_json::from_value(value).unwrap()
}
fn status_signal() -> Value {
json!({
"kind": "status_page", "episode_id": "episode-1", "level": "major_outage",
"evidence": "fresh", "observed_at": "2026-10-26T12:00:00Z",
"title": "Incident inc1", "detail": null, "reported_by": "Example", "url": null,
"components": [],
"incidents": [{
"id": "inc1", "name": "Incident inc1", "level": "major_outage",
"phase": "investigating", "latest_update": null, "url": null,
"started_at": null, "updated_at": "2026-10-26T12:00:00Z"
}],
"maintenances": [], "updated_at": "2026-10-26T12:00:00Z"
})
}
fn instance_signal() -> Value {
json!({
"kind": "instance", "episode_id": "episode-2", "level": "major_outage",
"evidence": "fresh", "observed_at": "2026-10-26T12:00:00Z",
"title": "flux: scaled to zero", "detail": null, "reported_by": "Provider 1",
"url": null,
"components": [{"name": "flux", "level": "major_outage", "detail": "scaled to zero"}],
"incidents": [], "maintenances": [], "updated_at": "2026-10-26T12:00:00Z"
})
}
fn broken() -> ComponentHealth {
ComponentHealth {
name: "flux".to_owned(),
level: HealthLevel::MajorOutage,
detail: Some("scaled to zero".to_owned()),
}
}
fn operational() -> Value {
statuspage_body(&[("web", "operational")], json!([]))
}
fn outage() -> Value {
statuspage_body(
&[("web", "major_outage")],
json!([{
"id": "inc1", "name": "Incident inc1", "status": "investigating",
"impact": "critical", "updated_at": "2026-10-22T10:00:00Z",
"components": [{"id": "web"}],
"incident_updates": [{"body": "We are investigating", "created_at": "2026-10-22T10:00:00Z"}]
}]),
)
}
fn source(feed: &FeedStub) -> StatusSource {
StatusSource {
name: "Example".to_owned(),
page_url: Url::parse("https://status.example.com/").unwrap(),
feed_url: feed.feed_url(),
format: FeedFormat::Statuspage,
components: ComponentFilter::All,
}
}
struct Rig {
factory: Arc<StubFactory>,
http: HttpClientManager,
state: NotificationState,
controller: bool,
status_pages: bool,
version: i64,
}
impl Rig {
fn new() -> Self {
Self {
factory: Arc::new(StubFactory::default()),
http: HttpClientManager::new().unwrap(),
state: NotificationState::default(),
controller: false,
status_pages: true,
version: 1,
}
}
fn stub(&self, id: i64, build: impl FnOnce(StubProvider) -> StubProvider) -> Arc<StubProvider> {
self.factory.queue(build(StubProvider::new(id)))
}
fn config(&self, id: i64) -> ProviderConfig {
ProviderConfig {
id: Some(id),
name: format!("Provider {id}"),
provider_type: "github".to_owned(),
token: "secret-token".to_owned(),
config: HashMap::new(),
refresh_interval: 30,
version: Some(self.version),
}
}
async fn scan(&mut self, ids: &[i64]) -> Vec<LocalNotification> {
self.scan_within(ids, Duration::from_secs(20)).await
}
async fn scan_within(&mut self, ids: &[i64], deadline: Duration) -> Vec<LocalNotification> {
let configs: Vec<ProviderConfig> = ids.iter().map(|id| self.config(*id)).collect();
let factory = Arc::clone(&self.factory);
let build: ProviderFactory = Arc::new(move |config| factory.provide(config));
let controller = self.controller;
scan(
&configs,
&build,
&move |_| controller,
&self.http,
self.status_pages,
&mut self.state,
deadline,
)
.await
.unwrap()
}
}
fn fail(stub: &StubProvider, error: DomainError) {
*stub.pipelines.lock().unwrap() = Err(error);
}
fn network() -> DomainError {
DomainError::NetworkError("down".to_owned())
}
#[tokio::test]
async fn a_cold_scan_baselines_silently_and_later_failures_notify() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Failed, 1)])
});
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Failed, 2)]);
let found = rig.scan(&[1]).await;
assert_eq!(
found,
[LocalNotification {
key: "pipeline:github__1__api".to_owned(),
title: "api failed".to_owned(),
body: BODY.to_owned(),
pipeline_id: Some("github__1__api".to_owned()),
provider_id: Some(1),
}]
);
}
#[tokio::test]
async fn rehydrated_state_keeps_the_baseline_and_holds_no_credentials() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Success, 1)])
});
rig.scan(&[1]).await;
rig.state = reload(&rig.state);
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Failed, 2)]);
assert_eq!(titles(&rig.scan(&[1]).await), ["api failed"]);
rig.state = reload(&rig.state);
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Failed, 3)]);
assert_eq!(titles(&rig.scan(&[1]).await), ["api failed"]);
assert!(!serde_json::to_string(&rig.state)
.unwrap()
.contains("secret-token"));
}
#[tokio::test]
async fn an_event_then_a_scan_notify_once() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Success, 1)])
});
rig.scan(&[1]).await;
let failed = run(1, "api", PipelineStatus::Failed, 2);
let first = rig.state.observe_event(&failed_event(vec![failed.clone()]));
let repeated = rig.state.observe_event(&failed_event(vec![failed.clone()]));
stub.set_pipelines(vec![failed]);
assert_eq!(titles(&first), ["api failed"]);
assert!(repeated.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn a_scan_then_an_event_notify_once() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Success, 1)])
});
rig.scan(&[1]).await;
let failed = run(1, "api", PipelineStatus::Failed, 2);
stub.set_pipelines(vec![failed.clone()]);
let scanned = rig.scan(&[1]).await;
assert_eq!(titles(&scanned), ["api failed"]);
assert!(rig
.state
.observe_event(&failed_event(vec![failed.clone()]))
.is_empty());
assert!(rig
.state
.observe_event(&updated(1, vec![failed]))
.is_empty());
}
#[test]
fn updated_and_failed_events_for_one_failure_notify_once() {
let mut state = NotificationState::default();
state.seed(&[run(1, "api", PipelineStatus::Success, 1)], &[summary(1)]);
let failed = run(1, "api", PipelineStatus::Failed, 2);
let first = state.observe_event(&updated(1, vec![failed.clone()]));
let second = state.observe_event(&failed_event(vec![failed]));
assert_eq!(titles(&first), ["api failed"]);
assert_eq!(first[0].body, BODY);
assert!(second.is_empty());
}
#[test]
fn a_degraded_pipeline_notice_says_why() {
let mut state = NotificationState::default();
state.seed(&[run(1, "api", PipelineStatus::Success, 1)], &[summary(1)]);
let mut degraded = run(1, "api", PipelineStatus::Degraded, 1);
degraded.resource_state = Some(crate::domain::ResourceState::new(
"Degraded",
Some("Pod apps/api-7f9c: CrashLoopBackOff"),
));
let found = state.observe_event(&updated(1, vec![degraded]));
assert_eq!(titles(&found), ["api is degraded"]);
assert_eq!(
found[0].body,
format!("{BODY} ยท Pod apps/api-7f9c: CrashLoopBackOff")
);
}
#[test]
fn seeding_is_silent_and_replaces_older_state() {
let failed = run(1, "api", PipelineStatus::Failed, 1);
let mut state = NotificationState::default();
state.seed(&[run(2, "web", PipelineStatus::Success, 1)], &[summary(2)]);
state.seed(std::slice::from_ref(&failed), &[summary(1)]);
assert!(state
.observe_event(&failed_event(vec![failed.clone()]))
.is_empty());
assert!(state.observe_event(&updated(1, vec![failed])).is_empty());
assert!(!state.pipelines.contains_key(&2));
let next = run(1, "api", PipelineStatus::Failed, 2);
assert_eq!(
titles(&state.observe_event(&updated(1, vec![next]))),
["api failed"]
);
}
#[test]
fn an_overdue_event_notifies_once_per_run() {
let mut state = NotificationState::default();
state.seed(&[run(1, "api", PipelineStatus::Running, 1)], &[summary(1)]);
let mut slow = run(1, "api", PipelineStatus::Running, 1);
slow.overdue_run = Some("run-1".to_owned());
let first = state.observe_event(&updated(1, vec![slow.clone()]));
let repeated = state.observe_event(&updated(1, vec![slow.clone()]));
slow.overdue_run = Some("run-2".to_owned());
let next = state.observe_event(&updated(1, vec![slow]));
assert_eq!(titles(&first), ["api is running longer than usual"]);
assert!(repeated.is_empty());
assert_eq!(next.len(), 1);
}
#[tokio::test]
async fn a_scan_flags_a_running_pipeline_past_its_usual_duration_once() {
let mut rig = Rig::new();
let id = "github__1__api";
let mut runs = vec![running_run(id, 3)];
runs.extend((1..=3).map(|number| stub_run(id, number, 600)));
let stub = rig.stub(1, |stub| stub.with_runs(runs));
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Success, 1)]);
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Running, 2)]);
let found = rig.scan(&[1]).await;
assert_eq!(titles(&found), ["api is running longer than usual"]);
assert_eq!(found[0].key, "pipeline:github__1__api");
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn failed_scans_keep_the_baseline_and_an_outage_recovers_once() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Failed, 1)])
});
assert!(rig.scan(&[1]).await.is_empty());
fail(&stub, network());
assert!(rig.scan(&[1]).await.is_empty());
let outage = rig.scan(&[1]).await;
assert_eq!(titles(&outage), ["Can't reach Provider 1"]);
assert_eq!(outage[0].key, "connection:1");
assert_eq!(outage[0].provider_id, Some(1));
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Failed, 1)]);
let recovered = rig.scan(&[1]).await;
assert_eq!(titles(&recovered), ["Provider 1 is reachable again"]);
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Failed, 2)]);
assert_eq!(titles(&rig.scan(&[1]).await), ["api failed"]);
}
#[tokio::test]
async fn an_outage_survives_a_reload_and_is_not_repeated() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| stub);
rig.scan(&[1]).await;
fail(&stub, network());
rig.scan(&[1]).await;
assert_eq!(rig.scan(&[1]).await.len(), 1);
rig.state = reload(&rig.state);
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Success, 1)]);
assert_eq!(
titles(&rig.scan(&[1]).await),
["Provider 1 is reachable again"]
);
}
#[tokio::test]
async fn a_failed_scan_of_another_kind_never_fabricates_a_recovery() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| stub);
rig.scan(&[1]).await;
fail(&stub, network());
rig.scan(&[1]).await;
assert_eq!(rig.scan(&[1]).await.len(), 1);
fail(
&stub,
DomainError::AuthenticationFailed("revoked".to_owned()),
);
assert!(rig.scan(&[1]).await.is_empty());
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Success, 1)]);
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn a_changed_provider_version_starts_a_new_connection_episode() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| stub);
rig.scan(&[1]).await;
fail(&stub, network());
rig.scan(&[1]).await;
assert_eq!(rig.scan(&[1]).await.len(), 1);
rig.version = 2;
stub.set_pipelines(vec![run(1, "api", PipelineStatus::Success, 1)]);
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn removed_providers_lose_their_state_and_return_as_a_fresh_baseline() {
let mut rig = Rig::new();
rig.stub(1, |stub| stub);
let second = rig.stub(2, |stub| stub);
rig.scan(&[1, 2]).await;
assert!(rig.state.providers.contains_key(&2));
rig.scan(&[1]).await;
assert!(!rig.state.providers.contains_key(&2));
assert!(!rig.state.pipelines.contains_key(&2));
assert!(rig.state.pipelines.contains_key(&1));
second.set_pipelines(vec![run(2, "web", PipelineStatus::Failed, 5)]);
assert!(rig.scan(&[1, 2]).await.is_empty());
}
#[test]
fn a_removed_provider_event_and_retain_drop_state() {
let mut state = NotificationState::default();
state.seed(
&[
run(1, "api", PipelineStatus::Success, 1),
run(2, "web", PipelineStatus::Success, 1),
run(3, "app", PipelineStatus::Success, 1),
],
&[summary(1), summary(2), summary(3)],
);
state.observe_event(&CoreEvent::ProviderRemoved {
provider: summary(2),
timestamp: 0,
});
state.retain_providers(&HashSet::from([1]));
assert_eq!(state.pipelines.keys().copied().collect::<Vec<_>>(), [1]);
assert_eq!(state.providers.keys().copied().collect::<Vec<_>>(), [1]);
}
#[tokio::test]
async fn a_status_page_outage_alerts_once_and_recovers_only_after_calm_checks() {
let feed = FeedStub::start().await;
let mut rig = Rig::new();
rig.stub(1, |stub| stub.with_source(source(&feed)));
let key = feed_subject(&(feed.feed_url(), FeedFormat::Statuspage));
feed.respond_json(operational());
assert!(rig.scan(&[1]).await.is_empty());
feed.respond_json(outage());
let found = rig.scan(&[1]).await;
assert_eq!(titles(&found), ["Incident inc1"]);
assert_eq!(found[0].key, key);
assert_eq!(found[0].body, "We are investigating");
assert_eq!(found[0].provider_id, Some(1));
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig
.state
.observe_event(&alert_event(&key, "incident:inc1", "Incident inc1", false))
.is_empty());
feed.respond_json(operational());
assert!(rig.scan(&[1]).await.is_empty());
feed.respond(500, "");
assert!(rig.scan(&[1]).await.is_empty());
feed.respond_json(operational());
assert!(rig.scan(&[1]).await.is_empty());
let recovered = rig.scan(&[1]).await;
assert_eq!(titles(&recovered), ["Example is back to normal"]);
assert!(rig.scan(&[1]).await.is_empty());
}
#[test]
fn distinct_incidents_with_the_same_title_each_notify_once() {
let mut state = NotificationState::default();
let event = |id: &str| alert_event("status-page:feed", id, "Actions outage", false);
assert_eq!(
titles(&state.observe_event(&event("incident:a"))),
["Actions outage"]
);
assert!(state.observe_event(&event("incident:a")).is_empty());
assert_eq!(
titles(&state.observe_event(&event("incident:b"))),
["Actions outage"]
);
}
#[tokio::test]
async fn a_status_page_outage_seen_first_by_an_event_is_not_repeated_by_the_scan() {
let feed = FeedStub::start().await;
let mut rig = Rig::new();
rig.stub(1, |stub| stub.with_source(source(&feed)));
let key = feed_subject(&(feed.feed_url(), FeedFormat::Statuspage));
feed.respond_json(operational());
rig.scan(&[1]).await;
let event =
rig.state
.observe_event(&alert_event(&key, "incident:inc1", "Incident inc1", false));
assert_eq!(titles(&event), ["Incident inc1"]);
feed.respond_json(outage());
assert!(rig.scan(&[1]).await.is_empty());
rig.state = reload(&rig.state);
feed.respond_json(operational());
assert!(rig.scan(&[1]).await.is_empty());
assert_eq!(titles(&rig.scan(&[1]).await), ["Example is back to normal"]);
assert!(rig
.state
.observe_event(&alert_event(&key, "", "Example is back to normal", true))
.is_empty());
}
#[tokio::test]
async fn a_status_page_outage_present_at_the_first_scan_is_a_silent_baseline() {
let feed = FeedStub::start().await;
let mut rig = Rig::new();
rig.stub(1, |stub| stub.with_source(source(&feed)));
feed.respond_json(outage());
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
feed.respond_json(operational());
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn controller_health_alerts_and_recovers_without_trusting_failed_checks() {
let mut rig = Rig::new();
rig.controller = true;
let stub = rig.stub(1, |stub| stub);
let broken = ComponentHealth {
name: "flux".to_owned(),
level: HealthLevel::MajorOutage,
detail: Some("scaled to zero".to_owned()),
};
stub.set_health(Ok(Vec::new()));
assert!(rig.scan(&[1]).await.is_empty());
stub.set_health(Ok(vec![broken]));
let found = rig.scan(&[1]).await;
assert_eq!(titles(&found), ["flux: scaled to zero"]);
assert_eq!(found[0].key, "instance:1");
assert_eq!(
found[0].body,
"Reported by the controller health check of Provider 1."
);
stub.set_health(Ok(Vec::new()));
assert!(rig.scan(&[1]).await.is_empty());
stub.set_health(Err(network()));
assert!(rig.scan(&[1]).await.is_empty());
stub.set_health(Ok(Vec::new()));
assert!(rig.scan(&[1]).await.is_empty());
assert_eq!(
titles(&rig.scan(&[1]).await),
["Provider 1 controllers recovered"]
);
}
#[tokio::test]
async fn a_controller_outage_is_reported_when_pipeline_access_fails() {
let mut rig = Rig::new();
rig.controller = true;
let stub = rig.stub(1, |stub| {
stub.failing(DomainError::PermissionDenied("pipelines".to_owned()))
});
stub.set_health(Ok(Vec::new()));
rig.scan(&[1]).await;
stub.set_health(Ok(vec![broken()]));
let found = rig.scan(&[1]).await;
assert!(titles(&found).contains(&"flux: scaled to zero"));
}
#[tokio::test]
async fn a_controller_outage_survives_a_pipeline_fetch_that_outlasts_the_scan() {
let gate = Gate::new();
gate.open();
let mut rig = Rig::new();
rig.controller = true;
let stub = rig.stub(1, |stub| stub.gated(&gate));
rig.scan(&[1]).await;
gate.close();
stub.set_health(Ok(vec![broken()]));
let found = rig.scan_within(&[1], Duration::from_millis(300)).await;
assert_eq!(titles(&found), ["flux: scaled to zero"]);
}
#[tokio::test]
async fn hanging_controller_checks_cannot_starve_pipeline_fetches() {
let gate = Gate::new();
let mut rig = Rig::new();
rig.controller = true;
for id in 1..=4 {
rig.stub(id, |stub| stub.health_gated(&gate));
}
let last = rig.stub(5, |stub| stub);
let ids = [1, 2, 3, 4, 5];
rig.scan_within(&ids, Duration::from_millis(300)).await;
rig.scan_within(&ids, Duration::from_millis(300)).await;
assert_eq!(last.fetch_count(), 1);
}
#[tokio::test]
async fn a_provider_close_to_its_quota_is_skipped_until_the_reset() {
let mut rig = Rig::new();
let stub = rig.stub(1, |stub| stub);
*stub.rate_limit.lock().unwrap() = Some(RateLimit {
limit: 5000,
remaining: 10,
reset_at: Utc::now() + chrono::Duration::hours(1),
});
rig.scan(&[1]).await;
*stub.rate_limit.lock().unwrap() = None;
rig.scan(&[1]).await;
assert_eq!(stub.fetch_count(), 0);
}
#[tokio::test]
async fn a_scan_over_budget_returns_finished_notifications() {
let gate = Gate::new();
gate.open();
let mut rig = Rig::new();
let first = rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Success, 1)])
});
rig.stub(2, |stub| stub.gated(&gate));
rig.scan(&[1, 2]).await;
gate.close();
first.set_pipelines(vec![run(1, "api", PipelineStatus::Failed, 2)]);
let found = rig.scan_within(&[1, 2], Duration::from_millis(300)).await;
assert_eq!(titles(&found), ["api failed"]);
}
#[tokio::test]
async fn providers_starved_by_hanging_ones_go_first_on_the_next_scan() {
let gate = Gate::new();
let mut rig = Rig::new();
for id in 1..=4 {
rig.stub(id, |stub| stub.gated(&gate));
}
let last = rig.stub(5, |stub| stub);
let ids = [1, 2, 3, 4, 5];
rig.scan_within(&ids, Duration::from_millis(300)).await;
assert_eq!(last.fetch_count(), 0);
rig.scan_within(&ids, Duration::from_millis(300)).await;
assert_eq!(last.fetch_count(), 1);
}
#[tokio::test]
async fn a_bad_proxy_is_a_fatal_configuration_error() {
let mut state = NotificationState::default();
let result = crate::application::refresh_notifications(
&[],
"not a proxy url",
true,
&mut state,
Duration::from_secs(1),
)
.await;
assert!(result.is_err());
}
fn overdue_runs(pipeline_id: &str) -> Vec<PipelineRun> {
let mut runs = vec![running_run(pipeline_id, 3)];
runs.extend((1..=3).map(|number| stub_run(pipeline_id, number, 600)));
runs
}
#[tokio::test]
async fn running_pipelines_beyond_the_per_scan_cap_are_covered_by_later_scans() {
let mut rig = Rig::new();
let many: Vec<Pipeline> = (0..30)
.map(|n| run(1, &format!("p{n:02}"), PipelineStatus::Running, 1))
.collect();
rig.stub(1, |stub| {
stub.returning(many)
.with_runs(overdue_runs("github__1__p00"))
});
let checked = |rig: &Rig| {
rig.state.pipelines[&1]
.values()
.filter(|mark| mark.checked_at.is_some())
.count()
};
assert!(rig.scan(&[1]).await.is_empty());
assert_eq!(checked(&rig), 25);
assert!(rig.scan(&[1]).await.is_empty());
assert_eq!(checked(&rig), 30);
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn the_overdue_check_cap_applies_to_the_whole_scan() {
let mut rig = Rig::new();
for id in 1..=2 {
let running: Vec<Pipeline> = (0..20)
.map(|n| run(id, &format!("p{n:02}"), PipelineStatus::Running, 1))
.collect();
rig.stub(id, |stub| stub.returning(running));
}
let checked = |rig: &Rig| {
rig.state
.pipelines
.values()
.flat_map(|marks| marks.values())
.filter(|mark| mark.checked_at.is_some())
.count()
};
rig.scan(&[1, 2]).await;
assert_eq!(checked(&rig), 25);
rig.scan(&[1, 2]).await;
assert_eq!(checked(&rig), 40);
}
#[tokio::test]
async fn providers_beyond_the_overdue_cap_get_checks_on_the_next_scan() {
let mut rig = Rig::new();
let ids: Vec<i64> = (1..=30).collect();
for &id in &ids {
rig.stub(id, |stub| {
stub.returning(vec![run(id, "api", PipelineStatus::Running, 1)])
});
}
let checked = |rig: &Rig| {
rig.state
.pipelines
.values()
.flat_map(|marks| marks.values())
.filter(|mark| mark.checked_at.is_some())
.count()
};
rig.scan(&ids).await;
assert_eq!(checked(&rig), 25);
rig.scan(&ids).await;
assert_eq!(checked(&rig), 30);
}
#[tokio::test]
async fn overdue_checks_ask_a_batching_provider_once_per_scan() {
let mut rig = Rig::new();
let running: Vec<Pipeline> = (0..3)
.map(|n| run(1, &format!("p{n}"), PipelineStatus::Running, 1))
.collect();
let stub = rig.stub(1, |stub| stub.batching().returning(running));
rig.scan(&[1]).await;
let checked = rig.state.pipelines[&1]
.values()
.filter(|mark| mark.checked_at.is_some())
.count();
assert_eq!(checked, 3);
assert_eq!(stub.batch_fetch_count(), 1);
assert_eq!(stub.history_fetch_count(), 0);
}
#[tokio::test]
async fn a_pipeline_already_overdue_at_the_first_scan_stays_silent() {
let mut rig = Rig::new();
rig.stub(1, |stub| {
stub.returning(vec![run(1, "api", PipelineStatus::Running, 1)])
.with_runs(overdue_runs("github__1__api"))
});
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn hanging_run_history_does_not_block_outage_checks_within_the_deadline() {
let gate = Gate::new();
gate.open();
let mut rig = Rig::new();
rig.controller = true;
let stub = rig.stub(1, |stub| {
stub.run_gated(&gate)
.returning(vec![run(1, "api", PipelineStatus::Running, 1)])
});
stub.set_health(Ok(Vec::new()));
rig.scan(&[1]).await;
gate.close();
stub.set_health(Ok(vec![broken()]));
let found = rig.scan_within(&[1], Duration::from_millis(300)).await;
assert_eq!(titles(&found), ["flux: scaled to zero"]);
}
#[tokio::test]
async fn disabling_status_pages_drops_event_created_feed_state() {
let feed = FeedStub::start().await;
let mut rig = Rig::new();
rig.stub(1, |stub| stub.with_source(source(&feed)));
let key = feed_subject(&(feed.feed_url(), FeedFormat::Statuspage));
feed.respond_json(operational());
rig.scan(&[1]).await;
rig.state
.observe_event(&alert_event(&key, "incident:inc1", "Incident inc1", false));
assert!(rig.state.alerts.has(&key));
rig.status_pages = false;
let requests = feed.request_count();
assert!(rig.scan(&[1]).await.is_empty());
assert_eq!(feed.request_count(), requests);
assert!(!rig.state.alerts.has(&key));
assert!(!rig.state.delivered.contains_key(&key));
rig.status_pages = true;
for _ in 0..3 {
assert!(rig.scan(&[1]).await.is_empty());
}
}
#[tokio::test]
async fn a_status_page_outage_after_a_healthy_opt_in_alerts_on_the_first_scan() {
let feed = FeedStub::start().await;
let mut rig = Rig::new();
rig.stub(1, |stub| stub.with_source(source(&feed)));
rig.state.seed(&[], &[summary_with(1, Vec::new(), true)]);
rig.state = reload(&rig.state);
feed.respond_json(outage());
assert_eq!(titles(&rig.scan(&[1]).await), ["Incident inc1"]);
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn a_status_page_outage_present_at_opt_in_stays_silent() {
let feed = FeedStub::start().await;
let mut rig = Rig::new();
rig.stub(1, |stub| stub.with_source(source(&feed)));
rig.state
.seed(&[], &[summary_with(1, vec![status_signal()], true)]);
rig.state = reload(&rig.state);
feed.respond_json(outage());
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
feed.respond_json(operational());
for _ in 0..3 {
assert!(rig.scan(&[1]).await.is_empty());
}
}
#[tokio::test]
async fn a_controller_outage_after_a_healthy_opt_in_alerts_on_the_first_scan() {
let mut rig = Rig::new();
rig.controller = true;
let stub = rig.stub(1, |stub| stub);
stub.set_health(Ok(vec![broken()]));
rig.state.seed(&[], &[summary_with(1, Vec::new(), false)]);
rig.state = reload(&rig.state);
assert_eq!(titles(&rig.scan(&[1]).await), ["flux: scaled to zero"]);
assert!(rig.scan(&[1]).await.is_empty());
}
#[tokio::test]
async fn a_controller_outage_present_at_opt_in_stays_silent() {
let mut rig = Rig::new();
rig.controller = true;
let stub = rig.stub(1, |stub| stub);
stub.set_health(Ok(vec![broken()]));
rig.state
.seed(&[], &[summary_with(1, vec![instance_signal()], false)]);
rig.state = reload(&rig.state);
assert!(rig.scan(&[1]).await.is_empty());
assert!(rig.scan(&[1]).await.is_empty());
stub.set_health(Ok(Vec::new()));
for _ in 0..3 {
assert!(rig.scan(&[1]).await.is_empty());
}
}