From 0ad15247f6736959a3c6307e4eef0a7139c142bf Mon Sep 17 00:00:00 2001 From: jmaxwell Date: Tue, 16 Jul 2019 09:38:39 -0500 Subject: [PATCH] GH-1760 Deprecated findBySubjectAndVersionadded findBySubjectAndFormat Note also refactored and moved ServerController test class. Increased coverage of ServerController to 100%. polishing --- .../server/controllers/ServerController.java | 65 +- .../server/SchemaRegistryServerAvroTests.java | 286 --------- .../SchemaRegistryServerAvroTests.java | 594 ++++++++++++++++++ 3 files changed, 648 insertions(+), 297 deletions(-) delete mode 100644 spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java create mode 100644 spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/controllers/SchemaRegistryServerAvroTests.java diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java index 6b37b5bea..f9c80ee38 100644 --- a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java @@ -31,9 +31,11 @@ import org.springframework.cloud.stream.schema.server.support.UnsupportedFormatE import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; +import org.springframework.lang.NonNull; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -42,6 +44,8 @@ import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.util.UriComponentsBuilder; +import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE; + /** * @author Vinicius Carvalho * @author Ilayaperumal Gopinathan @@ -85,7 +89,7 @@ public class ServerController { List registeredEntities = this.repository .findBySubjectAndFormatOrderByVersion(schema.getSubject(), schema.getFormat()); - if (registeredEntities == null || registeredEntities.size() == 0) { + if (registeredEntities.isEmpty()) { schema.setVersion(1); result = this.repository.save(schema); } @@ -134,17 +138,45 @@ public class ServerController { return new ResponseEntity<>(schema.get(), HttpStatus.OK); } - @RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/{subject}/{format}") - public ResponseEntity> findBySubjectAndVersion( - @PathVariable("subject") String subject, + /** + *

+ * Find by {@link Schema#getSubject() subject} and {@link Schema#getFormat() format}. + * + * @param subject the {@link Schema#getSubject() subject}, must not be + * {@literal null}. + * @param format the {@link Schema#getFormat() format}, must not be {@literal null}. + * @return An {@link HttpStatus#OK} response populated with the list of {@link Schema + * Schemas}, in ascending order by {@link Schema#getVersion() version}, that matched + * the supplied {@link Schema#getSubject() subject} and {@link Schema#getFormat() + * format}. + * @deprecated use {@link #findBySubjectAndFormat(String, String)} + * @see GH-1760 + */ + @Deprecated + public ResponseEntity> findBySubjectAndVersion(@PathVariable("subject") String subject, @PathVariable("format") String format) { - List schemas = this.repository - .findBySubjectAndFormatOrderByVersion(subject, format); - if (schemas == null || schemas.size() == 0) { - throw new SchemaNotFoundException(String.format( - "No schemas found for subject %s and format %s", subject, format)); - } - return new ResponseEntity>(schemas, HttpStatus.OK); + return findBySubjectAndFormatOrderByVersionAsc(subject, format); + } + + /** + * Find by {@link Schema#getSubject() subject} and {@link Schema#getFormat() format}. + * + * @param subject the {@link Schema#getSubject() subject}, must not be + * {@literal null}. + * @param format the {@link Schema#getFormat() format}, must not be {@literal null}. + * @return An {@link HttpStatus#OK} response populated with the list of {@link Schema + * Schemas}, in ascending order by {@link Schema#getVersion() version}, that matched + * the supplied {@link Schema#getSubject() subject} and {@link Schema#getFormat() + * format}. + * + * @since 3.0.0 + */ + @GetMapping(produces = APPLICATION_JSON_VALUE, path = "/{subject}/{format}") + @NonNull + public ResponseEntity> findBySubjectAndFormat(@NonNull @PathVariable("subject") final String subject, + @NonNull @PathVariable("format") final String format) { + return findBySubjectAndFormatOrderByVersionAsc(subject, format); } @RequestMapping(value = "/{subject}/{format}/v{version}", method = RequestMethod.DELETE) @@ -190,6 +222,17 @@ public class ServerController { } + @NonNull + final ResponseEntity> findBySubjectAndFormatOrderByVersionAsc(@NonNull final String subject, + @NonNull final String format) { + List schemas = this.repository.findBySubjectAndFormatOrderByVersion(subject, format); + if (schemas.isEmpty()) { + throw new SchemaNotFoundException( + String.format("No schemas found for subject %s and format %s", subject, format)); + } + return new ResponseEntity<>(schemas, HttpStatus.OK); + } + private void deleteSchema(Schema schema) { if (schema == null) { throw new SchemaNotFoundException("Could not find Schema"); diff --git a/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java b/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java deleted file mode 100644 index 65ef4743f..000000000 --- a/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java +++ /dev/null @@ -1,286 +0,0 @@ -/* - * Copyright 2016-2017 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.schema.server; - -import java.util.List; - -import org.junit.Test; -import org.junit.runner.RunWith; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.test.web.client.TestRestTemplate; -import org.springframework.cloud.stream.schema.server.config.SchemaServerProperties; -import org.springframework.cloud.stream.schema.server.model.Schema; -import org.springframework.core.ParameterizedTypeReference; -import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpMethod; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; -import org.springframework.web.context.WebApplicationContext; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.springframework.test.annotation.DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD; - -/** - * @author Vinicius Carvalho - * @author Ilayaperumal Gopinathan - */ -@RunWith(SpringRunner.class) -// @checkstyle:off -@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT, properties = "spring.main.allow-bean-definition-overriding=true") -// @checkstyle:on -@DirtiesContext(classMode = AFTER_EACH_TEST_METHOD) -public class SchemaRegistryServerAvroTests { - - final String USER_SCHEMA_V1 = "{\"namespace\": \"example.avro\",\n" - + " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n" - + " {\"name\": \"name\", \"type\": \"string\"},\n" - + " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]}\n" - + " ]\n" + "}"; - - final String USER_SCHEMA_V2 = "{\"namespace\": \"example.avro\",\n" - + " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n" - + " {\"name\": \"name\", \"type\": \"string\"},\n" - + " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]},\n" - + " {\"name\": \"favorite_color\", \"type\": [\"string\", \"null\"]}\n" - + " ]\n" + "}"; - - @Autowired - private TestRestTemplate client; - - @Autowired - private SchemaServerProperties schemaServerProperties; - - @Autowired - private WebApplicationContext wac; - - @Test - public void testUnsupportedFormat() throws Exception { - Schema schema = new Schema(); - schema.setFormat("spring"); - schema.setSubject("boot"); - ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); - } - - @Test - public void testInvalidSchema() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("boot"); - schema.setDefinition("{}"); - ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); - } - - @Test - public void testUserSchemaV1() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("org.springframework.cloud.stream.schema.User"); - schema.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1)); - List location = response.getHeaders().get(HttpHeaders.LOCATION); - assertThat(location).isNotNull(); - ResponseEntity persistedSchema = this.client.getForEntity(location.get(0), - Schema.class); - assertThat(persistedSchema.getBody().getId()) - .isEqualTo(response.getBody().getId()); - - } - - @Test - public void testUserSchemaV2() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("org.springframework.cloud.stream.schema.User"); - schema.setDefinition(this.USER_SCHEMA_V1); - - Schema schema2 = new Schema(); - schema2.setFormat("avro"); - schema2.setSubject("org.springframework.cloud.stream.schema.User"); - schema2.setDefinition(this.USER_SCHEMA_V2); - - ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1)); - List location = response.getHeaders().get(HttpHeaders.LOCATION); - assertThat(location).isNotNull(); - - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", schema2, Schema.class); - assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(response2.getBody().getVersion()).isEqualTo(new Integer(2)); - List location2 = response2.getHeaders().get(HttpHeaders.LOCATION); - assertThat(location2).isNotNull(); - - } - - @Test - public void testIdempotentRegistration() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("org.springframework.cloud.stream.schema.User"); - schema.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1)); - List location = response.getHeaders().get(HttpHeaders.LOCATION); - assertThat(location).isNotNull(); - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response2.getBody().getId()).isEqualTo(response.getBody().getId()); - - } - - @Test - public void testSchemaNotfound() throws Exception { - ResponseEntity response = this.client - .getForEntity("http://localhost:8990/foo/avro/v42", Schema.class); - assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); - } - - @Test - public void testSchemaDeletionBySubjectFormatVersion() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("test"); - schema.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response1 = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - this.schemaServerProperties.setAllowSchemaDeletion(true); - this.client.delete("http://localhost:8990/test/avro/v1"); - ResponseEntity response2 = this.client - .getForEntity("http://localhost:8990/test/avro/v1", Schema.class); - assertThat(response2.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); - } - - @Test - public void testSchemaDeletionById() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("test"); - schema.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response1 = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - ResponseEntity response2 = this.client - .getForEntity("http://localhost:8990/test/avro/v1", Schema.class); - assertThat(response2.getStatusCode()).isEqualTo(HttpStatus.OK); - this.schemaServerProperties.setAllowSchemaDeletion(true); - this.client.delete("http://localhost:8990/schemas/1"); - ResponseEntity response3 = this.client - .getForEntity("http://localhost:8990/test/avro/1", Schema.class); - assertThat(response3.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); - } - - @Test - public void testSchemaDeletionBySubject() throws Exception { - Schema schema1 = new Schema(); - schema1.setFormat("avro"); - schema1.setSubject("test"); - schema1.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response1 = this.client - .postForEntity("http://localhost:8990/", schema1, Schema.class); - assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(this.client - .getForEntity("http://localhost:8990/test/avro/v1", Schema.class) - .getStatusCode()).isEqualTo(HttpStatus.OK); - this.client.getForEntity("http://localhost:8990/test/avro/1", Schema.class); - Schema schema2 = new Schema(); - schema2.setFormat("avro"); - schema2.setSubject("test"); - schema2.setDefinition(this.USER_SCHEMA_V2); - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", schema2, Schema.class); - assertThat(response2.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(this.client - .getForEntity("http://localhost:8990/test/avro/v2", Schema.class) - .getStatusCode()).isEqualTo(HttpStatus.OK); - this.schemaServerProperties.setAllowSchemaDeletion(true); - this.client.delete("http://localhost:8990/test"); - ResponseEntity response4 = this.client - .getForEntity("http://localhost:8990/test/avro/v1", Schema.class); - assertThat(response4.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); - ResponseEntity response5 = this.client - .getForEntity("http://localhost:8990/test/avro/v2", Schema.class); - assertThat(response5.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); - } - - @Test - public void testSchemaDeletionNotAllowed() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("test"); - schema.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response1 = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - ResponseEntity deleteBySubjectFormatVersion = this.client.exchange( - "http://localhost:8990/test/avro/v1", HttpMethod.DELETE, null, - Object.class); - assertThat(deleteBySubjectFormatVersion.getStatusCode()) - .isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); - ResponseEntity deleteBySubject = this.client.exchange( - "http://localhost:8990/test", HttpMethod.DELETE, null, Object.class); - assertThat(deleteBySubject.getStatusCode()) - .isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); - ResponseEntity deleteById = this.client.exchange( - "http://localhost:8990/schemas/1", HttpMethod.DELETE, null, Object.class); - assertThat(deleteById.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); - } - - @Test - public void testFindSchemasBySubjectAndVersion() throws Exception { - Schema v1 = new Schema(); - v1.setFormat("avro"); - v1.setSubject("test"); - v1.setDefinition(this.USER_SCHEMA_V1); - ResponseEntity response1 = this.client - .postForEntity("http://localhost:8990/", v1, Schema.class); - assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - - Schema v2 = new Schema(); - v2.setFormat("avro"); - v2.setSubject("test"); - v2.setDefinition(this.USER_SCHEMA_V2); - - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", v2, Schema.class); - assertThat(response2.getStatusCode().is2xxSuccessful()).isTrue(); - - ResponseEntity> schemaResponse = this.client.exchange( - "http://localhost:8990/test/avro", HttpMethod.GET, null, - new ParameterizedTypeReference>() { - }); - - assertThat(schemaResponse.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(schemaResponse.getBody().size()).isEqualTo(2); - } - -} diff --git a/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/controllers/SchemaRegistryServerAvroTests.java b/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/controllers/SchemaRegistryServerAvroTests.java new file mode 100644 index 000000000..d63b5d77b --- /dev/null +++ b/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/controllers/SchemaRegistryServerAvroTests.java @@ -0,0 +1,594 @@ +/* + * Copyright 2016-2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.schema.server.controllers; + +import java.net.URI; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Optional; +import java.util.stream.Stream; + +import org.apache.avro.Schema.Parser; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.web.ServerProperties; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.boot.web.server.Ssl; +import org.springframework.cloud.stream.schema.server.config.SchemaServerProperties; +import org.springframework.cloud.stream.schema.server.model.Schema; +import org.springframework.cloud.stream.schema.server.support.SchemaNotFoundException; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.web.util.UriComponentsBuilder; + +import static java.util.stream.Collectors.toList; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.springframework.test.annotation.DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD; + +/** + * @author Vinicius Carvalho + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringRunner.class) +// @checkstyle:off +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT, properties = "spring.main.allow-bean-definition-overriding=true") +// @checkstyle:on +@DirtiesContext(classMode = AFTER_EACH_TEST_METHOD) +public class SchemaRegistryServerAvroTests { + + private static final String AVRO_FORMAT_NAME = "avro"; + + private static final String AVRO_USER_DEFINITION_SCHEMA_V1 = "{\"namespace\": \"example.avro\",\n" + + " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n" + + " {\"name\": \"name\", \"type\": \"string\"},\n" + + " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]}\n" + + " ]\n" + "}"; + + private static final String AVRO_USER_DEFINTITION_SCHEMA_V2 = "{\"namespace\": \"example.avro\",\n" + + " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n" + + " {\"name\": \"name\", \"type\": \"string\"},\n" + + " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]},\n" + + " {\"name\": \"favorite_color\", \"type\": [\"string\", \"null\"]}\n" + + " ]\n" + "}"; + + private static final org.apache.avro.Schema AVRO_USER_AVRO_SCHEMA_V1 = new Parser() + .parse(AVRO_USER_DEFINITION_SCHEMA_V1); + + private static final org.apache.avro.Schema AVRO_USER_AVRO_SCHEMA_V2 = new Parser() + .parse(AVRO_USER_DEFINTITION_SCHEMA_V2); + + private static final String AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT = AVRO_USER_AVRO_SCHEMA_V1.getName() + .toLowerCase(); + + + private static final String AVRO_USER_SCHEMA_QUALIFED_NAME_STRATEGY_SUBJECT = AVRO_USER_AVRO_SCHEMA_V1 + .getFullName() + .toLowerCase(); + + private static final Schema AVRO_USER_REGISTRY_SCHEMA_V1 = toSchema( + AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT, + AVRO_FORMAT_NAME, AVRO_USER_AVRO_SCHEMA_V1.toString()); + + private static final Schema AVRO_USER_REGISTRY_SCHEMA_V2 = toSchema( + AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT, + AVRO_FORMAT_NAME, AVRO_USER_AVRO_SCHEMA_V2.toString()); + + private static final Schema AAVRO_USER_REGISTRY_SCHEMA_V1_WITH_QUAL_SUBJECT = toSchema( + AVRO_USER_SCHEMA_QUALIFED_NAME_STRATEGY_SUBJECT, + AVRO_FORMAT_NAME, AVRO_USER_AVRO_SCHEMA_V1.toString()); + @Autowired + private TestRestTemplate client; + + @Autowired + private SchemaServerProperties schemaServerProperties; + + @Autowired + private ServerController serverController; + + @Autowired + private ServerProperties serverProperties; + + private URI serverControllerUri; + + @Before + public void setUp() { + + String scheme = Optional.ofNullable(this.serverProperties.getSsl()) + .filter(Ssl::isEnabled) + .map(ssl -> "https").orElse("http"); + + Integer port = this.serverProperties.getPort(); + String contextPath = this.serverProperties.getServlet().getContextPath(); + + this.serverControllerUri = UriComponentsBuilder.newInstance().scheme(scheme) + .host("localhost") + .port(port) + .path(contextPath).build().toUri(); + + } + + @NonNull + static Schema toSchema(String subject, String format, String definition) { + Schema schema = new Schema(); + schema.setSubject(subject); + schema.setFormat(format); + schema.setDefinition(definition); + return schema; + } + + @Test + public void testUnsupportedFormat() throws Exception { + Schema schema = new Schema(); + schema.setFormat("spring"); + schema.setSubject("boot"); + ResponseEntity response = this.client + .postForEntity(this.serverControllerUri, schema, Schema.class); + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + } + + @Test + public void testInvalidSchema() throws Exception { + Schema schema = new Schema(); + schema.setFormat(AVRO_FORMAT_NAME); + schema.setSubject("boot"); + schema.setDefinition("{}"); + ResponseEntity response = this.client + .postForEntity(this.serverControllerUri, schema, Schema.class); + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + } + + @Test + public void testRegister1AvroSchema() { + + Schema schema = new Schema(); + schema.setFormat(AVRO_FORMAT_NAME); + schema.setSubject("org.springframework.cloud.stream.schema.User"); + schema.setDefinition(SchemaRegistryServerAvroTests.AVRO_USER_DEFINITION_SCHEMA_V1); + + registerSchemaAndAssertSuccess(schema, 1, 1); + + } + + @Test + public void testFindByIdFound() { + + ResponseEntity registerSchemaReponse = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + Schema registeredSchema = registerSchemaReponse.getBody(); + + URI findByIdUriId1 = this.serverControllerUri.resolve("/schemas/" + registeredSchema.getId()); + + ResponseEntity findByIdResponse = this.client + .getForEntity(findByIdUriId1, Schema.class); + + assertThat(findByIdResponse.getStatusCode().is2xxSuccessful()).isTrue(); + + Schema actual = findByIdResponse.getBody(); + assertSchema(registeredSchema, actual); + } + + @Test + public void testFindByIdNotFound() { + + registerSchemaAndAssertSuccess(AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + URI findByIdUriId1 = this.serverControllerUri.resolve("/schemas/" + 2); + + ResponseEntity response = this.client + .getForEntity(findByIdUriId1, Schema.class); + + final HttpStatus statusCode = response.getStatusCode(); + + assertThat(statusCode).isEqualTo(HttpStatus.NOT_FOUND); + + } + @Test + public void testUserSchemaV2() { + + registerSchemasAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, + AVRO_USER_REGISTRY_SCHEMA_V2); + } + + @Test + public void testIdempotentRegistration() { + + registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + + registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + } + + @Test + public void testSchemaNotfound() throws Exception { + ResponseEntity response = this.client + .getForEntity("http://localhost:8990/foo/avro/v42", Schema.class); + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + } + + @Test + public void testSchemaDeletionBySubjectFormatVersion() throws Exception { + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + this.schemaServerProperties.setAllowSchemaDeletion(true); + + URI subjectFormatVersionUri = this.serverControllerUri + .resolve(registerSchemaAndAssertSuccess.getHeaders().getLocation()); + + + ResponseEntity deleteResponse = this.client.exchange( + new RequestEntity<>(HttpMethod.DELETE, subjectFormatVersionUri), + Void.class); + + assertThat(deleteResponse.getStatusCode()).isEqualTo(HttpStatus.OK); + + ResponseEntity findBySubjectFormatVersionUriResponse = this.client + .getForEntity(subjectFormatVersionUri, Schema.class); + + assertThat(findBySubjectFormatVersionUriResponse.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + } + + @Test + public void testSchemaDeletionBySubjectFormatVersionNotFound() throws Exception { + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + this.schemaServerProperties.setAllowSchemaDeletion(true); + + URI subjectFormatVersionUri = this.serverControllerUri + .resolve(registerSchemaAndAssertSuccess.getHeaders().getLocation().toString().replace("v1", "v100")); + + ResponseEntity deleteResponse = this.client.exchange( + new RequestEntity<>(HttpMethod.DELETE, subjectFormatVersionUri), + Void.class); + + assertThat(deleteResponse.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + + } + + @Test + public void testSchemaDeletionBySubjectFormatVersionNotAllowed() throws Exception { + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + URI versionUri = this.serverControllerUri + .resolve(registerSchemaAndAssertSuccess.getHeaders().getLocation()); + + ResponseEntity deleteResponse = this.client.exchange(new RequestEntity<>(HttpMethod.DELETE, versionUri), + Void.class); + + assertThat(deleteResponse.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); + + } + + @Test + public void testSchemaDeletionById() throws Exception { + + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + this.schemaServerProperties.setAllowSchemaDeletion(true); + this.client.delete(this.serverControllerUri + .resolve("/schemas/" + registerSchemaAndAssertSuccess.getBody().getVersion())); + + ResponseEntity response3 = this.client + .getForEntity(registerSchemaAndAssertSuccess.getHeaders().getLocation(), Schema.class); + assertThat(response3.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + + } + + @Test + public void testSchemaDeletionByIdNotFound() throws Exception { + + registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + this.schemaServerProperties.setAllowSchemaDeletion(true); + + ResponseEntity deleteByIdResponse = this.client.exchange( + new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri + .resolve("/schemas/" + 2)), + Void.class); + + assertThat(deleteByIdResponse.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + } + + @Test + public void testSchemaDeletionByIdNotAllowed() throws Exception { + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + URI schemaIdUri = this.serverControllerUri + .resolve(this.serverControllerUri + .resolve("/schemas/" + registerSchemaAndAssertSuccess.getBody().getVersion())); + + ResponseEntity exchange = this.client.exchange(new RequestEntity<>(HttpMethod.DELETE, schemaIdUri), + Void.class); + + assertThat(exchange.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); + } + + @Test + public void testSchemaDeletionBySubject() { + Map>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, + AVRO_USER_REGISTRY_SCHEMA_V2, AAVRO_USER_REGISTRY_SCHEMA_V1_WITH_QUAL_SUBJECT); + + this.schemaServerProperties.setAllowSchemaDeletion(true); + + registerSchemaResponsesByFormatBySubject.forEach((subject, registerSchemaResponsesByFormat) -> { + + assertThat(registerSchemaResponsesByFormat).isNotEmpty(); + ResponseEntity deleteBySubject = this.client.exchange( + new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri + .resolve("/" + subject)), + Void.class); + + assertThat(deleteBySubject.getStatusCode()).isEqualTo(HttpStatus.OK); + + registerSchemaResponsesByFormat.forEach((format, registerSchemaResponses) -> { + + assertThat(registerSchemaResponses).isNotEmpty(); + + registerSchemaResponses.forEach(registerSchemaResponse -> { + + ResponseEntity shouldBe404Response = this.client.getForEntity( + registerSchemaResponse.getHeaders().getLocation(), + Schema.class); + + assertThat(shouldBe404Response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + + }); + }); + }); + + } + + @Test + public void testSchemaDeletionBySubjectNotFound() throws Exception { + + registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + this.schemaServerProperties.setAllowSchemaDeletion(true); + + ResponseEntity deleteBySubject = this.client.exchange( + new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri + .resolve("/foo")), + Void.class); + + assertThat(deleteBySubject.getStatusCode()) + .isEqualTo(HttpStatus.OK); + + + } + + @Test + public void testSchemaDeletionBySubjectNotAllowed() throws Exception { + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + Schema schema = registerSchemaAndAssertSuccess.getBody(); + + ResponseEntity deleteBySubject = this.client.exchange( + new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri + .resolve("/" + schema.getSubject())), + Void.class); + + assertThat(deleteBySubject.getStatusCode()) + .isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); + + } + + @Test + public void testFindSchemasBySubjectAndVersion() { + + Map>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, + AVRO_USER_REGISTRY_SCHEMA_V2); + + registerSchemaResponsesByFormatBySubject.forEach((subject, schemasByFormat) -> { + + assertThat(schemasByFormat).hasSize(1); + + schemasByFormat.forEach((format, schemas) -> { + assertThat(schemas).hasSize(2); + + @SuppressWarnings("deprecation") + final ResponseEntity> findBySubjectAndVersionResponseEntity = this.serverController + .findBySubjectAndVersion(subject, format); + + assertThat(findBySubjectAndVersionResponseEntity.getStatusCode().is2xxSuccessful()).isTrue(); + + final List schemaResponseBody = findBySubjectAndVersionResponseEntity.getBody(); + + assertThat(schemaResponseBody) + .zipSatisfy(schemas.stream().map(ResponseEntity::getBody) + .collect(toList()), this::assertSchema); + + }); + }); + + } + + @Test + public void testFindBySubjectAndFormatOrderByVersionAscNoMatch() { + String subject = "test"; + + String format = AVRO_FORMAT_NAME; + + assertThatExceptionOfType(SchemaNotFoundException.class).isThrownBy(() -> this.serverController + .findBySubjectAndFormatOrderByVersionAsc(subject, format)) + .withMessage("No schemas found for subject %s and format %s", subject, format) + .withNoCause(); + + + } + + @Test + public void testFindSchemasBySubjectAndFormat() { + + Map>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, + AVRO_USER_REGISTRY_SCHEMA_V2); + + registerSchemaResponsesByFormatBySubject.forEach((subject, schemasByFormat) -> { + + assertThat(schemasByFormat).hasSize(1); + + schemasByFormat.forEach((format, schemas) -> { + assertThat(schemas).hasSize(2); + + ResponseEntity> findBySubjectFormatResponse = this.client.exchange( + this.serverControllerUri.resolve("/" + subject + "/" + format), HttpMethod.GET, null, + new ParameterizedTypeReference>() { + }); + + assertThat(findBySubjectFormatResponse.getStatusCode().is2xxSuccessful()).isTrue(); + + final List schemaResponseBody = findBySubjectFormatResponse.getBody(); + + assertThat(schemaResponseBody) + .zipSatisfy(schemas.stream().map(ResponseEntity::getBody) + .collect(toList()), this::assertSchema); + + }); + }); + + } + + private Map>>> registerSchemasAndAssertSuccess( + @NonNull Schema... schemas) { + Map> versionsByFormatAndSubject = new HashMap<>(); + Map>>> result = new HashMap<>(); + int numOfSchemas = schemas.length; + int id = 0; + for (int i = 0; i < numOfSchemas; i++) { + Schema schema = schemas[i]; + id++; + String format = schema.getFormat(); + String subject = schema.getSubject(); + Integer version = versionsByFormatAndSubject + .compute(subject, + (_subject, currentValue) -> currentValue == null ? new HashMap<>() : currentValue) + .merge(format, 1, Integer::sum); + ResponseEntity registerSchemaResponse = registerSchemaAndAssertSuccess(schema, version, id); + result.compute(subject, + (_subject, currentValue) -> currentValue == null ? new HashMap<>() : currentValue) + + .compute(format, (_format, currentValue) -> { + List> value = currentValue == null ? new ArrayList<>() : currentValue; + value.add(registerSchemaResponse); + return value; + }); + } + Stream> asStream = result.entrySet().stream() + .map(Entry::getValue) + .map(Map::entrySet) + .flatMap(Collection::stream) + .map(Entry::getValue) + .flatMap(Collection::stream); + assertThat(asStream).hasSize(numOfSchemas); + return result; + + } + + @NonNull + private ResponseEntity registerSchemaAndAssertSuccess(@NonNull Schema schema, + @Nullable Integer expectedVersion, + @Nullable Integer expectedId) { + + ResponseEntity registerReponse = this.client + .postForEntity(this.serverControllerUri, schema, Schema.class); + + HttpStatus statusCode = registerReponse.getStatusCode(); + assertThat(statusCode.is2xxSuccessful()).isTrue(); + + Schema registeredSchema = registerReponse.getBody(); + assertSchema(schema, expectedVersion, expectedId, registeredSchema); + + HttpHeaders headers = registerReponse.getHeaders(); + assertLocation(headers, registeredSchema); + + return registerReponse; + } + + private void assertLocation(HttpHeaders headers, Schema registeredSchema) { + URI location = headers.getLocation(); + + assertThat(location).isNotNull(); + assertPersisted(location, registeredSchema); + } + + private void assertPersisted(URI location, Schema registeredSchema) { + + ResponseEntity findOneResponse = this.client.getForEntity(location, + Schema.class); + + HttpStatus statusCode = findOneResponse.getStatusCode(); + assertThat(statusCode.is2xxSuccessful()).isTrue(); + + Schema actual = findOneResponse.getBody(); + assertSchema(registeredSchema, registeredSchema.getVersion(), registeredSchema.getId(), actual); + + } + + private void assertSchema(@NonNull Schema expected, @NonNull Schema actual) { + + assertSchema(expected, expected.getVersion(), expected.getId(), actual); + } + + private void assertSchema(@NonNull Schema expected, Integer expectedVersion, Integer expectedId, + @NonNull Schema actual) { + + assertThat(actual).isEqualToIgnoringGivenFields(expected, "version", "id"); + if (expectedVersion != null) { + assertThat(actual.getVersion()).isEqualTo(expectedVersion); + } + if (expectedId != null) { + assertThat(actual.getId()).isEqualTo(expectedId); + } + } +}