diff --git a/data/qc/assayTransformCallback.R b/data/qc/assayTransformCallback.R index 7904609773..44937f1877 100644 --- a/data/qc/assayTransformCallback.R +++ b/data/qc/assayTransformCallback.R @@ -14,6 +14,8 @@ # limitations under the License. ## # Sleeps for the SleepSeconds run property, then queries exp.Data through the API before letting the import finish. +# A CallbackStatus of "report" or "fail" first records the script's working directory in the TransformWorkDirs list; "fail" then rejects the run. +# "edit" moves the Sync Callback Assay results of participant EditedByScript to visit 2, editing the run while its script runs. suppressMessages(library(Rlabkey)); labkey.setDefaults(apiKey = "${apikey}"); @@ -21,9 +23,25 @@ labkey.setDefaults(apiKey = "${apikey}"); run.props = labkey.transform.readRunPropertiesFile("${runInfo}"); sleep.seconds = as.numeric(labkey.transform.getRunPropertyValue(run.props, "SleepSeconds")); runprop.output.file = labkey.transform.getRunPropertyValue(run.props, "transformedRunPropertiesFile"); +errors.file = labkey.transform.getRunPropertyValue(run.props, "errorsFile"); +status = labkey.transform.getRunPropertyValue(run.props, "CallbackStatus"); + +if (status %in% c("report", "fail")) + labkey.insertRows(baseUrl = "${baseServerURL}", folderPath = "${containerPath}", schemaName = "lists", queryName = "TransformWorkDirs", toInsert = data.frame(WorkDir = dirname("${runInfo}"))); + +if (identical(status, "edit")) { + results = labkey.selectRows(baseUrl = "${baseServerURL}", folderPath = "${containerPath}", schemaName = "assay.General.Sync Callback Assay", queryName = "Data", colSelect = "RowId", colFilter = makeFilter(c("ParticipantID", "EQUAL", "EditedByScript")), colNameOpt = "fieldname"); + if (nrow(results) == 0) + stop("No Sync Callback Assay result for participant EditedByScript to edit"); + labkey.updateRows(baseUrl = "${baseServerURL}", folderPath = "${containerPath}", schemaName = "assay.General.Sync Callback Assay", queryName = "Data", toUpdate = data.frame(RowId = results$RowId, VisitID = 2)); +} Sys.sleep(sleep.seconds); rows = labkey.selectRows(baseUrl = "${baseServerURL}", folderPath = "${containerPath}", schemaName = "exp", queryName = "Data", colSelect = "RowId"); -writeLines(c(paste("CallbackStatus", "succeeded", sep = "\t"), paste("runComments", "set by transform", sep = "\t")), runprop.output.file); +if (identical(status, "fail")) { + writeLines(paste("error", "", "Transform script rejected the run", sep = "\t"), errors.file); +} else { + writeLines(c(paste("CallbackStatus", "succeeded", sep = "\t"), paste("runComments", "set by transform", sep = "\t")), runprop.output.file); +} diff --git a/src/org/labkey/test/tests/assay/AssayTransformCallbackTest.java b/src/org/labkey/test/tests/assay/AssayTransformCallbackTest.java index 9e80611b41..e48ba37c3d 100644 --- a/src/org/labkey/test/tests/assay/AssayTransformCallbackTest.java +++ b/src/org/labkey/test/tests/assay/AssayTransformCallbackTest.java @@ -15,16 +15,29 @@ */ package org.labkey.test.tests.assay; +import org.jetbrains.annotations.Nullable; +import org.json.JSONObject; import org.junit.Test; import org.junit.experimental.categories.Category; +import org.junit.function.ThrowingRunnable; import org.labkey.remoteapi.CommandException; +import org.labkey.remoteapi.assay.Batch; +import org.labkey.remoteapi.assay.Data; import org.labkey.remoteapi.assay.ImportRunCommand; import org.labkey.remoteapi.assay.ImportRunResponse; +import org.labkey.remoteapi.assay.Material; +import org.labkey.remoteapi.assay.Run; +import org.labkey.remoteapi.assay.SaveAssayBatchCommand; +import org.labkey.remoteapi.assay.SaveAssayBatchResponse; +import org.labkey.remoteapi.assay.SaveAssayRunsCommand; +import org.labkey.remoteapi.query.ContainerFilter; import org.labkey.remoteapi.query.Filter; import org.labkey.remoteapi.query.SelectRowsCommand; import org.labkey.remoteapi.query.SelectRowsResponse; +import org.labkey.remoteapi.query.Sort; import org.labkey.remoteapi.security.DeleteContainerCommand; import org.labkey.test.TestFileUtils; +import org.labkey.test.TestProperties; import org.labkey.test.WebTestHelper; import org.labkey.test.categories.Assays; import org.labkey.test.categories.Daily; @@ -33,24 +46,32 @@ import org.labkey.test.params.FieldDefinition; import org.labkey.test.params.assay.GeneralAssayDesign; import org.labkey.test.params.experiment.DataClassDefinition; +import org.labkey.test.params.experiment.SampleTypeDefinition; +import org.labkey.test.params.list.IntListDefinition; +import org.labkey.test.util.AuditLogHelper; import org.labkey.test.util.SiteSettingsHelper; +import org.labkey.test.util.core.webdav.WebDavUploadHelper; import java.io.File; import java.util.List; import java.util.Map; +import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; /** - * GH Issue 1707: an assay import must not hold a database transaction while its transform script runs, or a script - * that calls back into the server can deadlock with a concurrent folder delete. + * GH Issue 1707, GH Issue 1720: an assay import must not hold a database transaction while its transform script runs, + * or a script that calls back into the server can deadlock with a concurrent folder delete. Files saved before the + * script runs must still have their audit events recorded in the import's transaction. */ @Category({Assays.class, Daily.class}) public class AssayTransformCallbackTest extends AbstractAssayTransformTest @@ -61,7 +82,15 @@ public class AssayTransformCallbackTest extends AbstractAssayTransformTest private static final String SLEEP_SECONDS = "SleepSeconds"; private static final String CALLBACK_STATUS = "CallbackStatus"; private static final String TRANSFORM_COMMENT = "set by transform"; + private static final String WORK_DIRS_LIST = "TransformWorkDirs"; + private static final String WORK_DIR = "WorkDir"; + private static final String SAMPLE_TYPE = "Callback Samples"; + private static final String INPUT_SAMPLE = "S1"; + private static final String CYCLE_SAMPLE = "S2"; private static final String CALLBACK_DATA = "ParticipantID\tVisitID\nP1\t1\n"; + private static final String SCRIPT_REJECTED = "Transform script rejected the run"; + // The callback script's "edit" status updates the visit of this participant's results + private static final String EDITED_PARTICIPANT = "EditedByScript"; private static final String ASSAY_TRANSFORM_SCRIPT_TIMEOUT = "assayTransformScriptTimeout"; private static final int SCRIPT_SLEEP_SECONDS = 45; private static final int REQUEST_TIMEOUT_MS = (int) TimeUnit.MINUTES.toMillis(3); @@ -78,6 +107,21 @@ protected void doSetup() super.doSetup(); _syncAssayId = createCallbackAssay(SYNC_ASSAY, false); _backgroundAssayId = createCallbackAssay(BACKGROUND_ASSAY, true); + try + { + new IntListDefinition(WORK_DIRS_LIST, "Key") + .setFields(List.of(new FieldDefinition(WORK_DIR))) + .create(createDefaultConnection(), getProjectName()); + new SampleTypeDefinition(SAMPLE_TYPE) + .create(createDefaultConnection(), getProjectName()) + .addCustomRow(Map.of("Name", INPUT_SAMPLE)) + .addCustomRow(Map.of("Name", CYCLE_SAMPLE)) + .insertRows(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } } @Override @@ -90,30 +134,141 @@ protected void doCleanup(boolean afterTest) @Test public void testImportRunCallbackDuringFolderDelete() throws Exception { - String folderPath = createFolderWithDataClass("Sync Delete"); + assertFolderDeleteDuringCallback("Sync Delete", () -> importRun(_syncAssayId, SCRIPT_SLEEP_SECONDS).getRunId()); + } - ExecutorService executor = Executors.newSingleThreadExecutor(); - try - { - Future importFuture = executor.submit(() -> importRun(_syncAssayId, SCRIPT_SLEEP_SECONDS)); - sleep(5_000); - assertFalse("Import finished before the transform script's sleep ended", importFuture.isDone()); + @Test + public void testSaveBatchCallbackDuringFolderDelete() throws Exception + { + assertFolderDeleteDuringCallback("Save Batch Delete", () -> + saveBatch(batchOf(callbackRun(SCRIPT_SLEEP_SECONDS, "pending", "P1"))).getBatch().getRuns().getFirst().getId()); + } - deleteFolder(folderPath); - assertFalse("Folder delete should finish while the transform script is still running", importFuture.isDone()); + @Test + public void testSaveRunsCallbackDuringFolderDelete() throws Exception + { + assertFolderDeleteDuringCallback("Save Runs Delete", () -> { + SaveAssayRunsCommand command = new SaveAssayRunsByAssayIdCommand(_syncAssayId, List.of(callbackRun(SCRIPT_SLEEP_SECONDS, "pending", "P1"))); + command.setTimeout(REQUEST_TIMEOUT_MS); + return command.execute(createDefaultConnection(), getProjectName()).getRuns().getFirst().getId(); + }); + } - ImportRunResponse response = importFuture.get(REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); - SelectRowsResponse runs = selectRuns(SYNC_ASSAY, new Filter("RowId", response.getRunId())); - assertEquals("Runs with RowId " + response.getRunId(), 1, runs.getRowCount().intValue()); - assertEquals("Run property set by the transform script", "succeeded", runs.getRows().getFirst().get(CALLBACK_STATUS)); - assertEquals("Run comment set by the transform script", TRANSFORM_COMMENT, runs.getRows().getFirst().get("Comments")); - } - finally + @Test + public void testSaveBatchScriptFailureSavesNothing() throws Exception + { + int runsBefore = selectRuns(SYNC_ASSAY, null).getRowCount().intValue(); + int workDirsBefore = selectWorkDirs().size(); + + assertSaveRejected(batchOf(callbackRun(0, "report", "P1"), callbackRun(0, "fail", "P2")), SCRIPT_REJECTED); + + assertEquals("Runs after the second run's script failed", runsBefore, selectRuns(SYNC_ASSAY, null).getRowCount().intValue()); + List allWorkDirs = selectWorkDirs(); + List workDirs = allWorkDirs.subList(workDirsBefore, allWorkDirs.size()); + assertEquals("Transform script working directories reported", 2, workDirs.size()); + if (!TestProperties.isServerRemote()) { - executor.shutdownNow(); + for (String workDir : workDirs) + assertFalse("Transform script working directory should be deleted: " + workDir, new File(workDir).exists()); } } + @Test + public void testSaveBatchUpdatesExistingRun() throws Exception + { + Batch batch = saveBatch(batchOf(callbackRun(0, "pending", "P1"))).getBatch(); + int runId = batch.getRuns().getFirst().getId(); + + // A failing script elsewhere in the request must leave the existing run's results alone + assertSaveRejected(batchOf(batch.getId(), existingRun(runId, "pending", "P2"), callbackRun(0, "fail", "P3")), SCRIPT_REJECTED); + assertEquals("Results after a failed update", List.of("P1"), selectResultParticipants(runId)); + + int workDirsBefore = selectWorkDirs().size(); + saveBatch(batchOf(batch.getId(), existingRun(runId, "report", "P2"))); + assertEquals("Results after the update", List.of("P2"), selectResultParticipants(runId)); + assertEquals("Working directories reported by the update's transform script", workDirsBefore + 1, selectWorkDirs().size()); + SelectRowsResponse runs = selectRuns(SYNC_ASSAY, new Filter("RowId", runId)); + assertEquals("Run property set by the update's transform script", "succeeded", runs.getRows().getFirst().get(CALLBACK_STATUS)); + } + + @Test + public void testSaveBatchUpdateReusingStoredResults() throws Exception + { + Batch batch = saveBatch(batchOf(callbackRun(0, "pending", "P1"))).getBatch(); + int runId = batch.getRuns().getFirst().getId(); + int workDirsBefore = selectWorkDirs().size(); + + // The server compares the stored results its script saw with a read inside the transaction; unchanged ones must match + saveBatch(batchOf(batch.getId(), existingRunWithoutResults(runId, "report"))); + + assertEquals("Working directories reported by the update's transform script", workDirsBefore + 1, selectWorkDirs().size()); + assertEquals("Results after the update", List.of("P1"), selectResultParticipants(runId)); + } + + @Test + public void testSaveBatchRunChangedDuringScript() throws Exception + { + Batch batch = saveBatch(batchOf(callbackRun(0, "pending", EDITED_PARTICIPANT))).getBatch(); + int runId = batch.getRuns().getFirst().getId(); + + assertSaveRejected(batchOf(batch.getId(), existingRunWithoutResults(runId, "edit")), "changed while its transform scripts ran"); + + assertEquals("Visits after the script edited the run's results", List.of(2), selectResultVisits(runId)); + } + + @Test + public void testSaveDuplicateRunRejected() throws Exception + { + Batch batch = saveBatch(batchOf(callbackRun(0, "pending", "P1"))).getBatch(); + int runId = batch.getRuns().getFirst().getId(); + int workDirsBefore = selectWorkDirs().size(); + String duplicateError = "Run " + runId + " appears more than once"; + + assertSaveRejected(batchOf(batch.getId(), existingRun(runId, "report", "P2"), existingRun(runId, "report", "P3")), duplicateError); + + SaveAssayRunsCommand command = new SaveAssayRunsByAssayIdCommand(_syncAssayId, List.of(existingRun(runId, "report", "P2"), existingRun(runId, "report", "P3"))); + command.setTimeout(REQUEST_TIMEOUT_MS); + assertRejected(() -> command.execute(createDefaultConnection(), getProjectName()), duplicateError); + + assertEquals("Results after the rejected saves", List.of("P1"), selectResultParticipants(runId)); + assertEquals("Working directories reported by rejected saves", workDirsBefore, selectWorkDirs().size()); + } + + @Test + public void testSaveBatchMissingBatchRunsNoScripts() throws Exception + { + assertRejectedBeforeScripts(batchOf(Integer.MAX_VALUE, callbackRun(0, "report", "P1")), "Could not find assay batch"); + } + + @Test + public void testSaveBatchWithInputsAndOutputs() throws Exception + { + int workDirsBefore = selectWorkDirs().size(); + Run run = callbackRun(0, "report", "P1"); + run.setMaterialInputs(List.of(sample(INPUT_SAMPLE))); + run.setMaterialOutputs(List.of(sample("Derived " + INPUT_SAMPLE))); + run.setDataInputs(List.of(uploadedData("callbackInput.txt"))); + run.setDataOutputs(List.of(uploadedData("callbackOutput.txt"))); + + Run saved = saveBatch(batchOf(run)).getBatch().getRuns().getFirst(); + + assertEquals("Working directories reported by the transform script", workDirsBefore + 1, selectWorkDirs().size()); + assertEquals("Material inputs", List.of(INPUT_SAMPLE), saved.getMaterialInputs().stream().map(Material::getName).toList()); + assertEquals("Material outputs", List.of("Derived " + INPUT_SAMPLE), saved.getMaterialOutputs().stream().map(Material::getName).toList()); + assertEquals("Data inputs", List.of("callbackInput.txt"), saved.getDataInputs().stream().map(Data::getName).toList()); + assertEquals("Data outputs named callbackOutput.txt", 1, saved.getDataOutputs().stream().filter(data -> "callbackOutput.txt".equals(data.getName())).count()); + } + + @Test + public void testSaveBatchMaterialCycleRejectedBeforeScripts() throws Exception + { + Run run = callbackRun(0, "report", "P1"); + run.setMaterialInputs(List.of(sample(CYCLE_SAMPLE))); + run.setMaterialOutputs(List.of(sample(CYCLE_SAMPLE))); + + assertRejectedBeforeScripts(batchOf(run), "Circular input/output '" + CYCLE_SAMPLE + "'"); + } + @Test public void testBackgroundImportCallbackDuringFolderDelete() throws Exception { @@ -169,6 +324,58 @@ public void testBackgroundImportFromDataRows() throws Exception assertEquals("Run property set by the transform script", "succeeded", runs.getRows().getFirst().get(CALLBACK_STATUS)); } + @Test + public void testImportRunFileAuditInImportTransaction() throws Exception + { + File dataFile = writeDataFile("auditImportRun.tsv"); + importRun(_syncAssayId, dataFile, 0, "pending", null); + + Map audit = selectOnlyFileAudit(dataFile); + assertEquals("Transaction of the data file's audit event", new AuditLogHelper(this).getLastTransactionId(getProjectName()), audit.get("TransactionId")); + } + + @Test + public void testUploadWizardFileAuditInImportTransaction() throws Exception + { + File dataFile = writeDataFile("auditUploadWizard.tsv"); + beginAt(WebTestHelper.buildURL("assay", getProjectName(), "uploadWizard", Map.of("rowId", _syncAssayId))); + AssayImportPage importPage = new AssayImportPage(getDriver()); + importPage.setNamedInputText(SLEEP_SECONDS, "0"); + importPage.setNamedInputText(CALLBACK_STATUS, "pending"); + importPage.setDataFile(dataFile); + importPage.clickSaveAndFinish(); + + Map audit = selectOnlyFileAudit(dataFile); + assertEquals("Transaction of the data file's audit event", new AuditLogHelper(this).getLastTransactionId(getProjectName()), audit.get("TransactionId")); + } + + @Test + public void testFailedImportSkipsAuditOfDeletedFile() throws Exception + { + File dataFile = writeDataFile("auditFailedImport.tsv"); + assertImportRejected(dataFile, null); + + // The failed import deletes its data file, so an audit event would name a file that no longer exists + assertEquals("Audit events for the failed import's data file", List.of(), selectFileAudits(dataFile)); + } + + @Test + public void testFailedReRunAuditsKeptFile() throws Exception + { + int runId = importRun(_syncAssayId, writeDataFile("auditOriginalRun.tsv"), 0, "pending", null).getRunId(); + File reRunFile = writeDataFile("auditFailedReRun.tsv"); + assertImportRejected(reRunFile, runId); + + // A failed re-run keeps its data file, but the transaction its audit event was held for rolled back + Map audit = selectOnlyFileAudit(reRunFile); + assertNull("Transaction of the kept file's audit event", audit.get("TransactionId")); + if (!TestProperties.isServerRemote()) + { + File savedFile = new File((String) audit.get(AuditLogHelper.COL_FILE_AUDIT_DIRECTORY), (String) audit.get(AuditLogHelper.COL_FILE_AUDIT_FILE)); + assertTrue("Failed re-run should keep its data file: " + savedFile, savedFile.exists()); + } + } + @Test public void testTransformScriptTimeout() throws Exception { @@ -176,12 +383,7 @@ public void testTransformScriptTimeout() throws Exception try { - importRun(_syncAssayId, 90); - fail("Import should fail when the transform script exceeds the timeout"); - } - catch (CommandException e) - { - assertTrue("Unexpected error: " + e.getMessage(), e.getMessage().contains("Process killed after exceeding timeout")); + assertRejected(() -> importRun(_syncAssayId, 90), "Process killed after exceeding timeout"); } finally { @@ -190,6 +392,40 @@ public void testTransformScriptTimeout() throws Exception } } + private void assertFolderDeleteDuringCallback(String folderName, Callable save) throws Exception + { + String folderPath = createFolderWithDataClass(folderName); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try + { + Future saveFuture = executor.submit(save); + sleep(5_000); + assertStillRunning(saveFuture, "Save finished before the transform script's sleep ended"); + + deleteFolder(folderPath); + assertStillRunning(saveFuture, "Folder delete should finish while the transform script is still running"); + + int runId = saveFuture.get(REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); + SelectRowsResponse runs = selectRuns(SYNC_ASSAY, new Filter("RowId", runId)); + assertEquals("Runs with RowId " + runId, 1, runs.getRowCount().intValue()); + assertEquals("Run property set by the transform script", "succeeded", runs.getRows().getFirst().get(CALLBACK_STATUS)); + assertEquals("Run comment set by the transform script", TRANSFORM_COMMENT, runs.getRows().getFirst().get("Comments")); + } + finally + { + executor.shutdownNow(); + } + } + + /** A save that already failed rethrows its own error instead of failing on timing. */ + private static void assertStillRunning(Future future, String message) throws Exception + { + if (future.isDone()) + future.get(); + assertFalse(message, future.isDone()); + } + private int createCallbackAssay(String name, boolean backgroundUpload) { try @@ -197,7 +433,10 @@ private int createCallbackAssay(String name, boolean backgroundUpload) int protocolId = new GeneralAssayDesign(name) .setBatchFields(List.of(), false) .setRunFields(List.of(new FieldDefinition(SLEEP_SECONDS, FieldDefinition.ColumnType.Integer), new FieldDefinition(CALLBACK_STATUS)), true) - .addProtocolTransformer(protocol -> protocol.setBackgroundUpload(backgroundUpload)) + .addProtocolTransformer(protocol -> { + protocol.setBackgroundUpload(backgroundUpload); + protocol.setEditableResults(true); + }) .createAssay(getProjectName(), createDefaultConnection()) .getProtocolId(); ReactAssayDesignerPage.beginAt(this, getProjectName(), protocolId, "general", "") @@ -232,14 +471,170 @@ private void deleteFolder(String folderPath) throws Exception private ImportRunResponse importRun(int assayId, int sleepSeconds) throws Exception { - File dataFile = new File(TestFileUtils.ensureTestTempDir(getClass().getSimpleName()), "callbackData.tsv"); + return importRun(assayId, writeDataFile("callbackData.tsv"), sleepSeconds, "pending", null); + } + + private ImportRunResponse importRun(int assayId, File dataFile, int sleepSeconds, String callbackStatus, @Nullable Integer reRunId) throws Exception + { + ImportRunCommand command = new ImportRunCommand(assayId, dataFile) + { + @Override + protected Map createParameterMap() + { + Map params = super.createParameterMap(); + if (reRunId != null) + params.put("reRunId", reRunId); + return params; + } + }; + command.setProperties(Map.of(SLEEP_SECONDS, sleepSeconds, CALLBACK_STATUS, callbackStatus)); + command.setTimeout(REQUEST_TIMEOUT_MS); + return command.execute(createDefaultConnection(), getProjectName()); + } + + private void assertImportRejected(File dataFile, @Nullable Integer reRunId) + { + assertRejected(() -> importRun(_syncAssayId, dataFile, 0, "fail", reRunId), SCRIPT_REJECTED); + } + + private static void assertRejected(ThrowingRunnable request, String expectedError) + { + CommandException e = assertThrows("Request should fail with: " + expectedError, CommandException.class, request); + assertTrue("Unexpected error: " + e.getMessage(), e.getMessage().contains(expectedError)); + } + + private File writeDataFile(String name) throws Exception + { + File dataFile = new File(TestFileUtils.ensureTestTempDir(getClass().getSimpleName()), name); TestFileUtils.writeFile(dataFile, CALLBACK_DATA); - ImportRunCommand command = new ImportRunCommand(assayId, dataFile); - command.setProperties(Map.of(SLEEP_SECONDS, sleepSeconds, CALLBACK_STATUS, "pending")); + return dataFile; + } + + private List> selectFileAudits(File providedFile) throws Exception + { + return new AuditLogHelper(this).getAuditLogsFromLKS(getProjectName(), AuditLogHelper.AuditEvent.FILE_SYSTEM_EVENT, + List.of("TransactionId", AuditLogHelper.COL_FILE_AUDIT_FILE, AuditLogHelper.COL_FILE_AUDIT_DIRECTORY), + List.of(new Filter(AuditLogHelper.COL_FILE_AUDIT_PROVIDED_FILE, providedFile.getName())), null, ContainerFilter.CurrentAndSubfolders).getRows(); + } + + private Map selectOnlyFileAudit(File providedFile) throws Exception + { + List> audits = selectFileAudits(providedFile); + assertEquals("Audit events for " + providedFile.getName(), 1, audits.size()); + return audits.getFirst(); + } + + private static Run callbackRun(int sleepSeconds, String callbackStatus, String participantId) + { + Run run = new Run(); + run.setProperties(Map.of(SLEEP_SECONDS, sleepSeconds, CALLBACK_STATUS, callbackStatus)); + run.setResultData(List.of(Map.of("ParticipantID", participantId, "VisitID", 1))); + return run; + } + + private static Material sample(String name) + { + Material material = new Material(); + material.setName(name); + material.setSampleSetName(SAMPLE_TYPE); + return material; + } + + private Data uploadedData(String name) throws Exception + { + File file = new File(TestFileUtils.ensureTestTempDir(getClass().getSimpleName()), name); + TestFileUtils.writeFile(file, name); + new WebDavUploadHelper(getProjectName()).uploadFile(file); + Data data = new Data(); + data.setName(name); + data.setPipelinePath(name); + return data; + } + + private static Run existingRun(int runId, String callbackStatus, String participantId) + { + Run run = callbackRun(0, callbackStatus, participantId); + run.setId(runId); + return run; + } + + /** Run always posts dataInputs, so the server replaces the run's content using its stored results. */ + private static Run existingRunWithoutResults(int runId, String callbackStatus) + { + Run run = new Run(); + run.setId(runId); + run.setProperties(Map.of(SLEEP_SECONDS, 0, CALLBACK_STATUS, callbackStatus)); + return run; + } + + private static Batch batchOf(Run... runs) + { + return batchOf(null, runs); + } + + /** A null batchId saves a new batch. */ + private static Batch batchOf(@Nullable Integer batchId, Run... runs) + { + Batch batch = new Batch(); + if (batchId != null) + batch.setId(batchId); + batch.setRuns(List.of(runs)); + return batch; + } + + private SaveAssayBatchResponse saveBatch(Batch batch) throws Exception + { + SaveAssayBatchCommand command = new SaveAssayBatchCommand(_syncAssayId, batch); command.setTimeout(REQUEST_TIMEOUT_MS); return command.execute(createDefaultConnection(), getProjectName()); } + private void assertSaveRejected(Batch batch, String expectedError) + { + assertRejected(() -> saveBatch(batch), expectedError); + } + + private void assertRejectedBeforeScripts(Batch batch, String expectedError) throws Exception + { + int runsBefore = selectRuns(SYNC_ASSAY, null).getRowCount().intValue(); + int workDirsBefore = selectWorkDirs().size(); + + assertSaveRejected(batch, expectedError); + + assertEquals("Runs after the rejected save", runsBefore, selectRuns(SYNC_ASSAY, null).getRowCount().intValue()); + assertEquals("Working directories reported by the rejected save", workDirsBefore, selectWorkDirs().size()); + } + + private List selectResultParticipants(int runId) throws Exception + { + return selectResultColumn(runId, "ParticipantID", String.class::cast); + } + + private List selectResultVisits(int runId) throws Exception + { + return selectResultColumn(runId, "VisitID", value -> ((Number) value).intValue()); + } + + private List selectResultColumn(int runId, String column, Function mapper) throws Exception + { + SelectRowsCommand command = new SelectRowsCommand("assay.General." + SYNC_ASSAY, "Data"); + command.setColumns(List.of(column)); + command.addFilter(new Filter("Run/RowId", runId)); + return command.execute(createDefaultConnection(), getProjectName()).getRows().stream() + .map(row -> mapper.apply(row.get(column))) + .toList(); + } + + private List selectWorkDirs() throws Exception + { + SelectRowsCommand command = new SelectRowsCommand("lists", WORK_DIRS_LIST); + command.setColumns(List.of(WORK_DIR)); + command.addSort("Key", Sort.Direction.ASCENDING); + return command.execute(createDefaultConnection(), getProjectName()).getRows().stream() + .map(row -> (String) row.get(WORK_DIR)) + .toList(); + } + private Map importDataRowsFromBrowser(String folderPath, Boolean saveDataAsFile) { return executeAsyncScript(""" @@ -292,4 +687,24 @@ private static boolean isActiveAndStarted(String status) { return status != null && !status.endsWith("WAITING") && !List.of("COMPLETE", "ERROR", "CANCELLED").contains(status); } + + /** SaveAssayRunsCommand only identifies a protocol by name, which the server treats as a non-assay protocol */ + private static class SaveAssayRunsByAssayIdCommand extends SaveAssayRunsCommand + { + private final int _assayId; + + SaveAssayRunsByAssayIdCommand(int assayId, List runs) + { + super(null, runs); + _assayId = assayId; + } + + @Override + public JSONObject getJsonObject() + { + JSONObject json = super.getJsonObject(); + json.put("assayId", _assayId); + return json; + } + } }