From af81f8dd9ecc883ceeac23a46bb2b101b360871d Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 19 Aug 2019 19:08:09 -0400 Subject: [PATCH] Migrating PR from core Spring Cloud Stream https://github.com/spring-cloud/spring-cloud-stream/commit/0ad15247f6736959a3c6307e4eef0a7139c142bf --- .../controllers/ServerController.java | 84 ++- .../server/SchemaRegistryServerAvroTests.java | 615 +++++++++++++----- 2 files changed, 527 insertions(+), 172 deletions(-) diff --git a/spring-cloud-schema-registry-core/src/main/java/org/springframework/cloud/schema/registry/controllers/ServerController.java b/spring-cloud-schema-registry-core/src/main/java/org/springframework/cloud/schema/registry/controllers/ServerController.java index 2d10fcb..6faf3fa 100644 --- a/spring-cloud-schema-registry-core/src/main/java/org/springframework/cloud/schema/registry/controllers/ServerController.java +++ b/spring-cloud-schema-registry-core/src/main/java/org/springframework/cloud/schema/registry/controllers/ServerController.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2019 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. @@ -31,9 +31,11 @@ import org.springframework.cloud.schema.registry.support.UnsupportedFormatExcept 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,9 +44,12 @@ 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 + * @author Jeff Maxwell */ @RestController @RequestMapping(path = "${spring.cloud.stream.schema.server.path:}") @@ -57,8 +62,8 @@ public class ServerController { private final SchemaServerProperties schemaServerProperties; public ServerController(SchemaRepository repository, - Map validators, - SchemaServerProperties schemaServerProperties) { + Map validators, + SchemaServerProperties schemaServerProperties) { Assert.notNull(repository, "cannot be null"); Assert.notEmpty(validators, "cannot be empty"); this.repository = repository; @@ -68,7 +73,7 @@ public class ServerController { @RequestMapping(method = RequestMethod.POST, path = "/", consumes = "application/json", produces = "application/json") public synchronized ResponseEntity register(@RequestBody Schema schema, - UriComponentsBuilder builder) { + UriComponentsBuilder builder) { SchemaValidator validator = this.validators.get(schema.getFormat()); if (validator == null) { @@ -85,7 +90,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); } @@ -115,8 +120,8 @@ public class ServerController { @RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/{subject}/{format}/v{version}") public ResponseEntity findOne(@PathVariable("subject") String subject, - @PathVariable("format") String format, - @PathVariable("version") Integer version) { + @PathVariable("format") String format, + @PathVariable("version") Integer version) { Schema schema = this.repository.findOneBySubjectAndFormatAndVersion(subject, format, version); if (schema == null) { @@ -134,23 +139,51 @@ 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, - @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); + /** + *

+ * 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) { + 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) public void delete(@PathVariable("subject") String subject, - @PathVariable("format") String format, - @PathVariable("version") Integer version) { + @PathVariable("format") String format, + @PathVariable("version") Integer version) { if (this.schemaServerProperties.isAllowSchemaDeletion()) { Schema schema = this.repository.findOneBySubjectAndFormatAndVersion(subject, format, version); @@ -190,6 +223,17 @@ public class ServerController { } + @NonNull + public 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-schema-registry-server/src/test/java/org/springframework/cloud/schema/registry/server/SchemaRegistryServerAvroTests.java b/spring-cloud-schema-registry-server/src/test/java/org/springframework/cloud/schema/registry/server/SchemaRegistryServerAvroTests.java index 4fbab86..9d1ff9a 100644 --- a/spring-cloud-schema-registry-server/src/test/java/org/springframework/cloud/schema/registry/server/SchemaRegistryServerAvroTests.java +++ b/spring-cloud-schema-registry-server/src/test/java/org/springframework/cloud/schema/registry/server/SchemaRegistryServerAvroTests.java @@ -16,25 +16,46 @@ package org.springframework.cloud.schema.registry.server; +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.assertj.core.api.Assertions; +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.schema.registry.config.SchemaServerProperties; +import org.springframework.cloud.schema.registry.controllers.ServerController; import org.springframework.cloud.schema.registry.model.Schema; +import org.springframework.cloud.schema.registry.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.context.WebApplicationContext; +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 @@ -44,22 +65,49 @@ import org.springframework.web.context.WebApplicationContext; // @checkstyle:off @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT, properties = "spring.main.allow-bean-definition-overriding=true") // @checkstyle:on -@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) +@DirtiesContext(classMode = AFTER_EACH_TEST_METHOD) public class SchemaRegistryServerAvroTests { - final String USER_SCHEMA_V1 = "{\"namespace\": \"example.avro\",\n" + 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" + "}"; - final String USER_SCHEMA_V2 = "{\"namespace\": \"example.avro\",\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; @@ -67,7 +115,38 @@ public class SchemaRegistryServerAvroTests { private SchemaServerProperties schemaServerProperties; @Autowired - private WebApplicationContext wac; + 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 { @@ -75,83 +154,84 @@ public class SchemaRegistryServerAvroTests { schema.setFormat("spring"); schema.setSubject("boot"); ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - Assertions.assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + .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"); + schema.setFormat(AVRO_FORMAT_NAME); schema.setSubject("boot"); schema.setDefinition("{}"); ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - Assertions.assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + .postForEntity(this.serverControllerUri, schema, Schema.class); + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); } @Test - public void testUserSchemaV1() throws Exception { + public void testRegister1AvroSchema() { + Schema schema = new Schema(); - schema.setFormat("avro"); + schema.setFormat(AVRO_FORMAT_NAME); 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); - Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1)); - List location = response.getHeaders().get(HttpHeaders.LOCATION); - Assertions.assertThat(location).isNotNull(); - ResponseEntity persistedSchema = this.client.getForEntity(location.get(0), - Schema.class); - Assertions.assertThat(persistedSchema.getBody().getId()) - .isEqualTo(response.getBody().getId()); + schema.setDefinition(SchemaRegistryServerAvroTests.AVRO_USER_DEFINITION_SCHEMA_V1); + + registerSchemaAndAssertSuccess(schema, 1, 1); } @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); + public void testFindByIdFound() { - Schema schema2 = new Schema(); - schema2.setFormat("avro"); - schema2.setSubject("org.springframework.cloud.stream.schema.User"); - schema2.setDefinition(this.USER_SCHEMA_V2); + ResponseEntity registerSchemaReponse = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); - ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1)); - List location = response.getHeaders().get(HttpHeaders.LOCATION); - Assertions.assertThat(location).isNotNull(); + Schema registeredSchema = registerSchemaReponse.getBody(); - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", schema2, Schema.class); - Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.assertThat(response2.getBody().getVersion()).isEqualTo(new Integer(2)); - List location2 = response2.getHeaders().get(HttpHeaders.LOCATION); - Assertions.assertThat(location2).isNotNull(); + 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 testIdempotentRegistration() throws Exception { - Schema schema = new Schema(); - schema.setFormat("avro"); - schema.setSubject("org.springframework.cloud.stream.schema.User"); - schema.setDefinition(this.USER_SCHEMA_V1); + public void testFindByIdNotFound() { + + registerSchemaAndAssertSuccess(AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + + URI findByIdUriId1 = this.serverControllerUri.resolve("/schemas/" + 2); + ResponseEntity response = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1)); - List location = response.getHeaders().get(HttpHeaders.LOCATION); - Assertions.assertThat(location).isNotNull(); - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", schema, Schema.class); - Assertions.assertThat(response2.getBody().getId()).isEqualTo(response.getBody().getId()); + .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); } @@ -159,126 +239,357 @@ public class SchemaRegistryServerAvroTests { public void testSchemaNotfound() throws Exception { ResponseEntity response = this.client .getForEntity("http://localhost:8990/foo/avro/v42", Schema.class); - Assertions.assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + 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); - Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + 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); - Assertions.assertThat(response2.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + + 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 { - 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); - Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - ResponseEntity response2 = this.client - .getForEntity("http://localhost:8990/test/avro/v1", Schema.class); - Assertions.assertThat(response2.getStatusCode()).isEqualTo(HttpStatus.OK); + + + ResponseEntity registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + this.schemaServerProperties.setAllowSchemaDeletion(true); - this.client.delete("http://localhost:8990/schemas/1"); + this.client.delete(this.serverControllerUri + .resolve("/schemas/" + registerSchemaAndAssertSuccess.getBody().getVersion())); + ResponseEntity response3 = this.client - .getForEntity("http://localhost:8990/test/avro/1", Schema.class); - Assertions.assertThat(response3.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + .getForEntity(registerSchemaAndAssertSuccess.getHeaders().getLocation(), 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); - Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.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); - Assertions.assertThat(response2.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.assertThat(this.client - .getForEntity("http://localhost:8990/test/avro/v2", Schema.class) - .getStatusCode()).isEqualTo(HttpStatus.OK); + public void testSchemaDeletionByIdNotFound() throws Exception { + + registerSchemaAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1); + this.schemaServerProperties.setAllowSchemaDeletion(true); - this.client.delete("http://localhost:8990/test"); - ResponseEntity response4 = this.client - .getForEntity("http://localhost:8990/test/avro/v1", Schema.class); - Assertions.assertThat(response4.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); - ResponseEntity response5 = this.client - .getForEntity("http://localhost:8990/test/avro/v2", Schema.class); - Assertions.assertThat(response5.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + + 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 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); - Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); - ResponseEntity deleteBySubjectFormatVersion = this.client.exchange( - "http://localhost:8990/test/avro/v1", HttpMethod.DELETE, null, - Object.class); - Assertions.assertThat(deleteBySubjectFormatVersion.getStatusCode()) - .isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); - ResponseEntity deleteBySubject = this.client.exchange( - "http://localhost:8990/test", HttpMethod.DELETE, null, Object.class); - Assertions.assertThat(deleteBySubject.getStatusCode()) - .isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); - ResponseEntity deleteById = this.client.exchange( - "http://localhost:8990/schemas/1", HttpMethod.DELETE, null, Object.class); - Assertions.assertThat(deleteById.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED); + 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 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); - Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue(); + public void testSchemaDeletionBySubject() { + Map>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess( + AVRO_USER_REGISTRY_SCHEMA_V1, + AVRO_USER_REGISTRY_SCHEMA_V2, AAVRO_USER_REGISTRY_SCHEMA_V1_WITH_QUAL_SUBJECT); - Schema v2 = new Schema(); - v2.setFormat("avro"); - v2.setSubject("test"); - v2.setDefinition(this.USER_SCHEMA_V2); + this.schemaServerProperties.setAllowSchemaDeletion(true); - ResponseEntity response2 = this.client - .postForEntity("http://localhost:8990/", v2, Schema.class); - Assertions.assertThat(response2.getStatusCode().is2xxSuccessful()).isTrue(); + 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); - ResponseEntity> schemaResponse = this.client.exchange( - "http://localhost:8990/test/avro", HttpMethod.GET, null, - new ParameterizedTypeReference>() { }); + }); + }); - Assertions.assertThat(schemaResponse.getStatusCode().is2xxSuccessful()).isTrue(); - Assertions.assertThat(schemaResponse.getBody().size()).isEqualTo(2); } + @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); + } + } }