test_utils.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 time
import datetime
from multiprocessing import Process, Event
import django

import logging
import typing

logger = logging.getLogger(__name__)
from lofar.sas.tmss.tmss.tmssapp.models import Subtask, SubtaskState

def assertDataWithUrls(self, data, expected):
    """
    object instances get returned as urls, check that the value is part of that url
    """
    # TODO: Make this smarter, this only checks for matching pk!

    from django.db import models

    for k, v in expected.items():
        if isinstance(v, models.Model):
            v = str(v.pk)
            v = v.replace(' ', '%20')
            err_msg = "The value '%s' (key is %s) is not in expected %s" % (str(v), str(data[k]), k)
            self.assertTrue(str(v) in data[k], err_msg)

        elif isinstance(v, datetime.datetime):
            # URL (data[k]) is string but the test_data object (v) is datetime format, convert latter to string format to compare
            self.assertEqual(v.isoformat(), data[k])
        else:
            self.assertEqual(v, data[k])


def assertUrlList(self, url_list, expected_objects):
    """
    object instances get returned as urls, check that the expected projects are in that list
    """

    # TODO: Make this smarter, this only checks for matching pk!

    from django.db import models
    self.assertEqual(len(url_list), len(expected_objects))
    for v in expected_objects:
        if isinstance(v, models.Model):
            v = str(v.pk)
            v = v.replace(' ', '%20')
            self.assertTrue(any(str(v) in myurl for myurl in url_list))
        else:
            raise ValueError('Expected item is not a Django model instance: %s' % v)


def minimal_json_schema(title:str="my title", description:str="my description", id:str="http://example.com/foo/bar.json", properties:dict={}, required=[]):
    return {"$schema": "http://json-schema.org/draft-06/schema#",
            "$id": id,
            "title": title,
            "description": description,
            "type": "object",
            "properties": properties,
            "required": required,
            "default": {}
            }


def set_subtask_state_following_allowed_transitions(subtask: typing.Union[Subtask, int], desired_state_value:str) -> Subtask:
    '''helper function to set subtask state to the desired_state_value following allowed transitions
    Please note that this function is meant to be used in unit/intgration tests only to "simulate" subtask going
    from one state to the desired state, and thus preventing repetitive code like set_state A, set state B ... etc'''
    if isinstance(subtask, int):
        # the given subtask is an id. Fetch object.
        subtask = Subtask.objects.get(id=subtask)

    # end states that we cannot get out of accoring to the design
    END_STATE_VALUES = (SubtaskState.Choices.FINISHED.value, SubtaskState.Choices.UNSCHEDULABLE.value, SubtaskState.Choices.OBSOLETE.value)

    while subtask.state.value != desired_state_value and (subtask.state.value not in END_STATE_VALUES):
        # handle "unsuccessful path" to cancelled/canceling end state
        if desired_state_value in (SubtaskState.Choices.CANCELLED.value, SubtaskState.Choices.CANCELLING.value) and \
                subtask.state.value not in (SubtaskState.Choices.DEFINING.value,
                                            SubtaskState.Choices.QUEUEING.value,
                                            SubtaskState.Choices.STARTING.value,
                                            SubtaskState.Choices.FINISHING.value,
                                            SubtaskState.Choices.CANCELLING.value):
            subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.CANCELLING.value)

        # handle "unsuccessful path" to error end state
        elif desired_state_value == SubtaskState.Choices.ERROR.value and subtask.state.value in (SubtaskState.Choices.DEFINING.value,
                                                                                                 SubtaskState.Choices.QUEUEING.value,
                                                                                                 SubtaskState.Choices.STARTING.value,
                                                                                                 SubtaskState.Choices.FINISHING.value,
                                                                                                 SubtaskState.Choices.CANCELLING.value):
            subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.ERROR.value)

        # handle "unsuccessful path" to OBSOLETE end state (via CANCELLED)
        elif desired_state_value == SubtaskState.Choices.OBSOLETE.value:
            if subtask.state.value == SubtaskState.Choices.DEFINING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.DEFINED.value)
            elif subtask.state.value in (SubtaskState.Choices.DEFINED.value, SubtaskState.Choices.SCHEDULED.value, SubtaskState.Choices.QUEUED.value, SubtaskState.Choices.ERROR.value):
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.CANCELLING.value)
            elif subtask.state.value == SubtaskState.Choices.CANCELLING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.CANCELLED.value)
            elif subtask.state.value == SubtaskState.Choices.CANCELLED.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.OBSOLETE.value)

        # handle "unsuccessful path" to unschedulable end state
        elif desired_state_value == SubtaskState.Choices.UNSCHEDULABLE.value and subtask.state.value == SubtaskState.Choices.SCHEDULING.value:
            subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.UNSCHEDULABLE.value)

        # handle reverse path to unscheduling
        elif desired_state_value == SubtaskState.Choices.UNSCHEDULING.value and subtask.state.value in (SubtaskState.Choices.SCHEDULED.value):
            subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.UNSCHEDULING.value)
        else:
            # handle "normal successful path"
            if subtask.state.value == SubtaskState.Choices.DEFINING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.DEFINED.value)
            elif subtask.state.value == SubtaskState.Choices.DEFINED.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.SCHEDULING.value)
            elif subtask.state.value == SubtaskState.Choices.SCHEDULING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.SCHEDULED.value)
            elif subtask.state.value == SubtaskState.Choices.SCHEDULED.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.QUEUEING.value)
            elif subtask.state.value == SubtaskState.Choices.QUEUEING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.QUEUED.value)
            elif subtask.state.value == SubtaskState.Choices.QUEUED.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.STARTING.value)
            elif subtask.state.value == SubtaskState.Choices.STARTING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.STARTED.value)
            elif subtask.state.value == SubtaskState.Choices.STARTED.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.FINISHING.value)
            elif subtask.state.value == SubtaskState.Choices.FINISHING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.FINISHED.value)
            elif subtask.state.value == SubtaskState.Choices.CANCELLING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.CANCELLED.value)
            elif subtask.state.value == SubtaskState.Choices.UNSCHEDULING.value:
                subtask.state = SubtaskState.objects.get(value=SubtaskState.Choices.DEFINED.value)

        subtask.save()
        # loop, check in while statement at top if we reached the desired state already.

    return subtask