From 894661df4f8d234a407e678c033d20c873d8ae5f Mon Sep 17 00:00:00 2001 From: Utkarsh Upadhyay Date: Mon, 20 Apr 2015 21:53:04 +0200 Subject: [PATCH] Add test to check for failure in case of contention. 100% coverage! --- seqfile/tests.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/seqfile/tests.py b/seqfile/tests.py index 0c33a01..01bfacc 100644 --- a/seqfile/tests.py +++ b/seqfile/tests.py @@ -115,11 +115,11 @@ def test_findNextFile_with_files_non_consecutive_fnamegen(): # Monkey patch _doAtomicFileCreation for testing _S._oldAtomicCreation = _S._doAtomicFileCreation -def _testAtomicCreationFactory(folder): +def _testAtomicCreationFactory(predicate): def _testAtomicCreation(filePath): # Create only the first file in the sequence - if filePath == join(folder, fnameGen(0)): + if predicate(filePath): # Create the file to simulate concurrent file creation _os.close( _os.open( filePath, _os.O_CREAT | _os.O_EXCL ) ) @@ -130,8 +130,18 @@ def _testAtomicCreation(filePath): def test_findNextFile_concurrent(): with tempDir() as d: - _S._doAtomicFileCreation = _testAtomicCreationFactory(d) + predicate = lambda filePath: filePath == join(d, fnameGen(0)) + _S._doAtomicFileCreation = _testAtomicCreationFactory(predicate) # File created must be the second file in the sequence assert _S.findNextFile(d, prefix, suffix) == join(d, fnameGen(1)) +@raises(OSError) +def test_findNextFile_concurrent_max_attempts_fail(): + with tempDir() as d: + # Always create the file, maxattempts will be exhausted + predicate = lambda _filePath: True + _S._doAtomicFileCreation = _testAtomicCreationFactory(predicate) + _S.findNextFile(d, prefix, suffix) + +