Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor: singleton is now a context manager #71

Merged
merged 2 commits into from
Oct 11, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 10 additions & 3 deletions src/tendo/singleton.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,9 @@ def __init__(self, flavor_id="", lockfile=""):
self.lockfile = os.path.normpath(
tempfile.gettempdir() + '/' + basename)

logger.debug("SingleInstance lockfile: " + self.lockfile)
logger.debug(f"SingleInstance lockfile: {self.lockfile}")

def __enter__(self):
if sys.platform == 'win32':
try:
# file already exists, we try to remove (in case previous
Expand All @@ -67,10 +69,13 @@ def __init__(self, flavor_id="", lockfile=""):
"Another instance is already running, quitting.")
raise SingleInstanceException()
self.initialized = True
return self

def __del__(self):
def __exit__(self, exc_type, exc_value, exc_tb):
if not self.initialized:
return
if exc_value is not None:
logger.warning("Error: %s" % exc_value, exc_info=True)
try:
if sys.platform == 'win32':
if hasattr(self, 'fd'):
Expand All @@ -85,7 +90,9 @@ def __del__(self):
if logger:
logger.warning(e)
else:
print("Unloggable error: %s" % e)
print(f"Unloggable error: {e}")
if exc_value is not None:
raise e from exc_value
sys.exit(-1)


Expand Down
56 changes: 29 additions & 27 deletions src/tendo/tests/test_singleton.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,46 +13,48 @@ def f(name):
tmp = logger.level
logger.setLevel(logging.CRITICAL) # we do not want to see the warning
try:
me2 = SingleInstance(flavor_id=name) # noqa
with SingleInstance(flavor_id=name):
pass
except SingleInstanceException:
sys.exit(-1)
logger.setLevel(tmp)
pass


def test_1():
me = SingleInstance(flavor_id="test-1")
del me # now the lock should be removed
def test_singleton_from_script():
with SingleInstance(flavor_id="test-1"):
pass
# now the lock should be removed
assert True


def test_2():
def test_singleton_from_process():
p = Process(target=f, args=("test-2",))
p.start()
p.join()
# the called function should succeed
assert p.exitcode == 0, "%s != 0" % p.exitcode


def test_3():
me = SingleInstance(flavor_id="test-3") # noqa -- me should still kept
p = Process(target=f, args=("test-3",))
p.start()
p.join()
# the called function should fail because we already have another
# instance running
assert p.exitcode != 0, "%s != 0 (2nd execution)" % p.exitcode
# note, we return -1 but this translates to 255 meanwhile we'll
# consider that anything different from 0 is good
p = Process(target=f, args=("test-3",))
p.start()
p.join()
# the called function should fail because we already have another
# instance running
assert p.exitcode != 0, "%s != 0 (3rd execution)" % p.exitcode


def test_4():
def test_multiple_singletons_from_process():
with SingleInstance(flavor_id="test-3"):
p = Process(target=f, args=("test-3",))
p.start()
p.join()
# the called function should fail because we already have another
# instance running
assert p.exitcode != 0, "%s != 0 (2nd execution)" % p.exitcode
# note, we return -1 but this translates to 255 meanwhile we'll
# consider that anything different from 0 is good
p = Process(target=f, args=("test-3",))
p.start()
p.join()
# the called function should fail because we already have another
# instance running
assert p.exitcode != 0, "%s != 0 (3rd execution)" % p.exitcode


def test_singleton_lock_file():
lockfile = '/tmp/foo.lock'
me = SingleInstance(lockfile=lockfile)
assert me.lockfile == lockfile
with SingleInstance(lockfile=lockfile) as me:
print(me)
assert me.lockfile == lockfile