GitPython/test/git/test_utils.py at 0.2 · og-python-scripts/GitPython · GitHub

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

# test_utils.py

# This module is part of GitPython and is released under

# the BSD License: http://www.opensource.org/licenses/bsd-license.php

import os

import tempfile

from test.testlib import *

from git.utils import *

from git import *

from git.cmd import dashify

import time

class TestUtils(TestCase):

def setup(self):

self.testdict = {

"string": "42",

"int": 42,

"array": [ 42 ],

}

def test_it_should_dashify(self):

assert_equal('this-is-my-argument', dashify('this_is_my_argument'))

assert_equal('foo', dashify('foo'))

def test_lock_file(self):

my_file = tempfile.mktemp()

lock_file = LockFile(my_file)

assert not lock_file._has_lock()

# release lock we don't have - fine

lock_file._release_lock()

# get lock

lock_file._obtain_lock_or_raise()

assert lock_file._has_lock()

# concurrent access

other_lock_file = LockFile(my_file)

assert not other_lock_file._has_lock()

self.failUnlessRaises(IOError, other_lock_file._obtain_lock_or_raise)

lock_file._release_lock()

assert not lock_file._has_lock()

other_lock_file._obtain_lock_or_raise()

self.failUnlessRaises(IOError, lock_file._obtain_lock_or_raise)

# auto-release on destruction

del(other_lock_file)

lock_file._obtain_lock_or_raise()

lock_file._release_lock()

def test_blocking_lock_file(self):

my_file = tempfile.mktemp()

lock_file = BlockingLockFile(my_file)

lock_file._obtain_lock()

# next one waits for the lock

start = time.time()

wait_time = 0.1

wait_lock = BlockingLockFile(my_file, 0.05, wait_time)

self.failUnlessRaises(IOError, wait_lock._obtain_lock)

elapsed = time.time() - start

assert elapsed <= wait_time + 0.02 # some extra time it may cost

def _cmp_contents(self, file_path, data):

# raise if data from file at file_path

# does not match data string

fp = open(file_path, "rb")

try:

assert fp.read() == data

finally:

fp.close()

def test_safe_operation(self):

my_file = tempfile.mktemp()

orig_data = "hello"

new_data = "world"

my_file_fp = open(my_file, "wb")

my_file_fp.write(orig_data)

my_file_fp.close()

try:

cwrite = ConcurrentWriteOperation(my_file)

# didn't start writing, doesnt matter

cwrite._end_writing(False)

cwrite._end_writing(True)

assert not cwrite._is_writing()

# write data and fail

stream = cwrite._begin_writing()

assert cwrite._is_writing()

stream.write(new_data)

cwrite._end_writing(successful=False)

self._cmp_contents(my_file, orig_data)

assert not os.path.exists(stream.name)

# write data - concurrently

ocwrite = ConcurrentWriteOperation(my_file)

stream = cwrite._begin_writing()

self.failUnlessRaises(IOError, ocwrite._begin_writing)

stream.write("world")

cwrite._end_writing(successful=True)

self._cmp_contents(my_file, new_data)

assert not os.path.exists(stream.name)

# could test automatic _end_writing on destruction

finally:

os.remove(my_file)

# END final cleanup

GitPython/test/git/test_utils.py at 0.2 · og-python-scripts/GitPython · GitHub

Repository navigation

Expand file tree

Breadcrumbs

test_utils.py

Latest commit

History

Breadcrumbs

test_utils.py

File metadata and controls