@Override
    @SuppressWarnings("unchecked")
    public Output run(RunContext runContext) throws Exception {
        Logger logger = runContext.logger();
        String renderedNamespace = runContext.render(this.namespace);
        String renderedDestination = runContext.render(destination);

        final Namespace namespace = runContext.storage().namespace(renderedNamespace);

        List<String> renderedFiles;
        if (files instanceof String filesString) {
            renderedFiles = List.of(runContext.render(filesString));
        } else if (files instanceof List<?> filesList) {
            renderedFiles = runContext.render((List<String>) filesList);
        } else {
            throw new IllegalArgumentException("The files property must be a string or a list of strings");
        }

        Map<String, URI> downloaded = namespace.findAllFilesMatching(PathMatcherPredicate.matches(renderedFiles))
            .stream()
            .map(Rethrow.throwFunction(file -> {
                try (InputStream is = runContext.storage().getFile(file.uri())) {
                    URI uri = runContext.storage().putFile(is, renderedDestination + file.path());
                    logger.debug(String.format("Downloaded %s", uri));
                    return new AbstractMap.SimpleEntry<>(file.path(true).toString(), uri);
                }
            }))
            .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue));
        runContext.metric(Counter.of("downloaded", downloaded.size()));
        return Output.builder().files(downloaded).build();
    }
@Test
    void shouldDownloadNamespaceFile() throws Exception {
        String namespaceId = "io.kestra." + IdUtils.create();
        DownloadFiles downloadFiles = DownloadFiles.builder()
            .id(DownloadFiles.class.getSimpleName())
            .type(DownloadFiles.class.getName())
            .files(List.of("**test1.txt"))
            .namespace("{{ inputs.namespace }}")
            .build();

        final RunContext runContext = TestsUtils.mockRunContext(this.runContextFactory, downloadFiles, Map.of("namespace", namespaceId));
        final Namespace namespace = runContext.storage().namespace(namespaceId);

        namespace.putFile(Path.of("/a/b/test1.txt"), new ByteArrayInputStream("1".getBytes(StandardCharsets.UTF_8)));
        namespace.putFile(Path.of("/a/b/test2.txt"), new ByteArrayInputStream("2".getBytes(StandardCharsets.UTF_8)));

        DownloadFiles.Output output = downloadFiles.run(runContext);

        assertThat(output.getFiles().size(), is(1));
        assertThat(output.getFiles().get("/a/b/test1.txt"), notNullValue());

    }