diff --git a/docs/migrations/25-04.md b/docs/migrations/25-04.md index 205ed72e09..00eea9829b 100644 --- a/docs/migrations/25-04.md +++ b/docs/migrations/25-04.md @@ -37,6 +37,14 @@ The third preview of workflow outputs introduces the following breaking changes See {ref}`workflow-output-def` to learn more about the workflow output definition. +

Data lineage

+ +This release introduces built-in provenance tracking, also known as *data lineage*. When `lineage.enabled` is set to `true` in your configuration, Nextflow will record every workflow run, task execution, output file, and the links between them. + +You can explore this lineage from the command line using the {ref}`cli-lineage` command. Additionally, you can refer to files in the lineage store from a Nextflow script using the `lid://` path prefix as well as the {ref}`channel-from-lineage` channel factory. + +See the {ref}`cli-lineage` command and {ref}`config-lineage` config scope for details. + ## Enhancements

Improved inspect command

diff --git a/docs/reference/channel.md b/docs/reference/channel.md index 5d3b54c95f..fc824512d1 100644 --- a/docs/reference/channel.md +++ b/docs/reference/channel.md @@ -58,6 +58,37 @@ But when more than one argument is provided, they are always managed as *single* channel.from( [1, 2], [5,6], [7,9] ) ``` +(channel-from-lineage)= + +## fromLineage + +:::{versionadded} 25.04.0 +::: + +:::{warning} *Experimental: may change in a future release.* +::: + +The `channel.fromLineage` factory creates a channel that emits files from the {ref}`cli-lineage` store that match the given key-value params: + +```nextflow +channel + .fromLineage(workflowRun: 'lid://0d1d1622ced3e4edc690bec768919b45', labels: ['alpha', 'beta']) + .view() +``` + +The above snippet emits files published by the given workflow run that are labeled as `alpha` and `beta`. + +Available options: + +`labels` +: List of labels associated with the desired files. + +`taskRun` +: LID of the task run that produced the desired files. + +`workflowRun` +: LID of the workflow run that produced the desired files. + (channel-fromlist)= ## fromList diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 67cc43d104..88db1aebf4 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -703,7 +703,7 @@ $ nextflow lineage SUBCOMMAND [arg ..] **Description** -The `lineage` command is used to inspect lineage metadata. +The `lineage` command is used to inspect lineage metadata. Data lineage can be enabled by setting `lineage.enabled` to `true` in your Nextflow configuration (see the {ref}`config-lineage` config scope for details). **Options** @@ -720,31 +720,35 @@ TIMESTAMP RUN NAME SESSION ID 2025-04-22 14:45:43 backstabbing_heyrovsky 21bc4fad-e8b8-447d-9410-388f926a711f lid://c914d714877cc5c882c55a5428b510b1 ``` -View a metadata description. +View a lineage record. ```console $ nextflow lineage view ``` -View a metadata description fragment. A fragment can be a property of a metadata description (e.g., `output` or `params`) or a set of nested properties separated by a `.` (e.g., `workflow.repository`). +The output of a workflow run can be shown by appending `#output` to the workflow run LID: ```console -$ nextflow lineage view +$ nextflow lineage view lid://c914d714877cc5c882c55a5428b510b1#output ``` -Find a specific metadata description that matches a URL-like query string. The query string consists of `key=value` statements separated by `&`, where keys are defined similarly to the `fragments` used in the `view` command. +:::{tip} +You can use the [jq](https://jqlang.org/) command-line tool to apply further queries and transformations on the resulting lineage record. +::: + +Find all lineage records that match a set of key-value pairs: ```console -$ nextflow lineage find "" +$ nextflow lineage find = = ... ``` -Display a git-style diff between two metadata descriptions. +Display a git-style diff between two lineage records. ```console $ nextflow lineage diff ``` -Render the lineage graph for a workflow or task output in an HTML file. (default file path: `./lineage.html`). +Render the lineage graph for a workflow or task output as an HTML file. (default file path: `./lineage.html`). ```console $ nextflow lineage render [html-file-path] diff --git a/docs/reference/config.md b/docs/reference/config.md index 4dd1941524..bbac6272d3 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -1123,7 +1123,7 @@ See the {ref}`k8s-page` page for more details. ## `lineage` -The `lineage` scope controls the generation of lineage metadata. +The `lineage` scope controls the generation of {ref}`cli-lineage` metadata. The following settings are available: diff --git a/modules/nextflow/src/main/groovy/nextflow/Channel.groovy b/modules/nextflow/src/main/groovy/nextflow/Channel.groovy index ee1bb55f43..3d3ff69b58 100644 --- a/modules/nextflow/src/main/groovy/nextflow/Channel.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/Channel.groovy @@ -40,6 +40,7 @@ import nextflow.datasource.SraExplorer import nextflow.exception.AbortOperationException import nextflow.extension.CH import nextflow.extension.GroupTupleOp +import nextflow.extension.LinExtension import nextflow.extension.MapOp import nextflow.file.DirListener import nextflow.file.DirWatcher @@ -47,6 +48,7 @@ import nextflow.file.DirWatcherV2 import nextflow.file.FileHelper import nextflow.file.FilePatternSplitter import nextflow.file.PathVisitor +import nextflow.plugin.Plugins import nextflow.plugin.extension.PluginExtensionProvider import nextflow.util.Duration import nextflow.util.TestOnly @@ -657,4 +659,23 @@ class Channel { fromPath0Future = future.exceptionally(Channel.&handlerException) } + static DataflowWriteChannel fromLineage(Map params) { + checkParams('fromLineage', params, LinExtension.PARAMS) + final result = CH.create() + if( NF.isDsl2() ) { + session.addIgniter { fromLineage0(result, params) } + } + else { + fromLineage0(result, params ) + } + return result + } + + private static void fromLineage0(DataflowWriteChannel channel, Map params) { + final linExt = Plugins.getExtension(LinExtension) + if( !linExt ) + throw new IllegalStateException("Unable to load lineage extensions.") + final future = CompletableFuture.runAsync(() -> linExt.fromLineage(session, channel, params)) + future.exceptionally(this.&handlerException) + } } diff --git a/modules/nextflow/src/main/groovy/nextflow/extension/LinExtension.groovy b/modules/nextflow/src/main/groovy/nextflow/extension/LinExtension.groovy new file mode 100644 index 0000000000..d65010131c --- /dev/null +++ b/modules/nextflow/src/main/groovy/nextflow/extension/LinExtension.groovy @@ -0,0 +1,43 @@ +/* + * Copyright 2013-2025, Seqera Labs + * + * Licensed 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 nextflow.extension + +import groovyx.gpars.dataflow.DataflowWriteChannel +import nextflow.Session + +/** + * Interface for nf-lineage extensions. + * + * @author Jorge Ejarque params) +} diff --git a/modules/nextflow/src/test/groovy/nextflow/cli/CmdLineageTest.groovy b/modules/nextflow/src/test/groovy/nextflow/cli/CmdLineageTest.groovy index 4db9155246..da7787cb59 100644 --- a/modules/nextflow/src/test/groovy/nextflow/cli/CmdLineageTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/cli/CmdLineageTest.groovy @@ -179,7 +179,7 @@ class CmdLineageTest extends Specification { then: stdout.size() == 1 - stdout[0] == "Error loading lid://12345 - Lineage object 12345 not found" + stdout[0] == "Error loading lid://12345 - Lineage record 12345 not found" cleanup: folder?.deleteDir() @@ -280,45 +280,10 @@ class CmdLineageTest extends Specification { def entry = new FileOutput("path/to/file",new Checksum("45372qe","nextflow","standard"), "lid://123987/file.bam", "lid://12345", "lid://123987/", 1234, time, time, null) def jsonSer = encoder.encode(entry) - def expectedOutput = jsonSer - lidFile.text = jsonSer - when: - def lidCmd = new CmdLineage(launcher: launcher, args: ["view", "lid:///?type=FileOutput"]) - lidCmd.run() - def stdout = capture - .toString() - .readLines()// remove the log part - .findResults { line -> !line.contains('DEBUG') ? line : null } - .findResults { line -> !line.contains('INFO') ? line : null } - .findResults { line -> !line.contains('plugin') ? line : null } - - then: - stdout.size() == expectedOutput.readLines().size() - stdout.join('\n') == expectedOutput - - cleanup: - folder?.deleteDir() - } - - def 'should show query results'(){ - given: - def folder = Files.createTempDirectory('test').toAbsolutePath() - def configFile = folder.resolve('nextflow.config') - configFile.text = "lineage.enabled = true\nlineage.store.location = '$folder'".toString() - def lidFile = folder.resolve("12345/.data.json") - Files.createDirectories(lidFile.parent) - def launcher = Mock(Launcher){ - getOptions() >> new CliOptions(config: [configFile.toString()]) - } - def encoder = new LinEncoder().withPrettyPrint(true) - def time = OffsetDateTime.now() - def entry = new FileOutput("path/to/file",new Checksum("45372qe","nextflow","standard"), - "lid://123987/file.bam", "lid://12345", "lid://123987/", 1234, time, time, null) - def jsonSer = encoder.encode(entry) - def expectedOutput = jsonSer + def expectedOutput = '[\n "lid://12345"\n]' lidFile.text = jsonSer when: - def lidCmd = new CmdLineage(launcher: launcher, args: ["view", "lid:///?type=FileOutput"]) + def lidCmd = new CmdLineage(launcher: launcher, args: ["find", "type=FileOutput"]) lidCmd.run() def stdout = capture .toString() diff --git a/modules/nf-lang/src/main/java/nextflow/script/types/ChannelFactory.java b/modules/nf-lang/src/main/java/nextflow/script/types/ChannelFactory.java index 826d1fcb39..b9dd747de3 100644 --- a/modules/nf-lang/src/main/java/nextflow/script/types/ChannelFactory.java +++ b/modules/nf-lang/src/main/java/nextflow/script/types/ChannelFactory.java @@ -31,6 +31,8 @@ public interface ChannelFactory { Channel fromFilePairs(Map opts, String pattern, Closure grouping); + Channel fromLineage(Map opts); + Channel fromList(Collection values); Channel fromPath(Map opts, String pattern); diff --git a/modules/nf-lineage/src/main/nextflow/lineage/DefaultLinStore.groovy b/modules/nf-lineage/src/main/nextflow/lineage/DefaultLinStore.groovy index 3ebc1b4ff5..6b1f0dd6fe 100644 --- a/modules/nf-lineage/src/main/nextflow/lineage/DefaultLinStore.groovy +++ b/modules/nf-lineage/src/main/nextflow/lineage/DefaultLinStore.groovy @@ -93,15 +93,7 @@ class DefaultLinStore implements LinStore { void close() throws IOException { } @Override - Map search(String queryString) { - def params = null - if (queryString) { - params = LinUtils.parseQuery(queryString) - } - return searchAllFiles(params) - } - - private Map searchAllFiles(Map> params) { + Map search(Map> params) { final results = new HashMap() Files.walkFileTree(location, new FileVisitor() { diff --git a/modules/nf-lineage/src/main/nextflow/lineage/LinExtensionImpl.groovy b/modules/nf-lineage/src/main/nextflow/lineage/LinExtensionImpl.groovy new file mode 100644 index 0000000000..ac92fe817c --- /dev/null +++ b/modules/nf-lineage/src/main/nextflow/lineage/LinExtensionImpl.groovy @@ -0,0 +1,75 @@ +/* + * Copyright 2013-2025, Seqera Labs + * + * Licensed 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 nextflow.lineage + +import groovy.transform.CompileStatic +import groovy.util.logging.Slf4j +import groovyx.gpars.dataflow.DataflowWriteChannel +import nextflow.Channel +import nextflow.Session +import nextflow.extension.LinExtension +import nextflow.lineage.fs.LinPathFactory +import nextflow.lineage.model.FileOutput +import nextflow.lineage.serde.LinSerializable + +import static nextflow.lineage.fs.LinPath.* + +/** + * Lineage channel extensions + * + * @author Jorge Ejarque + */ +@CompileStatic +@Slf4j +class LinExtensionImpl implements LinExtension { + + @Override + void fromLineage(Session session, DataflowWriteChannel channel, Map opts) { + final queryParams = buildQueryParams(opts) + log.trace("Querying lineage with params: $queryParams") + new LinPropertyValidator().validateQueryParams(queryParams.keySet()) + final store = getStore(session) + emitSearchResults(channel, store.search(queryParams)) + channel.bind(Channel.STOP) + } + + private static Map> buildQueryParams(Map opts) { + final queryParams = [type: [FileOutput.class.simpleName] ] + if( opts.workflowRun ) + queryParams['workflowRun'] = [opts.workflowRun as String] + if( opts.taskRun ) + queryParams['taskRun'] = [opts.taskRun as String] + if( opts.labels ) + queryParams['labels'] = opts.labels as List + return queryParams + } + + protected LinStore getStore(Session session) { + final store = LinStoreFactory.getOrCreate(session) + if( !store ) { + throw new Exception("Lineage store not found - Check Nextflow configuration") + } + return store + } + + private void emitSearchResults(DataflowWriteChannel channel, Map results) { + if( !results ) { + return + } + results.keySet().forEach { channel.bind( LinPathFactory.create( asUriString(it) ) ) } + } +} diff --git a/modules/nf-lineage/src/main/nextflow/lineage/LinStore.groovy b/modules/nf-lineage/src/main/nextflow/lineage/LinStore.groovy index 3f826b7a0a..2dc6974f6e 100644 --- a/modules/nf-lineage/src/main/nextflow/lineage/LinStore.groovy +++ b/modules/nf-lineage/src/main/nextflow/lineage/LinStore.groovy @@ -55,9 +55,9 @@ interface LinStore extends Closeable { /** * Search for lineage entries. - * @queryString Json-path like query string. (Only simple and nested field operators are supported(No array, wildcards,etc.) - * @return Key-lineage entry pairs fulfilling the queryString + * @param params Map of query params + * @return Key-lineage entry pairs fulfilling the query params */ - Map search(String queryString) + Map search(Map> params) } diff --git a/modules/nf-lineage/src/main/nextflow/lineage/LinUtils.groovy b/modules/nf-lineage/src/main/nextflow/lineage/LinUtils.groovy index 8f3f3c34f8..d1129745b5 100644 --- a/modules/nf-lineage/src/main/nextflow/lineage/LinUtils.groovy +++ b/modules/nf-lineage/src/main/nextflow/lineage/LinUtils.groovy @@ -16,13 +16,15 @@ package nextflow.lineage +import static nextflow.lineage.fs.LinFileSystemProvider.* +import static nextflow.lineage.fs.LinPath.* + import java.nio.file.attribute.FileTime import java.time.OffsetDateTime import java.time.ZoneId import groovy.transform.CompileStatic import groovy.util.logging.Slf4j -import nextflow.lineage.fs.LinPath import nextflow.lineage.model.TaskRun import nextflow.lineage.model.WorkflowRun import nextflow.lineage.serde.LinEncoder @@ -40,45 +42,37 @@ class LinUtils { private static final String[] EMPTY_ARRAY = new String[] {} /** - * Query a lineage store. + * Get a lineage record or fragment from the Lineage store. * - * @param store lineage store to query. - * @param uri Query to perform in a URI-like format. - * Format 'lid://[?QueryString][#fragment]' where: - * - Key: Element where the query will be applied. '/' indicates query will be applied in all the elements of the lineage store. - * - QueryString: all param-value pairs that the lineage element should fulfill in a URI's query string format. + * @param store Lineage store. + * @param uri Object or fragment to retrieve in URI-like format. + * Format 'lid://[#fragment]' where: + * - Key: Metadata Element key * - Fragment: Element fragment to retrieve. - * @return Collection of object fulfilling the query + * @return Lineage record or fragment. */ - static Collection query(LinStore store, URI uri) { - String key = uri.authority ? uri.authority + uri.path : uri.path - if (key == LinPath.SEPARATOR) { - return globalSearch(store, uri) - } else { - final parameters = uri.query ? parseQuery(uri.query) : null - final children = parseChildrenFromFragment(uri.fragment) - return searchPath(store, key, parameters, children ) - } - } + static Object getMetadataObject(LinStore store, URI uri) { + if( uri.scheme != SCHEME ) + throw new IllegalArgumentException("Invalid LID URI - scheme is different for $SCHEME") + final key = uri.authority ? uri.authority + uri.path : uri.path + if( key == SEPARATOR ) + throw new IllegalArgumentException("Cannot get record from the root LID URI") + if ( uri.query ) + log.warn("Query string is not supported for Lineage URI: `$uri` -- it will be ignored") - private static Collection globalSearch(LinStore store, URI uri) { - final results = store.search(uri.query).values() - if (results && uri.fragment) { - // If fragment is defined get the property of the object indicated by the fragment - return filterResults(results, uri.fragment) - } - return results + final children = parseChildrenFromFragment(uri.fragment) + return getMetadataObject0(store, key, children ) } - private static List filterResults(Collection results, String fragment) { - final filteredResults = [] - results.forEach { - final output = navigate(it, fragment) - if (output) { - filteredResults.add(output) - } + private static Object getMetadataObject0(LinStore store, String key, String[] children = []) { + final record = store.load(key) + if (!record) { + throw new FileNotFoundException("Lineage record $key not found") + } + if (children && children.size() > 0) { + return getSubObject(store, key, record, children) } - return filteredResults + return record } /** @@ -96,127 +90,67 @@ class LinUtils { } /** - * Search for objects inside a description + * Get a lineage sub-record. * - * @param store lineage store - * @param key lineage key where to perform the search - * @param params Parameter-value pairs to be evaluated in the key - * @param children Sub-objects to evaluate and retrieve - * @return List of object - */ - protected static List searchPath(LinStore store, String key, Map> params, String[] children = []) { - final object = store.load(key) - if (!object) { - throw new FileNotFoundException("Lineage object $key not found") - } - final results = new LinkedList() - if (children && children.size() > 0) { - treatSubObject(store, key, object, children, params, results) - } else { - treatObject(object, params, results) - } - - return results - } - - private static void treatSubObject(LinStore store, String key, LinSerializable object, String[] children, Map> params, LinkedList results) { - final output = getSubObject(store, key, object, children) - if (!output) { - throw new FileNotFoundException("Lineage object $key#${children.join('.')} not found") - } - treatObject(output, params, results) - } - - /** - * Get a metadata sub-object. + * If the requested sub-record is the workflow or task outputs, retrieves the outputs from the outputs description. * - * If the requested sub-object is the workflow or task outputs, retrieves the outputs from the outputs description. - * - * @param store Store to retrieve lineage metadata objects. - * @param key Parent metadata key. - * @param object Parent object. - * @param children Array of string in indicating the properties to navigate to get the sub-object. - * @return Sub-object or null in it does not exist. + * @param store Store to retrieve lineage records. + * @param key Parent key. + * @param record Parent record. + * @param children Array of string in indicating the properties to navigate to get the sub-record. + * @return Sub-record or null in it does not exist. */ - static Object getSubObject(LinStore store, String key, LinSerializable object, String[] children) { - if( isSearchingOutputs(object, children) ) { + static Object getSubObject(LinStore store, String key, LinSerializable record, String[] children) { + if( isSearchingOutputs(record, children) ) { // When asking for a Workflow or task output retrieve the outputs description final outputs = store.load("${key}#output") if (!outputs) return null return navigate(outputs, children.join('.')) } - return navigate(object, children.join('.')) + return navigate(record, children.join('.')) } /** * Check if the Lid pseudo path or query is for Task or Workflow outputs. * - * @param object Parent Lid metadata object - * @param children Array of string in indicating the properties to navigate to get the sub-object. + * @param record Parent lineage record + * @param children Array of string in indicating the properties to navigate to get the sub-record. * @return return 'true' if the parent is a Task/Workflow run and the first element in children is 'outputs'. Otherwise 'false' */ - static boolean isSearchingOutputs(LinSerializable object, String[] children) { - return (object instanceof WorkflowRun || object instanceof TaskRun) && children && children[0] == 'output' + static boolean isSearchingOutputs(LinSerializable record, String[] children) { + return (record instanceof WorkflowRun || record instanceof TaskRun) && children && children[0] == 'output' } /** - * Evaluates object or the objects in a collection matches a set of parameter-value pairs. It includes in the results collection in case of match. + * Evaluates record or the records in a collection matches a set of parameter-value pairs. It includes in the results collection in case of match. * - * @param object Object or collection of objects to evaluate - * @param params parameter-value pairs to evaluate in each object - * @param results results collection to include the matching objects + * @param record Object or collection of records to evaluate + * @param params parameter-value pairs to evaluate in each record + * @param results results collection to include the matching records */ - protected static void treatObject(def object, Map> params, List results) { + protected static void treatObject(def record, Map> params, List results) { if (params) { - if (object instanceof Collection) { - (object as Collection).forEach { treatObject(it, params, results) } - } else if (checkParams(object, params)) { - results.add(object) + if (record instanceof Collection) { + (record as Collection).forEach { treatObject(it, params, results) } + } else if (checkParams(record, params)) { + results.add(record) } } else { - results.add(object) + results.add(record) } } /** - * Parses a query string and store them in parameter-value Map. - * If the queryString contains repeated params returned value is a List containing all values. - * It also allows values including "=". First '=' will be used as separators and other will be included as value. - * - * @param queryString URI-like query string. (e.g. param1=value1¶m2=value2). - * @return Map containing the parameter-value pairs of the query string - */ - static Map> parseQuery(String queryString) { - if( !queryString ) { - return [:] - } - Map> params = [:].withDefault { [] } - - queryString.split('&').each { pair -> - def idx = pair.indexOf('=') - if( idx < 0 ) - throw new IllegalArgumentException("Parameter $pair doesn't contain '=' separator") - final key = URLDecoder.decode(pair[0..> params) { + static boolean checkParams(Object record, Map> params) { for( final entry : params.entrySet() ) { - final value = navigate(object, entry.key) + final value = navigate(record, entry.key) if( !checkParam(value, entry.value) ) { return false } @@ -234,7 +168,7 @@ class LinUtils { return colValue.collect { it.toString() }.containsAll(expected) } - //Single object can't be compared with collection with one of more elements + //Single record can't be compared with collection with one of more elements if( expected.size() > 1 ) { return false } @@ -243,16 +177,16 @@ class LinUtils { } /** - * Retrieves the sub-object or value indicated by a path. + * Retrieves the sub-record or value indicated by a path. * * @param obj Object to navigate * @param path Elements path separated by '.' e.g. field.subfield - * @return sub-object / value + * @return sub-record / value */ static Object navigate(Object obj, String path) { if (!obj) return null - // type has been replaced by class when evaluating LidSerializable objects + // type has been replaced by class when evaluating LidSerializable records if (obj instanceof LinSerializable && path == 'type') return obj.getClass()?.simpleName try { @@ -261,7 +195,7 @@ class LinUtils { } } catch (Throwable e) { - log.debug("Error navigating to $path in object", e) + log.debug("Error navigating to $path in record", e) return null } } @@ -285,8 +219,8 @@ class LinUtils { private static Object navigateCollection(Collection collection, String key) { final results = [] - for (Object object : collection) { - final res = getSubPath(object, key) + for (Object record : collection) { + final res = getSubPath(record, key) if (res) results.add(res) } @@ -294,7 +228,7 @@ class LinUtils { log.trace("No property found for $key") return null } - // Return a single object if only ine results is found. + // Return a single record if only ine results is found. return results.size() == 1 ? results[0] : results } @@ -325,14 +259,14 @@ class LinUtils { /** * Helper function to unify the encoding of outputs when querying and navigating the lineage pseudoFS. - * Outputs can include LinSerializable objects, collections or parts of these objects. - * LinSerializable objects can be encoded with the LinEncoder, but collections or parts of - * these objects require to extend the GsonEncoder. + * Outputs can include LinSerializable records, collections or parts of these records. + * LinSerializable records can be encoded with the LinEncoder, but collections or parts of + * these records require to extend the GsonEncoder. * * @param output Output to encode * @return Output encoded as a JSON string */ - static String encodeSearchOutputs(Object output, boolean prettyPrint) { + static String encodeSearchOutputs(Object output, boolean prettyPrint = false) { if (output instanceof LinSerializable) { return new LinEncoder().withPrettyPrint(prettyPrint).encode(output) } else { diff --git a/modules/nf-lineage/src/main/nextflow/lineage/cli/LinCommandImpl.groovy b/modules/nf-lineage/src/main/nextflow/lineage/cli/LinCommandImpl.groovy index 43290e46ff..c059e1566c 100644 --- a/modules/nf-lineage/src/main/nextflow/lineage/cli/LinCommandImpl.groovy +++ b/modules/nf-lineage/src/main/nextflow/lineage/cli/LinCommandImpl.groovy @@ -28,6 +28,7 @@ import nextflow.cli.CmdLineage import nextflow.config.ConfigMap import nextflow.dag.MermaidHtmlRenderer import nextflow.lineage.LinHistoryRecord +import nextflow.lineage.LinPropertyValidator import nextflow.lineage.LinStore import nextflow.lineage.LinStoreFactory import nextflow.lineage.LinUtils @@ -98,13 +99,12 @@ class LinCommandImpl implements CmdLineage.LinCommand { return } try { - def entries = LinUtils.query(store, new URI(args[0])) - if( !entries ) { - println "No entries found for ${args[0]}" + def entry = LinUtils.getMetadataObject(store, new URI(args[0])) + if( !entry ) { + println "No entry found for ${args[0]}" return } - entries = entries.size() == 1 ? entries[0] : entries - println LinUtils.encodeSearchOutputs(entries, true) + println LinUtils.encodeSearchOutputs(entry, true) } catch (Throwable e) { println "Error loading ${args[0]} - ${e.message}" } @@ -319,9 +319,25 @@ class LinCommandImpl implements CmdLineage.LinCommand { return } try { - println LinUtils.encodeSearchOutputs(store.search(args[0]).keySet().collect {asUriString(it)}, true) + final params = parseFindArgs(args) + new LinPropertyValidator().validateQueryParams(params.keySet()) + println LinUtils.encodeSearchOutputs( store.search(params).keySet().collect { asUriString(it) }, true ) } catch (Throwable e){ println "Error searching for ${args[0]}. ${e.message}" } } + + private Map> parseFindArgs(List args){ + Map> params = [:].withDefault { [] } + + args.collectEntries { pair -> + final idx = pair.indexOf('=') + if( idx < 0 ) + throw new IllegalArgumentException("Parameter $pair doesn't contain '=' separator") + final key = URLDecoder.decode(pair[0.. SUPPORTED_CHECKSUM_ALGORITHMS=["nextflow"] + static public final List SUPPORTED_CHECKSUM_ALGORITHMS = ["nextflow"] static public final String SEPARATOR = '/' public static final String LID_PROT = "${SCHEME}://" - static private final String[] EMPTY = new String[] {} + static private final String[] EMPTY = new String[]{} private LinFileSystem fileSystem @@ -72,12 +73,25 @@ class LinPath implements Path, LogicalDataPath { throw new IllegalArgumentException("Invalid LID URI - scheme is different for $SCHEME") } this.fileSystem = fs + setFieldsFormURI(uri) + // Check if query and fragment are with filePath + if( query == null && fragment == null ) + setFieldsFormURI(new URI(toUriString())) + // Warn if query is specified + if( query ) + log.warn("Query string is not supported for Lineage URI: `$uri` -- it will be ignored") + // Validate fragment + if( fragment ) + new LinPropertyValidator().validate(fragment.tokenize('.')) + } + + private void setFieldsFormURI(URI uri){ this.query = uri.query this.fragment = uri.fragment - this.filePath = resolve0( fs, norm0("${uri.authority?:''}${uri.path}") ) + this.filePath = resolve0(fileSystem, norm0("${uri.authority?:''}${uri.path}") ) } - protected LinPath(String query, String fragment, String filepath, LinFileSystem fs){ + protected LinPath(String query, String fragment, String filepath, LinFileSystem fs) { this.fileSystem = fs this.query = query this.fragment = fragment @@ -100,9 +114,9 @@ class LinPath implements Path, LogicalDataPath { return path && path.startsWith(LID_PROT) } - private static String buildPath(String first, String[] more){ + private static String buildPath(String first, String[] more) { first = norm0(first) - if (more){ + if( more ) { final morePath = norm0(more).join(SEPARATOR) return first.isEmpty() ? morePath : first + SEPARATOR + morePath } @@ -117,25 +131,25 @@ class LinPath implements Path, LogicalDataPath { } protected static void validateChecksum(Checksum checksum, Path hashedPath) { - if( !checksum) + if( !checksum ) return - if( ! isAlgorithmSupported(checksum.algorithm) ) { + if( !isAlgorithmSupported(checksum.algorithm) ) { log.warn("Checksum of '$hashedPath' can't be validated. Algorithm '${checksum.algorithm}' is not supported") return } final hash = checksum.mode ? CacheHelper.hasher(hashedPath, CacheHelper.HashMode.of(checksum.mode.toString().toLowerCase())).hash().toString() : CacheHelper.hasher(hashedPath).hash().toString() - if (hash != checksum.value) + if( hash != checksum.value ) log.warn("Checksum of '$hashedPath' does not match with the one stored in the metadata") } - protected static isAlgorithmSupported( String algorithm ){ + protected static isAlgorithmSupported(String algorithm) { return algorithm && algorithm in SUPPORTED_CHECKSUM_ALGORITHMS } @TestOnly - protected String getFilePath(){ this.filePath } + protected String getFilePath() { this.filePath } /** * Finds the target path of a LinPath. @@ -149,20 +163,20 @@ class LinPath implements Path, LogicalDataPath { * IllegalArgumentException if the filepath, filesystem or its LinStore are null. * FileNotFoundException if the filePath or children are not found in the LinStore. */ - protected static Path findTarget(LinFileSystem fs, String filePath, boolean resultsAsPath, String[] children=[]) throws Exception { + protected static Path findTarget(LinFileSystem fs, String filePath, boolean resultsAsPath, String[] children = []) throws Exception { if( !fs ) throw new IllegalArgumentException("Cannot get target path for a relative lineage path") if( filePath.isEmpty() || filePath == SEPARATOR ) - throw new IllegalArgumentException("Cannot get target path for an empty lineage path") + throw new IllegalArgumentException("Cannot get target path for an empty lineage path (lid:///)") final store = fs.getStore() if( !store ) throw new Exception("Lineage store not found - Check Nextflow configuration") final object = store.load(filePath) - if ( object ){ + if( object ) { if( object instanceof FileOutput ) { return getTargetPathFromOutput(object, children) } - if( resultsAsPath ){ + if( resultsAsPath ) { return getMetadataAsTargetPath(object, fs, filePath, children) } } else { @@ -180,11 +194,11 @@ class LinPath implements Path, LogicalDataPath { throw new FileNotFoundException("Target path '$filePath' does not exist") } - protected static Path getMetadataAsTargetPath(LinSerializable results, LinFileSystem fs, String filePath, String[] children){ + protected static Path getMetadataAsTargetPath(LinSerializable results, LinFileSystem fs, String filePath, String[] children) { if( !results ) { throw new FileNotFoundException("Target path '$filePath' does not exist") } - if (children && children.size() > 0) { + if( children && children.size() > 0 ) { return getSubObjectAsPath(fs, filePath, results, children) } else { return generateLinMetadataPath(fs, filePath, results, children) @@ -209,13 +223,12 @@ class LinPath implements Path, LogicalDataPath { throw new FileNotFoundException("Target path '$key#output' does not exist") } return generateLinMetadataPath(fs, key, outputs, children) - } - else { + } else { return generateLinMetadataPath(fs, key, object, children) } } - private static LinMetadataPath generateLinMetadataPath(LinFileSystem fs, String key, Object object, String[] children){ + private static LinMetadataPath generateLinMetadataPath(LinFileSystem fs, String key, Object object, String[] children) { def creationTime = toFileTime(navigate(object, 'createdAt') as OffsetDateTime ?: OffsetDateTime.now()) final output = children ? navigate(object, children.join('.')) : object if( !output ) { @@ -229,19 +242,19 @@ class LinPath implements Path, LogicalDataPath { // return the real path stored in the metadata validateDataOutput(lidObject) def realPath = FileHelper.toCanonicalPath(lidObject.path as String) - if (children && children.size() > 0) + if( children && children.size() > 0 ) realPath = realPath.resolve(children.join(SEPARATOR)) - if (!realPath.exists()) + if( !realPath.exists() ) throw new FileNotFoundException("Target path '$realPath' does not exist") return realPath } - private static boolean isEmptyBase(LinFileSystem fs, String base){ + private static boolean isEmptyBase(LinFileSystem fs, String base) { return !base || base == SEPARATOR || (fs && base == "..") } private static String resolve0(LinFileSystem fs, String base, String[] more) { - if( isEmptyBase(fs,base) ) { + if( isEmptyBase(fs, base) ) { return resolveEmptyPathCase(fs, more as List) } if( base.contains(SEPARATOR) ) { @@ -253,8 +266,8 @@ class LinPath implements Path, LogicalDataPath { return more ? result.resolve(more.join(SEPARATOR)).toString() : result.toString() } - private static String resolveEmptyPathCase(LinFileSystem fs, List more ){ - switch(more.size()) { + private static String resolveEmptyPathCase(LinFileSystem fs, List more) { + switch( more.size() ) { case 0: return "/" case 1: @@ -265,7 +278,7 @@ class LinPath implements Path, LogicalDataPath { } static private String norm0(String path) { - if( !path || path==SEPARATOR) + if( !path || path == SEPARATOR ) return "" //Remove repeated elements path = Path.of(path.trim()).normalize().toString() @@ -273,12 +286,12 @@ class LinPath implements Path, LogicalDataPath { if( path.startsWith(SEPARATOR) ) path = path.substring(1) if( path.endsWith(SEPARATOR) ) - path = path.substring(0,path.size()-1) + path = path.substring(0, path.size() - 1) return path } - + static private String[] norm0(String... path) { - for( int i=0; i1 ) - return subpath(0,c-1) - if( c==1 ) - return new LinPath(fileSystem,SEPARATOR) + if( c > 1 ) + return subpath(0, c - 1) + if( c == 1 ) + return new LinPath(fileSystem, SEPARATOR) return null } @@ -322,21 +335,21 @@ class LinPath implements Path, LogicalDataPath { @Override Path getName(int index) { - if( index<0 ) + if( index < 0 ) throw new IllegalArgumentException("Path name index cannot be less than zero - offending value: $index") final path = Path.of(filePath) - if (index == path.nameCount - 1){ - return new LinPath( fragment, query, path.getName(index).toString(), null) + if( index == path.nameCount - 1 ) { + return new LinPath( query, fragment, path.getName(index).toString(), null) } - return new LinPath(index==0 ? fileSystem : null, path.getName(index).toString()) + return new LinPath(index == 0 ? fileSystem : null, path.getName(index).toString()) } @Override Path subpath(int beginIndex, int endIndex) { - if( beginIndex<0 ) + if( beginIndex < 0 ) throw new IllegalArgumentException("subpath begin index cannot be less than zero - offending value: $beginIndex") final path = Path.of(filePath) - return new LinPath(beginIndex==0 ? fileSystem : null, path.subpath(beginIndex, endIndex).toString()) + return new LinPath(beginIndex == 0 ? fileSystem : null, path.subpath(beginIndex, endIndex).toString()) } @Override @@ -369,7 +382,7 @@ class LinPath implements Path, LogicalDataPath { if( LinPath.class != other.class ) throw new ProviderMismatchException() - final that = (LinPath)other + final that = (LinPath) other if( that.fileSystem && this.fileSystem != that.fileSystem ) return other @@ -388,7 +401,7 @@ class LinPath implements Path, LogicalDataPath { final scheme = FileHelper.getUrlProtocol(path) if( !scheme ) { // consider the path as a lid relative path - return resolve(new LinPath(null,path)) + return resolve(new LinPath(null, path)) } if( scheme != SCHEME ) { throw new ProviderMismatchException() @@ -413,12 +426,12 @@ class LinPath implements Path, LogicalDataPath { // Compare 'filePath' as relative paths path = Path.of(filePath).relativize(Path.of(lidOther.filePath)) } - return new LinPath(lidOther.query, lidOther.fragment, path.getNameCount()>0 ? path.toString() : SEPARATOR, null) + return new LinPath(lidOther.query, lidOther.fragment, path.getNameCount() > 0 ? path.toString() : SEPARATOR, null) } @Override URI toUri() { - return asUri("${SCHEME}://${filePath}${query ? '?' + query: ''}${fragment ? '#'+ fragment : ''}") + return asUri("${SCHEME}://${filePath}${query ? '?' + query : ''}${fragment ? '#' + fragment : ''}") } String toUriString() { @@ -440,22 +453,22 @@ class LinPath implements Path, LogicalDataPath { } /** - * Get the path associated to a DataOutput metadata. + * Get the path associated with a FileOutput record. * - * @return Path associated to a DataOutput - * @throws FileNotFoundException if the metadata associated to the LinPath does not exist or its type is not a DataOutput. + * @return Path associated with a FileOutput record + * @throws FileNotFoundException if the record does not exist or its type is not a FileOutput. */ protected Path getTargetPath() { return findTarget(fileSystem, filePath, false, parseChildrenFromFragment(fragment)) } /** - * Get the path associated to any metadata object. + * Get the path associated with a lineage record. * - * @return Path associated to a DataOutput or LinMetadataFile with the metadata object for other types. - * @throws FileNotFoundException if the metadata associated to the LinPath does not exist + * @return Path associated with a FileOutput record, or LinMetadataFile with the lineage record for other types. + * @throws FileNotFoundException if the record does not exist */ - protected Path getTargetOrMetadataPath(){ + protected Path getTargetOrMetadataPath() { return findTarget(fileSystem, filePath, true, parseChildrenFromFragment(fragment)) } @@ -479,7 +492,7 @@ class LinPath implements Path, LogicalDataPath { if( LinPath.class != other.class ) { return false } - final that = (LinPath)other + final that = (LinPath) other return this.fileSystem == that.fileSystem && this.filePath.equals(that.filePath) } @@ -488,24 +501,24 @@ class LinPath implements Path, LogicalDataPath { */ @Override int hashCode() { - return Objects.hash(fileSystem,filePath) + return Objects.hash(fileSystem, filePath) } static URI asUri(String path) { - if (!path) + if( !path ) throw new IllegalArgumentException("Missing 'path' argument") - if (!path.startsWith(LID_PROT)) + if( !path.startsWith(LID_PROT) ) throw new IllegalArgumentException("Invalid LID file system path URI - it must start with '${LID_PROT}' prefix - offendinf value: $path") - if (path.startsWith(LID_PROT + SEPARATOR) && path.length() > 7) - throw new IllegalArgumentException("Invalid LID file system path URI - make sure the schema prefix does not container more than two slash characters - offending value: $path") - if (path == LID_PROT) //Empty path case + if( path.startsWith(LID_PROT + SEPARATOR) && path.length() > 7 ) + throw new IllegalArgumentException("Invalid LID file system path URI - make sure the schema prefix does not container more than two slash characters or a query in the root '/' - offending value: $path") + if( path == LID_PROT ) //Empty path case return new URI("lid:///") return new URI(path) } @Override String toString() { - return "$filePath${query ? '?' + query: ''}${fragment ? '#'+ fragment : ''}".toString() + return "$filePath${query ? '?' + query : ''}${fragment ? '#' + fragment : ''}".toString() } } diff --git a/modules/nf-lineage/src/resources/META-INF/extensions.idx b/modules/nf-lineage/src/resources/META-INF/extensions.idx index 53c350a1be..5b327e222d 100644 --- a/modules/nf-lineage/src/resources/META-INF/extensions.idx +++ b/modules/nf-lineage/src/resources/META-INF/extensions.idx @@ -15,5 +15,6 @@ # nextflow.lineage.DefaultLinStoreFactory +nextflow.lineage.LinExtensionImpl nextflow.lineage.LinObserverFactory nextflow.lineage.cli.LinCommandImpl diff --git a/modules/nf-lineage/src/test/nextflow/lineage/DefaultLinStoreTest.groovy b/modules/nf-lineage/src/test/nextflow/lineage/DefaultLinStoreTest.groovy index a471b3c932..da1f407099 100644 --- a/modules/nf-lineage/src/test/nextflow/lineage/DefaultLinStoreTest.groovy +++ b/modules/nf-lineage/src/test/nextflow/lineage/DefaultLinStoreTest.groovy @@ -123,7 +123,7 @@ class DefaultLinStoreTest extends Specification { lidStore.save(key4, value4) when: - def results = lidStore.search("type=FileOutput&labels=value2") + def results = lidStore.search( [type:['FileOutput'], labels:['value2']]) then: results.size() == 2 results.keySet().containsAll([key2,key3]) diff --git a/modules/nf-lineage/src/test/nextflow/lineage/LinExtensionImplTest.groovy b/modules/nf-lineage/src/test/nextflow/lineage/LinExtensionImplTest.groovy new file mode 100644 index 0000000000..3a7d4c4eaf --- /dev/null +++ b/modules/nf-lineage/src/test/nextflow/lineage/LinExtensionImplTest.groovy @@ -0,0 +1,111 @@ +/* + * Copyright 2013-2025, Seqera Labs + * + * Licensed 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 nextflow.lineage + +import java.nio.file.Path +import java.time.Instant +import java.time.OffsetDateTime +import java.time.ZoneOffset + +import nextflow.Channel +import nextflow.Session +import nextflow.extension.CH +import nextflow.lineage.config.LineageConfig +import nextflow.lineage.fs.LinPathFactory +import nextflow.lineage.model.Checksum +import nextflow.lineage.model.DataPath +import nextflow.lineage.model.FileOutput +import nextflow.lineage.model.Parameter +import nextflow.lineage.model.Workflow +import nextflow.lineage.model.WorkflowRun +import spock.lang.Specification +import spock.lang.TempDir + +import static nextflow.lineage.fs.LinPath.* + +/** + * Lineage channel extensions tests + * + * @author Jorge Ejarque + */ +class LinExtensionImplTest extends Specification { + + @TempDir + Path tempDir + + Path storeLocation + Map configMap + + def setup() { + storeLocation = tempDir.resolve("store") + configMap = [linage: [enabled: true, store: [location: storeLocation.toString()]]] + } + + def 'should return global query results' () { + given: + def uniqueId = UUID.randomUUID() + def time = OffsetDateTime.ofInstant(Instant.ofEpochMilli(1234567), ZoneOffset.UTC) + def mainScript = new DataPath("file://path/to/main.nf", new Checksum("78910", "nextflow", "standard")) + def workflow = new Workflow([mainScript],"https://nextflow.io/nf-test/", "123456" ) + def key = "testKey" + def value1 = new WorkflowRun(workflow, uniqueId.toString(), "test_run", [ new Parameter("String", "param1", "value1"), new Parameter("String", "param2", "value2")] ) + def key2 = "testKey2" + def value2 = new FileOutput("/path/tp/file1", new Checksum("78910", "nextflow", "standard"), "testkey", "testkey", "taskid", 1234, time, time, ["value1","value2"]) + def key3 = "testKey3" + def value3 = new FileOutput("/path/tp/file2", new Checksum("78910", "nextflow", "standard"), "testkey", "testkey", null, 1234, time, time, ["value2", "value3"]) + def key4 = "testKey4" + def value4 = new FileOutput("/path/tp/file", new Checksum("78910", "nextflow", "standard"), "testkey", "testkey", "taskid", 1234, time, time, ["value4","value3"]) + def lidStore = new DefaultLinStore() + def session = Mock(Session) { + getConfig() >> configMap + } + lidStore.open(LineageConfig.create(session)) + lidStore.save(key, value1) + lidStore.save(key2, value2) + lidStore.save(key3, value3) + lidStore.save(key4, value4) + def linExt = Spy(new LinExtensionImpl()) + when: + def results = CH.create() + linExt.fromLineage(session, results, [labels: ["value2", "value3"]]) + then: + linExt.getStore(session) >> lidStore + and: + results.val == LinPathFactory.create( asUriString(key3) ) + results.val == Channel.STOP + + when: + results = CH.create() + linExt.fromLineage(session, results, [taskRun: "taskid", labels: ["value4"]]) + then: + linExt.getStore(session) >> lidStore + and: + results.val == LinPathFactory.create( asUriString(key4) ) + results.val == Channel.STOP + + when: + results = CH.create() + linExt.fromLineage(session, results, [workflowRun: "testkey", taskRun: "taskid", labels: ["value2"]]) + then: + linExt.getStore(session) >> lidStore + and: + results.val == LinPathFactory.create( asUriString(key2) ) + results.val == Channel.STOP + + + } +} diff --git a/modules/nf-lineage/src/test/nextflow/lineage/LinUtilsTest.groovy b/modules/nf-lineage/src/test/nextflow/lineage/LinUtilsTest.groovy index 5cd05e75ca..0cd7a6edeb 100644 --- a/modules/nf-lineage/src/test/nextflow/lineage/LinUtilsTest.groovy +++ b/modules/nf-lineage/src/test/nextflow/lineage/LinUtilsTest.groovy @@ -67,7 +67,7 @@ class LinUtilsTest extends Specification{ } - def 'should query'() { + def 'should get lineage record'() { given: def uniqueId = UUID.randomUUID() def mainScript = new DataPath("file://path/to/main.nf", new Checksum("78910", "nextflow", "standard")) @@ -81,32 +81,30 @@ class LinUtilsTest extends Specification{ lidStore.save("$key#output", outputs1) when: - List params = LinUtils.query(lidStore, new URI('lid://testKey#params')) + def params = LinUtils.getMetadataObject(lidStore, new URI('lid://testKey#params')) then: - params.size() == 1 - params[0] instanceof List - (params[0] as List).size() == 2 + params instanceof List + (params as List).size() == 2 when: - List outputs = LinUtils.query(lidStore, new URI('lid://testKey#output')) + def outputs = LinUtils.getMetadataObject(lidStore, new URI('lid://testKey#output')) then: - outputs.size() == 1 - outputs[0] instanceof List - def param = (outputs[0] as List)[0] as Parameter + outputs instanceof List + def param = (outputs as List)[0] as Parameter param.name == "output" when: - LinUtils.query(lidStore, new URI('lid://testKey#no-exist')) + LinUtils.getMetadataObject(lidStore, new URI('lid://testKey#no-exist')) then: thrown(IllegalArgumentException) when: - LinUtils.query(lidStore, new URI('lid://testKey#outputs.no-exist')) + LinUtils.getMetadataObject(lidStore, new URI('lid://testKey#outputs.no-exist')) then: thrown(IllegalArgumentException) when: - LinUtils.query(lidStore, new URI('lid://no-exist#something')) + LinUtils.getMetadataObject(lidStore, new URI('lid://no-exist#something')) then: thrown(IllegalArgumentException) } @@ -123,6 +121,7 @@ class LinUtilsTest extends Specification{ "" | [] } + def "should check params in an object"() { given: def obj = [ "type": "value", "workflow": ["repository": "subvalue"], "output" : [ ["path":"/to/file"],["path":"file2"] ], "labels": ["a","b"] ] @@ -147,20 +146,6 @@ class LinUtilsTest extends Specification{ } - def 'should parse query' (){ - expect: - LinUtils.parseQuery(PARAMS) == EXPECTED - where: - PARAMS | EXPECTED - "type=value" | ["type": ["value"]] - "workflow.repository=subvalue" | ["workflow.repository": ["subvalue"]] - "type=value1&taskRun=value2" | ["type": ["value1"], "taskRun": ["value2"]] - "type=val with space" | ["type": ["val with space"]] - "type=value&labels=a&labels=b" | ["type": ["value"], "labels":["a","b"]] - "" | [:] - null | [:] - } - def "should navigate in object params"() { given: def obj = [ @@ -201,23 +186,6 @@ class LinUtilsTest extends Specification{ [["nested": ["subfield": "match"]], ["nested": ["subfield": "other"]]] | ["nested.subfield": ["match"]] | [["nested": ["subfield": "match"]]] } - def "Should search path"() { - given: - def uniqueId = UUID.randomUUID() - def mainScript = new DataPath("file://path/to/main.nf", new Checksum("78910", "nextflow", "standard")) - def workflow = new Workflow([mainScript], "https://nextflow.io/nf-test/", "123456") - def key = "testKey" - def value1 = new WorkflowRun(workflow, uniqueId.toString(), "test_run", [new Parameter("String", "param1", "value1"), new Parameter("String", "param2", "value2")]) - def lidStore = new DefaultLinStore() - lidStore.open(config) - lidStore.save(key, value1) - when: - def result = LinUtils.searchPath(lidStore, key, ["name":["param1"]], ["params"] as String[]) - - then: - result == [new Parameter("String", "param1", "value1")] - } - def 'should navigate' (){ def uniqueId = UUID.randomUUID() def mainScript = new DataPath("file://path/to/main.nf", new Checksum("78910", "nextflow", "standard")) diff --git a/modules/nf-lineage/src/test/nextflow/lineage/cli/LinCommandImplTest.groovy b/modules/nf-lineage/src/test/nextflow/lineage/cli/LinCommandImplTest.groovy index 61295cd328..9d0fa1ba10 100644 --- a/modules/nf-lineage/src/test/nextflow/lineage/cli/LinCommandImplTest.groovy +++ b/modules/nf-lineage/src/test/nextflow/lineage/cli/LinCommandImplTest.groovy @@ -152,7 +152,7 @@ class LinCommandImplTest extends Specification{ then: stdout.size() == 1 - stdout[0] == "Error loading lid://12345 - Lineage object 12345 not found" + stdout[0] == "Error loading lid://12345 - Lineage record 12345 not found" } def 'should get lineage lid content' (){ @@ -278,7 +278,7 @@ class LinCommandImplTest extends Specification{ outputHtml.text == expectedOutput } - def 'should show query results'(){ + def 'should show an error if trying to do a query'(){ given: def lidFile = storeLocation.resolve("12345/.data.json") Files.createDirectories(lidFile.parent) @@ -287,7 +287,7 @@ class LinCommandImplTest extends Specification{ def entry = new FileOutput("path/to/file",new Checksum("45372qe","nextflow","standard"), "lid://123987/file.bam", "lid://123987/", null, 1234, time, time, null) def jsonSer = encoder.encode(entry) - def expectedOutput = jsonSer + def expectedOutput = "Error loading lid:///?type=FileOutput - Cannot get record from the root LID URI" lidFile.text = jsonSer when: new LinCommandImpl().describe(configMap, ["lid:///?type=FileOutput"]) @@ -303,35 +303,6 @@ class LinCommandImplTest extends Specification{ stdout.join('\n') == expectedOutput } - def 'should show query with fragment'(){ - given: - def lidFile = storeLocation.resolve("12345/.data.json") - Files.createDirectories(lidFile.parent) - def lidFile2 = storeLocation.resolve("67890/.data.json") - Files.createDirectories(lidFile2.parent) - def encoder = new LinEncoder().withPrettyPrint(true) - def time = OffsetDateTime.ofInstant(Instant.ofEpochMilli(123456789), ZoneOffset.UTC) - def entry = new FileOutput("path/to/file",new Checksum("45372qe","nextflow","standard"), - "lid://123987/file.bam", "lid://123987/", null, 1234, time, time, null) - def entry2 = new FileOutput("path/to/file2",new Checksum("42472qet","nextflow","standard"), - "lid://123987/file2.bam", "lid://123987/", null, 1235, time, time, null) - def expectedOutput1 = '[\n "path/to/file",\n "path/to/file2"\n]' - def expectedOutput2 = '[\n "path/to/file2",\n "path/to/file"\n]' - lidFile.text = encoder.encode(entry) - lidFile2.text = encoder.encode(entry2) - when: - new LinCommandImpl().describe(configMap, ["lid:///?type=FileOutput#path"]) - def stdout = capture - .toString() - .readLines()// remove the log part - .findResults { line -> !line.contains('DEBUG') ? line : null } - .findResults { line -> !line.contains('INFO') ? line : null } - .findResults { line -> !line.contains('plugin') ? line : null } - - then: - stdout.join('\n') == expectedOutput1 || stdout.join('\n') == expectedOutput2 - } - def 'should diff'(){ given: def lidFile = storeLocation.resolve("12345/.data.json") @@ -414,7 +385,7 @@ class LinCommandImplTest extends Specification{ when: def config = new ConfigMap() new LinCommandImpl().log(config) - new LinCommandImpl().describe(config, ["lid:///?type=FileOutput"]) + new LinCommandImpl().describe(config, ["lid:///12345"]) new LinCommandImpl().render(config, ["lid://12345", "output.html"]) new LinCommandImpl().diff(config, ["lid://89012", "lid://12345"]) @@ -439,18 +410,23 @@ class LinCommandImplTest extends Specification{ Files.createDirectories(lidFile.parent) def lidFile2 = storeLocation.resolve("123987/file2.bam/.data.json") Files.createDirectories(lidFile2.parent) + def lidFile3 = storeLocation.resolve(".meta/123987/file3.bam/.data.json") + Files.createDirectories(lidFile3.parent) def encoder = new LinEncoder().withPrettyPrint(true) def time = OffsetDateTime.ofInstant(Instant.ofEpochMilli(123456789), ZoneOffset.UTC) def entry = new FileOutput("path/to/file",new Checksum("45372qe","nextflow","standard"), - "lid://123987/file.bam", "lid://123987/", null, 1234, time, time, null) + "lid://123987/file.bam", "lid://123987/", null, 1234, time, time, ["experiment=test"]) def entry2 = new FileOutput("path/to/file2",new Checksum("42472qet","nextflow","standard"), + "lid://123987/file2.bam", "lid://123987/", null, 1235, time, time, ["experiment=test"]) + def entry3 = new FileOutput("path/to/file3",new Checksum("42472qet","nextflow","standard"), "lid://123987/file2.bam", "lid://123987/", null, 1235, time, time, null) def expectedOutput1 = '[\n "lid://123987/file.bam",\n "lid://123987/file2.bam"\n]' def expectedOutput2 = '[\n "lid://123987/file2.bam",\n "lid://123987/file.bam"\n]' lidFile.text = encoder.encode(entry) lidFile2.text = encoder.encode(entry2) + lidFile3.text = encoder.encode(entry3) when: - new LinCommandImpl().find(configMap, ["type=FileOutput"]) + new LinCommandImpl().find(configMap, ["type=FileOutput", "labels=experiment=test"]) def stdout = capture .toString() .readLines()// remove the log part diff --git a/modules/nf-lineage/src/test/nextflow/lineage/fs/LinPathTest.groovy b/modules/nf-lineage/src/test/nextflow/lineage/fs/LinPathTest.groovy index 8766f0cd9b..112a326da6 100644 --- a/modules/nf-lineage/src/test/nextflow/lineage/fs/LinPathTest.groovy +++ b/modules/nf-lineage/src/test/nextflow/lineage/fs/LinPathTest.groovy @@ -72,13 +72,36 @@ class LinPathTest extends Specification { path.query == QUERY where: - URI_STRING | PATH | QUERY | FRAGMENT - "lid://1234/hola" | "1234/hola" | null | null - "lid://1234/hola#frag.sub" | "1234/hola" | null | "frag.sub" - "lid://1234/#frag.sub" | "1234" | null | "frag.sub" - "lid://1234/?q=a&b=c" | "1234" | "q=a&b=c" | null - "lid://1234/?q=a&b=c#frag.sub" | "1234" | "q=a&b=c" | "frag.sub" - "lid:///" | "/" | null | null + URI_STRING | PATH | QUERY | FRAGMENT + "lid://1234/hola" | "1234/hola" | null | null + "lid://1234/hola#workflow.repository" | "1234/hola" | null | "workflow.repository" + "lid://1234/#workflow.repository" | "1234" | null | "workflow.repository" + "lid://1234/?q=a&b=c" | "1234" | "q=a&b=c" | null + "lid://1234/?q=a&b=c#workflow.repository" | "1234" | "q=a&b=c" | "workflow.repository" + "lid:///" | "/" | null | null + } + + def 'should throw exception if fragment contains an unknown property'() { + when: + new LinPath(fs, new URI ("lid://1234/hola#no-exist")) + then: + thrown(IllegalArgumentException) + + } + + def 'should warn if query is specified'() { + when: + new LinPath(fs, new URI("lid://1234/hola?query")) + def stdout = capture + .toString() + .readLines()// remove the log part + .findResults { line -> !line.contains('DEBUG') ? line : null } + .findResults { line -> !line.contains('INFO') ? line : null } + .findResults { line -> !line.contains('plugin') ? line : null } + + then: + stdout.size() == 1 + stdout[0].endsWith("Query string is not supported for Lineage URI: `lid://1234/hola?query` -- it will be ignored") } def 'should create correct lid Path' () { @@ -245,10 +268,13 @@ class LinPathTest extends Specification { } def 'should get file name' () { - when: - def lid1 = new LinPath(fs, '1234567890/this/file.bam') - then: - lid1.getFileName() == new LinPath(null, 'file.bam') + expect: + new LinPath(fs, PATH).getFileName() == EXPECTED + where: + PATH | EXPECTED + '1234567890/this/file.bam' | new LinPath(null, 'file.bam') + '12345/hola?query#output' | new LinPath("query", "output", "hola", null) + } def 'should get file parent' () { @@ -280,11 +306,12 @@ class LinPathTest extends Specification { expect: new LinPath(fs, PATH).getName(INDEX) == EXPECTED where: - PATH | INDEX | EXPECTED - '123' | 0 | new LinPath(fs, '123') - '123/a' | 1 | new LinPath(null, 'a') - '123/a/' | 1 | new LinPath(null, 'a') - '123/a/b' | 2 | new LinPath(null, 'b') + PATH | INDEX | EXPECTED + '123' | 0 | new LinPath(fs, '123') + '123/a' | 1 | new LinPath(null, 'a') + '123/a/' | 1 | new LinPath(null, 'a') + '123/a/b' | 2 | new LinPath(null, 'b') + '123/a?q#output' | 1 | new LinPath(null, 'a?q#output') } @Unroll