codabench/src/apps/api/tests/test_cleanup.py at 9d295aaa651cc899dd9f7060bf61aa6dcb2332d2 · codalab/codabench · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
from rest_framework.test import APITestCase
from django.urls import reverse
import json
from factories import (
    UserFactory,
    CompetitionFactory,
    PhaseFactory,
    TaskFactory,
    SubmissionFactory,
    DataFactory
)
from competitions.models import Submission
from datasets.models import Data


class CleanUpTests(APITestCase):
    def setUp(self):

        # Create a user
        user = UserFactory(username='test_user', password='test_user')

        # Create a competition
        comp = CompetitionFactory(created_by=user)

        # Create used tasks
        self.used_tasks = [
            TaskFactory(created_by=user),
            TaskFactory(created_by=user)
        ]

        # Create unused task
        self.unused_tasks = [
            TaskFactory(created_by=user),
            TaskFactory(created_by=user)
        ]

        # Create phase with used tasks
        phase = PhaseFactory(competition=comp, tasks=self.used_tasks)

        # Create used-failed submission
        self.failed_submissions = [SubmissionFactory(
            phase=phase,
            owner=user,
            status=Submission.FAILED,
            data=DataFactory(created_by=user, type=Data.SUBMISSION, competition=comp)
        )]

        # Create unused submission
        self.unused_submissions = [
            DataFactory(created_by=user, type=Data.SUBMISSION),
            DataFactory(created_by=user, type=Data.SUBMISSION)
        ]

        # Create unused datasets and programs
        self.unused_datasets_programs = [
            DataFactory(created_by=user, type=Data.INGESTION_PROGRAM),
            DataFactory(created_by=user, type=Data.SCORING_PROGRAM),
            DataFactory(created_by=user, type=Data.INPUT_DATA),
            DataFactory(created_by=user, type=Data.REFERENCE_DATA)
        ]

        self.client.login(username='test_user', password='test_user')

    def test_cleanup_stats(self):

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_tasks"] == len(self.unused_tasks)
        assert content["unused_datasets_programs"] == len(self.unused_datasets_programs)
        assert content["unused_submissions"] == len(self.unused_submissions)
        assert content["failed_submissions"] == len(self.failed_submissions)

    def test_delete_unused_tasks(self):

        url = reverse('delete_unused_tasks')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused tasks deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_tasks"] == 0

    def test_delete_unused_datasets(self):

        url = reverse('delete_unused_datasets')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused datasets and programs deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_datasets_programs"] == 0

    def test_delete_unused_submissions(self):

        url = reverse('delete_unused_submissions')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused submissions deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_submissions"] == 0

    def test_delete_failed_submissions(self):

        url = reverse('delete_failed_submissions')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Failed submissions deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["failed_submissions"] == 0