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()