migration_signed_facts_tests.rs (8343B)
1 use super::{ 2 tests::{connection, establish_runtime_version, pragma}, 3 *, 4 }; 5 use crate::authored::signed_fact_fixture; 6 use radroots_storage::{ 7 atomic::AtomicCommitDisposition, 8 authored_atomic::{AuthoredAtomicCommand, AuthoredAtomicOutcome, AuthoredAtomicReceipt}, 9 }; 10 11 const FAILING_V15: &str = concat!( 12 include_str!("migration/runtime/0015_authored_signed_facts.up.sql"), 13 "\nINSERT INTO missing_fixture_table VALUES (1);" 14 ); 15 16 fn plan(current: u32, fail: bool) -> MigrationPlan { 17 MigrationPlan { 18 database: RUNTIME_DATABASE, 19 application_id: RUNTIME_APPLICATION_ID, 20 set_application_id_sql: SET_RUNTIME_APPLICATION_ID, 21 minimum_version: runtime::MINIMUM_VERSION, 22 current_version: current, 23 steps: runtime::MIGRATIONS 24 .iter() 25 .take(current as usize) 26 .map(|step| MigrationStep { 27 version: step.version(), 28 sql: if fail && step.version() == 15 { 29 FAILING_V15 30 } else { 31 runtime::migration_sql(step.version()).unwrap() 32 }, 33 owned_objects: step.owned_objects(), 34 }) 35 .collect(), 36 } 37 } 38 39 pub(super) async fn seed(connection: &mut SqliteConnection) -> (Vec<u8>, Vec<u8>, Vec<u8>) { 40 let (command, _) = signed_fact_fixture::prepare(); 41 let AuthoredAtomicCommand::Prepare(prepared) = &command else { 42 unreachable!() 43 }; 44 let operation = prepared.operation(); 45 let artifact = &prepared.artifacts()[0]; 46 let operation_snapshot = serde_json::to_vec(operation).unwrap(); 47 let artifact_snapshot = serde_json::to_vec(artifact).unwrap(); 48 sqlx::query("INSERT INTO radroots_runtime_authored_operations (operation_id, artifact_count, created_at_unix_ms, updated_at_unix_ms, revision, snapshot) VALUES (?, 1, 10, 10, 1, ?)") 49 .bind(operation.operation_id().as_bytes().as_slice()).bind(operation_snapshot).execute(&mut *connection).await.unwrap(); 50 sqlx::query("INSERT INTO radroots_runtime_authored_artifacts (artifact_id, operation_id, ordinal, origin, signing_state, admission_state, plan_wire, created_at_unix_ms, updated_at_unix_ms, revision, snapshot) VALUES (?, ?, 0, 'planned', 'planned', 'pending', ?, 10, 10, 1, ?)") 51 .bind(artifact.artifact_id().as_bytes().as_slice()).bind(operation.operation_id().as_bytes().as_slice()) 52 .bind(artifact.plan().unwrap().wire_json()).bind(&artifact_snapshot).execute(&mut *connection).await.unwrap(); 53 for delivery in prepared.delivery_plans() { 54 sqlx::query("INSERT INTO radroots_runtime_authored_delivery_plans (plan_id, artifact_id, request_digest, state, attempt_count, created_at_unix_ms, updated_at_unix_ms, revision, snapshot) VALUES (?, ?, ?, 'pending', 0, 10, 10, 1, ?)") 55 .bind(delivery.plan_id().as_bytes().as_slice()).bind(delivery.artifact_id().as_bytes().as_slice()) 56 .bind(delivery.request_digest().as_slice()).bind(serde_json::to_vec(delivery).unwrap()).execute(&mut *connection).await.unwrap(); 57 for (ordinal, target) in delivery.intent().target_set().targets().iter().enumerate() { 58 sqlx::query("INSERT INTO radroots_runtime_authored_delivery_targets (plan_id, ordinal, target_fingerprint, target_snapshot) VALUES (?, ?, ?, ?)") 59 .bind(delivery.plan_id().as_bytes().as_slice()).bind(i64::try_from(ordinal).unwrap()) 60 .bind(target.fingerprint().as_str()).bind(serde_json::to_vec(target).unwrap()).execute(&mut *connection).await.unwrap(); 61 } 62 } 63 let outcome = AuthoredAtomicOutcome::Prepared { 64 operation: operation.clone(), 65 artifacts: prepared.artifacts().to_vec(), 66 delivery_plans: prepared.delivery_plans().to_vec(), 67 }; 68 let receipt = AuthoredAtomicReceipt::new( 69 &command, 70 AtomicCommitDisposition::Committed, 71 10, 72 outcome.clone(), 73 ) 74 .unwrap(); 75 let receipt_bytes = serde_json::to_vec(&serde_json::json!({"outcome": outcome})).unwrap(); 76 sqlx::query("INSERT INTO radroots_runtime_authored_atomic_commits (commit_id, commit_digest, phase, target_id, requested_at_unix_ms, committed_at_unix_ms, receipt) VALUES (?, ?, 'prepare', ?, 10, 10, ?)") 77 .bind(receipt.commit_id().as_bytes().as_slice()).bind(receipt.digest().as_bytes().as_slice()) 78 .bind(operation.operation_id().as_bytes().as_slice()).bind(&receipt_bytes).execute(&mut *connection).await.unwrap(); 79 ( 80 artifact_snapshot, 81 receipt_bytes, 82 receipt.commit_id().as_bytes().to_vec(), 83 ) 84 } 85 86 pub(super) async fn retained( 87 connection: &mut SqliteConnection, 88 expected: &(Vec<u8>, Vec<u8>, Vec<u8>), 89 ) { 90 let artifact: Vec<u8> = 91 sqlx::query_scalar("SELECT snapshot FROM radroots_runtime_authored_artifacts") 92 .fetch_one(&mut *connection) 93 .await 94 .unwrap(); 95 let (receipt, id): (Vec<u8>, Vec<u8>) = 96 sqlx::query_as("SELECT receipt, commit_id FROM radroots_runtime_authored_atomic_commits") 97 .fetch_one(&mut *connection) 98 .await 99 .unwrap(); 100 assert_eq!((artifact, receipt, id), *expected); 101 } 102 103 #[tokio::test] 104 async fn v15_preserves_v14_rows_and_receipts_and_prior_schema_policy_fails_closed() { 105 let mut connection = connection().await; 106 establish_runtime_version(&mut connection, 14).await; 107 let old = seed(&mut connection).await; 108 assert!(matches!( 109 migrate(&mut connection, OpenMode::ReadOnly, &plan(15, false)).await, 110 Err(Error::SchemaMigrationRequired { 111 actual: 14, 112 current: 15, 113 .. 114 }) 115 )); 116 assert_eq!( 117 migrate( 118 &mut connection, 119 OpenMode::ReadWriteExisting, 120 &plan(15, false) 121 ) 122 .await 123 .unwrap() 124 .applied(), 125 1 126 ); 127 assert_eq!(pragma(&mut connection, "user_version").await, 15); 128 retained(&mut connection, &old).await; 129 let stop: Option<String> = 130 sqlx::query_scalar("SELECT signing_stop FROM radroots_runtime_authored_artifacts") 131 .fetch_one(&mut connection) 132 .await 133 .unwrap(); 134 assert_eq!(stop, None); 135 for mode in [OpenMode::ReadOnly, OpenMode::ReadWriteExisting] { 136 assert!(matches!( 137 migrate(&mut connection, mode, &plan(14, false)).await, 138 Err(Error::SchemaTooNew { 139 actual: 15, 140 supported: 14, 141 .. 142 }) 143 )); 144 } 145 assert_eq!( 146 migrate(&mut connection, OpenMode::ReadOnly, &plan(15, false)) 147 .await 148 .unwrap() 149 .applied(), 150 0 151 ); 152 connection.close().await.unwrap(); 153 } 154 155 #[tokio::test] 156 async fn failed_v15_migration_rolls_back_column_guard_version_and_historical_bytes() { 157 let mut connection = connection().await; 158 establish_runtime_version(&mut connection, 14).await; 159 let old = seed(&mut connection).await; 160 assert!( 161 migrate( 162 &mut connection, 163 OpenMode::ReadWriteExisting, 164 &plan(15, true) 165 ) 166 .await 167 .is_err() 168 ); 169 assert_eq!(pragma(&mut connection, "user_version").await, 14); 170 retained(&mut connection, &old).await; 171 assert!( 172 sqlx::query("SELECT signing_stop FROM radroots_runtime_authored_artifacts") 173 .fetch_all(&mut connection) 174 .await 175 .is_err() 176 ); 177 assert_eq!(sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM sqlite_schema WHERE name = 'radroots_runtime_authored_artifacts_signed_fact_guard'").fetch_one(&mut connection).await.unwrap(), 0); 178 migrate_runtime(&mut connection, OpenMode::ReadWriteExisting) 179 .await 180 .unwrap(); 181 retained(&mut connection, &old).await; 182 connection.close().await.unwrap(); 183 } 184 185 #[test] 186 fn signed_fact_decision_binds_the_exact_successor_migration() { 187 let decision: serde_json::Value = serde_json::from_str(include_str!( 188 "../../../contracts/architecture/decisions/authored_signed_facts.v1.json" 189 )) 190 .unwrap(); 191 let migration = runtime::MIGRATIONS[14]; 192 assert_eq!(decision["migration"]["version"], migration.version()); 193 assert_eq!(decision["migration"]["name"], migration.name()); 194 assert_eq!(decision["migration"]["sha256"], migration.up_sha256()); 195 }