test_dataproduct.py

import django.test as dtest
import rest_framework.test as rtest
from django.contrib.auth.models import User
import rest_framework.status as response_status
from django.db import IntegrityError

from lofardata.models import DataProduct

test_object_value = dict(obs_id='12345', oid_source='SAS', dataproduct_source='lofar',
                         dataproduct_type='observation',
                         project='LT10_10',
                         location='lta-head.lofar.psnc.pl',
                         activity='observation',
                         surl='srm://lta-head.lofar.psnc.pl:4884/subfolder/some_nice.tar',
                         filesize=40,
                         antenna_set="HBA Dual Inner",
                         instrument_filter= "110-190 MHz",
                         dysco_compression=True)


class TestDatabaseInteraction(dtest.TestCase):
    def test_insert(self):
        test_values = dict(test_object_value)
        dp = DataProduct(**test_values)
        dp.save()
        self.assertTrue(dp.pk is not None, 'Failed saving object')
        dp = DataProduct.objects.get(obs_id='12345')
        for field in test_values:
            self.assertEqual(getattr(dp, field), test_values.get(field))

    def test_insert_with_helper_function(self):
        test_values = dict(test_object_value)

        dp = DataProduct.insert_dataproduct(**test_values)

        for field in test_values:
            self.assertEqual(test_object_value.get(field), getattr(dp, field), msg=f'Field {field} does not coincide')

        self.assertEqual('lta-head.lofar.psnc.pl', dp.location)

    def test_insert_non_unique_surl(self):
        test_values = dict(test_object_value)
        DataProduct.insert_dataproduct(**test_values)

        try:
            DataProduct.insert_dataproduct(**test_values)
        except IntegrityError:
            self.fail(
                "Inserting a dataproduct with a surl that's already present raised an Integrity error unexpectedly. It should pass quietly!")


class TestRESTAPI(rtest.APITestCase):
    def setUp(self):
        self.user = User.objects.create_superuser('admin')
        self.client.force_authenticate(self.user)

    def test_insert_not_allowed(self):
        client = rtest.APIClient()
        response = client.post('/ldvspec/api/v1/insert_dataproduct/', data={}, format='json')
        self.assertEqual(response_status.HTTP_403_FORBIDDEN, response.status_code)

    def test_insert_flat_error(self):
        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data={}, format='json')
        self.assertEqual(response_status.HTTP_400_BAD_REQUEST, response.status_code)

    def test_insert_flat_single(self):
        test_payload = dict(test_object_value)
        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')
        self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)

    def test_insert_flat_multi(self):
        test_payload_one = dict(test_object_value)
        test_payload_two = dict(test_object_value)
        test_payload_two['obs_id'] = '9876'
        test_payload_two['surl'] = 'srm://surfsara.nl:4884/x-men/wolverine.tar'

        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=[test_payload_one, test_payload_two],
                                    format='json')

        self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
        self.assertTrue(DataProduct.objects.count() == 2, 'Not all dataproduct have been inserted')

    def test_insert_flat_multi_insert_single(self):
        test_payload = dict(test_object_value)
        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')

        self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
        self.assertTrue(DataProduct.objects.count() == 1, 'Not all dataproduct have been inserted')

    def test_insert_flat_all_valid_locations(self):
        test_payload_one = dict(test_object_value)
        test_payload_two = dict(test_object_value)
        test_payload_three = dict(test_object_value)
        test_payload_four = dict(test_object_value)
        test_payload_one['location'] = 'lta-head.lofar.psnc.pl'
        test_payload_two['location'] = 'srm.grid.sara.nl'
        test_payload_two['surl'] = 'srm://surfsara.nl:4884/subfolder/some_nice_two.tar'
        test_payload_three['location'] = 'lofar-srm.fz-juelich.de'
        test_payload_three['surl'] = 'srm://surfsara.nl:4884/subfolder/some_nice_three.tar'
        test_payload_four['location'] = 'srm.target.rug.nl'
        test_payload_four['surl'] = 'srm://surfsara.nl:4884/subfolder/some_nice_four.tar'

        payload = [test_payload_one, test_payload_two, test_payload_three, test_payload_four]
        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/',
                                    data=payload,
                                    format='json')

        self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
        self.assertTrue(DataProduct.objects.count() == 4, 'Not all dataproduct have been inserted')
        locations = [x['location'] for x in DataProduct.objects.values('location')]
        for pl in payload:
            self.assertTrue(pl['location'] in locations, 'Location not present in the db')

    def test_insert_invalid_location(self):
        test_payload = dict(test_object_value)
        test_payload['location'] = 'srm.grid.invalid.nl'

        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')

        self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
        self.assertTrue(DataProduct.objects.count() == 1, 'The invalid ')
        locations = [x['location'] for x in DataProduct.objects.values('location')]
        self.assertEqual(1, len(locations))
        self.assertEqual(locations[0], 'Unknown')

    def test_insert_missing_location(self):
        test_payload = dict(test_object_value)
        test_payload.pop('location')

        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')

        self.assertEqual(response_status.HTTP_400_BAD_REQUEST, response.status_code)