206 lines
8.0 KiB
Go
206 lines
8.0 KiB
Go
package service
|
|
|
|
import (
|
|
"strings"
|
|
"testing"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
func TestCoreServiceArtifactTransferWorkflow(t *testing.T) {
|
|
svc, sessionToken := newRegisteredArtifactTransferService(t)
|
|
payload := []byte("artifact payload for upload")
|
|
openRequest := validArtifactTransferOpen(sessionToken, payload, 8)
|
|
|
|
opened, err := svc.OpenArtifactTransfer(openRequest)
|
|
if err != nil {
|
|
t.Fatalf("open artifact transfer: %v", err)
|
|
}
|
|
if !opened.Accepted || opened.TransferID == "" || opened.TotalChunks != 4 || opened.NextMissingChunkIndex != 0 {
|
|
t.Fatalf("unexpected open response: %+v", opened)
|
|
}
|
|
if opened.Artifact.State != domain.ArtifactStateUploading {
|
|
t.Fatalf("expected uploading artifact, got %+v", opened.Artifact)
|
|
}
|
|
|
|
duplicateOpen, err := svc.OpenArtifactTransfer(openRequest)
|
|
if err != nil {
|
|
t.Fatalf("duplicate open artifact transfer: %v", err)
|
|
}
|
|
if !duplicateOpen.Duplicate || duplicateOpen.TransferID != opened.TransferID {
|
|
t.Fatalf("expected duplicate open response, got %+v", duplicateOpen)
|
|
}
|
|
|
|
firstChunk := validArtifactChunk(sessionToken, opened.TransferID, payload, 0, 8)
|
|
firstAck, err := svc.UploadArtifactChunk(firstChunk)
|
|
if err != nil {
|
|
t.Fatalf("upload first chunk: %v", err)
|
|
}
|
|
if !firstAck.Accepted || firstAck.NextMissingChunkIndex != 1 || len(firstAck.ReceivedChunkIndexes) != 1 || firstAck.ReceivedChunkIndexes[0] != 0 {
|
|
t.Fatalf("unexpected first chunk ack: %+v", firstAck)
|
|
}
|
|
|
|
duplicateAck, err := svc.UploadArtifactChunk(firstChunk)
|
|
if err != nil {
|
|
t.Fatalf("upload duplicate chunk: %v", err)
|
|
}
|
|
if !duplicateAck.Duplicate {
|
|
t.Fatalf("expected duplicate chunk ack, got %+v", duplicateAck)
|
|
}
|
|
|
|
status, err := svc.QueryArtifactTransferStatus(domain.ArtifactTransferStatusQuery{RunEndpointID: "run-local", SessionToken: sessionToken, TransferID: opened.TransferID, ArtifactID: "artifact-1"})
|
|
if err != nil {
|
|
t.Fatalf("query artifact transfer status: %v", err)
|
|
}
|
|
if status.NextMissingChunkIndex != 1 || len(status.ReceivedChunkIndexes) != 1 {
|
|
t.Fatalf("unexpected status after first chunk: %+v", status)
|
|
}
|
|
|
|
for index := 1; index < opened.TotalChunks; index++ {
|
|
if _, err := svc.UploadArtifactChunk(validArtifactChunk(sessionToken, opened.TransferID, payload, index, 8)); err != nil {
|
|
t.Fatalf("upload chunk %d: %v", index, err)
|
|
}
|
|
}
|
|
completed, err := svc.CompleteArtifactTransfer(domain.ArtifactTransferComplete{RunEndpointID: "run-local", SessionToken: sessionToken, TransferID: opened.TransferID, ArtifactID: "artifact-1", Checksum: validator.BytesChecksum(payload), SizeBytes: int64(len(payload))})
|
|
if err != nil {
|
|
t.Fatalf("complete artifact transfer: %v", err)
|
|
}
|
|
if !completed.Accepted || !completed.Completed || completed.Artifact.State != domain.ArtifactStateAvailable {
|
|
t.Fatalf("unexpected complete response: %+v", completed)
|
|
}
|
|
artifact, err := svc.GetArtifact("artifact-1")
|
|
if err != nil {
|
|
t.Fatalf("get completed artifact: %v", err)
|
|
}
|
|
if artifact.State != domain.ArtifactStateAvailable || artifact.Checksum != validator.BytesChecksum(payload) {
|
|
t.Fatalf("expected available artifact, got %+v", artifact)
|
|
}
|
|
}
|
|
|
|
func TestCoreServiceRejectsInvalidArtifactTransferChunks(t *testing.T) {
|
|
svc, sessionToken := newRegisteredArtifactTransferService(t)
|
|
payload := []byte("artifact payload")
|
|
opened, err := svc.OpenArtifactTransfer(validArtifactTransferOpen(sessionToken, payload, 8))
|
|
if err != nil {
|
|
t.Fatalf("open artifact transfer: %v", err)
|
|
}
|
|
|
|
badChecksum := validArtifactChunk(sessionToken, opened.TransferID, payload, 0, 8)
|
|
badChecksum.Checksum = validator.BytesChecksum([]byte("different"))
|
|
_, err = svc.UploadArtifactChunk(badChecksum)
|
|
if err == nil || !strings.Contains(err.Error(), "checksum") {
|
|
t.Fatalf("expected checksum rejection, got %v", err)
|
|
}
|
|
|
|
firstChunk := validArtifactChunk(sessionToken, opened.TransferID, payload, 0, 8)
|
|
if _, err := svc.UploadArtifactChunk(firstChunk); err != nil {
|
|
t.Fatalf("upload first chunk: %v", err)
|
|
}
|
|
conflict := firstChunk
|
|
conflict.Payload = []byte("ARTIFACT")
|
|
conflict.Checksum = validator.BytesChecksum(conflict.Payload)
|
|
_, err = svc.UploadArtifactChunk(conflict)
|
|
if err == nil || !strings.Contains(err.Error(), "conflicts") {
|
|
t.Fatalf("expected conflicting chunk rejection, got %v", err)
|
|
}
|
|
|
|
_, err = svc.CompleteArtifactTransfer(domain.ArtifactTransferComplete{RunEndpointID: "run-local", SessionToken: sessionToken, TransferID: opened.TransferID, ArtifactID: "artifact-1", Checksum: validator.BytesChecksum(payload), SizeBytes: int64(len(payload))})
|
|
if err == nil || !strings.Contains(err.Error(), "missing chunks") {
|
|
t.Fatalf("expected missing chunk completion rejection, got %v", err)
|
|
}
|
|
}
|
|
|
|
func TestCoreServiceRejectsInvalidArtifactTransferOwnerAndSession(t *testing.T) {
|
|
svc, sessionToken := newRegisteredArtifactTransferService(t)
|
|
payload := []byte("artifact payload")
|
|
invalidSession := validArtifactTransferOpen("stale-token", payload, 8)
|
|
_, err := svc.OpenArtifactTransfer(invalidSession)
|
|
if err == nil || !strings.Contains(err.Error(), "sessionToken") {
|
|
t.Fatalf("expected invalid session rejection, got %v", err)
|
|
}
|
|
|
|
otherHello := validRunControlHello()
|
|
otherHello.RunEndpointID = "run-other"
|
|
otherHello.DisplayName = "Other Run"
|
|
otherHello.CapabilityReport.Capabilities = []string{"control.hello", "control.heartbeat", "process.start", "logs.read"}
|
|
otherHello.CapabilityReport.Fingerprint = "cap-other"
|
|
other, err := svc.RegisterRunHello(otherHello)
|
|
if err != nil {
|
|
t.Fatalf("register other run: %v", err)
|
|
}
|
|
invalidOwner := validArtifactTransferOpen(other.SessionToken, payload, 8)
|
|
invalidOwner.RunEndpointID = "run-other"
|
|
invalidOwner.ArtifactID = "artifact-other"
|
|
invalidOwner.IdempotencyKey = "artifact-upload-other"
|
|
_, err = svc.OpenArtifactTransfer(invalidOwner)
|
|
if err == nil || !strings.Contains(err.Error(), "owner job") {
|
|
t.Fatalf("expected owner mismatch rejection, got %v", err)
|
|
}
|
|
|
|
validOwner := validArtifactTransferOpen(sessionToken, payload, 8)
|
|
validOwner.OwnerKind = domain.ArtifactOwnerKindPlatform
|
|
_, err = svc.OpenArtifactTransfer(validOwner)
|
|
if err == nil || !strings.Contains(err.Error(), "ownerKind") {
|
|
t.Fatalf("expected owner kind rejection, got %v", err)
|
|
}
|
|
}
|
|
|
|
func newRegisteredArtifactTransferService(t *testing.T) (*CoreService, string) {
|
|
t.Helper()
|
|
svc := newTestCoreService()
|
|
plugin, endpoint := createPluginAndRunEndpoint(t, svc)
|
|
instance, err := svc.CreateServerInstance(domain.ServerInstance{
|
|
ID: "server-1",
|
|
PluginID: plugin.ID,
|
|
RunEndpointID: endpoint.ID,
|
|
Name: "SCUM #1",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create server instance: %v", err)
|
|
}
|
|
if _, err := svc.CreateJob(domain.Job{ID: "job-1", ServerInstanceID: instance.ID, RunEndpointID: endpoint.ID, Capability: "process.start", IdempotencyKey: "idem-start"}); err != nil {
|
|
t.Fatalf("create job: %v", err)
|
|
}
|
|
hello, err := svc.RegisterRunHello(validRunControlHello())
|
|
if err != nil {
|
|
t.Fatalf("register run hello: %v", err)
|
|
}
|
|
return svc, hello.SessionToken
|
|
}
|
|
|
|
func validArtifactTransferOpen(sessionToken string, payload []byte, chunkSize int) domain.ArtifactTransferOpen {
|
|
return domain.ArtifactTransferOpen{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: sessionToken,
|
|
ArtifactID: "artifact-1",
|
|
Direction: domain.ArtifactTransferDirectionUpload,
|
|
OwnerKind: domain.ArtifactOwnerKindJob,
|
|
OwnerID: "job-1",
|
|
SizeBytes: int64(len(payload)),
|
|
ChunkSizeBytes: chunkSize,
|
|
Checksum: validator.BytesChecksum(payload),
|
|
IdempotencyKey: "artifact-upload-1",
|
|
}
|
|
}
|
|
|
|
func validArtifactChunk(sessionToken string, transferID string, payload []byte, index int, chunkSize int) domain.ArtifactChunkUpload {
|
|
offset := index * chunkSize
|
|
end := offset + chunkSize
|
|
if end > len(payload) {
|
|
end = len(payload)
|
|
}
|
|
chunkPayload := payload[offset:end]
|
|
return domain.ArtifactChunkUpload{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: sessionToken,
|
|
TransferID: transferID,
|
|
ArtifactID: "artifact-1",
|
|
ChunkIndex: index,
|
|
Offset: int64(offset),
|
|
SizeBytes: len(chunkPayload),
|
|
Checksum: validator.BytesChecksum(chunkPayload),
|
|
Payload: chunkPayload,
|
|
}
|
|
}
|