Paired metadata redact

q2cli/builtin/tools.py

@@ -1749,3 +1749,42 @@ def signature_verify(input_path, name):
     click.echo(CONFIG.cfg_style('success',
                                 f'Signature {name} on Result '
                                 f'{input_path} verified successfully.'))
+
+
+@tools.command(
+    name='redact-metadata',
+    short_help='Remove metadata from a Result.',
+    help='Remove all metadata from a Result and return the Result '
+         'otherwise unchanged.'
+)
+@click.option(
+    '--input-path',
+    required=True,
+    help='The path to the Result to remove metadata from.'
+)
+@click.option(
+    '--output-path',
+    required=True,
+    help='Path to save the Result with redacted metadata to.'
+)
+def redact_metadata(input_path, output_path):
+    from rachis.sdk.result import Result
+    from q2cli.core.config import CONFIG
+
+    result = Result.load(input_path)
+
+    try:
+        result.redact_metadata()
+    except ValueError as e:
+        click.echo(CONFIG.cfg_style('error', str(e)), err=True)
+        raise click.Abort()
+
+    result.save(output_path)
+
+    click.echo(
+        CONFIG.cfg_style(
+            'success',
+            f'Succesfully redacted metadata from {input_path}, and saved to '
+            f'{output_path}.'
+        )
+    )

q2cli/tests/test_tools.py

@@ -21,6 +21,7 @@
 from click.testing import CliRunner
 from qiime2 import Artifact, Metadata
 from qiime2.core.testing.util import get_dummy_plugin
+from qiime2.core.testing.type import IntSequence1, IntSequence2, SingleInt
 from qiime2.metadata.base import SUPPORTED_COLUMN_TYPES
 from qiime2.core.cache import Cache
 from qiime2.sdk.result import Result
@@ -2013,5 +2014,81 @@ def test_make_report_duplicate_names(self):
         self.assertIn('Multiple files share the same name', result.output)
 
 
+class TestRedactMetadata(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tempdir = tempfile.mkdtemp(prefix='qiime2-q2cli-test-temp-')
+
+        metadata_path = os.path.join(cls.tempdir, 'metadata.tsv')
+        with open(metadata_path, 'w') as fh:
+            fh.write('sample-id\tbarcode-sequence\n')
+            fh.write('1\tACT')
+        dummy_md = Metadata.load(metadata_path)
+
+        pm = PluginManager()
+        identity_with_metadata = pm.plugins['dummy-plugin'].actions[
+            'identity_with_metadata'
+        ]
+
+        cls.artifact1_fp = os.path.join(cls.tempdir, 'a1.qza')
+        cls.artifact2_fp = os.path.join(cls.tempdir, 'a2.qza')
+        cls.artifact3_fp = os.path.join(cls.tempdir, 'a3.qza')
+
+        artifact1 = Artifact.import_data(IntSequence1, [0, 6, 7])
+        artifact1, = identity_with_metadata(artifact1, dummy_md)
+        artifact1.save(cls.artifact1_fp)
+        artifact2 = Artifact.import_data(IntSequence2, [3, 4, 5])
+        artifact2, = identity_with_metadata(artifact2, dummy_md)
+        artifact2.save(cls.artifact2_fp)
+        artifact3 = Artifact.import_data(SingleInt, 9)
+        artifact3.save(cls.artifact3_fp)
+
+        cls.runner = CliRunner()
+        cls.redacted_fp = os.path.join(cls.tempdir, 'redacted')
+
+    def test_with_metadata(self):
+        redacted_fp = self.redacted_fp + '.qza'
+        result = self.runner.invoke(tools, [
+            'redact-metadata', '--input-path', self.artifact1_fp,
+            '--output-path', redacted_fp
+        ])
+
+        self.assertEqual(result.exit_code, 0)
+        success = f'Succesfully redacted metadata from {self.artifact1_fp}, ' \
+                  f'and saved to {redacted_fp}.\n'
+        self.assertEqual(success, result.output)
+
+    def test_fails_already_redacted(self):
+        '''
+        Redacting metadata from an Artifact twice should fail.
+        '''
+        redacted_fp = self.redacted_fp + '.qza'
+        self.runner.invoke(tools, [
+            'redact-metadata', '--input-path', self.artifact2_fp,
+            '--output-path', redacted_fp
+        ])
+
+        result = self.runner.invoke(tools, [
+            'redact-metadata', '--input-path', redacted_fp,
+            '--output-path', os.path.join(self.tempdir, 'bad')
+        ])
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('only redacted metadata', result.output)
+
+    def test_fails_no_metadata(self):
+        '''
+        Redacting metadata from an Artifact with no metadata should fail.
+        '''
+        redacted_fp = self.redacted_fp + '.qza'
+        result = self.runner.invoke(tools, [
+            'redact-metadata', '--input-path', self.artifact3_fp,
+            '--output-path', redacted_fp
+        ])
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Result without metadata', result.output)
+
+
 if __name__ == "__main__":
     unittest.main()