use std::io::{Read as _, Write as _};
use std::net::TcpListener;
use std::sync::{Arc, LazyLock, Mutex, MutexGuard, PoisonError};
use std::thread;
use onetaskgraph_github_projects::accounting::{Accounting, Budget, Outcome, RateLimit, Request};
use onetaskgraph_live::{Allowance, RETAINED_BUFFER, Unaffordable};
use serde_json::json;
#[allow(dead_code)]
mod journey;
#[allow(dead_code)]
mod lane;
use journey::budget;
const LIMIT: u64 = 4_321;
const RESETS_AT: u64 = 1_775_000_000;
const HEADERS_RESET_AT: u64 = 1_775_000_777;
const ALREADY_EXCEEDED: &str = "API rate limit already exceeded for user ID 1.";
#[derive(Clone)]
enum FirstCall {
Answer {
status: &'static str,
headers: String,
body: String,
},
HangUp,
}
impl FirstCall {
fn reporting(remaining: u64) -> Self {
Self::Answer {
status: "200 OK",
headers: graphql_headers(remaining, LIMIT.saturating_sub(remaining)),
body: json!({"data": {}}).to_string(),
}
}
}
fn graphql_headers(remaining: u64, used: u64) -> String {
format!(
"x-ratelimit-limit: {LIMIT}\r\nx-ratelimit-used: {used}\r\n\
x-ratelimit-remaining: {remaining}\r\nx-ratelimit-reset: {HEADERS_RESET_AT}\r\n\
x-ratelimit-resource: graphql\r\n"
)
}
struct Standin {
host: String,
first_call: Arc<Mutex<FirstCall>>,
asked: Arc<Mutex<Vec<String>>>,
}
impl Standin {
fn start() -> Self {
let listener = TcpListener::bind("127.0.0.1:0").expect("a stand-in listener");
let host = format!("http://{}", listener.local_addr().unwrap());
let first_call = Arc::new(Mutex::new(FirstCall::reporting(LIMIT)));
let asked = Arc::new(Mutex::new(Vec::new()));
let (scripted, recorded) = (Arc::clone(&first_call), Arc::clone(&asked));
thread::spawn(move || {
for stream in listener.incoming() {
let mut stream = stream.expect("a stand-in connection");
let mut request = [0_u8; 16384];
let read = stream.read(&mut request).expect("a stand-in request");
let head = String::from_utf8_lossy(&request[..read]).into_owned();
let line = head.lines().next().unwrap_or_default().to_owned();
let mut parts = line.split(' ');
let called = format!(
"{} {}",
parts.next().unwrap_or_default(),
parts.next().unwrap_or_default()
);
let mut asked = recorded.lock().unwrap();
asked.push(called.clone());
let first_graphql_of_this_run = asked
.iter()
.rev()
.take_while(|call| *call != "GET /rate_limit")
.filter(|call| *call == "POST /graphql")
.count()
== 1;
drop(asked);
let (status, headers, body) = match called.as_str() {
"GET /rate_limit" => (
"200 OK",
format!(
"x-ratelimit-limit: {LIMIT}\r\nx-ratelimit-used: 0\r\n\
x-ratelimit-remaining: {LIMIT}\r\nx-ratelimit-resource: core\r\n"
),
budget::documented_answer(LIMIT, LIMIT, LIMIT, RESETS_AT).to_string(),
),
"POST /graphql" if first_graphql_of_this_run => {
match scripted.lock().unwrap().clone() {
FirstCall::Answer {
status,
headers,
body,
} => (status, headers, body),
FirstCall::HangUp => continue,
}
}
_ => (
"404 Not Found",
String::new(),
json!({"message": "this stand-in answers the allowance read and the first real call alone"})
.to_string(),
),
};
let response = format!(
"HTTP/1.1 {status}\r\nContent-Type: application/json\r\n{headers}\
Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(response.as_bytes());
}
});
Self {
host,
first_call,
asked,
}
}
}
static STANDIN: LazyLock<Standin> = LazyLock::new(|| {
let standin = Standin::start();
journey::against(journey::Endpoints {
graphql: format!("{}/graphql", standin.host),
rest_host: standin.host.clone(),
source: Some(json!({"endpoint": format!("{}/graphql", standin.host)})),
});
standin
});
static ONE_RUN_AT_A_TIME: Mutex<()> = Mutex::new(());
struct Drive {
panic: Box<dyn std::any::Any + Send>,
asked: Vec<String>,
}
impl Drive {
fn message(&self) -> String {
self.panic
.downcast_ref::<String>()
.cloned()
.unwrap_or_else(|| panic!("the run carries its message: {:?}", self.panic))
}
}
fn drive(first_call: FirstCall) -> (Drive, MutexGuard<'static, ()>) {
let turn = take_the_turn();
let standin = &*STANDIN;
*standin.first_call.lock().unwrap() = first_call;
let before = standin.asked.lock().unwrap().len();
let panic = thread::spawn(|| {
block_on(journey::run(journey::Nomination {
token: "test-token".to_owned(),
owner: "octo-org".to_owned(),
project_number: 7,
repository: "acme/work".to_owned(),
}));
})
.join()
.expect_err("every drive of this stand-in ends in a panic, declined or failed");
let asked = standin.asked.lock().unwrap()[before..].to_vec();
(Drive { panic, asked }, turn)
}
fn take_the_turn() -> MutexGuard<'static, ()> {
ONE_RUN_AT_A_TIME
.lock()
.unwrap_or_else(PoisonError::into_inner)
}
fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("a runtime for the journey")
.block_on(future)
}
fn estimated() -> u64 {
budget::estimate()[&Budget::Graphql]
}
fn the_free_read_and_one_real_call() -> Vec<String> {
vec!["GET /rate_limit".to_owned(), "POST /graphql".to_owned()]
}
#[test]
fn a_journey_whose_first_call_carries_less_than_the_endpoint_claimed_does_not_run() {
let buffer = RETAINED_BUFFER.of(LIMIT);
let remaining = estimated() + buffer - 1;
let (drive, _turn) = drive(FirstCall::reporting(remaining));
let message = drive.message();
assert!(message.contains("DID NOT RUN"), "{message}");
assert!(
message.contains("not a test failure in the code under test"),
"{message}"
);
assert!(!message.contains("mutation schema drifted"), "{message}");
for figure in [
"the allowance read GET /rate_limit".to_owned(),
format!("claimed {LIMIT} of {LIMIT} points remaining"),
budget::first_call_headers("mutation schema introspection"),
format!("reported {remaining} of {LIMIT} remaining"),
format!("estimated to spend {}", estimated()),
format!("retained buffer is {buffer}"),
format!("resets at {HEADERS_RESET_AT}"),
"nothing here waits for it".to_owned(),
] {
assert!(
message.contains(&figure),
"{figure} is missing from {message}"
);
}
assert_eq!(drive.asked, the_free_read_and_one_real_call());
if std::env::var_os(budget::FOLLOW_THROUGH).is_some() {
std::panic::resume_unwind(drive.panic);
}
}
#[test]
fn a_journey_whose_first_call_is_refused_for_a_rate_limit_is_declined_rather_than_failed() {
let (drive, _turn) = drive(FirstCall::Answer {
status: "403 Forbidden",
headers: graphql_headers(0, LIMIT + 42),
body: json!({"message": ALREADY_EXCEEDED}).to_string(),
});
let message = drive.message();
assert!(message.contains("DID NOT RUN"), "{message}");
assert!(!message.contains("mutation schema drifted"), "{message}");
assert!(
message.contains("it was refused for a rate limit"),
"{message}"
);
assert!(
message.contains("carried no allowance this session could read"),
"{message}"
);
assert!(
message.contains(&format!("claimed {LIMIT} of {LIMIT} points remaining")),
"{message}"
);
assert!(
message.contains(&format!(
"By the first reading that budget resets at {RESETS_AT}"
)),
"{message}"
);
assert_eq!(drive.asked, the_free_read_and_one_real_call());
}
#[test]
fn a_first_call_refused_with_the_budget_still_showing_room_is_declined_as_the_secondary_limiter() {
let (drive, _turn) = drive(FirstCall::Answer {
status: "403 Forbidden",
headers: graphql_headers(LIMIT - 1, 1),
body: json!({"message": "You have exceeded a secondary rate limit."}).to_string(),
});
let message = drive.message();
assert!(message.contains("DID NOT RUN"), "{message}");
assert!(
message.contains(&format!(
"it was refused for a rate limit, its headers reporting {} of {LIMIT} remaining",
LIMIT - 1
)),
"{message}"
);
assert_eq!(drive.asked, the_free_read_and_one_real_call());
}
#[test]
fn an_answered_first_call_carrying_no_readable_allowance_declines_rather_than_assuming_the_claim() {
for headers in [String::new(), graphql_headers(0, LIMIT + 42)] {
let (drive, _turn) = drive(FirstCall::Answer {
status: "200 OK",
headers: headers.clone(),
body: json!({"data": {}}).to_string(),
});
let message = drive.message();
assert!(message.contains("DID NOT RUN"), "{headers:?}: {message}");
assert!(
message.contains(
"it was answered and its headers carried no allowance this session could read"
),
"{headers:?}: {message}"
);
assert!(message.contains("not one it may assume"), "{message}");
assert_eq!(drive.asked, the_free_read_and_one_real_call());
}
}
#[test]
fn a_first_call_that_never_reached_the_host_fails_as_itself_rather_than_declining() {
let (drive, _turn) = drive(FirstCall::HangUp);
let message = drive.message();
assert!(message.contains("mutation schema drifted"), "{message}");
assert!(message.contains("could not reach GitHub"), "{message}");
assert!(!message.contains("DID NOT RUN"), "{message}");
assert_eq!(drive.asked, the_free_read_and_one_real_call());
}
#[test]
fn a_first_call_carrying_the_allowance_the_endpoint_claimed_lets_the_session_go_on() {
let (drive, _turn) = drive(FirstCall::reporting(LIMIT - 1));
let message = drive.message();
assert!(message.contains("mutation schema drifted"), "{message}");
assert!(!message.contains("DID NOT RUN"), "{message}");
assert!(
drive.asked.len() > 2,
"a session the second reading affords goes on past its first real call: {:?}",
drive.asked
);
assert_eq!(drive.asked[..2], the_free_read_and_one_real_call()[..]);
}
fn recorded_first_call(outcome: Outcome, headers: &[(&str, &str)]) -> Request {
let headers: Vec<(String, String)> = headers
.iter()
.map(|(name, value)| ((*name).to_owned(), (*value).to_owned()))
.collect();
Request::graphql(
"query { __typename }",
&json!({}),
Some("mutation schema introspection"),
None,
)
.finished(
outcome,
RateLimit::read(|name| {
headers
.iter()
.find(|(held, _)| held == name)
.map(|(_, value)| value.clone())
}),
)
}
fn admitted_on_the_whole_allowance(into: &Accounting) -> budget::Admitted {
block_on(budget::precondition("test-token", &STANDIN.host, into))
.expect("a whole allowance admits this session on the first reading")
}
#[test]
fn the_second_reading_is_decided_on_the_call_recorded_right_after_the_allowance_read() {
let _turn = take_the_turn();
let into = Accounting::new();
let admitted = admitted_on_the_whole_allowance(&into);
into.record(recorded_first_call(
Outcome::Answered,
&[
("x-ratelimit-limit", &LIMIT.to_string()),
("x-ratelimit-remaining", "0"),
("x-ratelimit-used", &LIMIT.to_string()),
("x-ratelimit-resource", "graphql"),
],
));
let declined = budget::recheck(&admitted, &into)
.expect_err("nothing left on the second reading declines the session");
let Some(Unaffordable::Contradicted(readings)) = declined.unaffordable_because() else {
panic!("a second reading that refuses is a contradiction: {declined:?}");
};
assert_eq!(readings.metered, budget::metered(Budget::Graphql));
assert_eq!(
readings.claimed,
Allowance::read(LIMIT, LIMIT, RESETS_AT).expect("the claim")
);
assert_eq!(readings.claimed_by, budget::allowance_read());
assert_eq!(
readings.carried,
Ok(Allowance::read(LIMIT, 0, RESETS_AT).expect("nothing left"))
);
assert_eq!(
readings.carried_by,
budget::first_call_headers("mutation schema introspection")
);
assert_eq!(readings.estimated_cost, estimated());
assert_eq!(readings.retained_buffer, RETAINED_BUFFER.of(LIMIT));
let again = admitted_on_the_whole_allowance(&into);
into.record(recorded_first_call(
Outcome::Answered,
&[
("x-ratelimit-limit", &LIMIT.to_string()),
("x-ratelimit-remaining", &(LIMIT - 1).to_string()),
("x-ratelimit-used", "1"),
("x-ratelimit-resource", "graphql"),
],
));
budget::recheck(&again, &into).expect("the call after the second read affords it");
assert!(budget::recheck(&admitted, &into).is_err());
}
#[test]
fn a_first_call_refused_for_something_other_than_a_rate_limit_is_decided_on_its_headers() {
let _turn = take_the_turn();
let into = Accounting::new();
let admitted = admitted_on_the_whole_allowance(&into);
into.record(recorded_first_call(
Outcome::Refused,
&[
("x-ratelimit-limit", &LIMIT.to_string()),
("x-ratelimit-remaining", &(LIMIT - 1).to_string()),
("x-ratelimit-used", "1"),
("x-ratelimit-reset", &HEADERS_RESET_AT.to_string()),
("x-ratelimit-resource", "graphql"),
],
));
budget::recheck(&admitted, &into).expect("headers with room let the refusal be a refusal");
let admitted = admitted_on_the_whole_allowance(&into);
into.record(recorded_first_call(
Outcome::Refused,
&[
("x-ratelimit-limit", &LIMIT.to_string()),
("x-ratelimit-remaining", "0"),
("x-ratelimit-used", &LIMIT.to_string()),
("x-ratelimit-resource", "graphql"),
],
));
assert!(
budget::recheck(&admitted, &into).is_err(),
"headers reporting nothing left decline the session whatever refused the call"
);
let admitted = admitted_on_the_whole_allowance(&into);
into.record(recorded_first_call(Outcome::Refused, &[]));
budget::recheck(&admitted, &into)
.expect("a refusal that says nothing about the budget does not decline on it");
}