test_import_tasks.py
6.73 KB
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
# -*- coding: utf8 -*-
# This file is part of PyBossa.
#
# Copyright (C) 2015 SciFabric LTD.
#
# PyBossa is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# PyBossa is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with PyBossa. If not, see <http://www.gnu.org/licenses/>.
from default import Test, with_context, flask_app
from pybossa.jobs import import_tasks, task_repo, get_autoimport_jobs
from pybossa.model.task import Task
from pybossa.importers import ImportReport
from factories import ProjectFactory, TaskFactory, UserFactory
from mock import patch
class TestImportTasksJob(Test):
@with_context
@patch('pybossa.jobs.importer.create_tasks')
def test_it_creates_the_new_tasks(self, create):
project = ProjectFactory.create()
form_data = {'type': 'csv', 'csv_url': 'http://google.es'}
import_tasks(project.id, **form_data)
create.assert_called_once_with(task_repo, project.id, **form_data)
@with_context
@patch('pybossa.jobs.send_mail')
@patch('pybossa.jobs.importer.create_tasks')
def test_sends_email_to_user_with_result_on_success(self, create, send_mail):
create.return_value = ImportReport(message='1 new task was imported successfully', metadata=None, total=1)
project = ProjectFactory.create()
form_data = {'type': 'csv', 'csv_url': 'http://google.es'}
subject = 'Tasks Import to your project %s' % project.name
body = 'Hello,\n\n1 new task was imported successfully to your project %s!\n\nAll the best,\nThe PyBossa team.' % project.name
email_data = dict(recipients=[project.owner.email_addr],
subject=subject, body=body)
import_tasks(project.id, **form_data)
send_mail.assert_called_once_with(email_data)
@with_context
@patch('pybossa.jobs.send_mail')
@patch('pybossa.jobs.importer.create_tasks')
def test_it_adds_import_metadata_to_autoimporter_if_is_autoimport_job(self, create, send_mail):
create.return_value = ImportReport(message='1 new task was imported successfully', metadata="meta", total=1)
form_data = {'type': 'csv', 'csv_url': 'http://google.es'}
project = ProjectFactory.create(info=dict(autoimporter=form_data))
subject = 'Tasks Import to your project %s' % project.name
body = 'Hello,\n\n1 new task was imported successfully to your project %s!\n\nAll the best,\nThe PyBossa team.' % project.name
email_data = dict(recipients=[project.owner.email_addr],
subject=subject, body=body)
import_tasks(project.id, from_auto=True, **form_data)
autoimporter = project.get_autoimporter()
assert autoimporter.get('last_import_meta') == 'meta', autoimporter
@with_context
@patch('pybossa.jobs.send_mail')
@patch('pybossa.jobs.importer.create_tasks')
def test_it_does_not_add_import_metadata_to_autoimporter_if_is_import_job(self, create, send_mail):
create.return_value = ImportReport(message='1 new task was imported successfully', metadata="meta", total=1)
form_data = {'type': 'csv', 'csv_url': 'http://google.es'}
project = ProjectFactory.create(info=dict(autoimporter=form_data))
subject = 'Tasks Import to your project %s' % project.name
body = 'Hello,\n\n1 new task was imported successfully to your project %s!\n\nAll the best,\nThe PyBossa team.' % project.name
email_data = dict(recipients=[project.owner.email_addr],
subject=subject, body=body)
import_tasks(project.id, from_auto=False, **form_data)
autoimporter = project.get_autoimporter()
assert autoimporter.get('last_import_meta') == None, autoimporter
class TestAutoimportJobs(Test):
@with_context
def test_autoimport_jobs_no_autoimporter(self):
"""Test JOB autoimport does not return projects without autoimporter."""
user = UserFactory.create(pro=True)
ProjectFactory.create(owner=user)
jobs_generator = get_autoimport_jobs()
jobs = []
for job in jobs_generator:
jobs.append(job)
msg = "There should be 0 jobs."
assert len(jobs) == 0, msg
@with_context
def test_autoimport_jobs_with_autoimporter(self):
"""Test JOB autoimport jobs returns projects with autoimporter."""
user = UserFactory.create(pro=True)
project = ProjectFactory.create(owner=user,info=dict(autoimporter='foobar'))
jobs_generator = get_autoimport_jobs()
jobs = []
for job in jobs_generator:
jobs.append(job)
msg = "There should be 1 job."
assert len(jobs) == 1, msg
job = jobs[0]
msg = "It sould be the same project."
assert job['args'][0] == project.id, msg
msg = "It sould be created as an auto import job."
assert job['args'][1] == True, msg
msg = "There sould be the kwargs."
assert job['kwargs'] == 'foobar', msg
@with_context
@patch.dict(flask_app.config, {'PRO_FEATURES': {'autoimporter': True}})
def test_autoimport_jobs_without_pro_when_only_pro(self):
"""Test JOB autoimport jobs does not return normal user owned projects
if autoimporter is only enabled for pros."""
ProjectFactory.create(info=dict(autoimporter='foobar'))
jobs_generator = get_autoimport_jobs()
jobs = []
for job in jobs_generator:
jobs.append(job)
msg = "There should be 0 jobs."
assert len(jobs) == 0, msg
@with_context
@patch.dict(flask_app.config, {'PRO_FEATURES': {'autoimporter': False}})
def test_autoimport_jobs_without_pro_when_for_everyone(self):
"""Test JOB autoimport jobs returns normal user owned projects
if autoimporter is enabled for everyone."""
project = ProjectFactory.create(info=dict(autoimporter='foobar'))
jobs_generator = get_autoimport_jobs()
jobs = []
for job in jobs_generator:
jobs.append(job)
msg = "There should be 1 job."
assert len(jobs) == 1, msg
job = jobs[0]
msg = "It sould be the same project."
assert job['args'][0] == project.id, msg
msg = "It sould be created as an auto import job."
assert job['args'][1] == True, msg
msg = "There sould be the kwargs."
assert job['kwargs'] == 'foobar', msg