2016-03-07 00:08:57 +00:00
|
|
|
# -*- coding: utf-8 -*-
|
2014-01-05 09:36:22 +00:00
|
|
|
# Copied from: https://github.com/pycontribs/tendo
|
|
|
|
# License: PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2
|
|
|
|
# Author: Sorin Sbarnea
|
|
|
|
# Changes: changed logging and formatting
|
2016-03-07 00:08:57 +00:00
|
|
|
|
2017-05-14 22:54:02 +00:00
|
|
|
from __future__ import absolute_import, division, print_function
|
|
|
|
|
2014-01-05 09:36:22 +00:00
|
|
|
from dosagelib import singleton
|
|
|
|
from multiprocessing import Process
|
|
|
|
|
|
|
|
|
|
|
|
def f(flavor_id):
|
|
|
|
return singleton.SingleInstance(flavor_id=flavor_id, exit_code=1)
|
|
|
|
|
|
|
|
|
2016-03-07 00:08:57 +00:00
|
|
|
class TestSingleton(object):
|
2014-01-05 09:36:22 +00:00
|
|
|
def test_1(self):
|
|
|
|
# test in current process
|
|
|
|
me = singleton.SingleInstance(flavor_id="test-1")
|
|
|
|
del me # now the lock should be removed
|
2016-03-07 00:08:57 +00:00
|
|
|
assert True
|
2014-01-05 09:36:22 +00:00
|
|
|
|
|
|
|
def test_2(self):
|
|
|
|
# test in current subprocess
|
|
|
|
p = Process(target=f, args=("test-2",))
|
|
|
|
p.start()
|
|
|
|
p.join()
|
|
|
|
# the called function should succeed
|
2016-03-07 00:08:57 +00:00
|
|
|
assert p.exitcode == 0
|
2014-01-05 09:36:22 +00:00
|
|
|
|
|
|
|
def test_3(self):
|
|
|
|
# test in current process and subprocess with failure
|
|
|
|
# start first instance
|
|
|
|
me = f("test-3")
|
|
|
|
# second instance
|
|
|
|
p = Process(target=f, args=("test-3",))
|
|
|
|
p.start()
|
|
|
|
p.join()
|
2016-03-07 00:08:57 +00:00
|
|
|
assert p.exitcode == 1
|
2014-01-05 09:36:22 +00:00
|
|
|
# third instance
|
|
|
|
p = Process(target=f, args=("test-3",))
|
|
|
|
p.start()
|
|
|
|
p.join()
|
2016-03-07 00:08:57 +00:00
|
|
|
assert p.exitcode == 1
|
2014-01-05 09:36:22 +00:00
|
|
|
del me # now the lock should be removed
|