This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 5de9bf9476 feat(file-service): complete the model file API for the
management UI (#7922)
5de9bf9476 is described below
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
+ }
+}
