test_wincc_models.py

from django.test import TestCase
from lofar.maintenance.monitoringdb.models.wincc import WinCCAntennaStatus

from datetime import datetime
import django.db.utils as db_utils
import pytz
import random


class WinCCAntennaStatusTest(TestCase):

    def test_antenna_status_insertion_auto_timestamp(self):
        antenna_status = WinCCAntennaStatus()
        antenna_status.station = 'CS001'
        antenna_status.antenna_type = 'HBA'
        antenna_status.antenna_id = 1
        antenna_status.status_code = 0

        before_insertion_timestamp = datetime.now(pytz.UTC)
        antenna_status.save()

        after_insertion_timestamp = datetime.now(pytz.UTC)

        saved_entity = WinCCAntennaStatus.objects.\
            filter(antenna_type=antenna_status.antenna_type, antenna_id=antenna_status.antenna_id).\
            order_by('-timestamp').first()
        self.assertEqual(saved_entity.status_code, antenna_status.status_code)

        self.assertGreaterEqual(saved_entity.timestamp, before_insertion_timestamp)
        self.assertLessEqual(saved_entity.timestamp, after_insertion_timestamp)

    def test_antenna_status_insertion_error_if_not_specified_station(self):
        with self.assertRaises(db_utils.IntegrityError):
            antenna_status = WinCCAntennaStatus()
            antenna_status.antenna_type = 'HBA'
            antenna_status.antenna_id = 1
            antenna_status.status_code = 0

            antenna_status.save()

    def test_antenna_status_insertion_error_if_not_specified_antenna_type(self):
        with self.assertRaises(db_utils.IntegrityError):
            antenna_status = WinCCAntennaStatus()
            antenna_status.station = 'CS001'
            antenna_status.antenna_id = 1
            antenna_status.status_code = 0
            antenna_status.save()

    def test_custom_query(self):
        for i in range(10):
            antenna_status = WinCCAntennaStatus()
            antenna_status.station = 'CS001'
            antenna_status.antenna_type = 'HBA'
            antenna_status.antenna_id = random.sample([1, 2], 1)[0]
            antenna_status.status_code = 0

            antenna_status.save()

        antenna_status = WinCCAntennaStatus()
        antenna_status.station = 'CS001'
        antenna_status.antenna_type = 'HBA'
        antenna_status.antenna_id = 1
        antenna_status.status_code = 0

        antenna_status.save()

        last_status = WinCCAntennaStatus()
        last_status.station = 'CS001'
        last_status.antenna_type = 'HBA'
        last_status.antenna_id = 1
        last_status.status_code = 0

        before_last_insertion_timestamp = datetime.now(pytz.UTC)
        last_status.save()

        saved_entity = WinCCAntennaStatus.objects. \
            filter(antenna_type=antenna_status.antenna_type, antenna_id=antenna_status.antenna_id). \
            order_by('-timestamp').first()
        self.assertGreaterEqual(saved_entity.timestamp, before_last_insertion_timestamp)

        result = WinCCAntennaStatus.objects.\
            latest_status_per_station_and_component_type('CS001',
            'HBA',
            datetime.now())

        self.assertIn('1', result)
        self.assertIn('2', result)
        self.assertGreaterEqual(result['1']["inserted_at"], before_last_insertion_timestamp)