diff --git a/src/database.rs b/src/database.rs index c7245eb..19c4329 100644 --- a/src/database.rs +++ b/src/database.rs @@ -325,13 +325,30 @@ impl ArchiveDatabaseRepository { /// "not found in Raft group"). pub fn read_state_verdict(&self, systemtender_db_name: &str) -> DbResult> { let db_config = self.base_config.with_database(systemtender_db_name); - let rows = match self.db.execute_query( + // The instruments column is rollout-order guarded, message-matched + // (the worker's own beat-guard pattern): a state row born before + // the column - fresh tender, worker has not stamped yet - reads + // the three verdict columns alone, instruments None. + let attempted = match self.db.execute_query( &db_config, "SELECT finished_at IS NOT NULL, \ EXTRACT(EPOCH FROM (now() - updated_at)), \ - beat_interval_secs FROM systemtender_state LIMIT 1", + beat_interval_secs, instruments \ + FROM systemtender_state LIMIT 1", true, ) { + Err(e) if e.0.contains("instruments") && e.0.contains("does not exist") => { + self.db.execute_query( + &db_config, + "SELECT finished_at IS NOT NULL, \ + EXTRACT(EPOCH FROM (now() - updated_at)), \ + beat_interval_secs FROM systemtender_state LIMIT 1", + true, + ) + } + other => other, + }; + let rows = match attempted { Ok(r) => r, Err(e) => { let error_text = e.0.clone(); @@ -364,6 +381,16 @@ impl ArchiveDatabaseRepository { None => Value::Null, }, ); + // The worker's JSON stamp, parsed leniently - a torn + // payload reads as absent, never as a failed read. + let mut instruments = Value::Null; + if let Some(Some(cell)) = row.get(3) { + if !cell.is_empty() { + instruments = + serde_json::from_str::(cell).unwrap_or(Value::Null); + } + } + m.insert("instruments".to_string(), instruments); return Ok(Some(Value::Object(m))); } } diff --git a/tests/port/instruments_read.rs b/tests/port/instruments_read.rs new file mode 100644 index 0000000..c2445cb --- /dev/null +++ b/tests/port/instruments_read.rs @@ -0,0 +1,68 @@ +//! Ported from tests/unit/test_instruments_read.py - rung 9, the +//! instruments door's read side: the verdict read picks up the +//! worker's JSON stamp with a rollout-order guard (message-matched, +//! the worker's own beat-guard pattern); a torn payload reads as +//! absent, never as a failed read; the missing-table wordings still +//! carry no verdict. + +use crate::support::{srow, FakeDb}; +use godon_controller::database::{ArchiveDatabaseRepository, DbConfig}; +use serde_json::json; +use std::sync::Arc; + +fn repo_with(db: Arc) -> ArchiveDatabaseRepository { + ArchiveDatabaseRepository::with_gateway(DbConfig::archive_from_env(), db) +} + +#[test] +fn verdict_carries_the_stamp() { + let db = Arc::new(FakeDb::rows(vec![srow(&[ + "f", + "12.0", + "10.0", + r#"{"strategies_run": ["hold"], "wish": null, "seek_spend": null}"#, + ])])); + let repo = repo_with(db); + let verdict = repo.read_state_verdict("db").unwrap().unwrap(); + assert_eq!( + verdict["instruments"], + json!({"strategies_run": ["hold"], "wish": null, "seek_spend": null}) + ); + assert_eq!(verdict["finished"], json!(false)); +} + +#[test] +fn rollout_order_falls_back() { + // A row born before the column: the three verdict columns alone, + // instruments None - one retry, no ladder. + let db = Arc::new(FakeDb::from_fn(|_db, query| { + if query.contains("instruments") { + return Err(godon_controller::database::DbError( + r#"column "instruments" does not exist"#.to_string(), + )); + } + Ok(Some(vec![srow(&["f", "12.0", "10.0"])])) + })); + let repo = repo_with(db.clone()); + let verdict = repo.read_state_verdict("db").unwrap().unwrap(); + assert_eq!(verdict["instruments"], json!(null)); + assert_eq!(db.queries().len(), 2); +} + +#[test] +fn missing_table_still_no_verdict() { + let db = Arc::new(FakeDb::err( + r#"relation "systemtender_state" does not exist"#, + )); + let repo = repo_with(db); + assert!(repo.read_state_verdict("db").unwrap().is_none()); +} + +#[test] +fn torn_payload_reads_absent() { + let db = Arc::new(FakeDb::rows(vec![srow(&["f", "12.0", "10.0", "{torn json"])])); + let repo = repo_with(db); + let verdict = repo.read_state_verdict("db").unwrap().unwrap(); + assert_eq!(verdict["instruments"], json!(null)); + assert_eq!(verdict["finished"], json!(false)); +} diff --git a/tests/port/main.rs b/tests/port/main.rs new file mode 100644 index 0000000..125c070 --- /dev/null +++ b/tests/port/main.rs @@ -0,0 +1,9 @@ +//! The python unit suite, ported module by module - rung 10's parity +//! bar. One file here per python file under tests/unit/; the fakes +//! python got from unittest.mock.patch live in support, injected +//! through the DbGateway / WindmillGateway / CausalClient seams +//! instead of module patching. + +mod support; + +mod instruments_read; diff --git a/tests/port/support.rs b/tests/port/support.rs new file mode 100644 index 0000000..1e0078a --- /dev/null +++ b/tests/port/support.rs @@ -0,0 +1,227 @@ +//! The shared fakes: what python's suite got from +//! `patch('controller.database.execute_query', ...)` and friends, +//! as gateways injected at construction. +//! +//! Rows are TEXT cells - the simple query protocol's shape - where +//! python's fakes handed typed tuples (psycopg2's decoded cells). A +//! python row `(False, 12.0, 'x')` ports as `srow(&["f", "12.0", +//! "x"])`. + +use godon_controller::database::{DbConfig, DbError, DbGateway, DbResult, TextRow}; +use godon_controller::wmill::{CausalClient, WindmillGateway}; +use serde_json::Value; +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; + +/// All-Some row of text cells. +pub fn srow(cells: &[&str]) -> TextRow { + cells.iter().map(|c| Some(c.to_string())).collect() +} + +/// Row with explicit NULLs: `row(&[Some("x"), None])`. +pub fn row(cells: &[Option<&str>]) -> TextRow { + cells.iter().map(|c| c.map(|s| s.to_string())).collect() +} + +pub type DbResponder = Box DbResult>> + Send + Sync>; + +/// The fake behind every ported test that patched execute_query. +/// Records (database, query) per call; answers from the responder. +pub struct FakeDb { + pub calls: Mutex>, + pub responder: DbResponder, +} + +impl FakeDb { + /// python's `side_effect=f(db_config, query)` - sees database name + /// and full query text. + pub fn from_fn( + f: impl Fn(&str, &str) -> DbResult>> + Send + Sync + 'static, + ) -> Self { + Self { + calls: Mutex::new(Vec::new()), + responder: Box::new(f), + } + } + + /// python's `return_value=[rows]` - every call answers the same. + pub fn rows(rows: Vec) -> Self { + Self::from_fn(move |_, _| Ok(Some(rows.clone()))) + } + + /// Answer nothing, successfully (the DDL/verb-write shape). + pub fn ok() -> Self { + Self::rows(Vec::new()) + } + + /// Always raise (python's `side_effect=Exception(...)`). + pub fn err(text: &str) -> Self { + let text = text.to_string(); + Self::from_fn(move |_, _| Err(DbError(text))) + } + + pub fn queries(&self) -> Vec { + self.calls + .lock() + .unwrap() + .iter() + .map(|(_, q)| q.clone()) + .collect() + } + + pub fn databases(&self) -> Vec { + self.calls + .lock() + .unwrap() + .iter() + .map(|(d, _)| d.clone()) + .collect() + } +} + +impl DbGateway for FakeDb { + fn execute_query( + &self, + db_config: &DbConfig, + query: &str, + _with_result: bool, + ) -> DbResult>> { + let db_name = db_config.database.clone().unwrap_or_default(); + self.calls + .lock() + .unwrap() + .push((db_name.clone(), query.to_string())); + (self.responder)(&db_name, query) + } + + fn execute_ddl_query(&self, db_config: &DbConfig, query: &str) -> DbResult<()> { + let db_name = db_config.database.clone().unwrap_or_default(); + self.calls + .lock() + .unwrap() + .push((db_name, query.to_string())); + Ok(()) + } +} + +fn pop(q: &Mutex>, default: T) -> T { + q.lock().unwrap().pop_front().unwrap_or(default) +} + +/// The fake behind python's `patch('...Windmill', return_value= +/// fake_client)`: per-call scripted outcomes (popped in order, +/// defaults when exhausted) plus a full record of what flew. +#[derive(Default)] +pub struct FakeWindmill { + pub cancel_results: Mutex>>, + pub job_results: Mutex, String>>>, + pub completed_results: Mutex, String>>>, + pub async_results: Mutex>>, + pub sync_results: Mutex>>, + pub cancelled: Mutex)>>, + pub dispatched: Mutex)>>, + pub sync_ran: Mutex>, +} + +impl FakeWindmill { + pub fn queue_async(&self, r: Result) -> &Self { + self.async_results.lock().unwrap().push_back(r); + self + } + pub fn queue_completed(&self, v: Option) -> &Self { + self.completed_results + .lock() + .unwrap() + .push_back(Ok(v)); + self + } + pub fn queue_job(&self, v: Option) -> &Self { + self.job_results.lock().unwrap().push_back(Ok(v)); + self + } +} + +impl WindmillGateway for FakeWindmill { + fn cancel_job(&self, job_id: &str, reason: Option<&str>) -> Result<(), String> { + self.cancelled + .lock() + .unwrap() + .push((job_id.to_string(), reason.map(|r| r.to_string()))); + pop(&self.cancel_results, Ok(())) + } + + fn get_job(&self, _job_id: &str) -> Result, String> { + pop(&self.job_results, Ok(None)) + } + + fn get_completed_maybe(&self, _job_id: &str) -> Result, String> { + pop(&self.completed_results, Ok(None)) + } + + fn run_script_by_path_async( + &self, + path: &str, + args: &Value, + tag: Option<&str>, + ) -> Result { + self.dispatched.lock().unwrap().push_back(( + path.to_string(), + args.clone(), + tag.map(|t| t.to_string()), + )); + pop(&self.async_results, Ok("job-fake".to_string())) + } + + fn run_script_by_path(&self, path: &str, args: &Value) -> Result { + self.sync_ran + .lock() + .unwrap() + .push((path.to_string(), args.clone())); + pop(&self.sync_results, Ok(Value::Null)) + } +} + +/// The fake behind python's urlopen/ask patches on the causal seam. +#[derive(Default)] +pub struct FakeCausal { + pub pages: Mutex>>, + pub asks: Mutex>>, + pub purge_results: Mutex>>, + pub asked: Mutex>, + pub purged: Mutex>, +} + +impl FakeCausal { + pub fn queue_page(&self, v: Option) -> &Self { + self.pages.lock().unwrap().push_back(v); + self + } + pub fn queue_ask(&self, r: Result) -> &Self { + self.asks.lock().unwrap().push_back(r); + self + } +} + +impl CausalClient for FakeCausal { + fn fetch_book_page(&self, _wish_id: &str) -> Option { + pop(&self.pages, None) + } + + fn ask_to_plan(&self, plan_request: &Value) -> Result { + self.asks.lock().unwrap().push_back(plan_request.clone()); + pop(&self.asks, Ok(Value::Null)) + } + + fn purge_curves(&self, systemtender_id: &str) -> Result { + self.purged.lock().unwrap().push_back(systemtender_id.to_string()); + pop(&self.purge_results, Ok(200)) + } +} + +/// Shared convenience: a config with a database name pinned, for +/// responders that branch on the db (python branched on db_config). +pub fn cfg_with_db(name: &str) -> DbConfig { + let mut c = DbConfig::archive_from_env(); + c.database = Some(name.to_string()); + c +}