This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7922-34eb44b21f3dd91340cb63310bb2e9dd31727d92 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 5de9bf94766d441e7d1655bf326a9208dc6769a5 Author: Tanishq Gandhi <[email protected]> AuthorDate: Wed Aug 26 04:16:48 2026 +0000 feat(file-service): complete the model file API for the management UI (#7922) ### What changes were proposed in this PR? #6869 and #6872 gave models create/list/access/version-upload. This PR adds the remaining read and staged-change endpoints, so the model management UI can be built as frontend-only work. Nine endpoints on `ModelResource`: | Endpoint | Purpose | | --- | --- | | `GET /model/presign-download` (+ `-s3`, and `public-` variants of both) | signed URL for one file — preview, and how a worker reads a model | | `GET /model/{mid}/diff` | list staged (uncommitted) changes | | `PUT /model/{mid}/diff?filePath=` | discard one staged change | | `POST /model/{mid}/existing-upload-files` | which files the repo already holds at the same size, so the client can skip re-uploading large weights | | `GET /model/{mid}/versionZip?mvid=\|latest=true` | download a whole version | | `GET /model/user-model-owners` | owner facet for the list page | None of it is a second copy of `DatasetResource`. Each piece moved into `ResourceUploadService`/`ResourceAccess` parameterised by `ResourceStorage`, and the dataset endpoints now call the same code: `DatasetResource` **−285 lines**, shared layer +226, `ModelResource` +233. Nine new endpoints for a net reduction in production code. Also fixes `listModels` reporting `size = 0` for models reached by grant while computing it for public ones. `framework` and `format` are now validated against a known set (400 otherwise) so a loader can dispatch on the value. They stay metadata only — a model is a folder, and no per-file extension check is applied. ### Any related issues, documentation, discussions? - Closes #6498 - Part of #6494 - Replaces #6974, which could not be reopened after the branch was rebuilt on the current stack - Stacked on #6872, which is stacked on #6869; base is `main`, so the diff carries those two PRs' commits until they merge. The other two parents, #7762 and #7764, have already landed. ### How was this PR tested? New `ModelApiForUiSpec` and `ModelDownloadResourceSpec` cover every endpoint, including negative cases (no access, private model anonymous, `mvid` and `latest` both/neither). The presign tests fetch the URL and compare bytes end to end through MinIO. ``` sbt "FileService/test" # 376/376 sbt "WorkflowCore/test" # 787/787 sbt scalafmtCheckAll "scalafixAll --check" ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --------- Co-authored-by: ali <[email protected]> Co-authored-by: Claude Opus 4.8 <[email protected]> --- .../texera/service/resource/DatasetResource.scala | 276 ++------------- .../texera/service/resource/ModelResource.scala | 237 +++++++++++-- .../texera/service/resource/ResourceAccess.scala | 20 ++ .../service/resource/ResourceUploadService.scala | 280 ++++++++++++++- .../texera/service/type/ResourceUploadTypes.scala | 36 ++ .../service/resource/DatasetResourceSpec.scala | 29 +- .../service/resource/ModelApiForUiSpec.scala | 380 +++++++++++++++++++++ .../resource/ModelDownloadResourceSpec.scala | 352 +++++++++++++++++++ .../resource/ModelResourcePermissionsSpec.scala | 12 +- .../service/resource/ModelUploadResourceSpec.scala | 31 +- .../service/resource/ResourceTestHelpers.scala | 61 ++++ 11 files changed, 1405 insertions(+), 309 deletions(-) diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala index 9f34dae154..f8bc532909 100644 --- a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala +++ b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala @@ -36,7 +36,6 @@ import org.apache.texera.dao.SqlServer.withTransaction import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum} import org.apache.texera.dao.jooq.generated.tables.Dataset.DATASET import org.apache.texera.dao.jooq.generated.tables.DatasetContributor.DATASET_CONTRIBUTOR -import org.apache.texera.dao.jooq.generated.tables.DatasetUserAccess.DATASET_USER_ACCESS import org.apache.texera.dao.jooq.generated.tables.DatasetVersion.DATASET_VERSION import org.apache.texera.dao.jooq.generated.tables.User.USER import org.apache.texera.dao.jooq.generated.tables.daos.{ @@ -49,7 +48,7 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{ DatasetUserAccess, DatasetVersion } -import org.apache.texera.service.`type`.LakeFSFileNode +import org.apache.texera.service.`type`.{Diff, ExistingUploadFilesRequest, LakeFSFileNode} import org.apache.texera.service.resource.DatasetAccessResource._ import org.apache.texera.service.resource.ResourceTables.{Dataset => DATASET_RESOURCE} import org.apache.texera.service.resource.DatasetResource.{context, _} @@ -57,13 +56,10 @@ import org.apache.texera.service.util.S3StorageClient import org.jooq.impl.DSL import org.jooq.{DSLContext, EnumType} -import java.io.{InputStream, OutputStream} -import java.net.{URI, URLDecoder} -import java.nio.charset.StandardCharsets -import java.nio.file.{Files, Paths} +import java.io.InputStream +import java.net.URI import java.util import java.util.Optional -import java.util.zip.{ZipEntry, ZipOutputStream} import scala.jdk.CollectionConverters._ import scala.jdk.OptionConverters._ import org.apache.commons.io.FilenameUtils @@ -264,16 +260,8 @@ object DatasetResource { contributors: Option[List[Contributor]] = None ) - case class Diff( - path: String, - pathType: String, - diffType: String, // "added", "removed", "changed", etc. - sizeBytes: Option[Long] // Size of the changed file (None for directories) - ) - - case class ExistingUploadFile(path: String, sizeBytes: Long) - - case class ExistingUploadFilesRequest(files: List[ExistingUploadFile]) + val ExistingUploadFilesRequest: org.apache.texera.service.`type`.ExistingUploadFilesRequest.type = + org.apache.texera.service.`type`.ExistingUploadFilesRequest case class DatasetDescriptionModification(did: Integer, description: String) @@ -820,28 +808,7 @@ class DatasetResource extends LazyLogging { @PathParam("did") did: Integer, @Auth user: SessionUser ): List[Diff] = { - val uid = user.getUid - withTransaction(context) { ctx => - if (!userHasReadAccess(ctx, did, uid)) { - throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE) - } - - // Retrieve staged (uncommitted) changes from LakeFS - val dataset = getDatasetByID(ctx, did) - val lakefsDiffs = withLakeFSErrorHandling { - LakeFSStorageClient.retrieveUncommittedObjects(dataset.getRepositoryName) - } - - // Convert LakeFS Diff objects to our custom Diff case class - lakefsDiffs.map(d => - new Diff( - d.getPath, - d.getPathType.getValue, - d.getType.getValue, - Option(d.getSizeBytes).map(_.longValue()) - ) - ) - } + ResourceUploadService.stagedChanges(ResourceStorage.Dataset, did, user.getUid) } @POST @@ -853,54 +820,13 @@ class DatasetResource extends LazyLogging { request: ExistingUploadFilesRequest, @Auth user: SessionUser ): Response = { - val uid = user.getUid - withTransaction(context) { ctx => - if (!userHasWriteAccess(ctx, did, uid)) { - throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE) - } - - val requested = Option(request) - .flatMap(request => Option(request.files)) - .getOrElse(List.empty) - .map { file => - val originalPath = file.path - val path = ResourceNaming.validateAndNormalizeFilePathOrThrow(originalPath) - if (file.sizeBytes < 0L) throw new BadRequestException("sizeBytes must be >= 0") - (path, originalPath, file.sizeBytes) - } - - val dataset = getDatasetByID(ctx, did) - val committed = getLatestDatasetVersion(ctx, did) - .map { v => - withLakeFSErrorHandling( - s"retrieving committed files of dataset '${dataset.getName}'" - ) { - LakeFSStorageClient - .retrieveObjectsOfVersion(dataset.getRepositoryName, v.getVersionHash) - .map(obj => obj.getPath -> obj.getSizeBytes.longValue()) - } - } - .getOrElse(List.empty) - - val staged = withLakeFSErrorHandling( - s"retrieving staged files of dataset '${dataset.getName}'" - ) { - LakeFSStorageClient.retrieveUncommittedObjects(dataset.getRepositoryName) - } - .filterNot(diff => Option(diff.getType).exists(_.getValue.equalsIgnoreCase("removed"))) - .flatMap(diff => Option(diff.getSizeBytes).map(size => diff.getPath -> size.longValue())) - - val existing = (committed ++ staged).toMap - val matches = requested - .collect { - case (path, originalPath, size) if existing.get(path).contains(size) => originalPath - } - .toList - .distinct - .sorted - - Response.ok(Map("filePaths" -> matches.asJava)).build() - } + ResourceUploadService.matchExistingUploads( + ResourceStorage.Dataset, + did, + user.getUid, + request, + ctx => getLatestDatasetVersion(ctx, did).map(_.getVersionHash) + ) } @PUT @@ -912,20 +838,12 @@ class DatasetResource extends LazyLogging { @QueryParam("filePath") encodedFilePath: String, @Auth user: SessionUser ): Response = { - val uid = user.getUid - withTransaction(context) { ctx => - if (!userHasWriteAccess(ctx, did, uid)) { - throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE) - } - val repositoryName = getDatasetByID(ctx, did).getRepositoryName - - // Decode the file path - val filePath = URLDecoder.decode(encodedFilePath, StandardCharsets.UTF_8.name()) - withLakeFSErrorHandling(s"resetting uncommitted changes of file '$filePath'") { - LakeFSStorageClient.resetObjectUploadOrDeletion(repositoryName, filePath) - } - Response.ok().build() - } + ResourceUploadService.resetStagedChange( + ResourceStorage.Dataset, + did, + encodedFilePath, + user.getUid + ) } /** @@ -1085,50 +1003,12 @@ class DatasetResource extends LazyLogging { throw new BadRequestException("Invalid parameters") } - // Retrieve dataset and version details - val datasetName = dataset.getName - val repositoryName = dataset.getRepositoryName - val versionHash = datasetVersion.getVersionHash - val objects = withLakeFSErrorHandling( - s"listing files of version '$versionHash' of dataset '$datasetName'" - ) { - LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName, versionHash) - } - - if (objects.isEmpty) { - return Response - .status(Response.Status.NOT_FOUND) - .entity(s"No objects found in version $versionHash of repository $repositoryName") - .build() - } - - // StreamingOutput for ZIP download - val streamingOutput = new StreamingOutput { - override def write(outputStream: OutputStream): Unit = { - val zipOut = new ZipOutputStream(outputStream) - try { - objects.foreach { obj => - val filePath = obj.getPath - val file = withLakeFSErrorHandling(s"downloading file '$filePath' for the zip") { - LakeFSStorageClient.getFileFromRepo(repositoryName, versionHash, filePath) - } - - zipOut.putNextEntry(new ZipEntry(filePath)) - Files.copy(Paths.get(file.toURI), zipOut) - zipOut.closeEntry() - } - } finally { - zipOut.close() - } - } - } - - val zipFilename = s"""attachment; filename="$datasetName-${datasetVersion.getName}.zip"""" - - Response - .ok(streamingOutput, "application/zip") - .header("Content-Disposition", zipFilename) - .build() + ResourceUploadService.versionZipResponse( + dataset.getRepositoryName, + datasetVersion.getVersionHash, + dataset.getName, + datasetVersion.getName + ) } } @@ -1182,17 +1062,10 @@ class DatasetResource extends LazyLogging { @GET @RolesAllowed(Array("REGULAR", "ADMIN")) @Path("/user-dataset-owners") - def retrieveOwners(@Auth user: SessionUser): util.List[String] = { - context - .selectDistinct(USER.EMAIL) - .from(USER) - .join(DATASET) - .on(DATASET.OWNER_UID.eq(USER.UID)) - .join(DATASET_USER_ACCESS) - .on(DATASET_USER_ACCESS.DID.eq(DATASET.DID)) - .where(DATASET_USER_ACCESS.UID.eq(user.getUid)) - .fetchInto(classOf[String]) - } + def retrieveOwners(@Auth user: SessionUser): util.List[String] = + withTransaction(context)(ctx => + ResourceAccess.ownerEmailsVisibleTo(ctx, DATASET_RESOURCE, user.getUid) + ) /** @see [[ResourceNaming.validateName]] */ private def validateDatasetName(name: String): Unit = @@ -1236,89 +1109,14 @@ class DatasetResource extends LazyLogging { repositoryName: String, commitHash: String, uid: Integer - ): Response = { - resolveDatasetAndPath(encodedUrl, repositoryName, commitHash, uid) match { - case Left(errorResponse) => - errorResponse - - case Right((resolvedRepositoryName, resolvedCommitHash, resolvedFilePath)) => - val url = withLakeFSErrorHandling( - s"generating a presigned URL for file '$resolvedFilePath'" - ) { - LakeFSStorageClient.getFilePresignedUrl( - resolvedRepositoryName, - resolvedCommitHash, - resolvedFilePath - ) - } - - Response.ok(Map("presignedUrl" -> url)).build() - } - } - - private def resolveDatasetAndPath( - encodedUrl: String, - repositoryName: String, - commitHash: String, - uid: Integer - ): Either[Response, (String, String, String)] = { - val decodedPathStr = URLDecoder.decode(encodedUrl, StandardCharsets.UTF_8.name()) - - (Option(repositoryName), Option(commitHash)) match { - case (Some(_), None) | (None, Some(_)) => - // Case 1: Only one parameter is provided (error case) - Left( - Response - .status(Response.Status.BAD_REQUEST) - .entity( - "Both repositoryName and commitHash must be provided together, or neither should be provided." - ) - .build() - ) - - case (Some(repositoryName), Some(commit)) => - // Case 2: repositoryName and commitHash are provided, validate access - val response = withTransaction(context) { ctx => - val datasetDao = new DatasetDao(ctx.configuration()) - val datasets = datasetDao.fetchByRepositoryName(repositoryName).asScala.toList - - if (datasets.isEmpty || !userHasReadAccess(ctx, datasets.head.getDid, uid)) - throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE) - - val dataset = datasets.head - // Standard read access check only - download restrictions handled per endpoint - // Non-download operations (viewing) should work for all public datasets - - (repositoryName, commit, decodedPathStr) - } - Right(response) - - case (None, None) => - // Case 3: Neither repositoryName nor commitHash are provided, resolve normally - val response = withTransaction(context) { ctx => - val fileUri = FileResolver.resolve(decodedPathStr) - val document = - DocumentFactory.openReadonlyDocument(fileUri).asInstanceOf[OnVersionedFileResource] - val datasetDao = new DatasetDao(ctx.configuration()) - val datasets = - datasetDao.fetchByRepositoryName(document.getRepositoryName()).asScala.toList - - if (datasets.isEmpty || !userHasReadAccess(ctx, datasets.head.getDid, uid)) - throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE) - - val dataset = datasets.head - // Standard read access check only - download restrictions handled per endpoint - // Non-download operations (viewing) should work for all public datasets - - ( - document.getRepositoryName(), - document.getVersionHash(), - document.getFileRelativePath() - ) - } - Right(response) - } - } + ): Response = + ResourceUploadService.presignedUrlResponse( + ResourceStorage.Dataset, + encodedUrl, + repositoryName, + commitHash, + uid + ) // === Multipart helpers === diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala b/file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala index ec7c1ac774..7c323928c8 100644 --- a/file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala +++ b/file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala @@ -28,7 +28,7 @@ import org.apache.texera.amber.core.storage.ResourceType import org.apache.texera.amber.core.storage.util.LakeFSStorageClient import org.apache.texera.auth.SessionUser import org.apache.texera.common.config.StorageConfig -import org.apache.texera.dao.SqlServer +import org.apache.texera.dao.{SiteSettings, SqlServer} import org.apache.texera.dao.SqlServer.withTransaction import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum import org.apache.texera.dao.jooq.generated.tables.Model.MODEL @@ -36,7 +36,7 @@ import org.apache.texera.dao.jooq.generated.tables.ModelVersion.MODEL_VERSION import org.apache.texera.dao.jooq.generated.tables.User.USER import org.apache.texera.dao.jooq.generated.tables.daos.{ModelDao, ModelUserAccessDao} import org.apache.texera.dao.jooq.generated.tables.pojos.{Model, ModelUserAccess, ModelVersion} -import org.apache.texera.service.`type`.LakeFSFileNode +import org.apache.texera.service.`type`.{Diff, ExistingUploadFilesRequest, LakeFSFileNode} import org.apache.texera.service.resource.ResourceTables.{Model => MODEL_RESOURCE} import org.apache.texera.service.resource.ModelAccessResource._ import org.apache.texera.service.resource.ModelResource.{context, _} @@ -52,7 +52,24 @@ import scala.jdk.OptionConverters._ object ModelResource { // MVP supports a single framework; stored on the model so later frameworks can be added. - private val DEFAULT_FRAMEWORK = "pytorch" + // Callers may omit the framework; it is a display label, not a validation gate for files. + val DEFAULT_FRAMEWORK = "pytorch" + + // Recognised values for the `framework` and `format` labels. They are metadata, not file + // checks -- a loader dispatches on them, so an unknown value is rejected at creation + // rather than surfacing later as an unloadable model. + val SUPPORTED_FRAMEWORKS: Set[String] = Set("pytorch", "tensorflow", "onnx", "sklearn") + + val SUPPORTED_FORMATS: Set[String] = + Set("torchscript", "state-dict", "safetensors", "onnx", "savedmodel", "joblib", "pickle") + + private def validateLabel(field: String, value: String, allowed: Set[String]): Unit = { + if (!allowed.contains(value)) { + throw new BadRequestException( + s"Unsupported $field '$value'. Supported values: ${allowed.toList.sorted.mkString(", ")}." + ) + } + } // Matches model_version.name VARCHAR(128). private val MAX_VERSION_NAME_LENGTH = 128 @@ -64,6 +81,9 @@ object ModelResource { .getInstance() .createDSLContext() + private def singleFileUploadMaxBytes(defaultMiB: Long = 20L): Long = + SiteSettings.getLong("single_file_upload_max_size_mib", defaultMiB) * 1024L * 1024L + /** * Helper function to get the model from DB using mid */ @@ -206,8 +226,14 @@ class ModelResource extends LazyLogging { model.setIsPublic(isModelPublic) model.setIsDownloadable(isModelDownloadable) model.setOwnerUid(uid) - model.setFramework(Option(request.framework).filter(_.nonEmpty).getOrElse(DEFAULT_FRAMEWORK)) - model.setFormat(request.format) + val framework = + Option(request.framework).map(_.trim).filter(_.nonEmpty).getOrElse(DEFAULT_FRAMEWORK) + validateLabel("framework", framework, SUPPORTED_FRAMEWORKS) + val format = Option(request.format).map(_.trim).filter(_.nonEmpty) + format.foreach(validateLabel("format", _, SUPPORTED_FORMATS)) + + model.setFramework(framework) + model.setFormat(format.orNull) // insert record and get created model with mid val createdModel = ResourceNaming.failOnDuplicateName(MODEL_RESOURCE.label) { @@ -418,28 +444,19 @@ class ModelResource extends LazyLogging { model = model, accessPrivilege = privilege, ownerEmail = ownerEmail, - size = 0 + size = repositorySizeOrZero(model) ) ), fromPublic = (model, ownerEmail) => - try { - Some( - DashboardModel( - isOwner = false, - model = model, - accessPrivilege = PrivilegeEnum.READ, - ownerEmail = ownerEmail, - size = LakeFSStorageClient.retrieveRepositorySize(model.getRepositoryName) - ) + Some( + DashboardModel( + isOwner = false, + model = model, + accessPrivilege = PrivilegeEnum.READ, + ownerEmail = ownerEmail, + size = repositorySizeOrZero(model) ) - } catch { - case e: io.lakefs.clients.sdk.ApiException => - logger.error( - s"LakeFS ApiException for model repository '${model.getRepositoryName}': ${e.getMessage}", - e - ) - None - } + ) ) }) } @@ -464,6 +481,180 @@ class ModelResource extends LazyLogging { withTransaction(context)(ctx => getDashboardModel(ctx, mid, None)) } + @GET + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/{mid}/versionZip") + def getModelVersionZip( + @PathParam("mid") mid: Integer, + @QueryParam("mvid") mvid: Integer, + @QueryParam("latest") latest: java.lang.Boolean, + @Auth user: SessionUser + ): Response = + withTransaction(context) { ctx => + if ((mvid != null && latest != null) || (mvid == null && latest == null)) { + throw new BadRequestException("Specify exactly one: mvid=<ID> OR latest=true") + } + + val uid = user.getUid + if (!userHasReadAccess(ctx, mid, uid)) { + throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE) + } + + val model = getModelByID(ctx, mid) + // Non-owners may download only while the owner leaves the model downloadable. + if (!userOwnModel(ctx, mid, uid) && !model.getIsDownloadable) { + throw new ForbiddenException("Model download is not allowed") + } + + // latest=false is not "give me the latest": only TRUE selects it, anything else is a 400. + val modelVersion = + if (mvid != null) getModelVersionByID(ctx, mid, mvid) + else if (java.lang.Boolean.TRUE.equals(latest)) + getLatestModelVersion(ctx, mid).getOrElse( + throw new NotFoundException(ERR_MODEL_VERSION_NOT_FOUND_MESSAGE) + ) + else throw new BadRequestException("Invalid parameters") + + ResourceUploadService.versionZipResponse( + model.getRepositoryName, + modelVersion.getVersionHash, + model.getName, + modelVersion.getName + ) + } + + /** Owner facet for the model list page. */ + @GET + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/user-model-owners") + def retrieveOwners(@Auth user: SessionUser): java.util.List[String] = + withTransaction(context)(ctx => + ResourceAccess.ownerEmailsVisibleTo(ctx, MODEL_RESOURCE, user.getUid) + ) + + // =========================================================================== + // Staged changes + // =========================================================================== + + @GET + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/{mid}/diff") + def getModelDiff( + @PathParam("mid") mid: Integer, + @Auth user: SessionUser + ): List[Diff] = + ResourceUploadService.stagedChanges(ResourceStorage.Model, mid, user.getUid) + + @PUT + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/{mid}/diff") + def resetModelFileDiff( + @PathParam("mid") mid: Integer, + @QueryParam("filePath") encodedFilePath: String, + @Auth user: SessionUser + ): Response = + ResourceUploadService.resetStagedChange( + ResourceStorage.Model, + mid, + encodedFilePath, + user.getUid + ) + + @POST + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/{mid}/existing-upload-files") + @Consumes(Array(MediaType.APPLICATION_JSON)) + def findExistingUploadFiles( + @PathParam("mid") mid: Integer, + request: ExistingUploadFilesRequest, + @Auth user: SessionUser + ): Response = + ResourceUploadService.matchExistingUploads( + ResourceStorage.Model, + mid, + user.getUid, + request, + ctx => getLatestModelVersion(ctx, mid).map(_.getVersionHash) + ) + + // =========================================================================== + // Presigned downloads + // =========================================================================== + + /** + * Resolves a presign request against the model tables and wraps the signed URL. + * The resolution itself is shared with datasets; only the descriptor differs. + */ + /** Size of a model's LakeFS repository, or 0 if LakeFS cannot answer. */ + private def repositorySizeOrZero(model: Model): Long = { + try { + LakeFSStorageClient.retrieveRepositorySize(model.getRepositoryName) + } catch { + case e: io.lakefs.clients.sdk.ApiException => + logger.error( + s"LakeFS ApiException for model repository '${model.getRepositoryName}': ${e.getMessage}", + e + ) + 0L + } + } + + private def generatePresignedResponse( + encodedUrl: String, + repositoryName: String, + commitHash: String, + uid: Integer + ): Response = + ResourceUploadService.presignedUrlResponse( + ResourceStorage.Model, + encodedUrl, + repositoryName, + commitHash, + uid + ) + + @GET + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/presign-download") + def getPresignedUrl( + @QueryParam("filePath") encodedUrl: String, + @QueryParam("repositoryName") repositoryName: String, + @QueryParam("commitHash") commitHash: String, + @Auth user: SessionUser + ): Response = + generatePresignedResponse(encodedUrl, repositoryName, commitHash, user.getUid) + + @GET + @RolesAllowed(Array("REGULAR", "ADMIN")) + @Path("/presign-download-s3") + def getPresignedUrlWithS3( + @QueryParam("filePath") encodedUrl: String, + @QueryParam("repositoryName") repositoryName: String, + @QueryParam("commitHash") commitHash: String, + @Auth user: SessionUser + ): Response = + generatePresignedResponse(encodedUrl, repositoryName, commitHash, user.getUid) + + @GET + @PermitAll + @Path("/public-presign-download") + def getPublicPresignedUrl( + @QueryParam("filePath") encodedUrl: String, + @QueryParam("repositoryName") repositoryName: String, + @QueryParam("commitHash") commitHash: String + ): Response = + generatePresignedResponse(encodedUrl, repositoryName, commitHash, null) + + @GET + @PermitAll + @Path("/public-presign-download-s3") + def getPublicPresignedUrlWithS3( + @QueryParam("filePath") encodedUrl: String, + @QueryParam("repositoryName") repositoryName: String, + @QueryParam("commitHash") commitHash: String + ): Response = + generatePresignedResponse(encodedUrl, repositoryName, commitHash, null) + // =========================================================================== // Versioning // =========================================================================== diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala index 6fffd1d335..3276d32b68 100644 --- a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala +++ b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala @@ -318,4 +318,24 @@ object ResourceAccess { s"You do not have access to ${resource.label} $id" ) } + + /** + * Emails of the owners of every resource the caller has an explicit grant on, for the + * owner facet on list pages. + */ + def ownerEmailsVisibleTo[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + uid: Integer + ): java.util.List[String] = + ctx + .selectDistinct(USER.EMAIL) + .from(USER) + .join(resource.table) + .on(resource.ownerUidField.eq(USER.UID)) + .join(resource.accessTable) + .on(resource.accessIdField.eq(resource.idField)) + .where(resource.accessUidField.eq(uid)) + .fetchInto(classOf[String]) + } diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceUploadService.scala b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceUploadService.scala index 693e25c510..b974cf31a8 100644 --- a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceUploadService.scala +++ b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceUploadService.scala @@ -20,9 +20,11 @@ package org.apache.texera.service.resource import jakarta.ws.rs._ -import jakarta.ws.rs.core.{HttpHeaders, Response} +import jakarta.ws.rs.core.{HttpHeaders, Response, StreamingOutput} import org.apache.texera.amber.core.storage.ResourceType +import org.apache.texera.amber.core.storage.model.OnVersionedFileResource import org.apache.texera.amber.core.storage.util.LakeFSStorageClient +import org.apache.texera.amber.core.storage.{DocumentFactory, FileResolver} import org.apache.texera.common.config.StorageConfig import org.apache.texera.dao.{SiteSettings, SqlServer} import org.apache.texera.dao.SqlServer.withTransaction @@ -56,9 +58,11 @@ import org.jooq.impl.DSL.{inline => inl} import org.jooq.{DSLContext, Record, Record2, Result, Table, TableField} import software.amazon.awssdk.services.s3.model.UploadPartResponse -import java.io.InputStream +import java.io.{InputStream, OutputStream} import java.net.URLDecoder import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Paths} +import java.util.zip.{ZipEntry, ZipOutputStream} import java.sql.SQLException import java.time.OffsetDateTime import java.util.Optional @@ -234,6 +238,278 @@ object ResourceUploadService { throw new NotFoundException(s"${s.resource.label.capitalize} $resourceId not found") ) + /** + * Resolves a presign-download request to (repositoryName, commitHash, relativePath), + * after confirming the caller may read the owning resource. + * + * A caller either addresses the file directly (repositoryName + commitHash) or hands over a + * logical path and lets the server resolve it; supplying exactly one of the pair is a client + * error. Both routes then converge: find the resource that owns the repository, check read + * access, and hand back the triple. `FileResolver` already dispatches on the path's + * resource-type prefix, so the logical-path route needs nothing resource-specific here. + */ + def resolveVersionedFile[R <: Record, A <: Record, S <: Record, P <: Record]( + s: ResourceStorage[R, A, S, P], + encodedUrl: String, + repositoryName: String, + commitHash: String, + uid: Integer + ): Either[Response, (String, String, String)] = { + val decodedPath = URLDecoder.decode(encodedUrl, StandardCharsets.UTF_8.name()) + + requireBothOrNeither(repositoryName, commitHash) match { + case Some(badRequest) => Left(badRequest) + case None => + Right(withTransaction(context) { ctx => + if (repositoryName != null) { + requireReadAccessToRepository(ctx, s, repositoryName, uid) + (repositoryName, commitHash, decodedPath) + } else { + val document = DocumentFactory + .openReadonlyDocument(FileResolver.resolve(decodedPath)) + .asInstanceOf[OnVersionedFileResource] + requireReadAccessToRepository(ctx, s, document.getRepositoryName(), uid) + ( + document.getRepositoryName(), + document.getVersionHash(), + document.getFileRelativePath() + ) + } + }) + } + } + + /** + * The whole presign-download endpoint: resolve the addressed file, check read + * access, and hand back the signed URL. Both resources' endpoints are this call. + */ + def presignedUrlResponse[R <: Record, A <: Record, S <: Record, P <: Record]( + s: ResourceStorage[R, A, S, P], + encodedUrl: String, + repositoryName: String, + commitHash: String, + uid: Integer + ): Response = + resolveVersionedFile(s, encodedUrl, repositoryName, commitHash, uid) match { + case Left(errorResponse) => errorResponse + case Right((repo, commit, path)) => presignedResponse(repo, commit, path) + } + + /** + * A caller either addresses a file directly (repositoryName + commitHash) or lets + * the server resolve it from a logical path (neither). Exactly one is a client error. + * + * @return Some(400 response) when only one of the two was provided, None otherwise + */ + private def requireBothOrNeither( + repositoryName: String, + commitHash: String + ): Option[Response] = + (Option(repositoryName), Option(commitHash)) match { + case (Some(_), None) | (None, Some(_)) => + Some( + Response + .status(Response.Status.BAD_REQUEST) + .entity( + "Both repositoryName and commitHash must be provided together, " + + "or neither should be provided." + ) + .build() + ) + case _ => None + } + + /** Wrap a LakeFS presigned URL in the response shape the frontend expects. */ + private def presignedResponse( + repositoryName: String, + commitHash: String, + filePath: String + ): Response = { + val url = withLakeFSErrorHandling( + s"generating a presigned URL for file '$filePath'" + ) { + LakeFSStorageClient.getFilePresignedUrl(repositoryName, commitHash, filePath) + } + + Response.ok(Map("presignedUrl" -> url)).build() + } + + /** Read access is checked on the resource that owns the repository, not on the file. */ + private def requireReadAccessToRepository[R <: Record, A <: Record, S <: Record, P <: Record]( + ctx: DSLContext, + s: ResourceStorage[R, A, S, P], + repositoryName: String, + uid: Integer + ): Unit = { + val id = ctx + .select(s.resource.idField) + .from(s.resource.table) + .where(s.repositoryNameField.eq(repositoryName)) + .fetchOne(s.resource.idField) + + if (id == null || !ResourceAccess.userHasReadAccess(ctx, s.resource, id, uid)) { + throw new ForbiddenException(noAccessMessage(s)) + } + } + + /** + * Streams every file of one committed version as a ZIP. + * + * Resolving which version, and whether the caller may download it, stays with the + * resource: the download flag is not part of [[ResourceTables]]. This is the part that + * is identical either way -- listing the version's objects and piping them into a zip. + */ + def versionZipResponse( + repositoryName: String, + versionHash: String, + displayName: String, + versionName: String + ): Response = { + val objects = withLakeFSErrorHandling( + s"listing files of version '$versionHash' of '$displayName'" + ) { + LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName, versionHash) + } + + if (objects.isEmpty) { + Response + .status(Response.Status.NOT_FOUND) + .entity(s"No objects found in version $versionHash of repository $repositoryName") + .build() + } else { + val streamingOutput = new StreamingOutput { + override def write(outputStream: OutputStream): Unit = { + val zipOut = new ZipOutputStream(outputStream) + try { + objects.foreach { obj => + val filePath = obj.getPath + val file = withLakeFSErrorHandling(s"downloading file '$filePath' for the zip") { + LakeFSStorageClient.getFileFromRepo(repositoryName, versionHash, filePath) + } + zipOut.putNextEntry(new ZipEntry(filePath)) + Files.copy(Paths.get(file.toURI), zipOut) + zipOut.closeEntry() + } + } finally { + zipOut.close() + } + } + } + + Response + .ok(streamingOutput, "application/zip") + .header("Content-Disposition", s"""attachment; filename="$displayName-$versionName.zip"""") + .build() + } + } + + /** Staged (uncommitted) changes in a resource's repository. Requires read access. */ + def stagedChanges[R <: Record, A <: Record, S <: Record, P <: Record]( + s: ResourceStorage[R, A, S, P], + resourceId: Integer, + uid: Integer + ): List[org.apache.texera.service.`type`.Diff] = + withTransaction(context) { ctx => + if (!ResourceAccess.userHasReadAccess(ctx, s.resource, resourceId, uid)) { + throw new ForbiddenException(noAccessMessage(s)) + } + val repositoryName = repositoryNameOf(ctx, s, resourceId) + withLakeFSErrorHandling { + LakeFSStorageClient.retrieveUncommittedObjects(repositoryName) + }.map(d => + org.apache.texera.service.`type`.Diff( + d.getPath, + d.getPathType.getValue, + d.getType.getValue, + Option(d.getSizeBytes).map(_.longValue()) + ) + ) + } + + /** Discards one staged change, restoring the file to its last committed state. */ + def resetStagedChange[R <: Record, A <: Record, S <: Record, P <: Record]( + s: ResourceStorage[R, A, S, P], + resourceId: Integer, + encodedFilePath: String, + uid: Integer + ): Response = + withTransaction(context) { ctx => + if (!ResourceAccess.userHasWriteAccess(ctx, s.resource, resourceId, uid)) { + throw new ForbiddenException(noAccessMessage(s)) + } + val repositoryName = repositoryNameOf(ctx, s, resourceId) + val filePath = URLDecoder.decode(encodedFilePath, StandardCharsets.UTF_8.name()) + withLakeFSErrorHandling(s"resetting uncommitted changes of file '$filePath'") { + LakeFSStorageClient.resetObjectUploadOrDeletion(repositoryName, filePath) + } + Response.ok().build() + } + + /** + * Reports which of the requested files the repository already holds at the same size, + * so a client can skip re-uploading them. Committed and staged files both count. A + * staged deletion contributes no match of its own, but it does not withdraw the + * committed file either -- behaviour carried over verbatim from the dataset endpoint + * this was extracted from. + * + * The latest committed version is read through `latestVersionHash` because the version + * table is the one part of this that is resource-specific. + */ + def matchExistingUploads[R <: Record, A <: Record, S <: Record, P <: Record]( + s: ResourceStorage[R, A, S, P], + resourceId: Integer, + uid: Integer, + request: org.apache.texera.service.`type`.ExistingUploadFilesRequest, + latestVersionHash: DSLContext => Option[String] + ): Response = + withTransaction(context) { ctx => + if (!ResourceAccess.userHasWriteAccess(ctx, s.resource, resourceId, uid)) { + throw new ForbiddenException(noAccessMessage(s)) + } + + val requested = Option(request) + .flatMap(r => Option(r.files)) + .getOrElse(List.empty) + .map { file => + val originalPath = file.path + val path = ResourceNaming.validateAndNormalizeFilePathOrThrow(originalPath) + if (file.sizeBytes < 0L) throw new BadRequestException("sizeBytes must be >= 0") + (path, originalPath, file.sizeBytes) + } + + val repositoryName = repositoryNameOf(ctx, s, resourceId) + val committed = latestVersionHash(ctx) + .map { hash => + withLakeFSErrorHandling( + s"retrieving committed files of ${s.resource.label} $resourceId" + ) { + LakeFSStorageClient + .retrieveObjectsOfVersion(repositoryName, hash) + .map(obj => obj.getPath -> obj.getSizeBytes.longValue()) + } + } + .getOrElse(List.empty) + + val staged = withLakeFSErrorHandling( + s"retrieving staged files of ${s.resource.label} $resourceId" + ) { + LakeFSStorageClient.retrieveUncommittedObjects(repositoryName) + } + .filterNot(diff => Option(diff.getType).exists(_.getValue.equalsIgnoreCase("removed"))) + .flatMap(diff => Option(diff.getSizeBytes).map(size => diff.getPath -> size.longValue())) + + val existing = (committed ++ staged).toMap + val matches = requested + .collect { + case (path, originalPath, size) if existing.get(path).contains(size) => originalPath + } + .toList + .distinct + .sorted + + Response.ok(Map("filePaths" -> matches.asJava)).build() + } + /** Removes one staged (uncommitted) file from a resource's repository. */ def deleteStagedFile[R <: Record, A <: Record, S <: Record, P <: Record]( s: ResourceStorage[R, A, S, P], diff --git a/file-service/src/main/scala/org/apache/texera/service/type/ResourceUploadTypes.scala b/file-service/src/main/scala/org/apache/texera/service/type/ResourceUploadTypes.scala new file mode 100644 index 0000000000..9a9a775067 --- /dev/null +++ b/file-service/src/main/scala/org/apache/texera/service/type/ResourceUploadTypes.scala @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.texera.service.`type` + +/** + * A staged (uncommitted) change in a versioned resource repository. + * + * @param sizeBytes size of the changed file, None for directories + */ +case class Diff( + path: String, + pathType: String, + diffType: String, // "added", "removed", "changed", etc. + sizeBytes: Option[Long] +) + +case class ExistingUploadFile(path: String, sizeBytes: Long) + +case class ExistingUploadFilesRequest(files: List[ExistingUploadFile]) diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala index 0dfb39909a..fd1a525331 100644 --- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala @@ -43,6 +43,7 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{ User } import org.apache.texera.service.MockLakeFS +import org.apache.texera.service.`type`.{ExistingUploadFile, ExistingUploadFilesRequest} import org.apache.texera.service.util.S3StorageClient import org.jooq.SQLDialect import org.jooq.impl.DSL @@ -791,12 +792,12 @@ class DatasetResourceSpec val resp = datasetResource.findExistingUploadFiles( dataset.getDid, - DatasetResource.ExistingUploadFilesRequest( + ExistingUploadFilesRequest( List( - DatasetResource.ExistingUploadFile("committed.csv", committed.length), - DatasetResource.ExistingUploadFile("staged.csv", staged.length), - DatasetResource.ExistingUploadFile("wrong-size.csv", staged.length + 1), - DatasetResource.ExistingUploadFile("missing.csv", 1L) + ExistingUploadFile("committed.csv", committed.length), + ExistingUploadFile("staged.csv", staged.length), + ExistingUploadFile("wrong-size.csv", staged.length + 1), + ExistingUploadFile("missing.csv", 1L) ) ), sessionUser @@ -838,8 +839,8 @@ class DatasetResourceSpec val requestPath = "folder/../committed.csv" val resp = datasetResource.findExistingUploadFiles( dataset.getDid, - DatasetResource.ExistingUploadFilesRequest( - List(DatasetResource.ExistingUploadFile(requestPath, committed.length)) + ExistingUploadFilesRequest( + List(ExistingUploadFile(requestPath, committed.length)) ), sessionUser ) @@ -862,7 +863,7 @@ class DatasetResourceSpec val resp = datasetResource.findExistingUploadFiles( dataset.getDid, - DatasetResource.ExistingUploadFilesRequest(null), + ExistingUploadFilesRequest(null), sessionUser ) @@ -874,8 +875,8 @@ class DatasetResourceSpec val ex = intercept[BadRequestException] { datasetResource.findExistingUploadFiles( baseDataset.getDid, - DatasetResource.ExistingUploadFilesRequest( - List(DatasetResource.ExistingUploadFile("bad-size.csv", -1L)) + ExistingUploadFilesRequest( + List(ExistingUploadFile("bad-size.csv", -1L)) ), sessionUser ) @@ -888,8 +889,8 @@ class DatasetResourceSpec val ex = intercept[ForbiddenException] { datasetResource.findExistingUploadFiles( multipartDataset.getDid, - DatasetResource.ExistingUploadFilesRequest( - List(DatasetResource.ExistingUploadFile("private.csv", 1L)) + ExistingUploadFilesRequest( + List(ExistingUploadFile("private.csv", 1L)) ), multipartNoWriteSessionUser ) @@ -919,8 +920,8 @@ class DatasetResourceSpec val ex = intercept[NotFoundException] { datasetResource.findExistingUploadFiles( dataset.getDid, - DatasetResource.ExistingUploadFilesRequest( - List(DatasetResource.ExistingUploadFile("missing.csv", 1L)) + ExistingUploadFilesRequest( + List(ExistingUploadFile("missing.csv", 1L)) ), sessionUser ) diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/ModelApiForUiSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/ModelApiForUiSpec.scala new file mode 100644 index 0000000000..cf55ca09fa --- /dev/null +++ b/file-service/src/test/scala/org/apache/texera/service/resource/ModelApiForUiSpec.scala @@ -0,0 +1,380 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.texera.service.resource + +import jakarta.ws.rs._ +import org.apache.texera.amber.core.storage.util.LakeFSStorageClient +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum +import org.apache.texera.dao.jooq.generated.tables.daos.UserDao +import org.apache.texera.dao.jooq.generated.tables.pojos.User +import org.apache.texera.service.MockLakeFS +import org.apache.texera.service.`type`.{ExistingUploadFile, ExistingUploadFilesRequest} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} + +import java.io.ByteArrayInputStream +import scala.jdk.CollectionConverters._ + +// Covers the model endpoints the UI needs beyond upload and download: the staged +// (uncommitted) change list and its per-file reset, the "already uploaded" probe +// that lets a client skip re-sending bytes, the owner facet for the list page, +// and the framework/format metadata labels. +class ModelApiForUiSpec + extends AnyFlatSpec + with Matchers + with MockTexeraDB + with MockLakeFS + with BeforeAndAfterAll + with BeforeAndAfterEach + with ResourceTestHelpers { + + private def mkUser(name: String): User = { + val user = new User + user.setName(name) + user.setEmail(s"[email protected]") + user.setRole(UserRoleEnum.ADMIN) + user + } + + private val ownerUser: User = mkUser("model_ui_owner") + private val strangerUser: User = mkUser("model_ui_stranger") + + lazy val modelResource = new ModelResource() + lazy val sessionUser = new SessionUser(ownerUser) + lazy val strangerSession = new SessionUser(strangerUser) + + override protected def beforeAll(): Unit = { + super.beforeAll() + initializeDBAndReplaceDSLContext() + val userDao = new UserDao(getDSLContext.configuration()) + userDao.insert(ownerUser) + userDao.insert(strangerUser) + } + + override protected def afterAll(): Unit = { + try shutdownDB() + finally super.afterAll() + } + + // ---------- helpers ---------- + private def newModel( + framework: String = "pytorch", + format: String = null, + isPublic: Boolean = false + ): ModelResource.DashboardModel = + modelResource.createModel( + ModelResource.CreateModelRequest( + modelName = uniqueName("ui-model"), + modelDescription = "for model-api-for-ui tests", + isModelPublic = isPublic, + isModelDownloadable = true, + framework = framework, + format = format + ), + sessionUser + ) + + private def upload(mid: Integer, path: String, bytes: Array[Byte]): Unit = + modelResource + .uploadOneFileToModel( + mid, + urlEnc(path), + "upload", + new ByteArrayInputStream(bytes), + mkHeaders(bytes.length.toLong), + sessionUser + ) + .getStatus shouldEqual 200 + + private def matchedPaths(response: core.Response): List[String] = + response.getEntity + .asInstanceOf[Map[String, java.util.List[String]]]("filePaths") + .asScala + .toList + + // =========================================================================== + // staged changes + // =========================================================================== + "getModelDiff" should "list an uploaded-but-uncommitted file as added" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + + val diffs = modelResource.getModelDiff(mid, sessionUser) + + diffs.map(_.path) should contain("model.pt") + val diff = diffs.find(_.path == "model.pt").get + diff.diffType shouldEqual "added" + diff.sizeBytes shouldEqual Some(64L) + } + + it should "report nothing once the changes are committed" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + + modelResource.getModelDiff(mid, sessionUser) shouldBe empty + } + + it should "report a deletion of a committed file as a staged change" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + modelResource.deleteModelFile(mid, urlEnc("model.pt"), sessionUser).getStatus shouldEqual 200 + + modelResource.getModelDiff(mid, sessionUser).map(_.diffType) should contain("removed") + } + + it should "refuse a caller with no access to the model" in { + val mid = newModel().model.getMid + + assertThrows[ForbiddenException] { + modelResource.getModelDiff(mid, strangerSession) + } + } + + "resetModelFileDiff" should "drop a staged upload" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + + modelResource + .resetModelFileDiff(mid, urlEnc("model.pt"), sessionUser) + .getStatus shouldEqual 200 + + modelResource.getModelDiff(mid, sessionUser) shouldBe empty + } + + it should "restore a committed file whose deletion was staged" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + modelResource.deleteModelFile(mid, urlEnc("model.pt"), sessionUser) + + modelResource.resetModelFileDiff(mid, urlEnc("model.pt"), sessionUser) + + modelResource.getModelDiff(mid, sessionUser) shouldBe empty + } + + it should "refuse a caller with no write access to the model" in { + val mid = newModel(isPublic = true).model.getMid + + assertThrows[ForbiddenException] { + modelResource.resetModelFileDiff(mid, urlEnc("model.pt"), strangerSession) + } + } + + // =========================================================================== + // existing-upload-files + // =========================================================================== + "findExistingUploadFiles" should "match a committed file of the same size" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + + val response = modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", 64L))), + sessionUser + ) + + matchedPaths(response) shouldEqual List("model.pt") + } + + it should "match a staged file that has not been committed yet" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + + val response = modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", 64L))), + sessionUser + ) + + matchedPaths(response) shouldEqual List("model.pt") + } + + it should "not match a file whose size differs" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + + val response = modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", 65L))), + sessionUser + ) + + matchedPaths(response) shouldBe empty + } + + // A staged deletion does not withdraw the committed file from the match set. This is + // the dataset endpoint's behaviour, preserved by the extraction rather than chosen here. + it should "still match a committed file whose deletion is staged" in { + val mid = newModel().model.getMid + upload(mid, "model.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + modelResource.deleteModelFile(mid, urlEnc("model.pt"), sessionUser) + + val response = modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", 64L))), + sessionUser + ) + + matchedPaths(response) shouldEqual List("model.pt") + } + + it should "report matches for a model with no version yet as empty" in { + val mid = newModel().model.getMid + + val response = modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", 64L))), + sessionUser + ) + + matchedPaths(response) shouldBe empty + } + + it should "reject a negative size" in { + val mid = newModel().model.getMid + + assertThrows[BadRequestException] { + modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", -1L))), + sessionUser + ) + } + } + + it should "reject a path that escapes the repository root" in { + val mid = newModel().model.getMid + + assertThrows[BadRequestException] { + modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("../secret.pt", 64L))), + sessionUser + ) + } + } + + it should "refuse a caller with no write access to the model" in { + val mid = newModel(isPublic = true).model.getMid + + assertThrows[ForbiddenException] { + modelResource.findExistingUploadFiles( + mid, + ExistingUploadFilesRequest(List(ExistingUploadFile("model.pt", 64L))), + strangerSession + ) + } + } + + // =========================================================================== + // owner facet + // =========================================================================== + "retrieveOwners" should "include the caller once they own a model" in { + newModel() + + modelResource.retrieveOwners(sessionUser).asScala should contain(ownerUser.getEmail) + } + + it should "not leak the owner of a private model to a stranger" in { + newModel(isPublic = false) + + modelResource.retrieveOwners(strangerSession).asScala should not contain ownerUser.getEmail + } + + // The facet is scoped by explicit grant, not by readability: a public model the caller + // was never granted does not put its owner in their filter list. Matches datasets. + it should "not surface the owner of a public model the caller holds no grant on" in { + newModel(isPublic = true) + + modelResource.retrieveOwners(strangerSession).asScala should not contain ownerUser.getEmail + } + + // =========================================================================== + // framework / format labels + // =========================================================================== + "createModel" should "accept every supported framework" in { + ModelResource.SUPPORTED_FRAMEWORKS.foreach { framework => + newModel(framework = framework).model.getFramework shouldEqual framework + } + } + + it should "accept every supported format" in { + ModelResource.SUPPORTED_FORMATS.foreach { format => + newModel(format = format).model.getFormat shouldEqual format + } + } + + it should "reject an unsupported framework" in { + val ex = intercept[BadRequestException](newModel(framework = "caffe")) + ex.getMessage should include("Unsupported framework 'caffe'") + } + + it should "reject an unsupported format" in { + assertThrows[BadRequestException](newModel(format = "gguf")) + } + + it should "fall back to the default framework when none is given" in { + newModel(framework = null).model.getFramework shouldEqual ModelResource.DEFAULT_FRAMEWORK + } + + it should "treat a blank framework as absent rather than invalid" in { + newModel(framework = " ").model.getFramework shouldEqual ModelResource.DEFAULT_FRAMEWORK + } + + it should "leave the format unset when none is given" in { + newModel(format = null).model.getFormat shouldBe null + } + + // =========================================================================== + // listing sizes + // =========================================================================== + "listModels" should "report the repository size of an owned model" in { + val model = newModel() + upload(model.model.getMid, "sized.pt", Array.fill[Byte](4096)(0x6)) + modelResource.createModelVersion("v1", model.model.getMid, sessionUser) + + val listed = modelResource + .listModels(sessionUser) + .find(_.model.getMid == model.model.getMid) + .getOrElse(fail("the owned model should be listed")) + + listed.isOwner shouldBe true + listed.size should be >= 4096L + } + + it should "still list a model whose repository size cannot be read" in { + // An unreadable size degrades to 0 rather than dropping the row. + val model = newModel() + LakeFSStorageClient.deleteRepo(model.model.getRepositoryName) + + val listed = modelResource.listModels(sessionUser).find(_.model.getMid == model.model.getMid) + + listed should not be empty + listed.get.size shouldEqual 0L + } +} diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/ModelDownloadResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/ModelDownloadResourceSpec.scala new file mode 100644 index 0000000000..9ee52955f8 --- /dev/null +++ b/file-service/src/test/scala/org/apache/texera/service/resource/ModelDownloadResourceSpec.scala @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.texera.service.resource + +import jakarta.ws.rs._ +import jakarta.ws.rs.core._ +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum +import org.apache.texera.dao.jooq.generated.tables.daos.UserDao +import org.apache.texera.dao.jooq.generated.tables.pojos.User +import org.apache.texera.service.MockLakeFS +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} + +import java.io.{ByteArrayInputStream, ByteArrayOutputStream} +import java.util.zip.ZipInputStream + +// Covers the model download surface: presigned single-file URLs (which +// ModelFileDocument also depends on) and the version zip. +class ModelDownloadResourceSpec + extends AnyFlatSpec + with Matchers + with MockTexeraDB + with MockLakeFS + with BeforeAndAfterAll + with BeforeAndAfterEach + with ResourceTestHelpers { + + private def mkUser(name: String): User = { + val user = new User + user.setName(name) + user.setEmail(s"[email protected]") + user.setRole(UserRoleEnum.ADMIN) + user + } + + private val ownerUser: User = mkUser("model_download_owner") + private val strangerUser: User = mkUser("model_download_stranger") + + lazy val modelResource = new ModelResource() + lazy val sessionUser = new SessionUser(ownerUser) + lazy val strangerSession = new SessionUser(strangerUser) + + override protected def beforeAll(): Unit = { + super.beforeAll() + initializeDBAndReplaceDSLContext() + val userDao = new UserDao(getDSLContext.configuration()) + userDao.insert(ownerUser) + userDao.insert(strangerUser) + } + + override protected def afterAll(): Unit = { + try shutdownDB() + finally super.afterAll() + } + + // ---------- helpers ---------- + private def newModel( + isPublic: Boolean = false, + isDownloadable: Boolean = true + ): ModelResource.DashboardModel = + modelResource.createModel( + ModelResource.CreateModelRequest( + modelName = uniqueName("dl-model"), + modelDescription = "for download tests", + isModelPublic = isPublic, + isModelDownloadable = isDownloadable, + framework = "pytorch", + format = null + ), + sessionUser + ) + + private def upload(mid: Integer, path: String, bytes: Array[Byte]): Unit = + modelResource + .uploadOneFileToModel( + mid, + urlEnc(path), + "upload", + new ByteArrayInputStream(bytes), + mkHeaders(bytes.length.toLong), + sessionUser + ) + .getStatus shouldEqual 200 + + /** Creates a model holding one committed file, returning it with its commit hash. */ + private def modelWithCommittedFile( + path: String = "model.pt", + bytes: Array[Byte] = Array.fill[Byte](1024)(0x7), + isPublic: Boolean = false, + isDownloadable: Boolean = true + ): (ModelResource.DashboardModel, String) = { + val model = newModel(isPublic, isDownloadable) + upload(model.model.getMid, path, bytes) + val version = modelResource.createModelVersion("v", model.model.getMid, sessionUser) + (model, version.modelVersion.getVersionHash) + } + + private def presignedUrlOf(response: Response): String = + response.getEntity.asInstanceOf[Map[String, String]]("presignedUrl") + + /** MinIO is bound to a fixed host port precisely so presigned URLs resolve here. */ + private def fetch(url: String): Array[Byte] = { + val stream = new java.net.URL(url).openStream() + try stream.readAllBytes() + finally stream.close() + } + // =========================================================================== + // presign-download + // =========================================================================== + "getPresignedUrl" should "return a presigned URL that serves the file's bytes" in { + val bytes = Array.tabulate[Byte](1024)(i => (i % 251).toByte) + val (model, commitHash) = modelWithCommittedFile(bytes = bytes) + + val response = modelResource.getPresignedUrl( + urlEnc("model.pt"), + model.model.getRepositoryName, + commitHash, + sessionUser + ) + + response.getStatus shouldEqual 200 + // The URL addresses the physical object, not the logical path, so the only + // meaningful assertion is that fetching it yields exactly what was uploaded. + val url = presignedUrlOf(response) + url should include("X-Amz-Signature") + fetch(url) shouldEqual bytes + } + + it should "refuse a caller with no access to the model" in { + val (model, commitHash) = modelWithCommittedFile() + + assertThrows[ForbiddenException] { + modelResource.getPresignedUrl( + urlEnc("model.pt"), + model.model.getRepositoryName, + commitHash, + strangerSession + ) + } + } + + it should "refuse a repository that belongs to no model" in { + assertThrows[ForbiddenException] { + modelResource.getPresignedUrl( + urlEnc("model.pt"), + "model-does-not-exist", + "deadbeef", + sessionUser + ) + } + } + + it should "reject a repositoryName without a commitHash" in { + val response = modelResource.getPresignedUrl(urlEnc("model.pt"), "model-1", null, sessionUser) + response.getStatus shouldEqual Response.Status.BAD_REQUEST.getStatusCode + } + + it should "reject a commitHash without a repositoryName" in { + val response = modelResource.getPresignedUrl(urlEnc("model.pt"), null, "abc123", sessionUser) + response.getStatus shouldEqual Response.Status.BAD_REQUEST.getStatusCode + } + + "getPresignedUrlWithS3" should "resolve the same file as the non-S3 variant" in { + val (model, commitHash) = modelWithCommittedFile() + + modelResource + .getPresignedUrlWithS3( + urlEnc("model.pt"), + model.model.getRepositoryName, + commitHash, + sessionUser + ) + .getStatus shouldEqual 200 + } + + "getPublicPresignedUrl" should "serve a file of a public model to an anonymous caller" in { + val (model, commitHash) = modelWithCommittedFile(isPublic = true) + + modelResource + .getPublicPresignedUrl(urlEnc("model.pt"), model.model.getRepositoryName, commitHash) + .getStatus shouldEqual 200 + } + + it should "refuse a file of a private model to an anonymous caller" in { + val (model, commitHash) = modelWithCommittedFile(isPublic = false) + + assertThrows[ForbiddenException] { + modelResource.getPublicPresignedUrl( + urlEnc("model.pt"), + model.model.getRepositoryName, + commitHash + ) + } + } + + // =========================================================================== + // versionZip + // =========================================================================== + "getModelVersionZip" should "stream a zip holding every file of the version" in { + val model = newModel() + val mid = model.model.getMid + upload(mid, "model.pt", Array.fill[Byte](512)(0x1)) + upload(mid, "tokenizer/vocab.txt", Array.fill[Byte](64)(0x2)) + val version = modelResource.createModelVersion("v1", mid, sessionUser) + + val response = + modelResource.getModelVersionZip(mid, version.modelVersion.getMvid, null, sessionUser) + + response.getStatus shouldEqual 200 + response.getHeaderString("Content-Disposition") should include(".zip") + + val buffer = new ByteArrayOutputStream() + response.getEntity.asInstanceOf[StreamingOutput].write(buffer) + + val zip = new ZipInputStream(new java.io.ByteArrayInputStream(buffer.toByteArray)) + val entries = Iterator + .continually(zip.getNextEntry) + .takeWhile(_ != null) + .map(_.getName) + .toList + zip.close() + + entries should contain allOf ("model.pt", "tokenizer/vocab.txt") + } + + it should "accept latest=true instead of an explicit version id" in { + val (model, _) = modelWithCommittedFile() + + modelResource + .getModelVersionZip(model.model.getMid, null, java.lang.Boolean.TRUE, sessionUser) + .getStatus shouldEqual 200 + } + + it should "reject being given both a version id and latest=true" in { + val (model, _) = modelWithCommittedFile() + val version = modelResource.retrieveLatestModelVersion(model.model.getMid, sessionUser) + + assertThrows[BadRequestException] { + modelResource.getModelVersionZip( + model.model.getMid, + version.modelVersion.getMvid, + java.lang.Boolean.TRUE, + sessionUser + ) + } + } + + it should "reject being given neither a version id nor latest=true" in { + val (model, _) = modelWithCommittedFile() + + assertThrows[BadRequestException] { + modelResource.getModelVersionZip(model.model.getMid, null, null, sessionUser) + } + } + + it should "report not-found for a version whose files were all deleted" in { + // Deleting the only file is itself a staged change, so a genuinely empty + // version can be committed — the zip of it has nothing to stream. + val model = newModel() + val mid = model.model.getMid + upload(mid, "only.pt", Array.fill[Byte](64)(0x1)) + modelResource.createModelVersion("v1", mid, sessionUser) + modelResource.deleteModelFile(mid, urlEnc("only.pt"), sessionUser).getStatus shouldEqual 200 + val emptyVersion = modelResource.createModelVersion("v2-empty", mid, sessionUser) + + val response = + modelResource.getModelVersionZip(mid, emptyVersion.modelVersion.getMvid, null, sessionUser) + + response.getStatus shouldEqual Response.Status.NOT_FOUND.getStatusCode + } + + it should "report not-found for a model that has no version yet" in { + val model = newModel() + + assertThrows[NotFoundException] { + modelResource.getModelVersionZip( + model.model.getMid, + null, + java.lang.Boolean.TRUE, + sessionUser + ) + } + } + + it should "refuse a caller with no access to the model" in { + val (model, _) = modelWithCommittedFile() + + assertThrows[ForbiddenException] { + modelResource.getModelVersionZip( + model.model.getMid, + null, + java.lang.Boolean.TRUE, + strangerSession + ) + } + } + + it should "refuse a non-owner when the model is not downloadable" in { + // readable because it is public, but the owner disabled downloads + val (model, _) = modelWithCommittedFile(isPublic = true, isDownloadable = false) + + assertThrows[ForbiddenException] { + modelResource.getModelVersionZip( + model.model.getMid, + null, + java.lang.Boolean.TRUE, + strangerSession + ) + } + } + + it should "reject latest=false instead of treating it as latest" in { + val (model, _) = modelWithCommittedFile() + + assertThrows[BadRequestException] { + modelResource.getModelVersionZip( + model.model.getMid, + null, + java.lang.Boolean.FALSE, + sessionUser + ) + } + } + + it should "still let the owner download a model that is not downloadable" in { + val (model, _) = modelWithCommittedFile(isPublic = false, isDownloadable = false) + + modelResource + .getModelVersionZip(model.model.getMid, null, java.lang.Boolean.TRUE, sessionUser) + .getStatus shouldEqual 200 + } +} diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/ModelResourcePermissionsSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/ModelResourcePermissionsSpec.scala index cc4e3ed2c4..01851aa64b 100644 --- a/file-service/src/test/scala/org/apache/texera/service/resource/ModelResourcePermissionsSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/resource/ModelResourcePermissionsSpec.scala @@ -29,7 +29,7 @@ import java.lang.reflect.Method // FileService registers RolesAllowedDynamicFeature, so each endpoint is enforced by its own method-level // annotation, and one carrying neither @PermitAll nor @RolesAllowed defaults to // OPEN. These tests pin that contract: every endpoint carries exactly one of the -// two, the non-public ones require REGULAR/ADMIN, and only the public read stays +// two, the non-public ones require REGULAR/ADMIN, and only the public reads stay // anonymous-accessible. class ModelResourcePermissionsSpec extends AnyFlatSpec with Matchers { @@ -43,7 +43,13 @@ class ModelResourcePermissionsSpec extends AnyFlatSpec with Matchers { private def rolesOf(m: Method): Option[RolesAllowed] = Option(m.getAnnotation(classOf[RolesAllowed])) - private val publicEndpointMethods: Set[String] = Set("getPublicModel") + // Anonymous-readable, mirroring the dataset side: the public model read plus the two + // presign routes a public model's files are fetched through. + private val publicEndpointMethods: Set[String] = Set( + "getPublicModel", + "getPublicPresignedUrl", + "getPublicPresignedUrlWithS3" + ) "ModelResource" should "expose HTTP endpoints (sanity check for the reflection scan)" in { endpointMethods should not be empty @@ -75,7 +81,7 @@ class ModelResourcePermissionsSpec extends AnyFlatSpec with Matchers { } } - it should "keep getPublicModel @PermitAll, and it alone" in { + it should "keep the public model endpoints @PermitAll, and them alone" in { val present = endpointMethods.map(_.getName).toSet.intersect(publicEndpointMethods) withClue("public endpoints missing from ModelResource (renamed?): ") { present shouldBe publicEndpointMethods diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/ModelUploadResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/ModelUploadResourceSpec.scala index 82444491a2..f4f300b7dc 100644 --- a/file-service/src/test/scala/org/apache/texera/service/resource/ModelUploadResourceSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/resource/ModelUploadResourceSpec.scala @@ -33,10 +33,8 @@ import org.scalatest.matchers.should.Matchers import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} import java.io.ByteArrayInputStream -import java.net.URLEncoder import java.nio.charset.StandardCharsets -import java.util.{Collections, Date, Locale, Optional} -import scala.util.Random +import java.util.Optional class ModelUploadResourceSpec extends AnyFlatSpec @@ -44,7 +42,8 @@ class ModelUploadResourceSpec with MockTexeraDB with MockLakeFS with BeforeAndAfterAll - with BeforeAndAfterEach { + with BeforeAndAfterEach + with ResourceTestHelpers { private val ownerUser: User = { val user = new User @@ -81,30 +80,6 @@ class ModelUploadResourceSpec } // ---------- helpers ---------- - private def urlEnc(raw: String): String = - URLEncoder.encode(raw, StandardCharsets.UTF_8.name()) - - private def uniqueName(prefix: String): String = - s"$prefix-${System.nanoTime()}-${Random.alphanumeric.take(6).mkString.toLowerCase}" - - /** Minimal HttpHeaders exposing only Content-Length, which the upload paths read. */ - private def mkHeaders(contentLength: Long): HttpHeaders = - new HttpHeaders { - private val headers = new MultivaluedHashMap[String, String]() - headers.putSingle(HttpHeaders.CONTENT_LENGTH, contentLength.toString) - override def getHeaderString(name: String): String = headers.getFirst(name) - override def getRequestHeaders: MultivaluedMap[String, String] = headers - override def getRequestHeader(name: String): java.util.List[String] = - Option(headers.get(name)).getOrElse(Collections.emptyList[String]()) - override def getAcceptableMediaTypes: java.util.List[MediaType] = Collections.emptyList() - override def getAcceptableLanguages: java.util.List[Locale] = Collections.emptyList() - override def getMediaType: MediaType = null - override def getLanguage: Locale = null - override def getCookies: java.util.Map[String, Cookie] = Collections.emptyMap() - override def getDate: Date = null - override def getLength: Int = contentLength.toInt - } - /** Creates a fresh model (provisions its LakeFS repo) and returns it. */ private def newModel(): ModelResource.DashboardModel = modelResource.createModel( diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/ResourceTestHelpers.scala b/file-service/src/test/scala/org/apache/texera/service/resource/ResourceTestHelpers.scala new file mode 100644 index 0000000000..b41c1399a6 --- /dev/null +++ b/file-service/src/test/scala/org/apache/texera/service/resource/ResourceTestHelpers.scala @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.texera.service.resource + +import jakarta.ws.rs.core._ + +import java.net.URLEncoder +import java.nio.charset.StandardCharsets +import java.util.{Collections, Date, Locale} +import scala.util.Random + +/** + * Plumbing shared by the resource specs that drive the JAX-RS methods directly + * rather than over HTTP: query-parameter encoding, collision-free resource names, + * and the HttpHeaders the upload paths read. + */ +trait ResourceTestHelpers { + + /** Endpoints take file paths as a single URL-encoded query parameter. */ + protected def urlEnc(raw: String): String = + URLEncoder.encode(raw, StandardCharsets.UTF_8.name()) + + /** Names are unique per owner, so specs sharing a DB must not collide. */ + protected def uniqueName(prefix: String): String = + s"$prefix-${System.nanoTime()}-${Random.alphanumeric.take(6).mkString.toLowerCase}" + + /** Minimal HttpHeaders exposing only Content-Length, which the upload paths read. */ + protected def mkHeaders(contentLength: Long): HttpHeaders = + new HttpHeaders { + private val headers = new MultivaluedHashMap[String, String]() + headers.putSingle(HttpHeaders.CONTENT_LENGTH, contentLength.toString) + override def getHeaderString(name: String): String = headers.getFirst(name) + override def getRequestHeaders: MultivaluedMap[String, String] = headers + override def getRequestHeader(name: String): java.util.List[String] = + Option(headers.get(name)).getOrElse(Collections.emptyList[String]()) + override def getAcceptableMediaTypes: java.util.List[MediaType] = Collections.emptyList() + override def getAcceptableLanguages: java.util.List[Locale] = Collections.emptyList() + override def getMediaType: MediaType = null + override def getLanguage: Locale = null + override def getCookies: java.util.Map[String, Cookie] = Collections.emptyMap() + override def getDate: Date = null + override def getLength: Int = contentLength.toInt + } +}
