Source Code for Module eoxserver.resources.processes.tests

  1  #----------------------------------------------------------------------- 
  2  # $Id: tests.py 2395 2013-03-22 18:13:55Z martin.paces $ 
  3  # 
  4  # Description:  
  5  # 
  6  #  unit test for asynchronous process API  
  7  # 
  8  #------------------------------------------------------------------------------- 
  9  # 
 10  # Project: EOxServer <http://eoxserver.org> 
 11  # Authors: Petr Blaha <petr.blaha@iguassu.cz> 
 12  # 
 13  #------------------------------------------------------------------------------- 
 14  # Copyright (C) 2011 Iguassu Software Systems, a.s  
 15  # 
 16  # Permission is hereby granted, free of charge, to any person obtaining a copy 
 17  # of this software and associated documentation files (the "Software"), to deal 
 18  # in the Software without restriction, including without limitation the rights 
 19  # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell  
 20  # copies of the Software, and to permit persons to whom the Software is  
 21  # furnished to do so, subject to the following conditions: 
 22  # 
 23  # The above copyright notice and this permission notice shall be included in all 
 24  # copies of this Software or works derived from this Software. 
 25  # 
 26  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 
 27  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 
 28  # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 
 29  # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 
 30  # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 
 31  # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 
 32  # THE SOFTWARE. 
 33  #------------------------------------------------------------------------------- 
 34   
 35  import uuid  
 36  import random 
 37  import unittest 
 38  import time 
 39   
 40  #---------------------------------------------------------------------- 
 41  # imports  
 42  from django.db import IntegrityError 
 43  from django.db.models import ProtectedError 
 44  from django.core.exceptions import ObjectDoesNotExist  
 45  from eoxserver.resources.processes.tracker import * 
 46   
 47  PROCESS_CLASS="TEST-PROCESS" 
 48  FASTER_CLASS="FAST-PROCESS" 
 49  ASYNC_HANDLER="apt_test.testHandler" 
 50  ASYNC_TIMEOUT=60 
 51  ASYNC_TIMERET=60 
 52  SERVER_ID=1 
 53  ID1="1111111" 
 54  ID2="2222222" 
 55  ID3="3333333" 
 56  ID4="4444444" 
 57  ID5="5555555" 
 58  ID6="6666666" 
 59  ID7="7777777" 
 60  ID8="8888888" 
 61  PARAM="1234568"  
 62  inputs={ 'key':'value' } 
 63   
 64 -class TestSequenceFunctions(unittest.TestCase): 
 65       
 66  #    def setUp(self): 
 67   
 68   
 69 -    def test_unregistred_protected(self): 
 70          registerTaskType( PROCESS_CLASS, ASYNC_HANDLER, ASYNC_TIMEOUT, ASYNC_TIMERET ) 
 71   
 72          # enqueue task for execution  
 73          enqueueTask( PROCESS_CLASS, ID1, inputs, PARAM )         
 74   
 75          # can not  unregisterTaskType if task exist 
 76          # self.assertRaises( TaskTypeHasInstances, unregisterTaskType, PROCESS_CLASS, True ) 
 77          self.assertRaises( ProtectedError, unregisterTaskType, PROCESS_CLASS ) 
 78          deleteTaskByIdentifier( PROCESS_CLASS, ID1 ) 
 79   
 80   
 81 -    def test_unregistred_taskType_force(self): 
 82          registerTaskType( PROCESS_CLASS, ASYNC_HANDLER, ASYNC_TIMEOUT, ASYNC_TIMERET ) 
 83   
 84          # enqueue task for execution  
 85          enqueueTask( PROCESS_CLASS, ID2, inputs, PARAM )         
 86   
 87          # force  unregisterTaskType 
 88          unregisterTaskType( PROCESS_CLASS, True ) 
 89          # delete  task 
 90          #deleteTaskByIdentifier( PROCESS_CLASS, ID2 ) 
 91   
 92   
 93 -    def test_unregistred_not_registred(self): 
 94          self.assertRaises( ObjectDoesNotExist, unregisterTaskType, "TEST-PROCESS-NOT_REGISTRED", True ) 
 95   
 96   
 97 -    def test_enqueue_not_registred(self): 
 98          # enqueue task for execution  
 99          self.assertRaises( ObjectDoesNotExist, enqueueTask, "TEST-PROCESS-NOT_REGISTRED", ID3, inputs, PARAM )         
100   
101   
102 -    def test_unregistred_taskType(self): 
103          registerTaskType( PROCESS_CLASS, ASYNC_HANDLER, ASYNC_TIMEOUT, ASYNC_TIMERET ) 
104   
105          # enqueue task for execution  
106          enqueueTask( PROCESS_CLASS, ID2, inputs, PARAM )         
107   
108          deleteTaskByIdentifier( PROCESS_CLASS, ID2) 
109          unregisterTaskType( PROCESS_CLASS ) 
110   
111   
112 -    def test_dequeue_more_task(self): 
113          registerTaskType( PROCESS_CLASS, ASYNC_HANDLER, ASYNC_TIMEOUT, ASYNC_TIMERET ) 
114   
115          # enue two task and dequee three  
116          # enqueue task for execution  
117          enqueueTask( PROCESS_CLASS, ID1, inputs, PARAM )         
118          enqueueTask( PROCESS_CLASS, ID2, inputs, PARAM )         
119   
120          dequeueTask( SERVER_ID )  
121          dequeueTask( SERVER_ID ) 
122   
123          # test for queue empty  
124          self.assertRaises( QueueEmpty, dequeueTask, SERVER_ID ) 
125          # delete  tasks 
126          deleteTaskByIdentifier( PROCESS_CLASS, ID2 ) 
127          deleteTaskByIdentifier( PROCESS_CLASS, ID1 ) 
128   
129   
130 -    def test_full_empty_queue(self): 
131          # enue more task then  QUEUE_SIZE 
132          for t in range ( 0, getMaxQueueSize() ) : 
133              enqueueTask( PROCESS_CLASS, t, inputs, PARAM ) 
134          # test for queue full 
135   
136          self.assertRaises( QueueFull, enqueueTask, PROCESS_CLASS, t+1, inputs, PARAM )  
137   
138          # clean tasks 
139          for t in range ( 0, getMaxQueueSize() ) : deleteTaskByIdentifier( PROCESS_CLASS, t ) 
140   
141          # test for queue empty  
142          self.assertRaises( QueueEmpty, dequeueTask, SERVER_ID ) 
143   
144   
145 -    def test_start_task(self): 
146          enqueueTask( PROCESS_CLASS, ID3, inputs, PARAM ) 
147          id = dequeueTask( SERVER_ID ) 
148          start = startTask( id[0] ) 
149   
150          # get the task ID and compare 
151          self.assertEqual( int( str( start[1] ) ), int( ID3 ) ) 
152          # delete task  
153          deleteTaskByIdentifier( PROCESS_CLASS, ID3 ) 
154   
155   
156      #  test both setTaskResponse and  getTaskResponse 
157 -    def test_responses(self): 
158          enqueueTask( PROCESS_CLASS, ID4, inputs, PARAM ) 
159          id = dequeueTask( SERVER_ID ) 
160          setTaskResponse( id[0], "123" )  
161   
162          response = getTaskResponse( PROCESS_CLASS, ID4 )[0] 
163          self.assertEqual( int( response ), 123 ) 
164          deleteTaskByIdentifier( PROCESS_CLASS, ID4 ) 
165   
166   
167 -    def test_task_duplicity(self): 
168   
169          enqueueTask( PROCESS_CLASS, ID5, inputs, PARAM ) 
170          # again 
171          self.assertRaises( IntegrityError, enqueueTask, PROCESS_CLASS, ID5, inputs, PARAM ) 
172   
173          deleteTaskByIdentifier( PROCESS_CLASS, ID5 ) 
174   
175   
176 -    def test_running_task(self): 
177          enqueueTask( PROCESS_CLASS, ID6, inputs, PARAM ) 
178          id = dequeueTask( SERVER_ID ) 
179          start = startTask( id[0] ) 
180   
181          # inspection  
182          status =  getTaskStatus( id[0] )  
183   
184          # status is  3:"RUNNING" 
185          self.assertEqual( status[0], 3 ) 
186          deleteTaskByIdentifier( PROCESS_CLASS, ID6 ) 
187   
188   
189 -    def test_reenqueue_task(self): 
190          enqueueTask( PROCESS_CLASS, ID7, inputs, PARAM ) 
191          id = dequeueTask( SERVER_ID ) 
192          reenqueueTask( id[0], "Run now" ) 
193          deleteTaskByIdentifier( PROCESS_CLASS, ID7 ) 
194   
195   
196 -    def test_three_task_zombie(self): 
197          registerTaskType( FASTER_CLASS, 2, 2 ) 
198          enqueueTask( FASTER_CLASS, ID1, inputs, PARAM )         
199          enqueueTask( FASTER_CLASS, ID2, inputs, PARAM )         
200          enqueueTask( FASTER_CLASS, ID3, inputs, PARAM )         
201   
202          id1 = dequeueTask( SERVER_ID ) 
203          id2 = dequeueTask( SERVER_ID ) 
204          id3 = dequeueTask( SERVER_ID ) 
205   
206          # first Task 
207          s2 =  getTaskStatus( id1[0] ) 
208   
209          # second Task is starting 
210          startTask( id2[0], "start running first" ) 
211          s2 =  getTaskStatus( id2[0] ) 
212   
213          # third Task is starting 
214          pauseTask( id3[0], "start running third" ) 
215          s3 =  getTaskStatus( id3[0] ) 
216   
217          time.sleep( 3 ) 
218          reenqueueZombieTasks( "Up" )  
219   
220          s1 =  getTaskStatus( id1[0] ) 
221          s2 =  getTaskStatus( id2[0] ) 
222          s3 =  getTaskStatus( id3[0] ) 
223   
224          # must be 1, 'ACCEPTED'  
225          self.assertEqual( int(s1[0]) +  int(s2[0]) +  int(s3[0]), 3 )         
226   
227          deleteTaskByIdentifier( PROCESS_CLASS, ID1 ) 
228          deleteTaskByIdentifier( PROCESS_CLASS, ID2 ) 
229          deleteTaskByIdentifier( PROCESS_CLASS, ID3 ) 
230   
231   
232 -    def test_set_task_status_failed(self): 
233          enqueueTask( PROCESS_CLASS, ID7, inputs, PARAM ) 
234          id = dequeueTask( SERVER_ID ) 
235          reenqueueTask( id[0], "Run now" ) 
236       
237          # start Task 
238          startTask( id[0], "start running" ) 
239          # pause task  
240          pauseTask( id[0] ) 
241   
242          taskStatus= TaskStatus( id[0] )  
243          taskStatus.setFailure( "failed" );  
244   
245          s = taskStatus.getStatus() 
246   
247          # must be 6, 'FAILED'  
248          self.assertEqual( int(s[0]), 6 )         
249   
250          deleteTaskByIdentifier( PROCESS_CLASS, ID7 ) 
251   
252   
253 -    def test_delete_retired_task(self): 
254          registerTaskType( FASTER_CLASS, ASYNC_HANDLER, 2, 2 ) 
255          enqueueTask( FASTER_CLASS, ID8, inputs, PARAM )         
256   
257          id = dequeueTask( SERVER_ID ) 
258   
259          # second Task is starting 
260          startTask( id[0], "go" ) 
261   
262          taskStatus= TaskStatus( id[0] )  
263          taskStatus.setSuccess( "OK" );  
264   
265          time.sleep( 5 ) 
266   
267          r = deleteRetiredTasks() 
268          
269          self.assertRaises( ObjectDoesNotExist, taskStatus.getStatus ) 
270   
271          deleteTaskByIdentifier( FASTER_CLASS, ID8 ) 
272