2 # Copyright (C) 2020 Jessica Tallon & Matt Molyneaux
4 # This file is part of Inboxen.
6 # Inboxen is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
11 # Inboxen is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
16 # You should have received a copy of the GNU Affero General Public License
17 # along with Inboxen. If not, see <http://www.gnu.org/licenses/>.
20 from datetime
import timedelta
21 from unittest
import mock
23 from django
.conf
import settings
24 from django
.urls
import reverse
25 from django
.utils
import timezone
27 from inboxen
.monitor
.models
import CheckItem
28 from inboxen
.monitor
.tasks
import check_tasks
29 from inboxen
.test
import InboxenTestCase
32 class CheckModelTestCase(InboxenTestCase
):
33 def test_creation(self
):
34 check
= CheckItem
.objects
.create_check(CheckItem
.SALMON
)
35 self
.assertEqual(check
.good
, True)
36 self
.assertEqual(check
.check_type
, CheckItem
.SALMON
)
38 def test_creation_bad_check(self
):
39 with self
.assertRaises(ValueError):
40 CheckItem
.objects
.create_check("test")
42 def test_creation_invalid_check(self
):
43 with self
.assertRaises(ValueError):
44 CheckItem
.objects
.create_check(120)
46 def test_check_not_today(self
):
47 check
= CheckItem
.objects
.create(check_type
=CheckItem
.SALMON
)
48 check
.when
= timezone
.now() - timedelta(days
=2)
50 self
.assertEqual(CheckItem
.objects
.check_ok(CheckItem
.SALMON
), False)
52 def test_check_not_good(self
):
53 CheckItem
.objects
.create(check_type
=CheckItem
.SALMON
, good
=False)
54 self
.assertEqual(CheckItem
.objects
.check_ok(CheckItem
.SALMON
), False)
56 def test_check_wrong_type(self
):
57 CheckItem
.objects
.create(check_type
=CheckItem
.CELERY
)
58 self
.assertEqual(CheckItem
.objects
.check_ok(CheckItem
.SALMON
), False)
60 def test_check_good(self
):
61 CheckItem
.objects
.create_check(CheckItem
.SALMON
)
62 self
.assertEqual(CheckItem
.objects
.check_ok(CheckItem
.SALMON
), True)
65 class CheckTaskTestCase(InboxenTestCase
):
67 check_tasks
.apply_async()
68 self
.assertEqual(CheckItem
.objects
.check_ok(CheckItem
.CELERY
), True)
71 self
.assertEqual(settings
.CELERY_BEAT_SCHEDULE
["monitor"]["task"], "inboxen.monitor.tasks.check_tasks")
74 class CheckSalmonViewTestCase(InboxenTestCase
):
76 CheckItem
.objects
.create_check(CheckItem
.SALMON
)
77 response
= self
.client
.get(reverse("monitor:salmon"))
78 self
.assertEqual(response
.status_code
, 200)
81 CheckItem
.objects
.create_check(CheckItem
.CELERY
)
82 response
= self
.client
.get(reverse("monitor:salmon"))
83 self
.assertEqual(response
.status_code
, 404)
86 class CheckCeleryViewTestCase(InboxenTestCase
):
87 @mock.patch("inboxen.monitor.views.app")
88 def test_no_rabbit(self
, app_mock
):
89 app_mock
.control
.broadcast
.side_effect
= Exception
90 CheckItem
.objects
.create_check(CheckItem
.CELERY
)
91 response
= self
.client
.get(reverse("monitor:celery"))
92 self
.assertEqual(response
.status_code
, 500)
94 @mock.patch("inboxen.monitor.views.app")
95 def test_no_workers(self
, app_mock
):
96 app_mock
.control
.broadcast
.return_value
= None
97 CheckItem
.objects
.create_check(CheckItem
.CELERY
)
98 response
= self
.client
.get(reverse("monitor:celery"))
99 self
.assertEqual(response
.status_code
, 502)
101 @mock.patch("inboxen.monitor.views.app")
102 def test_bad(self
, app_mock
):
103 app_mock
.control
.broadcast
.return_value
= {}
104 CheckItem
.objects
.create_check(CheckItem
.SALMON
)
105 response
= self
.client
.get(reverse("monitor:celery"))
106 self
.assertEqual(response
.status_code
, 404)
108 @mock.patch("inboxen.monitor.views.app")
109 def test_good(self
, app_mock
):
110 app_mock
.control
.broadcast
.return_value
= {}
111 CheckItem
.objects
.create_check(CheckItem
.CELERY
)
112 response
= self
.client
.get(reverse("monitor:celery"))
113 self
.assertEqual(response
.status_code
, 200)