alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-612 Test Mongo CDC multiple $set operations

Browse Source
This commit is contained in:
Renato Mefi 2018-11-08 13:43:17 +01:00 committed by Gunnar Morling
parent 7a4e1c48f1
commit d44a44aee5
1 changed files with 21 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/transforms/UnwrapFromMongoDbEnvelopeTestIT.java
View File

@ -136,6 +136,26 @@ public void shouldTransformEvents() throws InterruptedException, IOException {
assertThat(transformedUpdateValue.get("id")).isEqualTo(1);
assertThat(transformedUpdateValue.get("dataStr")).isEqualTo("bye");
// Test Update Multiple Fields
primary().execute("update", client -> {
client.getDatabase(DB_NAME).getCollection(COLLECTION_NAME).updateOne(RawBsonDocument.parse("{'_id' : 1}"),
RawBsonDocument.parse("{'$set': {'newStr': 'hello', 'dataInt': 456}}"));
});
records = consumeRecordsByTopic(1);
assertThat(records.recordsForTopic(TOPIC_NAME).size()).isEqualTo(1);
final SourceRecord updateMultipleRecord = records.recordsForTopic(TOPIC_NAME).get(0);
final SourceRecord transformedMultipleUpdate = transformation.apply(updateMultipleRecord);
final Struct transformedMultipleUpdateValue = (Struct)transformedMultipleUpdate.value();
assertThat(transformedMultipleUpdate.valueSchema().field("id").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
assertThat(transformedMultipleUpdate.valueSchema().field("newStr").schema()).isEqualTo(Schema.OPTIONAL_STRING_SCHEMA);
assertThat(transformedMultipleUpdate.valueSchema().field("dataInt").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
assertThat(transformedMultipleUpdateValue.get("id")).isEqualTo(1);
assertThat(transformedMultipleUpdateValue.get("newStr")).isEqualTo("hello");
assertThat(transformedMultipleUpdateValue.get("dataInt")).isEqualTo(456);
// Test update
primary().execute("delete", client -> {
client.getDatabase(DB_NAME).getCollection(COLLECTION_NAME).deleteOne(RawBsonDocument.parse("{'_id' : 1}"));