bin/appliskel/tests/concurrentSession/TestMinimalExample.py

   1 #!/usr/bin/env python
   2 # Copyright (C) 2013-2014  CEA/DEN, EDF R&D, OPEN CASCADE
   3 #
   4 # This library is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU Lesser General Public
   6 # License as published by the Free Software Foundation; either
   7 # version 2.1 of the License, or (at your option) any later version.
   8 #
   9 # This library is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this library; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  17 #
  18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  19 #
  20
  21 from PortManager import getPort, releasePort, stopServer
  22 import sys
  23 import multiprocessing
  24 import unittest
  25
  26 def port_reservation(prefered=None, test=None, expected=None):
  27   if prefered:
  28     port = getPort(prefered)
  29   else:
  30     port = getPort()
  31   print "port = %s"%port
  32
  33   if expected:
  34     print "expected = %s"%expected
  35     test.assertTrue(port == expected)
  36 #
  37
  38 class TestMinimalExample(unittest.TestCase):
  39   @classmethod
  40   def tearDownClass(cls):
  41     stopServer()
  42     stopServer() # no effect
  43   #
  44   def testSequential(self):
  45     print "BEGIN testSequential"
  46     processes = [
  47       multiprocessing.Process(target=port_reservation)
  48       for i in range(3)
  49       ]
  50
  51     for p in processes:
  52       p.start()
  53
  54     for p in processes:
  55       p.join()
  56
  57     # Try to get specific port number
  58     expected = 2872
  59     p = multiprocessing.Process(target=port_reservation, args=(2872, self,expected,))
  60     p.start()
  61     p.join()
  62
  63     # Try to get specific port number
  64     p = multiprocessing.Process(target=port_reservation, args=(2812, self,))
  65     p.start()
  66     p.join()
  67
  68     # Release port
  69     p = multiprocessing.Process(target=releasePort, args=(2812,))
  70     p.start()
  71     p.join()
  72
  73     # Try to get specific port number
  74     expected = 2812
  75     p = multiprocessing.Process(target=port_reservation, args=(2812, self,expected,))
  76     p.start()
  77     p.join()
  78
  79     print "END testSequential"
  80   #
  81   def testConcurrent(self):
  82     print "BEGIN testConcurrent"
  83     processes = [
  84       multiprocessing.Process(target=port_reservation)
  85       for i in range(3)
  86       ]
  87
  88     # Try to get specific port number
  89     p = multiprocessing.Process(target=port_reservation, args=(2852,))
  90     processes.append(p)
  91
  92     # Try to get specific port number
  93     p = multiprocessing.Process(target=port_reservation, args=(2812,))
  94     processes.append(p)
  95
  96     # Release port
  97     p = multiprocessing.Process(target=releasePort, args=(2812,))
  98     processes.append(p)
  99
 100     # Try to get specific port number
 101     p = multiprocessing.Process(target=port_reservation, args=(2812,))
 102     processes.append(p)
 103
 104     for p in processes:
 105       p.start()
 106
 107     for p in processes:
 108       p.join()
 109
 110     print "END testConcurrent"
 111   #
 112 #
 113
 114 unittest.main()