-
Notifications
You must be signed in to change notification settings - Fork 74
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add schema validation job endpoints (#5155)
* Add schema validation job endpoints --------- Co-authored-by: Simon Dumas <[email protected]>
- Loading branch information
Showing
34 changed files
with
535 additions
and
76 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
96 changes: 96 additions & 0 deletions
96
delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/SchemaJobRoutes.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.routes | ||
|
|
||
| import akka.http.scaladsl.model.{ContentTypes, StatusCodes} | ||
| import akka.http.scaladsl.server.Route | ||
| import akka.util.ByteString | ||
| import cats.effect.IO | ||
| import cats.implicits.catsSyntaxApplicativeError | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.RemoteContextResolution | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.utils.JsonKeyOrdering | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.directives.DeltaDirectives._ | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.directives.{AuthDirectives, FileResponse} | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.identities.Identities | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.projects.FetchContext | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.resources.Resources | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.schemas.job.SchemaValidationCoordinator | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.stream.StreamConverter | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.projections.{ProjectionErrors, Projections} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.query.SelectFilter | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.utils.StreamingUtils | ||
|
|
||
| /** | ||
| * Routes to trigger and get results from a schema validation job | ||
| */ | ||
| class SchemaJobRoutes( | ||
| identities: Identities, | ||
| aclCheck: AclCheck, | ||
| fetchContext: FetchContext, | ||
| schemaValidationCoordinator: SchemaValidationCoordinator, | ||
| projections: Projections, | ||
| projectionsErrors: ProjectionErrors | ||
| )(implicit | ||
| baseUri: BaseUri, | ||
| cr: RemoteContextResolution, | ||
| ordering: JsonKeyOrdering | ||
| ) extends AuthDirectives(identities, aclCheck) { | ||
|
|
||
| private def projectionName(project: ProjectRef) = SchemaValidationCoordinator.projectionMetadata(project).name | ||
|
|
||
| private def projectExists(project: ProjectRef) = fetchContext.onRead(project).void | ||
|
|
||
| private def streamValidationErrors(project: ProjectRef): IO[FileResponse] = { | ||
| IO.delay { | ||
| StreamConverter( | ||
| projectionsErrors | ||
| .failedElemEntries(projectionName(project), Offset.start) | ||
| .map(_.failedElemData) | ||
| .through(StreamingUtils.ndjson) | ||
| .map(ByteString(_)) | ||
| ) | ||
| } | ||
| }.map { s => | ||
| FileResponse("validation.json", ContentTypes.`application/json`, None, s) | ||
| } | ||
|
|
||
| def routes: Route = | ||
| baseUriPrefix(baseUri.prefix) { | ||
| pathPrefix("jobs") { | ||
| extractCaller { implicit caller => | ||
| pathPrefix("schemas") { | ||
| (pathPrefix("validation") & projectRef) { project => | ||
| authorizeFor(project, Permissions.schemas.run).apply { | ||
| concat( | ||
| (post & pathEndOrSingleSlash) { | ||
| emit( | ||
| StatusCodes.Accepted, | ||
| projectExists(project) >> schemaValidationCoordinator.run(project).start.void | ||
| ) | ||
| }, | ||
| (pathPrefix("statistics") & get & pathEndOrSingleSlash) { | ||
| emit( | ||
| projectExists(project) >> projections | ||
| .statistics( | ||
| project, | ||
| SelectFilter.latestOfEntity(Resources.entityType), | ||
| projectionName(project) | ||
| ) | ||
| ) | ||
| }, | ||
| (pathPrefix("errors") & get & pathEndOrSingleSlash) { | ||
| emit( | ||
| projectExists(project) >> streamValidationErrors(project).attemptNarrow[Nothing] | ||
| ) | ||
| } | ||
| ) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
160 changes: 160 additions & 0 deletions
160
delta/app/src/test/scala/ch/epfl/bluebrain/nexus/delta/routes/SchemaJobRoutesSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,160 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.routes | ||
|
|
||
| import akka.http.scaladsl.model.MediaRanges.`*/*` | ||
| import akka.http.scaladsl.model.StatusCodes | ||
| import akka.http.scaladsl.model.headers.{Accept, OAuth2BearerToken} | ||
| import akka.http.scaladsl.server.Route | ||
| import cats.effect.{IO, Ref} | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.nxv | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclSimpleCheck | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress.Root | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.generators.ProjectGen | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.identities.IdentitiesDummy | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.identities.model.Caller | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.projects.FetchContextDummy | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.schemas.job.SchemaValidationCoordinator | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.utils.BaseRouteSpec | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.{Anonymous, Authenticated, Group} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, ProjectRef} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.projections.{ProjectionErrors, Projections} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.Elem.FailedElem | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.{FailureReason, ProjectionProgress} | ||
|
|
||
| import java.time.Instant | ||
|
|
||
| class SchemaJobRoutesSpec extends BaseRouteSpec { | ||
|
|
||
| private val caller = Caller(alice, Set(alice, Anonymous, Authenticated(realm), Group("group", realm))) | ||
|
|
||
| private val asAlice = addCredentials(OAuth2BearerToken("alice")) | ||
|
|
||
| private val project = ProjectGen.project("org", "project") | ||
| private val rev = 1 | ||
| private val resourceId = nxv + "myid" | ||
|
|
||
| private val fetchContext = FetchContextDummy(List(project)) | ||
|
|
||
| private val identities = IdentitiesDummy(caller) | ||
|
|
||
| private val aclCheck = AclSimpleCheck((alice, Root, Set(Permissions.schemas.run))).accepted | ||
|
|
||
| private lazy val projections = Projections(xas, queryConfig, clock) | ||
| private lazy val projectionErrors = ProjectionErrors(xas, queryConfig, clock) | ||
|
|
||
| private val progress = ProjectionProgress(Offset.at(15L), Instant.EPOCH, 9000L, 400L, 30L) | ||
|
|
||
| private val runTrigger = Ref.unsafe[IO, Boolean](false) | ||
|
|
||
| private val schemaValidationCoordinator = new SchemaValidationCoordinator { | ||
| override def run(project: ProjectRef): IO[Unit] = runTrigger.set(true).void | ||
| } | ||
|
|
||
| private lazy val routes = Route.seal( | ||
| new SchemaJobRoutes( | ||
| identities, | ||
| aclCheck, | ||
| fetchContext, | ||
| schemaValidationCoordinator, | ||
| projections, | ||
| projectionErrors | ||
| ).routes | ||
| ) | ||
|
|
||
| override def beforeAll(): Unit = { | ||
| super.beforeAll() | ||
| val projectionMetadata = SchemaValidationCoordinator.projectionMetadata(project.ref) | ||
|
|
||
| val reason = FailureReason("ValidationFail", json"""{ "details": "..." }""") | ||
| val fail1 = FailedElem(EntityType("ACL"), resourceId, Some(project.ref), Instant.EPOCH, Offset.At(42L), reason, rev) | ||
| val fail2 = | ||
| FailedElem(EntityType("Schema"), resourceId, Some(project.ref), Instant.EPOCH, Offset.At(42L), reason, rev) | ||
|
|
||
| ( | ||
| projections.save(projectionMetadata, progress) >> | ||
| projectionErrors.saveFailedElems(projectionMetadata, List(fail1, fail2)) | ||
| ).accepted | ||
| } | ||
|
|
||
| "The schema validation job route" should { | ||
| "fail to start a validation job without permission" in { | ||
| Post("/v1/jobs/schemas/validation/org/project") ~> routes ~> check { | ||
| response.shouldBeForbidden | ||
| runTrigger.get.accepted shouldEqual false | ||
| } | ||
| } | ||
|
|
||
| "fail to start a validation job for an unknown project" in { | ||
| Post("/v1/jobs/schemas/validation/xxx/xxx") ~> asAlice ~> routes ~> check { | ||
| response.shouldFail(StatusCodes.NotFound, "ProjectNotFound") | ||
| runTrigger.get.accepted shouldEqual false | ||
| } | ||
| } | ||
|
|
||
| "start a validation job on the project with appropriate access" in { | ||
| Post("/v1/jobs/schemas/validation/org/project") ~> asAlice ~> routes ~> check { | ||
| response.status shouldEqual StatusCodes.Accepted | ||
| runTrigger.get.accepted shouldEqual true | ||
| } | ||
| } | ||
|
|
||
| "fail to get statistics on a validation job without permission" in { | ||
| Get("/v1/jobs/schemas/validation/org/project/statistics") ~> routes ~> check { | ||
| response.shouldBeForbidden | ||
| } | ||
| } | ||
|
|
||
| "fail to get statistics on a validation job for an unknown project" in { | ||
| Get("/v1/jobs/schemas/validation/xxx/xxx/statistics") ~> asAlice ~> routes ~> check { | ||
| response.shouldFail(StatusCodes.NotFound, "ProjectNotFound") | ||
| } | ||
| } | ||
|
|
||
| "get statistics on a validation job with appropriate access" in { | ||
| val expectedResponse = | ||
| json""" | ||
| { | ||
| "@context": "https://bluebrain.github.io/nexus/contexts/statistics.json", | ||
| "delayInSeconds" : 0, | ||
| "discardedEvents": 400, | ||
| "evaluatedEvents": 8570, | ||
| "failedEvents": 30, | ||
| "lastEventDateTime": "${Instant.EPOCH}", | ||
| "lastProcessedEventDateTime": "${Instant.EPOCH}", | ||
| "processedEvents": 9000, | ||
| "remainingEvents": 0, | ||
| "totalEvents": 9000 | ||
| }""" | ||
|
|
||
| Get("/v1/jobs/schemas/validation/org/project/statistics") ~> asAlice ~> routes ~> check { | ||
| response.status shouldEqual StatusCodes.OK | ||
| response.asJson shouldEqual expectedResponse | ||
| } | ||
| } | ||
|
|
||
| "fail to download errors on a validation job without permission" in { | ||
| Get("/v1/jobs/schemas/validation/org/project/errors") ~> routes ~> check { | ||
| response.shouldBeForbidden | ||
| } | ||
| } | ||
|
|
||
| "fail to download errors on a validation job for an unknown project" in { | ||
| Get("/v1/jobs/schemas/validation/xxx/xxx/errors") ~> asAlice ~> routes ~> check { | ||
| response.shouldFail(StatusCodes.NotFound, "ProjectNotFound") | ||
| } | ||
| } | ||
|
|
||
| "download errors on a validation job with appropriate access" in { | ||
| val expectedResponse = | ||
| s"""{"id":"$resourceId","project":"${project.ref}","offset":{"value":42,"@type":"At"},"rev":1,"reason":{"type":"ValidationFail","value":{"details":"..."}}} | ||
| |{"id":"$resourceId","project":"${project.ref}","offset":{"value":42,"@type":"At"},"rev":1,"reason":{"type":"ValidationFail","value":{"details":"..."}}} | ||
| |""".stripMargin | ||
|
|
||
| Get("/v1/jobs/schemas/validation/org/project/errors") ~> Accept(`*/*`) ~> asAlice ~> routes ~> check { | ||
| response.status shouldEqual StatusCodes.OK | ||
| response.asString shouldEqual expectedResponse | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.