Select Git revision
test_utils.py

TMSS-725: processed review comment. moved...
Jorrit Schaap authored
TMSS-725: processed review comment. moved set_subtask_state_following_allowed_transitions into test_utils so that it is clearly intended to be used in tests only, and not in normal tmss code. Took the opportunity to move the TMSSTestEnvironment into its own module as well to make a clear distinction between test-helper-methods (in test_utils.py) and the TMSSTestEnvironment (in test_environment.py). Adapted all tests with new import paths.
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_utils.py 7.34 KiB
#!/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], state_value:str) -> Subtask:
'''helper function to set subtask state following allowed transitions'''
if isinstance(subtask, int):
# the given subtask is an id. Fetch object.
subtask = Subtask.objects.get(id=subtask)
while subtask.state.value != state_value and (subtask.state.value not in (SubtaskState.Choices.FINISHED.value,
SubtaskState.Choices.ERROR.value,
SubtaskState.Choices.CANCELLED.value)):
# handle "unsuccessful path" to cancelled/canceling end state
if 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 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 reverse path to unscheduling
elif 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()
return subtask