Source code for apps.api.tests.test_measurement

from rest_framework.test import APITestCase
from rest_framework import status
from datetime import datetime, timedelta
from decimal import Decimal

from apps.core.models.device_type import DeviceType
from apps.core.models.area import Area
from apps.core.models.sensor import SensorEnum
from apps.core.models.measurement import (
    Device,
    Sensor,
    PhysicalUnit
)


[docs]class MeasurementTestCase(APITestCase):
[docs] @classmethod def setUpClass(cls) -> None: cls.device_type = DeviceType.objects.create(type_name='test_device_type', default_version='2021', description='test device type description') cls.area = Area.objects.create(office_name='test_office_name') cls.device = Device.objects.create(type=cls.device_type, battery_state=100, version='ver0', title='test_title', ip_address='2001:0db8:5b96:0000:0000:426f:8e17:642a', area=cls.area) cls.physical_unit = PhysicalUnit.objects.create(unit_name='AIR') cls.sensor = Sensor.objects.create(physical_unit=cls.physical_unit, sensor_type=SensorEnum.AIR)
# POST /measurements test cases
[docs] def test_create_single_measurement(self): data = [{ "device": self.device.id.__str__(), "physical_unit": self.physical_unit.id, "sensor": self.sensor.id.__str__(), "value": "0.00002" }] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data[0]['value'], '0.00002')
[docs] def test_create_multiple_measurements(self): data = [ { "device": self.device.id.__str__(), "physical_unit": self.physical_unit.id, "sensor": self.sensor.id.__str__(), "value": "0.00002" }, { "device": self.device.id.__str__(), "physical_unit": self.physical_unit.id, "sensor": self.sensor.id.__str__(), "value": "0.00004" }, { "device": self.device.id.__str__(), "physical_unit": self.physical_unit.id, "sensor": self.sensor.id.__str__(), "value": "0.00008" }] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data[0]['value'], "0.00002") self.assertEqual(response.data[1]['value'], "0.00004") self.assertEqual(response.data[2]['value'], "0.00008")
[docs] def test_post_empty_measurement_list(self): data = [] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] def test_post_empty_dict_in_measurement_list(self): data = [{}] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] def test_create_measurement_with_invalid_device_id(self): data = [{ "device": "invali_device_id", "physical_unit": self.physical_unit.id, "sensor": self.sensor.id.__str__(), "value": "0.00002" }] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] def test_create_measurement_with_invalid_physical_units_id(self): data = [{ "device": self.device.id.__str__(), "physical_unit": 'invalid_physical_unit_id', "sensor": self.sensor.id.__str__(), "value": "0.00002" }] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] def test_create_measurement_with_invalid_sensor_id(self): data = [{ "device": self.device.id.__str__(), "physical_unit": self.physical_unit.__str__(), "sensor": 'invalid_sensor_id', "value": "0.00002" }] response = self.client.post(f"/api/measurements", data=data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# GET /device/<uuid:device_id>/last_measurement test cases def __creat_measurement(self): data = [{ "device": self.device.id.__str__(), "physical_unit": self.physical_unit.id, "sensor": self.sensor.id.__str__(), "value": "0.00002" }] self.client.post(f"/api/measurements", data=data, format='json')
[docs] def test_get_last_measurement(self): self.__creat_measurement() response = self.client.get(f"/api/devices/{self.device.id}/last_measurement") self.assertEqual(response.status_code, status.HTTP_200_OK)
[docs] def test_get_last_measurement_with_invalid_device_id(self): self.__creat_measurement() invalid_device_id = 'invalid_device_id' response = self.client.get(f"/api/devices/{invalid_device_id}/last_measurement") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
# GET /device/<uuid:device_id>/measurements?time_from=<str:time_from>&time_to=<str:time_to> test cases
[docs] def test_get_measurement_from_period_with_parameters(self): time_to = datetime.now() + timedelta(seconds=10) time_from = time_to - timedelta(hours=10) self.__creat_measurement() url = f"/api/devices/{self.device.id}"\ f"/measurements?time_from={time_from.isoformat()}"\ f"&time_to={time_to.isoformat()}" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 1)
[docs] def test_get_measurement_from_period_without_parameters(self): self.__creat_measurement() url = f"/api/devices/{self.device.id}" \ f"/measurements" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 1)
[docs] def test_get_measurements_list_with_invalid_device_id(self): time_to = datetime.now() + timedelta(seconds=10) time_from = time_to - timedelta(hours=10) invalid_device_id = 'invalid_device_id' url = f"/api/devices/{invalid_device_id}" \ f"/measurements?time_from={time_from.isoformat()}" \ f"&time_to={time_to.isoformat()}" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
[docs] def test_get_measurements_list_with_invalid_time_from_parameter(self): time_to = datetime.now() + timedelta(seconds=10) time_from = 'invalid_time_from' url = f"/api/devices/{self.device.id}" \ f"/measurements?time_from={time_from}" \ f"&time_to={time_to.isoformat()}" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] def test_get_measurements_list_with_invalid_time_to_parameter(self): time_to = 'invalid_time_from' time_from = datetime.now() + timedelta(seconds=10) url = f"/api/devices/{self.device.id}" \ f"/measurements?time_from={time_from.isoformat()}" \ f"&time_to={time_to}" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] def test_get_measurements_list_with_invalid_time_from_bigger_than_time_to(self): time_to = datetime.now() time_from = datetime.now() + timedelta(seconds=10) url = f"/api/devices/{self.device.id}" \ f"/measurements?time_from={time_from.isoformat()}" \ f"&time_to={time_to}" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
[docs] @classmethod def tearDownClass(cls) -> None: cls.sensor.delete() cls.physical_unit.delete() cls.device.delete() cls.area.delete() cls.device_type.delete()