add tests for cli bag creation

Author: nkrabben

test_cli.py

@@ -0,0 +1,192 @@
+# encoding: utf-8
+
+from __future__ import absolute_import, division, print_function, unicode_literals
+import argparse
+
+import codecs
+import datetime
+import hashlib
+import logging
+import os
+import shutil
+import stat
+import sys
+import tempfile
+import unicodedata
+import unittest
+from unittest.mock import patch
+from os.path import join as j
+from io import StringIO
+
+import mock
+
+import bagit
+
+logging.basicConfig(filename="test.log", level=logging.DEBUG)
+stderr = logging.StreamHandler()
+stderr.setLevel(logging.WARNING)
+logging.getLogger().addHandler(stderr)
+
+# But we do want any exceptions raised in the logging path to be raised:
+logging.raiseExceptions = True
+
+
+def slurp_text_file(filename):
+    with bagit.open_text_file(filename) as f:
+        return f.read()
+
+
+class SelfCleaningTestCase(unittest.TestCase):
+    """TestCase subclass which cleans up self.tmpdir after each test"""
+
+    def setUp(self):
+        super(SelfCleaningTestCase, self).setUp()
+
+        self.starting_directory = (
+            os.getcwd()
+        )  # FIXME: remove this after we stop changing directories in bagit.py
+        self.tmpdir = tempfile.mkdtemp()
+        if os.path.isdir(self.tmpdir):
+            shutil.rmtree(self.tmpdir)
+        shutil.copytree("test-data", self.tmpdir)
+
+    def tearDown(self):
+        # FIXME: remove this after we stop changing directories in bagit.py
+        os.chdir(self.starting_directory)
+        if os.path.isdir(self.tmpdir):
+            # Clean up after tests which leave inaccessible files behind:
+
+            os.chmod(self.tmpdir, 0o700)
+
+            for dirpath, subdirs, filenames in os.walk(self.tmpdir, topdown=True):
+                for i in subdirs:
+                    os.chmod(os.path.join(dirpath, i), 0o700)
+
+            shutil.rmtree(self.tmpdir)
+
+        super(SelfCleaningTestCase, self).tearDown()
+
+class TestSingleProcessValidation(SelfCleaningTestCase):
+
+    @mock.patch('sys.stderr', new_callable=StringIO)
+    def test_missing_directory(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):
+        assert True
+
+    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()
+        )
+
+    def test_completeness_flag_without_validate(self):
+        assert True
+
+    def test_invalid_completeness_validate(self):
+        assert True
+
+    def test_valid_completeness_validate(self):
+        assert True
+
+    def test_valid_full_validate(self):
+        assert True
+
+    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.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)