TrafficControl implements Agent<KafkaRoad> { @Override public List<PatchOperation> inspectModel(String key, KafkaRoad model) { return handleModel(model); } @Autowired  TrafficControl(KafkaAdminClient kafkaAdminClient, @Value("${kafka.topic.prefix:road.}") String topicPrefix); @Override List<PatchOperation> newModel(String key, KafkaRoad newModel); @Override List<PatchOperation> updatedModel(String key, KafkaRoad oldModel, KafkaRoad newModel); @Override void deletedModel(String key, KafkaRoad oldModel); @Override List<PatchOperation> inspectModel(String key, KafkaRoad model);  }
@Test public void inspectRoad_does_nothing_when_topic_exists_and_is_correct() throws Exception { KafkaRoad model = new KafkaRoad("test_road", "road.test_road", RoadType.NORMAL, new TrafficControlStatus(true, 12, 3, ""), null, false); given(kafkaAdminClient.topicExists("road.test_road")).willReturn(true); given(kafkaAdminClient.topicDetails("road.test_road")).willReturn(new KafkaTopicDetails(RoadType.NORMAL, 12, 3)); List<PatchOperation> operations = trafficControl.inspectModel("test_road", model); assertTrue(operations.isEmpty()); }
@Test public void inspectRoad_updates_status_when_it_does_not_match_topic() throws Exception { KafkaRoad model = new KafkaRoad("test_road", "road.test_road", RoadType.NORMAL, new TrafficControlStatus(true, 6, 3, ""), null, false); given(kafkaAdminClient.topicExists("road.test_road")).willReturn(true); given(kafkaAdminClient.topicDetails("road.test_road")).willReturn(new KafkaTopicDetails(RoadType.NORMAL, 12, 3)); List<PatchOperation> operations = trafficControl.inspectModel("test_road", model); assertThat(operations.size(), is(1)); assertThat(operations.get(0).getOperation(), is(Operation.ADD)); assertThat(operations.get(0).getPath(), is("/status")); assertThat(operations.get(0).getValue(), is(new TrafficControlStatus(true, 12, 3, ""))); }
@Test public void inspectRoad_corrects_road_type() throws Exception { KafkaRoad model = new KafkaRoad("test_road", "road.test_road", RoadType.NORMAL, new TrafficControlStatus(true, 6, 3, ""), null, false); given(kafkaAdminClient.topicExists("road.test_road")).willReturn(true); given(kafkaAdminClient.topicDetails("road.test_road")).willReturn(new KafkaTopicDetails(RoadType.COMPACT, 6, 3)); List<PatchOperation> operations = trafficControl.inspectModel("test_road", model); assertThat(operations.size(), is(1)); assertThat(operations.get(0).getOperation(), is(Operation.ADD)); assertThat(operations.get(0).getPath(), is("/type")); assertThat(operations.get(0).getValue(), is(RoadType.COMPACT)); }