t_subtask_validation.py

#!/usr/bin/env python3

# Copyright (C) 2018    ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.    See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.

# $Id:  $

import os
import unittest
import requests

import logging
logger = logging.getLogger(__name__)
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)

# Do Mandatory setup:
# use setup/teardown magic for tmss test database
# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_database_unittest_setup module)
from lofar.sas.tmss.test.tmss_database_unittest_setup import *

from lofar.sas.tmss.tmss.tmssapp import models
from lofar.sas.tmss.tmss.exceptions import SpecificationException

class SubtaskValidationTest(unittest.TestCase):
    @staticmethod
    def create_subtask_template(schema):
        subtask_template_data = SubtaskTemplate_test_data()
        subtask_template_data['schema'] = schema
        return models.SubtaskTemplate.objects.create(**subtask_template_data)

    def test_validate_simple_string_schema_with_valid_specification(self):
        subtask_template = self.create_subtask_template('{"type": "string"}')
        specifications_doc = '"a random string"'
        subtask_data = Subtask_test_data(subtask_template, specifications_doc)

        subtask = models.Subtask.objects.create(**subtask_data)
        self.assertIsNotNone(subtask)

    def test_validate_simple_string_schema_with_invalid_specification(self):
        subtask_template = self.create_subtask_template('{"type": "string"}')
        specifications_doc = '42'
        subtask_data = Subtask_test_data(subtask_template, specifications_doc)

        with self.assertRaises(SpecificationException):
            models.Subtask.objects.create(**subtask_data)


    def test_validate_simple_string_schema_when_updating_valid_to_invalid_specification(self):
        subtask_template = self.create_subtask_template('{"type": "string"}')
        valid_spec = '"a random string"'
        subtask_data = Subtask_test_data(subtask_template, valid_spec)

        subtask = models.Subtask.objects.create(**subtask_data)
        self.assertIsNotNone(subtask)

        # updating the specification with an invalid should fail
        invalid_spec = '42'
        with self.assertRaises(SpecificationException):
            subtask.specifications_doc = invalid_spec
            subtask.save()
        self.assertEqual(invalid_spec, subtask.specifications_doc)

        # revert invalid update, and check
        subtask.refresh_from_db()
        self.assertEqual(valid_spec, subtask.specifications_doc)

    def test_validate_simple_string_schema_with_valid_specification(self):
        subtask_template = self.create_subtask_template('{"type": "string"}')
        specifications_doc = '"a random string"'
        subtask_data = Subtask_test_data(subtask_template, specifications_doc)

        subtask = models.Subtask.objects.create(**subtask_data)
        self.assertIsNotNone(subtask)

    def test_validate_flawed_json_schema(self):
        subtask_template = self.create_subtask_template('{ this is not a json object }')
        specifications_doc = '"a random string"'
        subtask_data = Subtask_test_data(subtask_template, specifications_doc)

        with self.assertRaises(SpecificationException) as context:
            models.Subtask.objects.create(**subtask_data)
        self.assertTrue('invalid json' in str(context.exception).lower())

    def test_validate_flawed_json_specification(self):
        subtask_template = self.create_subtask_template('{"type": "string"}')
        specifications_doc = '{ this is not a json object }'
        subtask_data = Subtask_test_data(subtask_template, specifications_doc)

        with self.assertRaises(SpecificationException) as context:
            models.Subtask.objects.create(**subtask_data)
        self.assertTrue('invalid json' in str(context.exception).lower())


if __name__ == "__main__":
    os.environ['TZ'] = 'UTC'
    unittest.main()