Merge pull request #82 from edina/fetch_allow_empty_folder

Fetch allow empty folder

nbexchange/plugin/fetch_assignment.py

@@ -1,3 +1,4 @@
+import glob
 import io
 import nbgrader.exchange.abc as abc
 import os
@@ -51,9 +52,14 @@ def init_dest(self):
         else:
             root = self.coursedir.assignment_id
         self.dest_path = os.path.abspath(os.path.join(self.assignment_dir, root))
-        if os.path.isdir(self.dest_path) and not self.replace_missing_files:
+        # Lets check there are no notebooks already in the dest_path dir
+        if (
+            os.path.isdir(self.dest_path)
+            and glob.glob(self.dest_path + "/*.ipynb")
+            and not self.replace_missing_files
+        ):
             self.fail(
-                f"You already have a copy of the assignment in this directory: {root}"
+                f"You already have notebook documents in directory: {root}. Please remove them before fetching again"
             )
         else:
             os.makedirs(os.path.dirname(self.dest_path + "/"), exist_ok=True)

nbexchange/tests/test_plugin_fetch_assignment.py

@@ -1,17 +1,17 @@
 import io
 import logging
 import os
+import pytest
 import shutil
 import tarfile
-from shutil import copyfile
 
-import pytest
 
 from mock import patch
+from shutil import copyfile
 
-from nbexchange.plugin import ExchangeFetchAssignment, Exchange
 from nbgrader.coursedir import CourseDirectory
-
+from nbgrader.exchange import ExchangeError
+from nbexchange.plugin import ExchangeFetchAssignment, Exchange
 from nbexchange.tests.utils import get_feedback_file
 
 logger = logging.getLogger(__file__)
@@ -161,3 +161,155 @@ def api_request(*args, **kwargs):
             )
     finally:
         shutil.rmtree(plugin.dest_path)
+
+
+@pytest.mark.gen_test
+def test_fetch_empty_folder_exists(plugin_config, tmpdir):
+    plugin_config.CourseDirectory.course_id = "no_course"
+    plugin_config.CourseDirectory.assignment_id = "assign_1_3"
+
+    plugin = ExchangeFetchAssignment(
+        coursedir=CourseDirectory(config=plugin_config), config=plugin_config
+    )
+    os.makedirs("assign_1_3")
+    try:
+
+        def api_request(*args, **kwargs):
+            tar_file = io.BytesIO()
+
+            with tarfile.open(fileobj=tar_file, mode="w:gz") as tar_handle:
+                tar_handle.add(
+                    notebook1_filename, arcname=os.path.basename(notebook1_filename)
+                )
+                tar_handle.add(
+                    notebook2_filename, arcname=os.path.basename(notebook2_filename)
+                )
+            tar_file.seek(0)
+
+            assert args[0] == (
+                f"assignment?course_id=no_course&assignment_id=assign_1_3"
+            )
+            assert "method" not in kwargs or kwargs.get("method").lower() == "get"
+            return type(
+                "Response",
+                (object,),
+                {
+                    "status_code": 200,
+                    "headers": {"content-type": "application/x-tar"},
+                    "content": tar_file.read(),
+                },
+            )
+
+        with patch.object(Exchange, "api_request", side_effect=api_request):
+            called = plugin.start()
+            print(f"dest_path {plugin.dest_path}")
+            assert os.path.exists(
+                os.path.join(plugin.dest_path, "assignment-0.6.ipynb")
+            )
+            assert os.path.exists(
+                os.path.join(plugin.dest_path, "assignment-0.6-2.ipynb")
+            )
+    finally:
+        shutil.rmtree(plugin.dest_path)
+
+
+@pytest.mark.gen_test
+def test_fetch_folder_exists_with_ipynb(plugin_config, tmpdir):
+    plugin_config.CourseDirectory.course_id = "no_course"
+    plugin_config.CourseDirectory.assignment_id = "assign_1_3"
+
+    plugin = ExchangeFetchAssignment(
+        coursedir=CourseDirectory(config=plugin_config), config=plugin_config
+    )
+    os.makedirs("assign_1_3")
+    with open("assign_1_3/decoy.ipynb", "w") as f:
+        f.write(" ")
+    try:
+
+        def api_request(*args, **kwargs):
+            tar_file = io.BytesIO()
+
+            with tarfile.open(fileobj=tar_file, mode="w:gz") as tar_handle:
+                tar_handle.add(
+                    notebook1_filename, arcname=os.path.basename(notebook1_filename)
+                )
+                tar_handle.add(
+                    notebook2_filename, arcname=os.path.basename(notebook2_filename)
+                )
+            tar_file.seek(0)
+
+            assert args[0] == (
+                f"assignment?course_id=no_course&assignment_id=assign_1_3"
+            )
+            assert "method" not in kwargs or kwargs.get("method").lower() == "get"
+            return type(
+                "Response",
+                (object,),
+                {
+                    "status_code": 200,
+                    "headers": {"content-type": "application/x-tar"},
+                    "content": tar_file.read(),
+                },
+            )
+
+        with patch.object(Exchange, "api_request", side_effect=api_request):
+            with pytest.raises(ExchangeError) as e_info:
+                called = plugin.start()
+            assert (
+                str(e_info.value)
+                == "You already have notebook documents in directory: assign_1_3. Please remove them before fetching again"
+            )
+    finally:
+        shutil.rmtree(plugin.dest_path)
+
+
+@pytest.mark.gen_test
+def test_fetch_folder_exists_with_other_file(plugin_config, tmpdir):
+    plugin_config.CourseDirectory.course_id = "no_course"
+    plugin_config.CourseDirectory.assignment_id = "assign_1_3"
+
+    plugin = ExchangeFetchAssignment(
+        coursedir=CourseDirectory(config=plugin_config), config=plugin_config
+    )
+    os.makedirs("assign_1_3")
+    with open("assign_1_3/decoy.txt", "w") as f:
+        f.write(" ")
+    try:
+
+        def api_request(*args, **kwargs):
+            tar_file = io.BytesIO()
+
+            with tarfile.open(fileobj=tar_file, mode="w:gz") as tar_handle:
+                tar_handle.add(
+                    notebook1_filename, arcname=os.path.basename(notebook1_filename)
+                )
+                tar_handle.add(
+                    notebook2_filename, arcname=os.path.basename(notebook2_filename)
+                )
+            tar_file.seek(0)
+
+            assert args[0] == (
+                f"assignment?course_id=no_course&assignment_id=assign_1_3"
+            )
+            assert "method" not in kwargs or kwargs.get("method").lower() == "get"
+            return type(
+                "Response",
+                (object,),
+                {
+                    "status_code": 200,
+                    "headers": {"content-type": "application/x-tar"},
+                    "content": tar_file.read(),
+                },
+            )
+
+        with patch.object(Exchange, "api_request", side_effect=api_request):
+            called = plugin.start()
+            print(f"dest_path {plugin.dest_path}")
+            assert os.path.exists(
+                os.path.join(plugin.dest_path, "assignment-0.6.ipynb")
+            )
+            assert os.path.exists(
+                os.path.join(plugin.dest_path, "assignment-0.6-2.ipynb")
+            )
+    finally:
+        shutil.rmtree(plugin.dest_path)