Skip to content
Snippets Groups Projects
Select Git revision
  • 4be3f41ccaa8c4757a3601ec422beb09d3e497d7
  • master default protected
  • L2SS-1914-fix_job_dispatch
  • TMSS-3170
  • TMSS-3167
  • TMSS-3161
  • TMSS-3158-Front-End-Only-Allow-Changing-Again
  • TMSS-3133
  • TMSS-3319-Fix-Templates
  • test-fix-deploy
  • TMSS-3134
  • TMSS-2872
  • defer-state
  • add-custom-monitoring-points
  • TMSS-3101-Front-End-Only
  • TMSS-984-choices
  • SDC-1400-Front-End-Only
  • TMSS-3079-PII
  • TMSS-2936
  • check-for-max-244-subbands
  • TMSS-2927---Front-End-Only-PXII
  • Before-Remove-TMSS
  • LOFAR-Release-4_4_318 protected
  • LOFAR-Release-4_4_317 protected
  • LOFAR-Release-4_4_316 protected
  • LOFAR-Release-4_4_315 protected
  • LOFAR-Release-4_4_314 protected
  • LOFAR-Release-4_4_313 protected
  • LOFAR-Release-4_4_312 protected
  • LOFAR-Release-4_4_311 protected
  • LOFAR-Release-4_4_310 protected
  • LOFAR-Release-4_4_309 protected
  • LOFAR-Release-4_4_308 protected
  • LOFAR-Release-4_4_307 protected
  • LOFAR-Release-4_4_306 protected
  • LOFAR-Release-4_4_304 protected
  • LOFAR-Release-4_4_303 protected
  • LOFAR-Release-4_4_302 protected
  • LOFAR-Release-4_4_301 protected
  • LOFAR-Release-4_4_300 protected
  • LOFAR-Release-4_4_299 protected
41 results

test_utils.py

Blame
  • Jorrit Schaap's avatar
    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.
    4be3f41c
    History
    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