test_case prune api

fix tests

conf/tests.py

@@ -1,4 +1,5 @@
 import hashlib
+from unittest import mock
 
 from django.utils import timezone
 
@@ -127,3 +128,24 @@ class LanguageListAPITest(APITestCase):
     def test_get_languages(self):
         resp = self.client.get(self.reverse("language_list_api"))
         self.assertSuccess(resp)
+
+
+class TestCasePruneAPITest(APITestCase):
+    def setUp(self):
+        self.url = self.reverse("prune_test_case_api")
+        self.create_super_admin()
+
+    def test_get_isolated_test_case(self):
+        resp = self.client.get(self.url)
+        self.assertSuccess(resp)
+
+    @mock.patch("conf.views.TestCasePruneAPI.delete_one")
+    @mock.patch("conf.views.os.listdir")
+    @mock.patch("conf.views.Problem")
+    def test_delete_test_case(self, mocked_problem, mocked_listdir, mocked_delete_one):
+        valid_id = "1172980672983b2b49820be3a741b109"
+        mocked_problem.return_value = [valid_id, ]
+        mocked_listdir.return_value = [valid_id, ".test", "aaa"]
+        resp = self.client.delete(self.url)
+        self.assertSuccess(resp)
+        mocked_delete_one.assert_called_once_with(valid_id)

conf/urls/admin.py

@@ -1,9 +1,10 @@
 from django.conf.urls import url
 
-from ..views import SMTPAPI, JudgeServerAPI, WebsiteConfigAPI
+from ..views import SMTPAPI, JudgeServerAPI, WebsiteConfigAPI, TestCasePruneAPI
 
 urlpatterns = [
     url(r"^smtp/?$", SMTPAPI.as_view(), name="smtp_admin_api"),
     url(r"^website/?$", WebsiteConfigAPI.as_view(), name="website_config_api"),
-    url(r"^judge_server/?$", JudgeServerAPI.as_view(), name="judge_server_api")
+    url(r"^judge_server/?$", JudgeServerAPI.as_view(), name="judge_server_api"),
+    url(r"^prune_test_case/?$", TestCasePruneAPI.as_view(), name="prune_test_case_api"),
 ]

conf/views.py

@@ -1,8 +1,13 @@
+import os
+import re
 import hashlib
+import shutil
 
 from django.utils import timezone
+from django.conf import settings
 
 from account.decorators import super_admin_required
+from problem.models import Problem
 from judge.dispatcher import process_pending_task
 from judge.languages import languages, spj_languages
 from options.options import SysOptions
@@ -117,3 +122,43 @@ class JudgeServerHeartbeatAPI(CSRFExemptAPIView):
 class LanguagesAPI(APIView):
     def get(self, request):
         return self.success({"languages": languages, "spj_languages": spj_languages})
+
+
+class TestCasePruneAPI(APIView):
+    @super_admin_required
+    def get(self, request):
+        """
+        return isolated test_case list
+        """
+        ret_data = []
+        dir_to_be_removed = self.get_orphan_ids()
+
+        # return an iterator
+        for d in os.scandir(settings.TEST_CASE_DIR):
+            if d.name in dir_to_be_removed:
+                ret_data.append({"id": d.name, "create_time": d.stat().st_ctime})
+        return self.success(ret_data)
+
+    @super_admin_required
+    def delete(self, request):
+        test_case_id = request.GET.get("id")
+        if test_case_id:
+            self.delete_one(test_case_id)
+            return self.success()
+        for id in self.get_orphan_ids():
+            self.delete_one(id)
+        return self.success()
+
+    @staticmethod
+    def get_orphan_ids():
+        db_ids = Problem.objects.all().values_list("test_case_id", flat=True)
+        disk_ids = os.listdir(settings.TEST_CASE_DIR)
+        test_case_re = re.compile(r"^[a-zA-Z0-9]{32}$")
+        disk_ids = filter(lambda f: test_case_re.match(f), disk_ids)
+        return list(set(disk_ids) - set(db_ids))
+
+    @staticmethod
+    def delete_one(id):
+        test_case_dir = os.path.join(settings.TEST_CASE_DIR, id)
+        if os.path.isdir(test_case_dir):
+            shutil.rmtree(test_case_dir, ignore_errors=True)