SpringOne2023解读-01-使用spring-cloud-contract与TestContainer构建可靠程序

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, classes = {BaristaApplication.class, BaseTestClass.TestConfig.class})
@Testcontainers
@AutoConfigureMessageVerifier
@ActiveProfiles("contracts")
public abstract class BaseTestClass {

    @Autowired
    KafkaHandler kafkaHandler;

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka"));
}

    static class KafkaMessageVerifier implements MessageVerifierReceiver<Message<?>> {

        private static final Log LOG = LogFactory.getLog(KafkaMessageVerifier.class);

        Map<String, BlockingQueue<Message<?>>> broker = new ConcurrentHashMap<>();


        @Override
        public Message receive(String destination, long timeout, TimeUnit timeUnit, @Nullable YamlContract contract) {
            broker.putIfAbsent(destination, new ArrayBlockingQueue<>(1));
            BlockingQueue<Message<?>> messageQueue = broker.get(destination);
            Message<?> message;
            try {
                message = messageQueue.poll(timeout, timeUnit);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            if (message != null) {
                LOG.info("Removed a message from a topic [" + destination + "]");
                LOG.info(message.getPayload().toString());
            }
            return message;
        }


        @KafkaListener(id = "baristaContractTestListener", topics = {"errors", "servings"})
        public void listen(ConsumerRecord payload, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
            LOG.info("Got a message from a topic [" + topic + "]");
            Map<String, Object> headers = new HashMap<>();
            new DefaultKafkaHeaderMapper().toHeaders(payload.headers(), headers);
            broker.putIfAbsent(topic, new ArrayBlockingQueue<>(1));
            BlockingQueue<Message<?>> messageQueue = broker.get(topic);
            messageQueue.add(MessageBuilder.createMessage(payload.value(), new MessageHeaders(headers)));
        }

        @Override
        public Message receive(String destination, YamlContract contract) {
            return receive(destination, 15, TimeUnit.SECONDS, contract);
        }

    }

package contracts

import org.springframework.cloud.contract.spec.Contract

/**
 * @author Olga Maciaszek-Sharma
 */

Contract.make {
	label("serving")
	input {
		triggeredBy("triggerServing()")
	}
	outputMessage {
		sentTo("servings")
		body([beverages:
					  [[uuid  : $(anyUuid()),
						coffee: [
								name         : "V60",
								coffeeContent: "500",
								device       : "V60"
						]],
					   [uuid  : $(anyUuid()),
						coffee: [
								name              : "Latte",
								coffeeContent     : "60",
								steamedMilkContent: "180",
								milkFoamContent   : "5"
						]]
		]])
		headers {
			messagingContentType(applicationJson())
			header 'testKey1', 'testValue1'
		}
	}

}

public void triggerServing() {
    Order order = new Order();
    order.add(new OrderEntry("latte", 1));
    order.add(new OrderEntry("v60", 2));
    //kafka 发送 order 消息
    kafkaHandler.process(order);
}

    @Test
	public void validate_shouldSendServing() throws Exception {
		// when:
			triggerServing();

		// then:
			ContractVerifierMessage response = contractVerifierMessaging.receive("servings",
					contract(this, "shouldSendServing.yml"));
			assertThat(response).isNotNull();

		// and:
			assertThat(response.getHeader("contentType")).isNotNull();
			assertThat(response.getHeader("contentType").toString()).isEqualTo("application/json");
			assertThat(response.getHeader("testKey1")).isNotNull();
			assertThat(response.getHeader("testKey1").toString()).isEqualTo("testValue1");

		// and:
			DocumentContext parsedJson = JsonPath.parse(contractVerifierObjectMapper.writeValueAsString(response.getPayload()));
			assertThatJson(parsedJson).array("['beverages']").contains("['uuid']").matches("[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['name']").isEqualTo("V60");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['coffeeContent']").isEqualTo("500");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['device']").isEqualTo("V60");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['name']").isEqualTo("Latte");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['coffeeContent']").isEqualTo("60");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['steamedMilkContent']").isEqualTo("180");
			assertThatJson(parsedJson).array("['beverages']").field("['coffee']").field("['milkFoamContent']").isEqualTo("5");
	}

@SpringBootApplication
public class LocalTestMain {
    public static void main(String[] args) {
        SpringApplication.from(Main::main)
                .with(TestContainerConfig.class)
                .run(args);
    }

    @TestConfiguration(proxyBeanMethods = false)
    static class TestContainerConfig {
        @Bean
        @ServiceConnection
        public MySQLContainer<?> mysqlContainer() {
            return new MySQLContainer<>("mysql");
        }

    }
}

@Container
static MySQLContainer mySQLContainer = new MySQLContainer("mysql");

@DynamicPropertySource
static void setProperties(DynamicPropertyRegistry registry) {
    registry.add("spring.datasource.url", mySQLContainer::getJdbcUrl);
    registry.add("spring.datasource.username", mySQLContainer::getUsername);
    registry.add("spring.datasource.password", mySQLContainer::getPassword);
}