draft_submission.rs (25936B)
1 use super::*; 2 use radroots_storage::{ 3 atomic::AtomicCommitId, 4 authored_draft::{ 5 AuthoredDraft, AuthoredDraftId, AuthoredDraftRevision, AuthoredDraftStage, 6 AuthoredDraftStore, 7 }, 8 authored_draft_query::AuthoredDraftScope, 9 authored_draft_submission::{AuthoredDraftSource, PrepareFromDraft}, 10 }; 11 use sha2::{Digest, Sha256}; 12 13 fn source() -> AuthoredDraft { 14 AuthoredDraft::initial( 15 AuthoredDraftId::new([8; 16]).unwrap(), 16 *authored_plan().author().as_bytes(), 17 "fixture.partial.v1", 18 b"unfinished 0.".to_vec(), 19 AuthoredDraftStage::Draft, 20 None, 21 9, 22 ) 23 .unwrap() 24 .with_scope(AuthoredDraftScope::new([5; 32]).unwrap()) 25 .unwrap() 26 } 27 fn request(source: &AuthoredDraft, key: u8, id: u8) -> PrepareFromDraft { 28 let AuthoredAtomicCommand::Prepare(base) = prepare(7).0 else { 29 unreachable!() 30 }; 31 let operation_id = OperationInstanceId::new([id; 16]).unwrap(); 32 let artifact_id = AuthoredArtifactId::new([id; 16]).unwrap(); 33 let delivery_id = AuthoredDeliveryPlanId::new([id; 16]).unwrap(); 34 let preparation = PrepareAuthoredOperation::new( 35 AuthoredOperation::new(operation_id, vec![artifact_id], 10).unwrap(), 36 vec![ 37 AuthoredArtifact::planned(artifact_id, operation_id, 0, &authored_plan(), 10).unwrap(), 38 ], 39 vec![ 40 AuthoredDeliveryPlan::new( 41 delivery_id, 42 artifact_id, 43 base.delivery_plans()[0].intent().clone(), 44 10, 45 ) 46 .unwrap(), 47 ], 48 base.input_digest(), 49 10, 50 ) 51 .unwrap(); 52 let payload = b"complete immutable semantic intent".to_vec(); 53 let mut intent = AuthoredDraft::reconstruct( 54 AuthoredDraftId::new([id; 16]).unwrap(), 55 AuthoredDraftRevision::INITIAL, 56 *source.author(), 57 "fixture.intent.v1", 58 payload.clone(), 59 Sha256::digest(&payload).into(), 60 AuthoredDraftStage::Queued, 61 Some(operation_id), 62 10, 63 10, 64 ) 65 .unwrap(); 66 if let Some(scope) = source.scope() { 67 intent = intent.with_scope(scope).unwrap(); 68 } 69 PrepareFromDraft::new( 70 AtomicCommitId::new([key; 16]).unwrap(), 71 AuthoredDraftSource::capture(source).unwrap(), 72 intent, 73 preparation, 74 ) 75 .unwrap() 76 } 77 fn command(request: PrepareFromDraft) -> AuthoredAtomicCommand { 78 AuthoredAtomicCommand::PrepareFromDraft(Box::new(request)) 79 } 80 81 #[test] 82 fn submitted_request_replays_before_cas_and_compares_actual_fields_without_trusting_digest() { 83 block_on(async { 84 let store = MemoryStorage::default(); 85 let source = source(); 86 store 87 .append_authored_draft(source.clone(), None) 88 .await 89 .unwrap(); 90 let request = request(&source, 4, 1); 91 let cmd = command(request.clone()); 92 let receipt = store.execute_authored(cmd.clone()).await.unwrap(); 93 assert_eq!(receipt.disposition(), AtomicCommitDisposition::Committed); 94 assert_eq!( 95 receipt.outcome(), 96 &AuthoredAtomicOutcome::Submitted(Box::new(request.clone())) 97 ); 98 assert_eq!( 99 store.authored_draft_head(source.draft_id()).await.unwrap(), 100 Some(source.clone()) 101 ); 102 assert_eq!( 103 store 104 .authored_draft_head(request.intent().draft_id()) 105 .await 106 .unwrap(), 107 Some(request.intent().clone()) 108 ); 109 let newer = source 110 .successor( 111 b"later editing".to_vec(), 112 AuthoredDraftStage::Draft, 113 None, 114 11, 115 ) 116 .unwrap(); 117 store 118 .append_authored_draft(newer, Some(source.revision())) 119 .await 120 .unwrap(); 121 let replay = store.execute_authored(cmd.clone()).await.unwrap(); 122 assert_eq!(replay.disposition(), AtomicCommitDisposition::Replay); 123 assert_eq!(replay.outcome(), receipt.outcome()); 124 // This is the exact ordinary command used by later Sync signing. 125 let ordinary = AuthoredAtomicCommand::Prepare(request.preparation().clone()); 126 let ordinary_receipt = store.execute_authored(ordinary.clone()).await.unwrap(); 127 assert_eq!( 128 ordinary_receipt.disposition(), 129 AtomicCommitDisposition::Replay 130 ); 131 assert!(matches!( 132 ordinary_receipt.outcome(), 133 AuthoredAtomicOutcome::Prepared { .. } 134 )); 135 assert!(!receipt.matches_command(&ordinary)); 136 assert!(!ordinary_receipt.matches_command(&cmd)); 137 let base = request.preparation(); 138 let old_plan = &base.delivery_plans()[0]; 139 let changed_plan = AuthoredDeliveryPlan::new( 140 old_plan.plan_id(), 141 old_plan.artifact_id(), 142 AuthoredDeliveryIntent::new( 143 "changed-delivery-request", 144 old_plan.intent().target_set().clone(), 145 SatisfactionPolicy::new(SatisfactionClass::Accepted, TargetPolicy::all()), 146 101, 147 ) 148 .unwrap(), 149 10, 150 ) 151 .unwrap(); 152 let changed = PrepareFromDraft::new( 153 request.command_id(), 154 request.source().clone(), 155 request.intent().clone(), 156 PrepareAuthoredOperation::new( 157 base.operation().clone(), 158 base.artifacts().to_vec(), 159 vec![changed_plan], 160 base.input_digest(), 161 10, 162 ) 163 .unwrap(), 164 ) 165 .unwrap(); 166 let changed = command(changed); 167 assert_eq!(cmd.commit_id(), changed.commit_id()); 168 assert_eq!( 169 cmd.digest(), 170 changed.digest(), 171 "caller digest is intentionally unchanged" 172 ); 173 assert_eq!( 174 store.execute_authored(changed).await, 175 Err(Error::AtomicCommitConflict) 176 ); 177 let fresh = command(self::request(&source, 6, 2)); 178 assert_eq!( 179 store.execute_authored(fresh.clone()).await, 180 Err(Error::DraftRevisionConflict) 181 ); 182 assert!( 183 store 184 .authored_receipt(fresh.commit_id()) 185 .await 186 .unwrap() 187 .is_none() 188 ); 189 assert!( 190 store 191 .authored_operation(OperationInstanceId::new([2; 16]).unwrap()) 192 .await 193 .unwrap() 194 .is_none() 195 ); 196 assert_eq!( 197 store 198 .authored_receipt(cmd.commit_id()) 199 .await 200 .unwrap() 201 .unwrap() 202 .outcome(), 203 receipt.outcome() 204 ); 205 }); 206 } 207 208 #[test] 209 fn intentional_identical_submissions_use_distinct_commands_and_operations() { 210 block_on(async { 211 let store = MemoryStorage::default(); 212 let source = source(); 213 store 214 .append_authored_draft(source.clone(), None) 215 .await 216 .unwrap(); 217 for (key, id) in [(1, 1), (2, 2)] { 218 let value = request(&source, key, id); 219 let receipt = store 220 .execute_authored(command(value.clone())) 221 .await 222 .unwrap(); 223 assert_eq!(receipt.disposition(), AtomicCommitDisposition::Committed); 224 assert_eq!( 225 receipt.commit_id(), 226 PrepareFromDraft::commit_id_for(source.author(), value.command_id()) 227 ); 228 } 229 assert_eq!( 230 store 231 .authored_draft_heads(*source.author(), 10) 232 .await 233 .unwrap() 234 .len(), 235 3 236 ); 237 }); 238 } 239 240 #[test] 241 fn submission_wire_and_receipts_validate_bindings_and_redact_payloads() { 242 let source = source(); 243 let request = request(&source, 1, 2); 244 let wire = serde_json::to_value(&request).unwrap(); 245 assert_eq!( 246 serde_json::from_value::<PrepareFromDraft>(wire.clone()).unwrap(), 247 request 248 ); 249 assert!(!format!("{:?}", command(request.clone())).contains("semantic intent")); 250 assert!(!format!("{request:?}").contains("atomic authored plan")); 251 assert_eq!(request.source().draft_id(), source.draft_id()); 252 assert_eq!(request.source().revision(), source.revision()); 253 assert_eq!(request.source().author(), source.author()); 254 assert_eq!(request.source().payload_schema(), source.payload_schema()); 255 assert_eq!(request.source().scope(), source.scope()); 256 assert_eq!(request.source().payload_sha256(), source.payload_sha256()); 257 let invalid = [ 258 ("/command_id", serde_json::json!([0; 16].to_vec())), 259 ("/source/author", serde_json::json!([0; 32].to_vec())), 260 ("/source/payload_schema", serde_json::json!(" invalid")), 261 ("/source/stage", serde_json::json!("queued")), 262 ("/source/created_at_unix_ms", serde_json::json!(0)), 263 ("/source/updated_at_unix_ms", serde_json::json!(8)), 264 ("/source/updated_at_unix_ms", serde_json::json!(11)), 265 ("/source/draft_id", serde_json::json!([2; 16].to_vec())), 266 ("/source/author", serde_json::json!([1; 32].to_vec())), 267 ("/source/scope", serde_json::Value::Null), 268 ("/intent/revision", serde_json::json!(2)), 269 ("/intent/operation_id", serde_json::json!([3; 16].to_vec())), 270 ("/intent/updated_at_unix_ms", serde_json::json!(11)), 271 ("/preparation/requested_at_unix_ms", serde_json::json!(0)), 272 ("/preparation/artifacts", serde_json::json!([])), 273 ]; 274 for (pointer, replacement) in invalid { 275 let mut invalid = wire.clone(); 276 *invalid.pointer_mut(pointer).unwrap() = replacement; 277 assert!( 278 serde_json::from_value::<PrepareFromDraft>(invalid).is_err(), 279 "{pointer}" 280 ); 281 } 282 let cmd = command(request.clone()); 283 let outcome = AuthoredAtomicOutcome::Submitted(Box::new(request)); 284 for (id, digest, at) in [ 285 (AtomicCommitId::new([7; 16]).unwrap(), cmd.digest(), 10), 286 (cmd.commit_id(), AtomicCommitDigest::new([7; 32]), 10), 287 (cmd.commit_id(), cmd.digest(), 9), 288 ] { 289 assert!( 290 AuthoredAtomicReceipt::from_durable_parts( 291 id, 292 digest, 293 AtomicCommitDisposition::Committed, 294 at, 295 outcome.clone() 296 ) 297 .is_err() 298 ); 299 } 300 } 301 302 #[test] 303 fn submission_rejects_preexisting_work_state_and_mismatched_captured_times() { 304 let source = source(); 305 let request = request(&source, 1, 2); 306 let base = request.preparation(); 307 let reject = |operation: AuthoredOperation, 308 artifacts: Vec<AuthoredArtifact>, 309 plans: Vec<AuthoredDeliveryPlan>| { 310 let preparation = 311 PrepareAuthoredOperation::new(operation, artifacts, plans, base.input_digest(), 10) 312 .unwrap(); 313 assert_eq!( 314 PrepareFromDraft::new( 315 request.command_id(), 316 request.source().clone(), 317 request.intent().clone(), 318 preparation 319 ), 320 Err(Error::AtomicWorkflowMismatch) 321 ); 322 }; 323 for (created, updated, revision) in [(9, 10, 1), (10, 11, 1), (10, 10, 2)] { 324 reject( 325 AuthoredOperation::reconstruct( 326 base.operation().operation_id(), 327 base.operation().artifact_ids().to_vec(), 328 created, 329 updated, 330 NonZeroU64::new(revision).unwrap(), 331 ) 332 .unwrap(), 333 base.artifacts().to_vec(), 334 base.delivery_plans().to_vec(), 335 ); 336 } 337 let artifact = &base.artifacts()[0]; 338 let mut already_signed = artifact.clone(); 339 already_signed 340 .record_signed(signed(&authored_plan()), 10) 341 .unwrap(); 342 let mut claimed = artifact.clone(); 343 claimed 344 .set_signing_claim( 345 WorkClaim::new( 346 [1; 16], 347 "existing worker", 348 NonZeroU64::MIN, 349 10, 350 20, 351 NonZeroU64::MIN, 352 ) 353 .unwrap(), 354 10, 355 ) 356 .unwrap(); 357 let mut later_wire = serde_json::to_value(artifact).unwrap(); 358 later_wire["updated_at_unix_ms"] = serde_json::json!(11); 359 let other_author = AuthoredEventPlan::from_generic( 360 GenericEventDraft::new( 361 "radroots.social.geochat.v1", 362 20_000, 363 1_800_000_100, 364 vec![], 365 "another author", 366 "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", 367 ) 368 .unwrap(), 369 ) 370 .unwrap(); 371 for changed in [ 372 AuthoredArtifact::imported_signed( 373 artifact.artifact_id(), 374 artifact.operation_id(), 375 0, 376 signed(&authored_plan()), 377 10, 378 ) 379 .unwrap(), 380 already_signed, 381 claimed, 382 AuthoredArtifact::planned( 383 artifact.artifact_id(), 384 artifact.operation_id(), 385 0, 386 &authored_plan(), 387 9, 388 ) 389 .unwrap(), 390 serde_json::from_value(later_wire).unwrap(), 391 AuthoredArtifact::planned( 392 artifact.artifact_id(), 393 artifact.operation_id(), 394 0, 395 &other_author, 396 10, 397 ) 398 .unwrap(), 399 ] { 400 reject( 401 base.operation().clone(), 402 vec![changed], 403 base.delivery_plans().to_vec(), 404 ); 405 } 406 let plan = &base.delivery_plans()[0]; 407 let mut cancelled = plan.clone(); 408 cancelled.cancel(10).unwrap(); 409 let mut later_wire = serde_json::to_value(plan).unwrap(); 410 later_wire["updated_at_unix_ms"] = serde_json::json!(11); 411 for changed in [ 412 cancelled, 413 AuthoredDeliveryPlan::new(plan.plan_id(), plan.artifact_id(), plan.intent().clone(), 9) 414 .unwrap(), 415 serde_json::from_value(later_wire).unwrap(), 416 ] { 417 reject( 418 base.operation().clone(), 419 base.artifacts().to_vec(), 420 vec![changed], 421 ); 422 } 423 let original = request.intent(); 424 for (stage, operation, created) in [ 425 (AuthoredDraftStage::Draft, None, 10), 426 (AuthoredDraftStage::Queued, original.operation_id(), 9), 427 ] { 428 let intent = AuthoredDraft::reconstruct( 429 original.draft_id(), 430 original.revision(), 431 *original.author(), 432 original.payload_schema(), 433 original.payload().to_vec(), 434 *original.payload_sha256(), 435 stage, 436 operation, 437 created, 438 10, 439 ) 440 .unwrap() 441 .with_scope(original.scope().unwrap()) 442 .unwrap(); 443 assert_eq!( 444 PrepareFromDraft::new( 445 request.command_id(), 446 request.source().clone(), 447 intent, 448 base.clone() 449 ), 450 Err(Error::AtomicWorkflowMismatch) 451 ); 452 } 453 let mut ready = serde_json::to_value(&request).unwrap(); 454 ready["intent"]["stage"] = serde_json::json!("ready_to_sign"); 455 let ready: PrepareFromDraft = serde_json::from_value(ready).unwrap(); 456 assert_eq!(ready.intent().stage(), AuthoredDraftStage::ReadyToSign); 457 let cmd = command(request.clone()); 458 let ordinary = AuthoredAtomicCommand::Prepare(base.clone()); 459 let submitted = AuthoredAtomicOutcome::Submitted(Box::new(request.clone())); 460 assert!( 461 AuthoredAtomicReceipt::new(&ordinary, AtomicCommitDisposition::Committed, 10, submitted) 462 .is_err() 463 ); 464 let prepared = AuthoredAtomicOutcome::Prepared { 465 operation: base.operation().clone(), 466 artifacts: base.artifacts().to_vec(), 467 delivery_plans: base.delivery_plans().to_vec(), 468 }; 469 assert!( 470 AuthoredAtomicReceipt::new(&cmd, AtomicCommitDisposition::Committed, 10, prepared).is_err() 471 ); 472 } 473 474 #[test] 475 fn submission_cannot_adopt_unassociated_existing_intent_or_operation() { 476 block_on(async { 477 let source = source(); 478 let request = request(&source, 1, 2); 479 let empty = MemoryStorage::default(); 480 assert_eq!( 481 empty.execute_authored(command(request.clone())).await, 482 Err(Error::DraftRevisionConflict) 483 ); 484 for existing_intent in [true, false] { 485 let store = MemoryStorage::default(); 486 store 487 .append_authored_draft(source.clone(), None) 488 .await 489 .unwrap(); 490 if existing_intent { 491 store 492 .append_authored_draft(request.intent().clone(), None) 493 .await 494 .unwrap(); 495 } else { 496 store 497 .execute_authored(AuthoredAtomicCommand::Prepare( 498 request.preparation().clone(), 499 )) 500 .await 501 .unwrap(); 502 } 503 assert_eq!( 504 store.execute_authored(command(request.clone())).await, 505 Err(if existing_intent { 506 Error::DraftRevisionConflict 507 } else { 508 Error::AtomicCommitConflict 509 }) 510 ); 511 assert!( 512 store 513 .authored_receipt(request.commit_id()) 514 .await 515 .unwrap() 516 .is_none() 517 ); 518 assert_eq!( 519 store.authored_draft_head(source.draft_id()).await.unwrap(), 520 Some(source.clone()) 521 ); 522 assert_eq!( 523 store 524 .authored_operation(request.preparation().operation().operation_id()) 525 .await 526 .unwrap() 527 .is_some(), 528 !existing_intent 529 ); 530 } 531 }); 532 } 533 534 #[test] 535 fn waiting_submission_requires_explicit_construction_and_retains_all_bindings() { 536 let source = source(); 537 let ready = request(&source, 1, 2); 538 for stage in [ 539 AuthoredDraftStage::Draft, 540 AuthoredDraftStage::MediaPreparing, 541 AuthoredDraftStage::MediaUploading, 542 AuthoredDraftStage::ReadyToSign, 543 AuthoredDraftStage::Queued, 544 AuthoredDraftStage::Cancelled, 545 ] { 546 let waiting = matches!( 547 stage, 548 AuthoredDraftStage::Draft 549 | AuthoredDraftStage::MediaPreparing 550 | AuthoredDraftStage::MediaUploading 551 ); 552 let intent = AuthoredDraft::reconstruct( 553 ready.intent().draft_id(), 554 AuthoredDraftRevision::INITIAL, 555 *source.author(), 556 ready.intent().payload_schema(), 557 ready.intent().payload().to_vec(), 558 *ready.intent().payload_sha256(), 559 stage, 560 (!waiting).then_some(ready.preparation().operation().operation_id()), 561 10, 562 10, 563 ) 564 .unwrap() 565 .with_scope(source.scope().unwrap()) 566 .unwrap(); 567 let result = PrepareFromDraft::new_waiting( 568 ready.command_id(), 569 ready.source().clone(), 570 intent.clone(), 571 ready.preparation().clone(), 572 ); 573 if !waiting { 574 assert_eq!(result, Err(Error::AtomicWorkflowMismatch)); 575 continue; 576 } 577 assert_eq!( 578 PrepareFromDraft::new( 579 ready.command_id(), 580 ready.source().clone(), 581 intent, 582 ready.preparation().clone() 583 ), 584 Err(Error::AtomicWorkflowMismatch) 585 ); 586 let request = result.unwrap(); 587 assert_eq!(request.intent().operation_id(), None); 588 assert_eq!(request.commit_id(), ready.commit_id()); 589 let wire = serde_json::to_value(&request).unwrap(); 590 assert_eq!( 591 serde_json::from_value::<PrepareFromDraft>(wire.clone()).unwrap(), 592 request 593 ); 594 for (pointer, value) in [ 595 ("/intent/operation_id", serde_json::json!([2; 16].to_vec())), 596 ("/intent/revision", serde_json::json!(2)), 597 ("/intent/author", serde_json::json!([1; 32].to_vec())), 598 ("/intent/scope", serde_json::Value::Null), 599 ("/intent/updated_at_unix_ms", serde_json::json!(11)), 600 ("/intent/stage", serde_json::json!("cancelled")), 601 ("/preparation/artifacts", serde_json::json!([])), 602 ] { 603 let mut invalid = wire.clone(); 604 *invalid.pointer_mut(pointer).unwrap() = value; 605 assert!( 606 serde_json::from_value::<PrepareFromDraft>(invalid).is_err(), 607 "{stage:?} {pointer}" 608 ); 609 } 610 block_on(async { 611 let store = MemoryStorage::default(); 612 store 613 .append_authored_draft(source.clone(), None) 614 .await 615 .unwrap(); 616 let expected = store 617 .execute_authored(command(request.clone())) 618 .await 619 .unwrap(); 620 // The caller records prerequisite progress without rewriting the captured request. 621 let progress = request 622 .intent() 623 .successor( 624 b"prerequisite verified".to_vec(), 625 AuthoredDraftStage::ReadyToSign, 626 Some(request.preparation().operation().operation_id()), 627 11, 628 ) 629 .unwrap(); 630 store 631 .append_authored_draft(progress.clone(), Some(request.intent().revision())) 632 .await 633 .unwrap(); 634 assert!( 635 progress 636 .successor( 637 b"changed semantic payload".to_vec(), 638 AuthoredDraftStage::Queued, 639 progress.operation_id(), 640 12 641 ) 642 .is_err() 643 ); 644 let queued = progress 645 .successor( 646 progress.payload().to_vec(), 647 AuthoredDraftStage::Queued, 648 progress.operation_id(), 649 12, 650 ) 651 .unwrap(); 652 store 653 .append_authored_draft(queued.clone(), Some(progress.revision())) 654 .await 655 .unwrap(); 656 let edit = source 657 .successor( 658 b"later composer edit".to_vec(), 659 AuthoredDraftStage::Draft, 660 None, 661 13, 662 ) 663 .unwrap(); 664 store 665 .append_authored_draft(edit.clone(), Some(source.revision())) 666 .await 667 .unwrap(); 668 let replay = store 669 .execute_authored(command(request.clone())) 670 .await 671 .unwrap(); 672 assert_eq!(replay.disposition(), AtomicCommitDisposition::Replay); 673 assert_eq!(replay.outcome(), expected.outcome()); 674 assert_eq!( 675 store.authored_draft_head(queued.draft_id()).await.unwrap(), 676 Some(queued) 677 ); 678 for changed_pointer in ["/intent/payload", "/source/payload_sha256", "/intent/stage"] { 679 let mut changed = wire.clone(); 680 if changed_pointer == "/intent/payload" { 681 let payload = b"different captured intent"; 682 changed["intent"]["payload"] = serde_json::json!(payload.to_vec()); 683 changed["intent"]["payload_sha256"] = 684 serde_json::json!(Sha256::digest(payload).to_vec()); 685 } else if changed_pointer == "/source/payload_sha256" { 686 changed["source"]["payload_sha256"] = serde_json::json!([7; 32].to_vec()); 687 } else { 688 changed["intent"]["stage"] = 689 serde_json::json!(if stage == AuthoredDraftStage::Draft { 690 "media_preparing" 691 } else { 692 "draft" 693 }); 694 } 695 let changed = serde_json::from_value::<PrepareFromDraft>(changed).unwrap(); 696 assert_eq!( 697 changed.preparation().input_digest(), 698 request.preparation().input_digest() 699 ); 700 if changed_pointer == "/intent/stage" { 701 assert_eq!( 702 command(changed.clone()).digest(), 703 command(request.clone()).digest() 704 ); 705 } 706 assert_eq!( 707 store.execute_authored(command(changed)).await, 708 Err(Error::AtomicCommitConflict) 709 ); 710 } 711 let mut fresh = wire.clone(); 712 fresh["command_id"] = serde_json::json!([3; 16].to_vec()); 713 let fresh = serde_json::from_value::<PrepareFromDraft>(fresh).unwrap(); 714 assert_eq!( 715 store.execute_authored(command(fresh.clone())).await, 716 Err(Error::DraftRevisionConflict) 717 ); 718 assert!( 719 store 720 .authored_receipt(fresh.commit_id()) 721 .await 722 .unwrap() 723 .is_none() 724 ); 725 assert_eq!( 726 store.authored_draft_head(source.draft_id()).await.unwrap(), 727 Some(edit) 728 ); 729 }); 730 } 731 }