405 lines
16 KiB
Rust
405 lines
16 KiB
Rust
use merge_request::*;
|
|
use rusqlite::{Connection, params};
|
|
use tempfile::TempDir;
|
|
|
|
fn setup() -> (TempDir, SqliteMergeRequestStore) {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("server.db");
|
|
let conn = Connection::open(&path).unwrap();
|
|
conn.execute_batch(r#"
|
|
PRAGMA foreign_keys=ON;
|
|
CREATE TABLE repositories(workspace_id TEXT NOT NULL,repository_id TEXT NOT NULL,PRIMARY KEY(workspace_id,repository_id));
|
|
CREATE TABLE typed_tickets(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,workflow_state TEXT NOT NULL,workflow_state_explicit INTEGER NOT NULL DEFAULT 1,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));
|
|
CREATE TABLE typed_ticket_events(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,event_index INTEGER NOT NULL,kind TEXT NOT NULL,author TEXT,at TEXT,status TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));
|
|
CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,event_index INTEGER NOT NULL,key TEXT NOT NULL,value TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id,event_index,key));
|
|
CREATE TABLE ticket_worker_assignments(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,assignment_id TEXT NOT NULL,runtime_id TEXT NOT NULL,worker_id TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id,assignment_id));
|
|
CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,assignment_id TEXT NOT NULL,runtime_id TEXT NOT NULL,worker_id TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));
|
|
"#).unwrap();
|
|
for ws in ["ws-a", "ws-b"] {
|
|
conn.execute("INSERT INTO repositories VALUES(?1,'repo')", params![ws])
|
|
.unwrap();
|
|
conn.execute(
|
|
"INSERT INTO typed_tickets VALUES(?1,'T1','inprogress',1,'t0')",
|
|
params![ws],
|
|
)
|
|
.unwrap();
|
|
conn.execute(
|
|
"INSERT INTO ticket_worker_assignments VALUES(?1,'T1','A1','R1','W1')",
|
|
params![ws],
|
|
)
|
|
.unwrap();
|
|
conn.execute(
|
|
"INSERT INTO ticket_current_worker_assignments VALUES(?1,'T1','A1','R1','W1')",
|
|
params![ws],
|
|
)
|
|
.unwrap();
|
|
}
|
|
drop(conn);
|
|
let store = SqliteMergeRequestStore::open(&path, "ws-a").unwrap();
|
|
(dir, store)
|
|
}
|
|
fn revision(id: &str, ordinal: u64, head: &str) -> MergeRequestRevision {
|
|
MergeRequestRevision {
|
|
revision_id: id.into(),
|
|
ordinal,
|
|
base_commit: "base".into(),
|
|
head_commit: head.into(),
|
|
head_tree: format!("tree-{head}"),
|
|
diff_digest: format!("sha256:diff-{head}"),
|
|
changed_paths: vec!["src/lib.rs".into()],
|
|
summary: format!("revision {id}"),
|
|
assignment_id: "A1".into(),
|
|
created_at: format!("t{ordinal}"),
|
|
}
|
|
}
|
|
fn open(store: &SqliteMergeRequestStore) {
|
|
store
|
|
.open_merge_request(OpenMergeRequest {
|
|
merge_request_id: "MR1".into(),
|
|
ticket_id: "T1".into(),
|
|
repository_id: "repo".into(),
|
|
revision: revision("V1", 1, "h1"),
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "t1".into(),
|
|
})
|
|
.unwrap();
|
|
}
|
|
fn attempt(store: &SqliteMergeRequestStore, id: &str, revision: &str, token: &str, child: &str) {
|
|
store
|
|
.register_reviewer_child_session(RegisterReviewerChildSession {
|
|
parent_runtime_id: "R1".into(),
|
|
parent_worker_id: "W1".into(),
|
|
child_session_id: child.into(),
|
|
now: "t".into(),
|
|
})
|
|
.unwrap();
|
|
store
|
|
.register_review_attempt(RegisterReviewAttempt {
|
|
attempt_id: id.into(),
|
|
ticket_id: "T1".into(),
|
|
revision_id: revision.into(),
|
|
parent_assignment_id: "A1".into(),
|
|
parent_runtime_id: "R1".into(),
|
|
parent_worker_id: "W1".into(),
|
|
child_session_id: child.into(),
|
|
capability_token: token.into(),
|
|
now: "t".into(),
|
|
})
|
|
.unwrap();
|
|
}
|
|
fn review(
|
|
store: &SqliteMergeRequestStore,
|
|
revision: &str,
|
|
token: &str,
|
|
decision: ReviewDecision,
|
|
) -> Result<MergeRequestReview> {
|
|
store.submit_review(SubmitReview {
|
|
ticket_id: "T1".into(),
|
|
revision_id: revision.into(),
|
|
capability_token: token.into(),
|
|
decision,
|
|
body: "evidence".into(),
|
|
findings: vec![],
|
|
now: "tr".into(),
|
|
})
|
|
}
|
|
|
|
#[test]
|
|
fn storage_allows_multiple_merge_requests_for_one_ticket() {
|
|
let (_dir, store) = setup();
|
|
open(&store);
|
|
store
|
|
.open_merge_request(OpenMergeRequest {
|
|
merge_request_id: "MR2".into(),
|
|
ticket_id: "T1".into(),
|
|
repository_id: "repo".into(),
|
|
revision: revision("V2", 1, "h2"),
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "t2".into(),
|
|
})
|
|
.unwrap();
|
|
let conn = Connection::open(store.db_path()).unwrap();
|
|
let count:i64=conn.query_row("SELECT COUNT(*) FROM merge_request_ticket_relations WHERE workspace_id='ws-a' AND ticket_id='T1'",[],|row|row.get(0)).unwrap();
|
|
assert_eq!(count, 2);
|
|
assert_eq!(
|
|
store
|
|
.show_for_ticket("T1")
|
|
.unwrap()
|
|
.unwrap()
|
|
.merge_request_id,
|
|
"MR2"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn bounded_context_rejects_oversized_revision_evidence() {
|
|
let (_dir, store) = setup();
|
|
let mut oversized = revision("V1", 1, "h1");
|
|
oversized.changed_paths = (0..=1_000).map(|i| format!("src/{i}.rs")).collect();
|
|
let result = store.open_merge_request(OpenMergeRequest {
|
|
merge_request_id: "MR1".into(),
|
|
ticket_id: "T1".into(),
|
|
repository_id: "repo".into(),
|
|
revision: oversized,
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "t".into(),
|
|
});
|
|
assert!(matches!(
|
|
result,
|
|
Err(MergeRequestError::TooLarge {
|
|
field: "revision.changed_paths",
|
|
..
|
|
})
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn rejected_v6_schema_missing_diff_digest_is_archived_before_fresh_v7() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("legacy.db");
|
|
let conn = Connection::open(&path).unwrap();
|
|
conn.execute_batch(
|
|
"CREATE TABLE merge_request_schema_migrations(version INTEGER PRIMARY KEY,name TEXT NOT NULL,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);\
|
|
INSERT INTO merge_request_schema_migrations(version,name) VALUES(6,'rejected_merge_request_v6');\
|
|
CREATE TABLE repositories(workspace_id TEXT NOT NULL,repository_id TEXT NOT NULL,PRIMARY KEY(workspace_id,repository_id));\
|
|
CREATE TABLE typed_tickets(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,workflow_state TEXT NOT NULL,workflow_state_explicit INTEGER NOT NULL DEFAULT 1,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));\
|
|
CREATE TABLE ticket_worker_assignments(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,assignment_id TEXT NOT NULL,runtime_id TEXT NOT NULL,worker_id TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id,assignment_id));\
|
|
CREATE TABLE merge_requests(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,repository_id TEXT NOT NULL,state TEXT NOT NULL,lifecycle_generation INTEGER NOT NULL,current_revision_id TEXT NOT NULL,created_at TEXT NOT NULL,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id));\
|
|
CREATE TABLE merge_request_ticket_relations(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,ticket_id TEXT NOT NULL,relation_kind TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id,ticket_id));\
|
|
CREATE TABLE merge_request_revisions(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,revision_id TEXT NOT NULL,ordinal INTEGER NOT NULL,base_commit TEXT NOT NULL,head_commit TEXT NOT NULL,head_tree TEXT NOT NULL,assignment_id TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id,revision_id));",
|
|
).unwrap();
|
|
drop(conn);
|
|
let store = SqliteMergeRequestStore::open(&path, "ws-a").unwrap();
|
|
assert!(store.show_for_ticket("missing").unwrap().is_none());
|
|
let conn = Connection::open(&path).unwrap();
|
|
let archived: i64 = conn.query_row("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='legacy_v6_merge_requests'",[],|row|row.get(0)).unwrap();
|
|
assert_eq!(archived, 1);
|
|
for table in [
|
|
"merge_request_review_attempts",
|
|
"merge_request_completion_operations",
|
|
] {
|
|
let present: i64 = conn
|
|
.query_row(
|
|
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?1",
|
|
params![table],
|
|
|row| row.get(0),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(present, 1);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn request_changes_new_revision_resets_and_exact_completion_replay_converges() {
|
|
let (_dir, store) = setup();
|
|
open(&store);
|
|
attempt(&store, "AT1", "V1", "tok1", "child1");
|
|
review(&store, "V1", "tok1", ReviewDecision::RequestChanges).unwrap();
|
|
assert_eq!(
|
|
store.show_for_ticket("T1").unwrap().unwrap().review_status,
|
|
ReviewStatus::ChangesRequested
|
|
);
|
|
store
|
|
.add_revision(AddRevision {
|
|
ticket_id: "T1".into(),
|
|
expected_current_revision_id: "V1".into(),
|
|
revision: revision("V2", 2, "h2"),
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "t2".into(),
|
|
})
|
|
.unwrap();
|
|
assert_eq!(
|
|
store.show_for_ticket("T1").unwrap().unwrap().review_status,
|
|
ReviewStatus::Pending
|
|
);
|
|
assert!(review(&store, "V1", "tok1", ReviewDecision::Approve).is_err());
|
|
attempt(&store, "AT2", "V2", "tok2", "child2");
|
|
review(&store, "V2", "tok2", ReviewDecision::Approve).unwrap();
|
|
let input = CompleteMergeRequest {
|
|
operation_id: "OP1".into(),
|
|
ticket_id: "T1".into(),
|
|
expected_revision_id: "V2".into(),
|
|
assignment_id: "A1".into(),
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "tc".into(),
|
|
};
|
|
let first = store.complete(input.clone()).unwrap();
|
|
assert!(!first.replayed);
|
|
let replay = store.complete(input).unwrap();
|
|
assert!(replay.replayed);
|
|
assert!(matches!(
|
|
store.confirm_merge(MergeConfirmation {
|
|
ticket_id: "T1".into(),
|
|
expected_revision_id: "V2".into(),
|
|
authenticated_account_id: "runtime".into(),
|
|
actor_kind: "worker".into(),
|
|
explicit_confirmation: true,
|
|
now: "tm".into()
|
|
}),
|
|
Err(MergeRequestError::MergeConfirmationRequired)
|
|
));
|
|
let merged = store
|
|
.confirm_merge(MergeConfirmation {
|
|
ticket_id: "T1".into(),
|
|
expected_revision_id: "V2".into(),
|
|
authenticated_account_id: "account-1".into(),
|
|
actor_kind: "user".into(),
|
|
explicit_confirmation: true,
|
|
now: "tm".into(),
|
|
})
|
|
.unwrap();
|
|
assert_eq!(merged.state, MergeRequestState::Merged);
|
|
let conn = Connection::open(store.db_path()).unwrap();
|
|
assert_eq!(
|
|
conn.query_row(
|
|
"SELECT workflow_state FROM typed_tickets WHERE workspace_id='ws-a' AND ticket_id='T1'",
|
|
[],
|
|
|r| r.get::<_, String>(0)
|
|
)
|
|
.unwrap(),
|
|
"done"
|
|
);
|
|
assert_eq!(
|
|
conn.query_row(
|
|
"SELECT COUNT(*) FROM typed_ticket_events WHERE workspace_id='ws-a' AND ticket_id='T1'",
|
|
[],
|
|
|r| r.get::<_, i64>(0)
|
|
)
|
|
.unwrap(),
|
|
1
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn spoof_self_approval_replay_and_cross_workspace_are_rejected() {
|
|
let (_dir, store) = setup();
|
|
open(&store);
|
|
let mut bad = RegisterReviewAttempt {
|
|
attempt_id: "bad".into(),
|
|
ticket_id: "T1".into(),
|
|
revision_id: "V1".into(),
|
|
parent_assignment_id: "A1".into(),
|
|
parent_runtime_id: "R1".into(),
|
|
parent_worker_id: "W1".into(),
|
|
child_session_id: "W1".into(),
|
|
capability_token: "bad".into(),
|
|
now: "t".into(),
|
|
};
|
|
assert!(matches!(
|
|
store.register_review_attempt(bad.clone()),
|
|
Err(MergeRequestError::SelfApproval)
|
|
));
|
|
bad.child_session_id = "child".into();
|
|
assert!(matches!(
|
|
store.register_review_attempt(bad),
|
|
Err(MergeRequestError::InvalidReviewer)
|
|
));
|
|
attempt(&store, "AT", "V1", "secret", "child");
|
|
assert!(review(&store, "V1", "spoof", ReviewDecision::Approve).is_err());
|
|
review(&store, "V1", "secret", ReviewDecision::Approve).unwrap();
|
|
assert!(review(&store, "V1", "secret", ReviewDecision::Approve).is_err());
|
|
let other = SqliteMergeRequestStore::open_verified(store.db_path(), "ws-b").unwrap();
|
|
assert!(other.show_for_ticket("T1").unwrap().is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn reopen_resets_approval_and_merge_requires_authenticated_explicit_user() {
|
|
let (_dir, store) = setup();
|
|
open(&store);
|
|
attempt(&store, "AT", "V1", "token", "child");
|
|
review(&store, "V1", "token", ReviewDecision::Approve).unwrap();
|
|
store.close("T1", "V1", "tc").unwrap();
|
|
let reopened = store.reopen("T1", "V1", "tr").unwrap();
|
|
assert_eq!(reopened.review_status, ReviewStatus::Pending);
|
|
let denied = store.confirm_merge(MergeConfirmation {
|
|
ticket_id: "T1".into(),
|
|
expected_revision_id: "V1".into(),
|
|
authenticated_account_id: "user".into(),
|
|
actor_kind: "user".into(),
|
|
explicit_confirmation: false,
|
|
now: "tm".into(),
|
|
});
|
|
assert!(matches!(
|
|
denied,
|
|
Err(MergeRequestError::MergeConfirmationRequired)
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn concurrent_exact_completion_replays_commit_one_ticket_side_effect() {
|
|
let (_dir, store) = setup();
|
|
open(&store);
|
|
attempt(&store, "AT", "V1", "token", "child");
|
|
review(&store, "V1", "token", ReviewDecision::Approve).unwrap();
|
|
let input = CompleteMergeRequest {
|
|
operation_id: "OP-concurrent".into(),
|
|
ticket_id: "T1".into(),
|
|
expected_revision_id: "V1".into(),
|
|
assignment_id: "A1".into(),
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "t".into(),
|
|
};
|
|
let left_store = store.clone();
|
|
let left_input = input.clone();
|
|
let left = std::thread::spawn(move || left_store.complete(left_input));
|
|
let right_store = store.clone();
|
|
let right = std::thread::spawn(move || right_store.complete(input));
|
|
let outcomes = [
|
|
left.join().unwrap().unwrap(),
|
|
right.join().unwrap().unwrap(),
|
|
];
|
|
assert_eq!(
|
|
outcomes.iter().filter(|outcome| !outcome.replayed).count(),
|
|
1
|
|
);
|
|
assert_eq!(
|
|
outcomes.iter().filter(|outcome| outcome.replayed).count(),
|
|
1
|
|
);
|
|
let conn = Connection::open(store.db_path()).unwrap();
|
|
let events: i64 = conn
|
|
.query_row(
|
|
"SELECT COUNT(*) FROM typed_ticket_events WHERE workspace_id='ws-a' AND ticket_id='T1'",
|
|
[],
|
|
|row| row.get(0),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(events, 1);
|
|
}
|
|
|
|
#[test]
|
|
fn operation_key_mismatch_and_assignment_takeover_are_fenced() {
|
|
let (_dir, store) = setup();
|
|
open(&store);
|
|
attempt(&store, "AT", "V1", "token", "child");
|
|
review(&store, "V1", "token", ReviewDecision::Approve).unwrap();
|
|
let mut input = CompleteMergeRequest {
|
|
operation_id: "OP".into(),
|
|
ticket_id: "T1".into(),
|
|
expected_revision_id: "V1".into(),
|
|
assignment_id: "A1".into(),
|
|
authenticated_runtime_id: "R1".into(),
|
|
authenticated_worker_id: "W1".into(),
|
|
now: "t".into(),
|
|
};
|
|
let conn = Connection::open(store.db_path()).unwrap();
|
|
conn.execute("UPDATE ticket_current_worker_assignments SET assignment_id='A2',runtime_id='R2',worker_id='W2' WHERE workspace_id='ws-a' AND ticket_id='T1'",[]).unwrap();
|
|
assert!(matches!(
|
|
store.complete(input.clone()),
|
|
Err(MergeRequestError::AssignmentMismatch)
|
|
));
|
|
conn.execute("UPDATE ticket_current_worker_assignments SET assignment_id='A1',runtime_id='R1',worker_id='W1' WHERE workspace_id='ws-a' AND ticket_id='T1'",[]).unwrap();
|
|
store.complete(input.clone()).unwrap();
|
|
input.expected_revision_id = "other".into();
|
|
assert!(matches!(
|
|
store.complete(input),
|
|
Err(MergeRequestError::OperationConflict)
|
|
));
|
|
}
|