CLI tests

bagit.py

@@ -855,7 +855,7 @@ def _validate_completeness(self):
             errors.append(e)
 
         if errors:
-            raise BagValidationError(_("Bag validation failed"), errors)
+            raise BagValidationError(_("Bag is incomplete"), errors)
 
     def _validate_entries(self, processes):
         """
@@ -1563,12 +1563,15 @@ def main():
     parser = _make_parser()
     args = parser.parse_args()
 
-    if args.processes < 0:
-        parser.error(_("The number of processes must be 0 or greater"))
+    if args.processes <= 0:
+        parser.error(_("The number of processes must be greater than 0"))
 
     if args.fast and not args.validate:
         parser.error(_("--fast is only allowed as an option for --validate!"))
 
+    if args.completeness_only and not args.validate:
+        parser.error(_("--completeness-only is only allowed as an option for --validate!"))
+
     _configure_logging(args)
 
     rc = 0
@@ -1585,6 +1588,8 @@ def main():
                 )
                 if args.fast:
                     LOGGER.info(_("%s valid according to Payload-Oxum"), bag_dir)
+                elif args.completeness_only:
+                    LOGGER.info(_("%s is complete and valid according to Payload-Oxum"), bag_dir)
                 else:
                     LOGGER.info(_("%s is valid"), bag_dir)
             except BagError as e:

test.py

@@ -16,6 +16,7 @@
 from os.path import join as j
 
 import mock
+from io import StringIO
 
 import bagit
 
@@ -218,7 +219,7 @@ def test_validation_completeness_error_details(self):
             got_exception = True
 
             exc_str = str(e)
-            self.assertIn("Bag validation failed: ", exc_str)
+            self.assertIn("Bag is incomplete: ", exc_str)
             self.assertIn(
                 "bag-info.txt exists in manifest but was not found on filesystem",
                 exc_str,
@@ -1100,6 +1101,194 @@ def test_fetch_malformed_url(self):
         self.assertEqual(expected_msg, str(cm.exception))
 
 
+class TestCLI(SelfCleaningTestCase):
+
+    @mock.patch('sys.stderr', new_callable=StringIO)
+    def test_directory_required(self, mock_stderr):
+        testargs = ["bagit.py"]
+
+        with self.assertRaises(SystemExit) as cm:
+            with mock.patch.object(sys, 'argv', testargs):
+                bagit.main()
+
+        self.assertEqual(cm.exception.code, 2)
+        self.assertIn(
+            "error: the following arguments are required: directory",
+            mock_stderr.getvalue()
+        )
+
+    @mock.patch('sys.stderr', new_callable=StringIO)
+    def test_not_enough_processes(self, mock_stderr):
+        testargs = ["bagit.py", "--processes", "0", self.tmpdir]
+
+        with self.assertRaises(SystemExit) as cm:
+            with mock.patch.object(sys, 'argv', testargs):
+                bagit.main()
+
+        self.assertEqual(cm.exception.code, 2)
+        self.assertIn(
+            "error: The number of processes must be greater than 0",
+            mock_stderr.getvalue()
+        )
+
+    @mock.patch('sys.stderr', new_callable=StringIO)
+    def test_fast_flag_without_validate(self, mock_stderr):
+        bag = bagit.make_bag(self.tmpdir)
+        testargs = ["bagit.py", "--fast", self.tmpdir]
+
+        with self.assertRaises(SystemExit) as cm:
+            with mock.patch.object(sys, 'argv', testargs):
+                bagit.main()
+
+        self.assertEqual(cm.exception.code, 2)
+        self.assertIn(
+            "error: --fast is only allowed as an option for --validate!",
+            mock_stderr.getvalue()
+        )
+
+    def test_invalid_fast_validate(self):
+        bag = bagit.make_bag(self.tmpdir)
+        os.remove(j(self.tmpdir, "data", "loc", "2478433644_2839c5e8b8_o_d.jpg"))
+        testargs = ["bagit.py", "--validate", "--completeness-only", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 1)
+        self.assertIn(
+            "%s is invalid: Payload-Oxum validation failed." % self.tmpdir,
+            captured.records[0].getMessage()
+        )
+
+    def test_valid_fast_validate(self):
+        bag = bagit.make_bag(self.tmpdir)
+        testargs = ["bagit.py", "--validate", "--fast", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 0)
+        self.assertEqual(
+            "%s valid according to Payload-Oxum" % self.tmpdir,
+            captured.records[0].getMessage()
+        )
+
+    @mock.patch('sys.stderr', new_callable=StringIO)
+    def test_completeness_flag_without_validate(self, mock_stderr):
+        bag = bagit.make_bag(self.tmpdir)
+        testargs = ["bagit.py", "--completeness-only", self.tmpdir]
+
+        with self.assertRaises(SystemExit) as cm:
+            with mock.patch.object(sys, 'argv', testargs):
+                bagit.main()
+
+        self.assertEqual(cm.exception.code, 2)
+        self.assertIn(
+            "error: --completeness-only is only allowed as an option for --validate!",
+            mock_stderr.getvalue()
+        )
+
+    def test_invalid_completeness_validate(self):
+        bag = bagit.make_bag(self.tmpdir)
+        old_path = j(self.tmpdir, "data", "README")
+        new_path = j(self.tmpdir, "data", "extra_file")
+        os.rename(old_path, new_path)
+
+        testargs = ["bagit.py", "--validate", "--completeness-only", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 1)
+        self.assertIn(
+            "%s is invalid: Bag is incomplete" % self.tmpdir,
+            captured.records[-1].getMessage()
+        )
+
+    def test_valid_completeness_validate(self):
+        bag = bagit.make_bag(self.tmpdir)
+        testargs = ["bagit.py", "--validate", "--completeness-only", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 0)
+        self.assertEqual(
+            "%s is complete and valid according to Payload-Oxum" % self.tmpdir,
+            captured.records[0].getMessage()
+        )
+
+    def test_invalid_full_validate(self):
+        bag = bagit.make_bag(self.tmpdir)
+        readme = j(self.tmpdir, "data", "README")
+        txt = slurp_text_file(readme)
+        txt = "A" + txt[1:]
+        with open(readme, "w") as r:
+            r.write(txt)
+
+        testargs = ["bagit.py", "--validate", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 1)
+        self.assertIn("Bag validation failed", captured.records[-1].getMessage())
+
+    def test_valid_full_validate(self):
+        bag = bagit.make_bag(self.tmpdir)
+        testargs = ["bagit.py", "--validate", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 0)
+        self.assertEqual(
+            "%s is valid" % self.tmpdir,
+            captured.records[-1].getMessage()
+        )
+
+    def test_failed_create_bag(self):
+        os.chmod(self.tmpdir, 0)
+
+        testargs = ["bagit.py", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        self.assertEqual(cm.exception.code, 1)
+        self.assertIn(
+            "Failed to create bag in %s" % self.tmpdir,
+            captured.records[-1].getMessage()
+        )
+
+    def test_create_bag(self):
+        testargs = ["bagit.py", self.tmpdir]
+
+        with self.assertLogs() as captured:
+            with self.assertRaises(SystemExit) as cm:
+                with mock.patch.object(sys, 'argv', testargs):
+                    bagit.main()
+
+        for rec in captured.records:
+            print(rec.getMessage())
+
+        self.assertEqual(cm.exception.code, 0)
+
+
 class TestUtils(unittest.TestCase):
     def setUp(self):
         super(TestUtils, self).setUp()
@@ -1118,5 +1307,7 @@ def test_force_unicode_int(self):
         self.assertIsInstance(bagit.force_unicode(1234), self.unicode_class)
 
 
+
+
 if __name__ == "__main__":
     unittest.main()