Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java (Java)

Code before:
        verify(context);
    }

    @Test
    public void shouldSendOldValuesIfConfigured() {
        final KTableTransformValues<String, String, String> transformValues =
            new KTableTransformValues<>(parent, new ExclamationValueTransformerSupplier(), null);

        expect(parent.enableSendingOldValues(true)).andReturn(true);
        replay(parent);

        transformValues.enableSendingOldValues(true);
        final Processor<String, Change<String>, String, Change<String>> processor = transformValues.get();
        processor.init(context);

        context.forward(new Record<>("Key", new Change<>("Key->newValue!", "Key->oldValue!"), 0));
        expectLastCall();
        replay(context);

        processor.process(new Record<>("Key", new Change<>("newValue", "oldValue"), 0));

        verify(context);
    }

    @Test
    public void shouldNotSetSendOldValuesOnParentIfMaterialized() {
        expect(parent.enableSendingOldValues(anyBoolean()))
            .andThrow(new AssertionError("Should not call enableSendingOldValues"))
            .anyTimes();

        replay(parent);

        new KTableTransformValues<>(parent, new NoOpValueTransformerWithKeySupplier<>(), QUERYABLE_NAME).enableSendingOldValues(true);

        verify(parent);
    }

    @Test
    public void shouldSetSendOldValuesOnParentIfNotMaterialized() {
        expect(parent.enableSendingOldValues(true)).andReturn(true);
        replay(parent);

        new KTableTransformValues<>(parent, new NoOpValueTransformerWithKeySupplier<>(), null).enableSendingOldValues(true);

        verify(parent);
    }

    @Test
    public void shouldTransformOnGetIfNotMaterialized() {
        final KTableTransformValues<String, String, String> transformValues =
            new KTableTransformValues<>(parent, new ExclamationValueTransformerSupplier(), null);

Review comment (suggestion):
This test checks nothing as far as I can tell because the QUERYABLE_NAME isn't equal to null, which is when we would reach the if-condition to throw the exception at.

There is another test which tests the behaviour when that value is null, but it doesn't throw an exception, so given that this is a deprecated I decided to not try to cover this scenario.

I have also not used verifyNoMoreInteractions, but if reviewers think it is necessary I can circle back and add it.

Code after:
        final KTableTransformValues<String, String, String> transformValues =
            new KTableTransformValues<>(parent, new ExclamationValueTransformerSupplier(), null);

        when(parent.enableSendingOldValues(true)).thenReturn(true);

        transformValues.enableSendingOldValues(true);
        final Processor<String, Change<String>, String, Change<String>> processor = transformValues.get();
        processor.init(context);

        doNothing().when(context).forward(new Record<>("Key", new Change<>("Key->newValue!", "Key->oldValue!"), 0));

        processor.process(new Record<>("Key", new Change<>("newValue", "oldValue"), 0));
    }

    @Test
    public void shouldNotSetSendOldValuesOnParentIfMaterialized() {
        new KTableTransformValues<>(parent, new NoOpValueTransformerWithKeySupplier<>(), QUERYABLE_NAME).enableSendingOldValues(true);

        verify(parent, never()).enableSendingOldValues(anyBoolean());
    }

    @Test
    public void shouldSetSendOldValuesOnParentIfNotMaterialized() {
        when(parent.enableSendingOldValues(true)).thenReturn(true);

        new KTableTransformValues<>(parent, new NoOpValueTransformerWithKeySupplier<>(), null).enableSendingOldValues(true);
    }

    @Test
    public void shouldTransformOnGetIfNotMaterialized() {
        final KTableTransformValues<String, String, String> transformValues =
            new KTableTransformValues<>(parent, new ExclamationValueTransformerSupplier(), null);

        when(parent.valueGetterSupplier()).thenReturn(parentGetterSupplier);
        when(parentGetterSupplier.get()).thenReturn(parentGetter);
        when(parentGetter.get("Key")).thenReturn(ValueAndTimestamp.make("Value", 73L));
        final ProcessorRecordContext recordContext = new ProcessorRecordContext(
            42L,
            23L,
            -1,
            "foo",
            new RecordHeaders()
        );
        when(context.recordContext()).thenReturn(recordContext);
        doNothing().when(context).setRecordContext(new ProcessorRecordContext(
            73L,
            -1L,
            -1,
            null,
            new RecordHeaders()
        ));