diff --git a/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaConsumer.java b/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaConsumer.java index ef09e0b..ee4683b 100644 --- a/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaConsumer.java +++ b/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaConsumer.java @@ -123,10 +123,10 @@ protected MatchContext buildMatchContext() { .matchMethod(context.getMatchMethod()) .matchFilePaths(context.getMatchFilePaths()) .matchKey(context.getExpectedRecordKey()) + .matchKeys(context.getExpectedRecordKeys()) .keyMatchStrategy(context.getKeyMatchStrategy()) .excludedFields(context.getExcludedFields()) .strictMatching(false) - .keyMatchStrategy(context.getKeyMatchStrategy()) .batchMatchStrategy(context.getBatchMatchStrategy()) .build(); } diff --git a/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaProducer.java b/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaProducer.java index 87ad8cd..a16c136 100644 --- a/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaProducer.java +++ b/src/main/java/io/github/ktestify/io/kafka/AbstractKafkaProducer.java @@ -145,11 +145,13 @@ protected String resolveSchema() throws IOException, RestClientException { } else { try { int version = Integer.parseInt(context.getSchemaVersion()); - logger.info("Resolving Avro schema for subject '{}' (version {}) from Schema Registry.", subject, version); + logger.info( + "Resolving Avro schema for subject '{}' (version {}) from Schema Registry.", subject, version); return client.getSchemaMetadata(subject, version).getSchema(); } catch (NumberFormatException e) { throw new ProducerException( - "Invalid schema version '" + context.getSchemaVersion() + "' for subject '" + subject + "'. Expected an integer.", + "Invalid schema version '" + context.getSchemaVersion() + "' for subject '" + subject + + "'. Expected an integer.", e); } } diff --git a/src/main/java/io/github/ktestify/io/kafka/ConsumerContext.java b/src/main/java/io/github/ktestify/io/kafka/ConsumerContext.java index 6adb3e0..9bd73ca 100644 --- a/src/main/java/io/github/ktestify/io/kafka/ConsumerContext.java +++ b/src/main/java/io/github/ktestify/io/kafka/ConsumerContext.java @@ -33,6 +33,7 @@ public final class ConsumerContext { private final Map properties; private final Consumer consumer; private final String expectedRecordKey; + private final List expectedRecordKeys; private final KeyMatchStrategy keyMatchStrategy; private final BatchMatchStrategy batchMatchStrategy; private final String matchMethod; @@ -49,6 +50,7 @@ private ConsumerContext( Map properties, Consumer consumer, String expectedRecordKey, + List expectedRecordKeys, KeyMatchStrategy keyMatchStrategy, BatchMatchStrategy batchMatchStrategy, String matchMethod, @@ -63,6 +65,7 @@ private ConsumerContext( this.properties = properties; this.consumer = consumer; this.expectedRecordKey = expectedRecordKey; + this.expectedRecordKeys = expectedRecordKeys != null ? expectedRecordKeys : Collections.emptyList(); this.keyMatchStrategy = keyMatchStrategy != null ? keyMatchStrategy : KeyMatchStrategy.EXACT; this.batchMatchStrategy = batchMatchStrategy != null ? batchMatchStrategy : BatchMatchStrategy.POSITIONAL; this.matchMethod = matchMethod; @@ -93,6 +96,7 @@ public static final class Builder { private Map properties; private Consumer consumer; private String expectedRecordKey; + private List expectedRecordKeys; private KeyMatchStrategy keyMatchStrategy; private BatchMatchStrategy batchMatchStrategy; private String matchMethod; @@ -124,6 +128,21 @@ public Builder expectedRecordKey(String expectedRecordKey) { return this; } + /** + * Sets the expected record keys for batch consumers. + * + *

When non-empty, {@code KafkaRecordFetcher} pre-filters records against this list of expected keys, and + * batch matchers assert each record's key alongside its value according to {@link #batchMatchStrategy}. + * + * @param expectedRecordKeys the expected record keys, or {@code null} for an empty list + * @return this builder + * @since 1.2.6 + */ + public Builder expectedRecordKeys(List expectedRecordKeys) { + this.expectedRecordKeys = expectedRecordKeys; + return this; + } + /** * Sets the strategy used to compare {@link #expectedRecordKey} against the actual record key during the * fetch-time pre-filter in {@code KafkaRecordFetcher.passesKeyFilter()}. @@ -235,6 +254,7 @@ public ConsumerContext build() { validatedProps, validatedConsumer, expectedRecordKey, + expectedRecordKeys, keyMatchStrategy, batchMatchStrategy, matchMethod, diff --git a/src/main/java/io/github/ktestify/io/kafka/KafkaRecordFetcher.java b/src/main/java/io/github/ktestify/io/kafka/KafkaRecordFetcher.java index 54a847d..2e3e62b 100644 --- a/src/main/java/io/github/ktestify/io/kafka/KafkaRecordFetcher.java +++ b/src/main/java/io/github/ktestify/io/kafka/KafkaRecordFetcher.java @@ -307,15 +307,20 @@ private List> pollUntilRecordFound(long deadlineNanos) throws } List> accumulated = new ArrayList<>(); + String keyFilterDesc = ""; + if (context.getExpectedRecordKeys() != null + && !context.getExpectedRecordKeys().isEmpty()) { + keyFilterDesc = context.getExpectedRecordKeys() + " (" + context.getKeyMatchStrategy() + ")"; + } else if (context.getExpectedRecordKey() != null) { + keyFilterDesc = "'" + context.getExpectedRecordKey() + "' (" + context.getKeyMatchStrategy() + ")"; + } log.info( "Polling topic '{}' in {} mode (target={} record(s), timeout={}ms, expectedKey={}).", namespacedTopic, batchMode ? "BATCH" : "SINGLE", targetSize, readTimeoutMs, - context.getExpectedRecordKey() != null - ? "'" + context.getExpectedRecordKey() + "' (" + context.getKeyMatchStrategy() + ")" - : ""); + keyFilterDesc); while (System.nanoTime() < deadlineNanos) { ConsumerRecords batch = kafkaConsumer.poll(Duration.ofMillis(pollIntervalMs)); @@ -412,7 +417,22 @@ private void release(ConsumedRecord record) { * context / properties using the configured {@link io.github.ktestify.match.KeyMatchStrategy}. */ private boolean passesKeyFilter(ConsumerRecord record) { - // Context takes priority over properties map + String recordKey = record.key() != null ? record.key().toString() : null; + + // 1. Batch expectedRecordKeys list (if non-empty) + List expectedKeys = context.getExpectedRecordKeys(); + if (expectedKeys != null && !expectedKeys.isEmpty()) { + for (String expectedKey : expectedKeys) { + if (context.getKeyMatchStrategy().matches(expectedKey, recordKey)) { + log.info(MESSAGE_CONSUMER_RECORD_MATCHES_EXPECTED_KEY, expectedKey); + return true; + } + } + log.debug(MESSAGE_CONSUMER_RECORD_DO_NOT_MATCHES_EXPECTED_KEY, recordKey, expectedKeys); + return false; + } + + // 2. Single expectedRecordKey (context takes priority over properties map) String expectedKey = context.getExpectedRecordKey(); if (expectedKey == null || expectedKey.isEmpty()) { expectedKey = properties.get("expectedRecordKey"); @@ -422,7 +442,6 @@ private boolean passesKeyFilter(ConsumerRecord record) { return true; // No filter configured: accept any key } - String recordKey = record.key() != null ? record.key().toString() : null; if (context.getKeyMatchStrategy().matches(expectedKey, recordKey)) { log.info(MESSAGE_CONSUMER_RECORD_MATCHES_EXPECTED_KEY, expectedKey); return true; diff --git a/src/main/java/io/github/ktestify/match/MatchContext.java b/src/main/java/io/github/ktestify/match/MatchContext.java index fbc9c76..8653ba9 100644 --- a/src/main/java/io/github/ktestify/match/MatchContext.java +++ b/src/main/java/io/github/ktestify/match/MatchContext.java @@ -70,6 +70,19 @@ public class MatchContext { /** Optional key whose value must match a specific value (key-value matching mode). */ String matchKey; + /** + * Expected record keys for batch matching. + * + *

When non-empty in batch matchers ({@code FileRecordMatcher}, {@code AvroFileRecordMatcher}), must have the + * same size as {@link #matchFilePaths}. In {@link BatchMatchStrategy#POSITIONAL} mode, record {@code i} must match + * both {@code matchKeys.get(i)} and {@code matchFilePaths.get(i)}; in {@link BatchMatchStrategy#LAZY} mode, each + * record claims the first unmatched {@code (key, file)} pair where both key and value match. + * + * @since 1.2.6 + */ + @Builder.Default + List matchKeys = Collections.emptyList(); + /** Expected value for {@link #matchKey}. */ String matchValue; diff --git a/src/main/java/io/github/ktestify/match/impl/AvroFileRecordMatcher.java b/src/main/java/io/github/ktestify/match/impl/AvroFileRecordMatcher.java index d488bc0..812b29f 100644 --- a/src/main/java/io/github/ktestify/match/impl/AvroFileRecordMatcher.java +++ b/src/main/java/io/github/ktestify/match/impl/AvroFileRecordMatcher.java @@ -16,10 +16,7 @@ package io.github.ktestify.match.impl; import io.github.ktestify.exceptions.ComparisonException; -import io.github.ktestify.match.BatchMatchStrategy; -import io.github.ktestify.match.MatchContext; -import io.github.ktestify.match.MatchResult; -import io.github.ktestify.match.RecordMatcher; +import io.github.ktestify.match.*; import io.github.ktestify.models.ConsumedRecord; import io.github.ktestify.utils.FileUtils; import io.github.ktestify.utils.serdes.AvroUtils; @@ -51,8 +48,13 @@ public MatchResult match(List> records, MatchConte if (filePaths == null || filePaths.isEmpty() || filePaths.stream().anyMatch(p -> p == null || p.isBlank())) { throw new ComparisonException("AvroFileRecordMatcher requires matchFilePath(s) to be set."); } + List expectedKeys = context.getMatchKeys() != null ? context.getMatchKeys() : List.of(); + if (!expectedKeys.isEmpty() && expectedKeys.size() != filePaths.size()) { + throw new ComparisonException("Expected record keys count (" + expectedKeys.size() + + ") does not match expected file count (" + filePaths.size() + ")."); + } - if (records.size() == 1 && filePaths.size() == 1) { + if (records.size() == 1 && filePaths.size() == 1 && expectedKeys.isEmpty()) { return matchSingle(records.getFirst(), filePaths.getFirst(), context); } @@ -66,9 +68,9 @@ public MatchResult match(List> records, MatchConte : BatchMatchStrategy.POSITIONAL; if (strategy == BatchMatchStrategy.LAZY) { - return matchBatchLazy(records, filePaths, context); + return matchBatchLazy(records, filePaths, expectedKeys, context); } - return matchBatchPositional(records, filePaths, context); + return matchBatchPositional(records, filePaths, expectedKeys, context); } private MatchResult matchSingle(ConsumedRecord record, String filePath, MatchContext context) { @@ -95,13 +97,37 @@ private MatchResult matchSingle(ConsumedRecord record, String fil } private MatchResult matchBatchPositional( - List> records, List filePaths, MatchContext context) { + List> records, + List filePaths, + List expectedKeys, + MatchContext context) { + KeyMatchStrategy keyStrategy = + context.getKeyMatchStrategy() != null ? context.getKeyMatchStrategy() : KeyMatchStrategy.EXACT; String lastExpected = ""; String lastActual = ""; for (int i = 0; i < records.size(); i++) { String filePath = filePaths.get(i); String expectedValue = FileUtils.getFileContent(FileUtils.getFile(filePath)); GenericRecord value = records.get(i).getValue(); + String actualKey = records.get(i).getKey(); + + if (!expectedKeys.isEmpty()) { + String expectedKey = expectedKeys.get(i); + if (!keyStrategy.matches(expectedKey, actualKey)) { + log.error( + "Avro record at index {} key mismatch: expected '{}' ({}), actual '{}'.", + i, + expectedKey, + keyStrategy, + actualKey); + return MatchResult.fail( + "Avro record at index " + i + " key does not match, expected: '" + expectedKey + + "', actual: '" + actualKey + "', strategy: " + keyStrategy + ".", + expectedKey, + actualKey != null ? actualKey : ""); + } + } + if (value == null) { log.error("Avro record at index {} has null value (tombstone), expected content of '{}'.", i, filePath); return MatchResult.nullValue(expectedValue); @@ -119,7 +145,15 @@ private MatchResult matchBatchPositional( expectedValue, actualValue); } - log.info("Avro record at index {} matches expected file '{}'.", i, filePath); + if (!expectedKeys.isEmpty()) { + log.info( + "Avro record at index {} matches expected key '{}' and file '{}'.", + i, + expectedKeys.get(i), + filePath); + } else { + log.info("Avro record at index {} matches expected file '{}'.", i, filePath); + } lastExpected = expectedValue; lastActual = actualValue; } @@ -127,7 +161,12 @@ private MatchResult matchBatchPositional( } private MatchResult matchBatchLazy( - List> records, List filePaths, MatchContext context) { + List> records, + List filePaths, + List expectedKeys, + MatchContext context) { + KeyMatchStrategy keyStrategy = + context.getKeyMatchStrategy() != null ? context.getKeyMatchStrategy() : KeyMatchStrategy.EXACT; List expectedContents = filePaths.stream() .map(path -> FileUtils.getFileContent(FileUtils.getFile(path))) .toList(); @@ -137,6 +176,7 @@ private MatchResult matchBatchLazy( for (int i = 0; i < records.size(); i++) { GenericRecord value = records.get(i).getValue(); + String actualKey = records.get(i).getKey(); if (value == null) { log.error("Avro record at index {} has null value (tombstone).", i); return MatchResult.nullValue(expectedContents.getFirst()); @@ -145,7 +185,9 @@ private MatchResult matchBatchLazy( int matchedIndex = -1; for (int j = 0; j < filePaths.size(); j++) { - if (!matchedFiles[j] && matchesExpectedContentQuietly(expectedContents.get(j), actualValue, context)) { + if (!matchedFiles[j] + && (expectedKeys.isEmpty() || keyStrategy.matches(expectedKeys.get(j), actualKey)) + && matchesExpectedContentQuietly(expectedContents.get(j), actualValue, context)) { matchedIndex = j; break; } @@ -153,7 +195,15 @@ private MatchResult matchBatchLazy( if (matchedIndex >= 0) { matchedFiles[matchedIndex] = true; - log.info("Avro record at index {} matched expected file '{}'.", i, filePaths.get(matchedIndex)); + if (!expectedKeys.isEmpty()) { + log.info( + "Avro record at index {} matched expected key '{}' and file '{}'.", + i, + expectedKeys.get(matchedIndex), + filePaths.get(matchedIndex)); + } else { + log.info("Avro record at index {} matched expected file '{}'.", i, filePaths.get(matchedIndex)); + } lastExpected = expectedContents.get(matchedIndex); lastActual = actualValue; } else { @@ -161,15 +211,19 @@ private MatchResult matchBatchLazy( String firstUnmatchedExpected = ""; for (int j = 0; j < filePaths.size(); j++) { if (!matchedFiles[j]) { - unmatchedFiles.add(filePaths.get(j)); + unmatchedFiles.add( + !expectedKeys.isEmpty() + ? expectedKeys.get(j) + " -> " + filePaths.get(j) + : filePaths.get(j)); if (firstUnmatchedExpected.isEmpty()) { firstUnmatchedExpected = expectedContents.get(j); } } } log.error( - "Avro record at index {} did not match any remaining expected file. Unmatched files: {}\nActual record:\n{}", + "Avro record at index {} (key='{}') did not match any remaining expected file. Unmatched files: {}\nActual record:\n{}", i, + actualKey, unmatchedFiles, actualValue); return MatchResult.fail( diff --git a/src/main/java/io/github/ktestify/match/impl/FileRecordMatcher.java b/src/main/java/io/github/ktestify/match/impl/FileRecordMatcher.java index 477611e..445559b 100644 --- a/src/main/java/io/github/ktestify/match/impl/FileRecordMatcher.java +++ b/src/main/java/io/github/ktestify/match/impl/FileRecordMatcher.java @@ -16,10 +16,7 @@ package io.github.ktestify.match.impl; import io.github.ktestify.exceptions.ComparisonException; -import io.github.ktestify.match.BatchMatchStrategy; -import io.github.ktestify.match.MatchContext; -import io.github.ktestify.match.MatchResult; -import io.github.ktestify.match.RecordMatcher; +import io.github.ktestify.match.*; import io.github.ktestify.models.ConsumedRecord; import io.github.ktestify.utils.FileUtils; import io.github.ktestify.utils.StringDiffUtils; @@ -48,8 +45,13 @@ public MatchResult match(List> records, MatchContext cont if (filePaths == null || filePaths.isEmpty() || filePaths.stream().anyMatch(p -> p == null || p.isBlank())) { throw new ComparisonException("FileRecordMatcher requires matchFilePath(s) to be set."); } + List expectedKeys = context.getMatchKeys() != null ? context.getMatchKeys() : List.of(); + if (!expectedKeys.isEmpty() && expectedKeys.size() != filePaths.size()) { + throw new ComparisonException("Expected record keys count (" + expectedKeys.size() + + ") does not match expected file count (" + filePaths.size() + ")."); + } - if (records.size() == 1 && filePaths.size() == 1) { + if (records.size() == 1 && filePaths.size() == 1 && expectedKeys.isEmpty()) { return matchSingle(records.getFirst(), filePaths.getFirst()); } @@ -63,9 +65,9 @@ public MatchResult match(List> records, MatchContext cont : BatchMatchStrategy.POSITIONAL; if (strategy == BatchMatchStrategy.LAZY) { - return matchBatchLazy(records, filePaths); + return matchBatchLazy(records, filePaths, expectedKeys, context); } - return matchBatchPositional(records, filePaths); + return matchBatchPositional(records, filePaths, expectedKeys, context); } private MatchResult matchSingle(ConsumedRecord record, String filePath) { @@ -90,13 +92,37 @@ private MatchResult matchSingle(ConsumedRecord record, String filePath) return MatchResult.fail("Record value does not match file '" + filePath + "'.", expected, actual); } - private MatchResult matchBatchPositional(List> records, List filePaths) { + private MatchResult matchBatchPositional( + List> records, + List filePaths, + List expectedKeys, + MatchContext context) { + KeyMatchStrategy keyStrategy = + context.getKeyMatchStrategy() != null ? context.getKeyMatchStrategy() : KeyMatchStrategy.EXACT; String lastExpected = ""; String lastActual = ""; for (int i = 0; i < records.size(); i++) { String filePath = filePaths.get(i); String expected = FileUtils.getFileContent(FileUtils.getFile(filePath)); String actual = records.get(i).getValue(); + String actualKey = records.get(i).getKey(); + + if (!expectedKeys.isEmpty()) { + String expectedKey = expectedKeys.get(i); + if (!keyStrategy.matches(expectedKey, actualKey)) { + log.error( + "Record at index {} key mismatch: expected '{}' ({}), actual '{}'.", + i, + expectedKey, + keyStrategy, + actualKey); + return MatchResult.fail( + "Record at index " + i + " key does not match, expected: '" + expectedKey + "', actual: '" + + actualKey + "', strategy: " + keyStrategy + ".", + expectedKey, + actualKey != null ? actualKey : ""); + } + } if (actual == null) { log.error("Record at index {} has null value (tombstone), expected content of '{}'.", i, filePath); @@ -112,14 +138,28 @@ private MatchResult matchBatchPositional(List> records, L return MatchResult.fail( "Record at index " + i + " does not match file '" + filePath + "'.", expected, actual); } - log.info("Record at index {} matches expected file '{}'.", i, filePath); + if (!expectedKeys.isEmpty()) { + log.info( + "Record at index {} matches expected key '{}' and file '{}'.", + i, + expectedKeys.get(i), + filePath); + } else { + log.info("Record at index {} matches expected file '{}'.", i, filePath); + } lastExpected = expected; lastActual = actual; } return MatchResult.pass(lastExpected, lastActual); } - private MatchResult matchBatchLazy(List> records, List filePaths) { + private MatchResult matchBatchLazy( + List> records, + List filePaths, + List expectedKeys, + MatchContext context) { + KeyMatchStrategy keyStrategy = + context.getKeyMatchStrategy() != null ? context.getKeyMatchStrategy() : KeyMatchStrategy.EXACT; List expectedContents = filePaths.stream() .map(path -> FileUtils.getFileContent(FileUtils.getFile(path))) .toList(); @@ -129,6 +169,7 @@ private MatchResult matchBatchLazy(List> records, List> records, List> records, List= 0) { matchedFiles[matchedIndex] = true; - log.info("Record at index {} matched expected file '{}'.", i, filePaths.get(matchedIndex)); + if (!expectedKeys.isEmpty()) { + log.info( + "Record at index {} matched expected key '{}' and file '{}'.", + i, + expectedKeys.get(matchedIndex), + filePaths.get(matchedIndex)); + } else { + log.info("Record at index {} matched expected file '{}'.", i, filePaths.get(matchedIndex)); + } lastExpected = expectedContents.get(matchedIndex); lastActual = actual; } else { @@ -152,15 +203,19 @@ private MatchResult matchBatchLazy(List> records, List " + filePaths.get(j) + : filePaths.get(j)); if (firstUnmatchedExpected.isEmpty()) { firstUnmatchedExpected = expectedContents.get(j); } } } log.error( - "Record at index {} did not match any remaining expected file. Unmatched files: {}", + "Record at index {} (key='{}') did not match any remaining expected file. Unmatched files: {}", i, + actualKey, unmatchedFiles); return MatchResult.fail( "Record at index " + i + " did not match any remaining expected file. Unmatched files: " diff --git a/src/test/java/io/github/ktestify/io/kafka/KafkaProducerConsumerUnitTest.java b/src/test/java/io/github/ktestify/io/kafka/KafkaProducerConsumerUnitTest.java index 79a0b32..e45b03a 100644 --- a/src/test/java/io/github/ktestify/io/kafka/KafkaProducerConsumerUnitTest.java +++ b/src/test/java/io/github/ktestify/io/kafka/KafkaProducerConsumerUnitTest.java @@ -153,6 +153,60 @@ public Map offsetsForTimes( assertTrue(consumer.call()); } + @Test + @DisplayName("RawKafkaConsumer fetches and matches batch records with expectedRecordKeys via MockConsumer") + void rawKafkaConsumerFetchesAndMatchesBatchWithExpectedRecordKeys() throws Exception { + KafkaRecordFetcher.clearMatchedRecords(); + TopicPartition tp = new TopicPartition("batch-out", 0); + MockConsumer mockConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) { + @Override + public Map offsetsForTimes( + Map timestampsToSearch, Duration timeout) { + return Map.of(tp, new OffsetAndTimestamp(0L, System.currentTimeMillis())); + } + }; + + File expectedFile1 = tempDir.resolve("expected1.json").toFile(); + File expectedFile2 = tempDir.resolve("expected2.json").toFile(); + Files.writeString(expectedFile1.toPath(), "{\"orderId\":\"ORD-1\"}", StandardCharsets.UTF_8); + Files.writeString(expectedFile2.toPath(), "{\"orderId\":\"ORD-2\"}", StandardCharsets.UTF_8); + + mockConsumer.schedulePollTask(() -> { + mockConsumer.rebalance(List.of(tp)); + mockConsumer.updateBeginningOffsets(Map.of(tp, 0L)); + mockConsumer.updateEndOffsets(Map.of(tp, 3L)); + }); + mockConsumer.schedulePollTask(() -> { + mockConsumer.addRecord( + new ConsumerRecord<>("batch-out", 0, 0L, "unrelated-key", "{\"orderId\":\"OTHER\"}")); + mockConsumer.addRecord(new ConsumerRecord<>("batch-out", 0, 1L, "key-2", "{\"orderId\":\"ORD-2\"}")); + mockConsumer.addRecord(new ConsumerRecord<>("batch-out", 0, 2L, "key-1", "{\"orderId\":\"ORD-1\"}")); + }); + + Topic topic = Topic.builder() + .topicName("batch-out") + .topicType(Topic.Type.OUTPUT) + .build(); + + ConsumerContext ctx = ConsumerContext.builder() + .topic(topic) + .consumer(mockConsumer) + .matchMethod(RecordMatcherFactory.METHOD_MATCH_FILE) + .matchFilePaths(List.of(expectedFile1.getAbsolutePath(), expectedFile2.getAbsolutePath())) + .expectedRecordKeys(List.of("key-1", "key-2")) + .keyMatchStrategy(KeyMatchStrategy.EXACT) + .batchMatchStrategy(BatchMatchStrategy.LAZY) + .isBatchConsumer(true) + .batchSize(2) + .readTimeout(2000L) + .consumerDeltaTime(10000L) + .referenceTimestamp(System.currentTimeMillis()) + .build(); + + RawKafkaConsumer consumer = new RawKafkaConsumer(ctx); + assertTrue(consumer.call()); + } + @Test @DisplayName("AvroKafkaConsumer resolves default matcher and rejects INPUT topic") void avroKafkaConsumerContextValidation() { diff --git a/src/test/java/io/github/ktestify/match/impl/AvroFileRecordMatcherTest.java b/src/test/java/io/github/ktestify/match/impl/AvroFileRecordMatcherTest.java index ba66a8c..cdb81c5 100644 --- a/src/test/java/io/github/ktestify/match/impl/AvroFileRecordMatcherTest.java +++ b/src/test/java/io/github/ktestify/match/impl/AvroFileRecordMatcherTest.java @@ -67,6 +67,15 @@ private static List> avroRecords(GenericRecord... .toList(); } + private static ConsumedRecord keyedAvroRecord(String key, GenericRecord record) { + return new ConsumedRecord<>("test-topic", 0, 0L, key, record, Instant.now(), Collections.emptyMap()); + } + + @SafeVarargs + private static List> keyedAvroRecords(ConsumedRecord... records) { + return List.of(records); + } + /** Builds a GenericRecord that matches expected-order.json */ private GenericRecord matchingRecord() { return buildOrder("ORD-001", "CUST-42", 99.99, "CREATED"); @@ -195,5 +204,74 @@ void throwsWhenRecordCountMismatchesFileCount() { MatchContext.builder().matchFilePaths(BATCH_FILES).build(); assertThrows(ComparisonException.class, () -> matcher.match(avroRecords(r1, r2), ctx)); } + + @Test + @DisplayName("passes in POSITIONAL mode when expectedRecordKeys (matchKeys) and files match in order") + void batchPositionalWithExpectedRecordKeysPass() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("k1", "k2", "k3")) + .batchMatchStrategy(BatchMatchStrategy.POSITIONAL) + .build(); + MatchResult result = matcher.match( + keyedAvroRecords(keyedAvroRecord("k1", r1), keyedAvroRecord("k2", r2), keyedAvroRecord("k3", r3)), + ctx); + assertTrue(result.isPassed()); + } + + @Test + @DisplayName("fails in POSITIONAL mode when expectedRecordKeys (matchKeys) do not match at index") + void batchPositionalWithExpectedRecordKeysFailOnKeyMismatch() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("k1", "k2", "k3")) + .batchMatchStrategy(BatchMatchStrategy.POSITIONAL) + .build(); + MatchResult result = matcher.match( + keyedAvroRecords( + keyedAvroRecord("k1", r1), keyedAvroRecord("wrong", r2), keyedAvroRecord("k3", r3)), + ctx); + assertFalse(result.isPassed()); + assertTrue(result.getDiff().contains("Avro record at index 1 key does not match")); + } + + @Test + @DisplayName("passes in LAZY mode when expectedRecordKeys and Avro files arrive out of order") + void batchLazyWithExpectedRecordKeysPassOutOfOrder() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("k1", "k2", "k3")) + .batchMatchStrategy(BatchMatchStrategy.LAZY) + .build(); + MatchResult result = matcher.match( + keyedAvroRecords(keyedAvroRecord("k3", r3), keyedAvroRecord("k1", r1), keyedAvroRecord("k2", r2)), + ctx); + assertTrue(result.isPassed()); + } + + @Test + @DisplayName("fails in LAZY mode when Avro value matches file j but key does not match key j") + void batchLazyWithExpectedRecordKeysFailWhenKeyAndFilePairedWrong() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("k1", "k2", "k3")) + .batchMatchStrategy(BatchMatchStrategy.LAZY) + .build(); + MatchResult result = matcher.match( + keyedAvroRecords(keyedAvroRecord("k2", r1), keyedAvroRecord("k1", r2), keyedAvroRecord("k3", r3)), + ctx); + assertFalse(result.isPassed()); + assertTrue(result.getDiff().contains("k1 ->")); + } + + @Test + @DisplayName("throws ComparisonException when matchKeys count does not equal matchFilePaths count") + void throwsWhenMatchKeysCountMismatchesFileCount() { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("k1", "k2")) + .build(); + assertThrows(ComparisonException.class, () -> matcher.match(avroRecords(r1, r2, r3), ctx)); + } } } diff --git a/src/test/java/io/github/ktestify/match/impl/FileRecordMatcherTest.java b/src/test/java/io/github/ktestify/match/impl/FileRecordMatcherTest.java index 283b3c3..345a640 100644 --- a/src/test/java/io/github/ktestify/match/impl/FileRecordMatcherTest.java +++ b/src/test/java/io/github/ktestify/match/impl/FileRecordMatcherTest.java @@ -50,6 +50,15 @@ private static List> rawRecords(String... values) { .toList(); } + private static ConsumedRecord keyedRawRecord(String key, String value) { + return new ConsumedRecord<>("test-topic", 0, 0L, key, value, Instant.now(), Collections.emptyMap()); + } + + @SafeVarargs + private static List> keyedRawRecords(ConsumedRecord... records) { + return List.of(records); + } + @Nested @DisplayName("Passing scenarios") class Passing { @@ -173,5 +182,87 @@ void throwsWhenRecordCountMismatchesFileCount() { assertThrows( ComparisonException.class, () -> matcher.match(rawRecords(BATCH_CONTENT_1, BATCH_CONTENT_2), ctx)); } + + @Test + @DisplayName("passes in POSITIONAL mode when expectedRecordKeys (matchKeys) and files match in order") + void batchPositionalWithExpectedRecordKeysPass() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("key-1", "key-2", "key-3")) + .batchMatchStrategy(BatchMatchStrategy.POSITIONAL) + .build(); + MatchResult result = matcher.match( + keyedRawRecords( + keyedRawRecord("key-1", BATCH_CONTENT_1), + keyedRawRecord("key-2", BATCH_CONTENT_2), + keyedRawRecord("key-3", BATCH_CONTENT_3)), + ctx); + assertTrue(result.isPassed()); + } + + @Test + @DisplayName("fails in POSITIONAL mode when expectedRecordKeys (matchKeys) do not match at index") + void batchPositionalWithExpectedRecordKeysFailOnKeyMismatch() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("key-1", "key-2", "key-3")) + .batchMatchStrategy(BatchMatchStrategy.POSITIONAL) + .build(); + MatchResult result = matcher.match( + keyedRawRecords( + keyedRawRecord("key-1", BATCH_CONTENT_1), + keyedRawRecord("wrong-key", BATCH_CONTENT_2), + keyedRawRecord("key-3", BATCH_CONTENT_3)), + ctx); + assertFalse(result.isPassed()); + assertTrue(result.getDiff().contains("Record at index 1 key does not match")); + } + + @Test + @DisplayName("passes in LAZY mode when expectedRecordKeys and files arrive out of order") + void batchLazyWithExpectedRecordKeysPassOutOfOrder() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("key-1", "key-2", "key-3")) + .batchMatchStrategy(BatchMatchStrategy.LAZY) + .build(); + MatchResult result = matcher.match( + keyedRawRecords( + keyedRawRecord("key-3", BATCH_CONTENT_3), + keyedRawRecord("key-1", BATCH_CONTENT_1), + keyedRawRecord("key-2", BATCH_CONTENT_2)), + ctx); + assertTrue(result.isPassed()); + } + + @Test + @DisplayName("fails in LAZY mode when value matches file j but key does not match key j") + void batchLazyWithExpectedRecordKeysFailWhenKeyAndFilePairedWrong() throws ComparisonException { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("key-1", "key-2", "key-3")) + .batchMatchStrategy(BatchMatchStrategy.LAZY) + .build(); + MatchResult result = matcher.match( + keyedRawRecords( + keyedRawRecord("key-2", BATCH_CONTENT_1), + keyedRawRecord("key-1", BATCH_CONTENT_2), + keyedRawRecord("key-3", BATCH_CONTENT_3)), + ctx); + assertFalse(result.isPassed()); + assertTrue(result.getDiff().contains("key-1 ->")); + } + + @Test + @DisplayName("throws ComparisonException when matchKeys count does not equal matchFilePaths count") + void throwsWhenMatchKeysCountMismatchesFileCount() { + MatchContext ctx = MatchContext.builder() + .matchFilePaths(BATCH_FILES) + .matchKeys(List.of("key-1", "key-2")) + .build(); + assertThrows( + ComparisonException.class, + () -> matcher.match(rawRecords(BATCH_CONTENT_1, BATCH_CONTENT_2, BATCH_CONTENT_3), ctx)); + } } }