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, DataLocation

test_object_value = dict(obs_id='12345', oid_source='SAS', dataproduct_source='lofar',
                         dataproduct_type='observation',
                         project='LT10_10',
                         activity='observation',
                         surl='srm://surfsara.nl:4884/subfolder/some_nice.tar',
                         filesize=40,
                         additional_meta={'dysco_compression': True})


class TestDatabaseInteraction(dtest.TestCase):
    def test_insert(self):
        location = DataLocation(name='sara', uri='srm://surfsara.nl')
        location.save()
        test_values = dict(test_object_value)
        test_values['location'] = location
        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('Sara', dp.location.name)
        self.assertEqual('surfsara.nl:4884', dp.location.uri)

    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)
        test_payload.pop('location')
        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 = dict(test_object_value)
        test_payload.pop('location')
        response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=[test_payload, test_payload],
                                    format='json')

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

    def test_insert_flat_multi_insert_single(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_201_CREATED, response.status_code)
        self.assertTrue(DataProduct.objects.count() == 1, 'Not all dataproduct have been inserted')
        self.assertTrue(DataLocation.objects.count() == 1, 'Not all dataproduct have been inserted')