~sirn/fanboi2

ref: 8b56ac65160b4b491c3a1bf0864cb47f9e996cd1 fanboi2/fanboi2/tests/test_tasks.py -rw-r--r-- 5.4 KiB
8b56ac65Kridsada Thanabulpong Fix broken error report due to Celery no longer raise original exceptions. 7 years ago
                                                                                
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
import mock
import transaction
import unittest
from fanboi2.models import DBSession
from fanboi2.tests import ModelMixin, TaskMixin


class TestAddTopicTask(TaskMixin, ModelMixin, unittest.TestCase):

    def _makeOne(self, *args, **kwargs):
        from fanboi2.tasks import add_topic
        return add_topic.delay(*args, **kwargs)

    def test_add_topic(self):
        import transaction
        from fanboi2.models import Topic
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            board_id = board.id  # board is not bound outside transaction!
        result = self._makeOne(request, board_id, 'Foobar', 'Hello, world!')
        topic = DBSession.query(Topic).first()
        self.assertTrue(result.successful())
        self.assertEqual(DBSession.query(Topic).count(), 1)
        self.assertEqual(DBSession.query(Topic).get(result.get()[1]), topic)
        self.assertEqual(topic.title, 'Foobar')
        self.assertEqual(topic.posts[0].body, 'Hello, world!')
        self.assertEqual(result.result, ('topic', topic.id))

    @mock.patch('fanboi2.utils.Akismet.spam')
    def test_add_topic_spam(self, akismet):
        from fanboi2.models import Topic
        akismet.return_value = True
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            board_id = board.id  # board is not bound outside transaction!
        result = self._makeOne(request, board_id, 'Foobar', 'Hello, world!')
        self.assertTrue(result.successful())
        self.assertEqual(DBSession.query(Topic).count(), 0)
        self.assertEqual(result.result, ('failure', 'spam'))

    @mock.patch('fanboi2.utils.Dnsbl.listed')
    def test_add_topic_dnsbl(self, dnsbl):
        from fanboi2.models import Topic
        dnsbl.return_value = True
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            board_id = board.id  # board is not bound outside transaction!
        result = self._makeOne(request, board_id, 'Foobar', 'Hello, world!')
        self.assertTrue(result.successful())
        self.assertEqual(DBSession.query(Topic).count(), 0)
        self.assertEqual(result.result, ('failure', 'dnsbl'))


class TestAddPostTask(TaskMixin, ModelMixin, unittest.TestCase):

    def _makeOne(self, *args, **kwargs):
        from fanboi2.tasks import add_post
        return add_post.delay(*args, **kwargs)

    def test_add_post(self):
        import transaction
        from fanboi2.models import Post
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            topic = self._makeTopic(board=board, title='Hello, world!')
            topic_id = topic.id  # topic is not bound outside transaction!
        result = self._makeOne(request, topic_id, 'Hi!', True)
        post = DBSession.query(Post).first()
        self.assertTrue(result.successful())
        self.assertEqual(DBSession.query(Post).count(), 1)
        self.assertEqual(DBSession.query(Post).get(result.get()[1]), post)
        self.assertEqual(post.body, 'Hi!')
        self.assertEqual(post.bumped, True)
        self.assertEqual(result.result, ('post', post.id))

    @mock.patch('fanboi2.utils.Akismet.spam')
    def test_add_post_spam(self, akismet):
        import transaction
        from fanboi2.models import Post
        akismet.return_value = True
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            topic = self._makeTopic(board=board, title='Hello, world!')
            topic_id = topic.id  # topic is not bound outside transaction!
        result = self._makeOne(request, topic_id, 'Hi!', True)
        self.assertTrue(result.successful())
        self.assertEqual(DBSession.query(Post).count(), 0)
        self.assertEqual(result.result, ('failure', 'spam'))

    def test_add_post_locked(self):
        import transaction
        from fanboi2.models import Post
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            topic = self._makeTopic(
                board=board,
                title='Hello, world!',
                status='locked')
            topic_id = topic.id  # topic is not bound outside transaction!
        result = self._makeOne(request, topic_id, 'Hi!', True)
        self.assertTrue(result.successful())
        self.assertEqual(DBSession.query(Post).count(), 0)
        self.assertEqual(result.result, ('failure', 'locked'))

    def test_add_post_retry(self):
        import transaction
        from sqlalchemy.exc import IntegrityError
        request = {'remote_addr': '127.0.0.1'}
        with transaction.manager:
            board = self._makeBoard(title='Foobar', slug='foobar')
            topic = self._makeTopic(board=board, title='Hello, world!')
            topic_id = topic.id  # topic is not bound outside transaction!
        with mock.patch('fanboi2.models.DBSession.flush') as dbs:
            dbs.side_effect = IntegrityError(None, None, None)
            result = self._makeOne(request, topic_id, 'Hi!', True)
        self.assertEqual(dbs.call_count, 5)
        self.assertFalse(result.successful())