test case

test-lock.py (text/x-python), 1.38 KB, created by Zac Medico on 2019-02-05 01:29:46 UTC

#!/usr/bin/env python
# Test case for https://bugs.gentoo.org/636798

import argparse
from contextlib import contextmanager
import logging
import os
import platform


from portage.locks import (
	lockfile,
	unlockfile,
)

@contextmanager
def default(lock_path):
	lock = lockfile(lock_path, wantnewlockfile=1)
	try:
		yield
	finally:
		unlockfile(lock)


def test_lock(lock_path, lock_mgr):
	while True:
		try:
			logging.debug('lock attempt')
			with lock_mgr(lock_path):
				logging.debug('lock acquired')
				with open(lock_path, 'wt+') as f:
					os.ftruncate(f.fileno(), 0)
					f.seek(0)

					token = '{node}-{pid}\n'.format(node=platform.node(), pid=os.getpid())
					f.write(token)
					f.flush()
					f.seek(0)
					content = f.read()
					if token != content:
						raise AssertionError("'{}' != '{}'".format(token, content))
					logging.debug('successful content assertion')
		finally:
			logging.debug('lock released')


def main():
	parser = argparse.ArgumentParser(description='Test case for https://bugs.gentoo.org/636798')
	parser.add_argument('lockfile', action='store', help='lockfile path')
	parser.add_argument('--debug', action='store_true')
	args = parser.parse_args()

	logging.basicConfig(
		level=(logging.DEBUG if args.debug else logging.INFO),
		style='{',
		format='[{name}] {levelname}: {message}',
	)

	test_lock(args.lockfile, lock_mgr=default)
	


if __name__ == '__main__':
	main()