Mitja Nikolaus | 03e412b | 2018-09-18 17:50:15 +0200 | [diff] [blame^] | 1 | """Tests for the heartbeats REST API.""" |
| 2 | |
| 3 | from django.urls import reverse |
| 4 | |
| 5 | from rest_framework import status |
| 6 | from rest_framework.test import APIClient |
| 7 | |
| 8 | from crashreports.tests.utils import HiccupCrashreportsAPITestCase, Dummy |
| 9 | |
| 10 | |
| 11 | class HeartbeatsTestCase(HiccupCrashreportsAPITestCase): |
| 12 | """Test cases for heartbeats.""" |
| 13 | |
| 14 | LIST_CREATE_URL = "api_v1_heartbeats" |
| 15 | RETRIEVE_URL = "api_v1_heartbeat" |
| 16 | LIST_CREATE_BY_UUID_URL = "api_v1_heartbeats_by_uuid" |
| 17 | RETRIEVE_BY_UUID_URL = "api_v1_heartbeat_by_uuid" |
| 18 | |
| 19 | @staticmethod |
| 20 | def _create_dummy_data(**kwargs): |
| 21 | return Dummy.heartbeat_data(**kwargs) |
| 22 | |
| 23 | def _post_multiple(self, client, data, count): |
| 24 | return [ |
| 25 | client.post(reverse(self.LIST_CREATE_URL), data) |
| 26 | for _ in range(count) |
| 27 | ] |
| 28 | |
| 29 | def _retrieve_single(self, user): |
| 30 | count = 5 |
| 31 | response = self._post_multiple(self.admin, self.data, count) |
| 32 | self.assertEqual(len(response), count) |
| 33 | self.assertEqual(response[0].status_code, status.HTTP_201_CREATED) |
| 34 | url = reverse(self.RETRIEVE_URL, args=[response[0].data["id"]]) |
| 35 | request = user.get(url) |
| 36 | return request.status_code |
| 37 | |
| 38 | def _retrieve_single_by_device(self, user): |
| 39 | count = 5 |
| 40 | response = self._post_multiple(self.user, self.data, count) |
| 41 | self.assertEqual(len(response), count) |
| 42 | self.assertEqual(response[0].status_code, status.HTTP_201_CREATED) |
| 43 | url = reverse( |
| 44 | self.RETRIEVE_BY_UUID_URL, |
| 45 | args=[self.uuid, response[0].data["device_local_id"]], |
| 46 | ) |
| 47 | request = user.get(url) |
| 48 | return request.status_code |
| 49 | |
| 50 | def setUp(self): |
| 51 | """Set up a device and some data.""" |
| 52 | super().setUp() |
| 53 | self.uuid, self.user, self.token = self._register_device() |
| 54 | self.data = self._create_dummy_data(uuid=self.uuid) |
| 55 | |
| 56 | def test_create_no_auth(self): |
| 57 | """Test creation without authentication.""" |
| 58 | noauth_client = APIClient() |
| 59 | response = noauth_client.post(reverse(self.LIST_CREATE_URL), self.data) |
| 60 | self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
| 61 | |
| 62 | def test_create_as_admin(self): |
| 63 | """Test creation as admin.""" |
| 64 | response = self.admin.post(reverse(self.LIST_CREATE_URL), self.data) |
| 65 | self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
| 66 | self.assertTrue(response.data["id"] > 0) |
| 67 | |
| 68 | def test_create_as_admin_not_existing_device(self): |
| 69 | """Test creation of heartbeat on non-existing device.""" |
| 70 | response = self.admin.post( |
| 71 | reverse(self.LIST_CREATE_URL), self._create_dummy_data() |
| 72 | ) |
| 73 | self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) |
| 74 | |
| 75 | def test_create_as_uuid_owner(self): |
| 76 | """Test creation as owner.""" |
| 77 | response = self.user.post( |
| 78 | reverse(self.LIST_CREATE_URL), |
| 79 | self._create_dummy_data(uuid=self.uuid), |
| 80 | ) |
| 81 | self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
| 82 | self.assertEqual(response.data["id"], -1) |
| 83 | |
| 84 | def test_create_as_uuid_not_owner(self): |
| 85 | """Test creation as non-owner.""" |
| 86 | uuid, _, _ = self._register_device() |
| 87 | response = self.user.post( |
| 88 | reverse(self.LIST_CREATE_URL), self._create_dummy_data(uuid=uuid) |
| 89 | ) |
| 90 | self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) |
| 91 | |
| 92 | def test_list(self): |
| 93 | """Test listing of heartbeats.""" |
| 94 | count = 5 |
| 95 | self._post_multiple(self.user, self.data, count) |
| 96 | response = self.admin.get(reverse(self.LIST_CREATE_URL)) |
| 97 | self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 98 | self.assertEqual(len(response.data["results"]), count) |
| 99 | |
| 100 | def test_retrieve_single_admin(self): |
| 101 | """Test retrieval as admin.""" |
| 102 | self.assertEqual(self._retrieve_single(self.admin), status.HTTP_200_OK) |
| 103 | |
| 104 | def test_retrieve_single_device_owner(self): |
| 105 | """Test retrieval as device owner.""" |
| 106 | self.assertEqual( |
| 107 | self._retrieve_single(self.user), status.HTTP_403_FORBIDDEN |
| 108 | ) |
| 109 | |
| 110 | def test_retrieve_single_noauth(self): |
| 111 | """Test retrieval without authentication.""" |
| 112 | noauth_client = APIClient() |
| 113 | self.assertEqual( |
| 114 | self._retrieve_single(noauth_client), status.HTTP_401_UNAUTHORIZED |
| 115 | ) |
| 116 | |
| 117 | def test_retrieve_single_by_device_admin(self): |
| 118 | """Test retrieval by device as admin.""" |
| 119 | self.assertEqual( |
| 120 | self._retrieve_single_by_device(self.admin), status.HTTP_200_OK |
| 121 | ) |
| 122 | |
| 123 | def test_retrieve_single_by_device_device_owner(self): |
| 124 | """Test retrieval by device as owner.""" |
| 125 | self.assertEqual( |
| 126 | self._retrieve_single_by_device(self.user), |
| 127 | status.HTTP_403_FORBIDDEN, |
| 128 | ) |
| 129 | |
| 130 | def test_retrieve_single_by_device_noauth(self): |
| 131 | """Test retrieval by device without authentication.""" |
| 132 | noauth_client = APIClient() |
| 133 | self.assertEqual( |
| 134 | self._retrieve_single_by_device(noauth_client), |
| 135 | status.HTTP_401_UNAUTHORIZED, |
| 136 | ) |
| 137 | |
| 138 | def test_list_by_uuid(self): |
| 139 | """Test listing of devices by UUID.""" |
| 140 | count = 5 |
| 141 | uuid, _, _ = self._register_device() |
| 142 | self._post_multiple(self.user, self.data, count) |
| 143 | self._post_multiple( |
| 144 | self.admin, self._create_dummy_data(uuid=uuid), count |
| 145 | ) |
| 146 | url = reverse(self.LIST_CREATE_BY_UUID_URL, args=[self.uuid]) |
| 147 | response = self.admin.get(url) |
| 148 | self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 149 | self.assertEqual(len(response.data["results"]), count) |
| 150 | |
| 151 | def test_list_noauth(self): |
| 152 | """Test listing of devices without authentication.""" |
| 153 | count = 5 |
| 154 | noauth_client = APIClient() |
| 155 | self._post_multiple(self.user, self.data, count) |
| 156 | response = noauth_client.get(reverse(self.LIST_CREATE_URL)) |
| 157 | self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
| 158 | |
| 159 | def test_list_device_owner(self): |
| 160 | """Test listing as device owner.""" |
| 161 | count = 5 |
| 162 | self._post_multiple(self.user, self.data, count) |
| 163 | response = self.user.get(reverse(self.LIST_CREATE_URL)) |
| 164 | self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) |
| 165 | |
| 166 | def test_no_radio_version(self): |
| 167 | """Test creation and retrieval without radio version.""" |
| 168 | data = self._create_dummy_data(uuid=self.uuid) |
| 169 | data.pop("radio_version") |
| 170 | response = self.user.post(reverse(self.LIST_CREATE_URL), data) |
| 171 | self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
| 172 | url = reverse(self.LIST_CREATE_BY_UUID_URL, args=[self.uuid]) |
| 173 | response = self.admin.get(url) |
| 174 | self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 175 | self.assertEqual(len(response.data["results"]), 1) |
| 176 | self.assertIsNone(response.data["results"][0]["radio_version"]) |
| 177 | |
| 178 | def test_radio_version_field(self): |
| 179 | """Test retrieval of radio version field.""" |
| 180 | response = self.user.post(reverse(self.LIST_CREATE_URL), self.data) |
| 181 | self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
| 182 | url = reverse(self.LIST_CREATE_BY_UUID_URL, args=[self.uuid]) |
| 183 | response = self.admin.get(url) |
| 184 | self.assertEqual(response.status_code, status.HTTP_200_OK) |
| 185 | self.assertEqual(len(response.data["results"]), 1) |
| 186 | self.assertEqual( |
| 187 | response.data["results"][0]["radio_version"], |
| 188 | self.data["radio_version"], |
| 189 | ) |
| 190 | |
| 191 | def test_send_non_existent_time(self): |
| 192 | """Test sending of heartbeat with non existent time. |
| 193 | |
| 194 | Test the resolution of a naive date-time in which the |
| 195 | Europe/Amsterdam daylight saving time transition moved the time |
| 196 | "forward". |
| 197 | """ |
| 198 | data = self._create_dummy_data(uuid=self.uuid) |
| 199 | # In 2017, the Netherlands changed from CET to CEST on March, |
| 200 | # 26 at 02:00 |
| 201 | data["date"] = "2017-03-26 02:34:56" |
| 202 | response = self.user.post(reverse(self.LIST_CREATE_URL), data) |
| 203 | self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
| 204 | |
| 205 | def test_send_ambiguous_time(self): |
| 206 | """Test sending of heartbeat with ambiguous time. |
| 207 | |
| 208 | Test the resolution of a naive date-time in which the |
| 209 | Europe/Amsterdam daylight saving time transition moved the time |
| 210 | "backward". |
| 211 | """ |
| 212 | data = self._create_dummy_data(uuid=self.uuid) |
| 213 | # In 2017, the Netherlands changed from CEST to CET on October, |
| 214 | # 29 at 03:00 |
| 215 | data["date"] = "2017-10-29 02:34:56" |
| 216 | response = self.user.post(reverse(self.LIST_CREATE_URL), data) |
| 217 | self.assertEqual(response.status_code, status.HTTP_201_CREATED) |