iframe-proxy | Sunbelt Computer Software

History

100 lines (83 loc) · 3.17 KB

Raw

100

# This module is part of GitDB and is released under

# the New BSD License: https://opensource.org/license/bsd-3-clause/

"""Test for object db"""

import tempfile

import os

from gitdb.test.lib import TestBase

from gitdb.util import (

to_hex_sha,

to_bin_sha,

NULL_HEX_SHA,

LockedFD

)

class TestUtils(TestBase):

def test_basics(self):

assert to_hex_sha(NULL_HEX_SHA) == NULL_HEX_SHA

assert len(to_bin_sha(NULL_HEX_SHA)) == 20

assert to_hex_sha(to_bin_sha(NULL_HEX_SHA)) == NULL_HEX_SHA.encode("ascii")

def _cmp_contents(self, file_path, data):

# raise if data from file at file_path

# does not match data string

with open(file_path, "rb") as fp:

assert fp.read() == data.encode("ascii")

def test_lockedfd(self):

my_file = tempfile.mktemp()

orig_data = "hello"

new_data = "world"

with open(my_file, "wb") as my_file_fp:

my_file_fp.write(orig_data.encode("ascii"))

try:

lfd = LockedFD(my_file)

lockfilepath = lfd._lockfilepath()

# cannot end before it was started

self.assertRaises(AssertionError, lfd.rollback)

self.assertRaises(AssertionError, lfd.commit)

# open for writing

assert not os.path.isfile(lockfilepath)

wfd = lfd.open(write=True)

assert lfd._fd is wfd

assert os.path.isfile(lockfilepath)

# write data and fail

os.write(wfd, new_data.encode("ascii"))

lfd.rollback()

assert lfd._fd is None

self._cmp_contents(my_file, orig_data)

assert not os.path.isfile(lockfilepath)

# additional call doesn't fail

lfd.commit()

lfd.rollback()

# test reading

lfd = LockedFD(my_file)

rfd = lfd.open(write=False)

assert os.read(rfd, len(orig_data)) == orig_data.encode("ascii")

assert os.path.isfile(lockfilepath)

# deletion rolls back

del(lfd)

assert not os.path.isfile(lockfilepath)

# write data - concurrently

lfd = LockedFD(my_file)

olfd = LockedFD(my_file)

assert not os.path.isfile(lockfilepath)

wfdstream = lfd.open(write=True, stream=True) # this time as stream

assert os.path.isfile(lockfilepath)

# another one fails

self.assertRaises(IOError, olfd.open)

wfdstream.write(new_data.encode("ascii"))

lfd.commit()

assert not os.path.isfile(lockfilepath)

self._cmp_contents(my_file, new_data)

# could test automatic _end_writing on destruction

finally:

os.remove(my_file)

# END final cleanup

# try non-existing file for reading

lfd = LockedFD(tempfile.mktemp())

try:

lfd.open(write=False)

except OSError:

assert not os.path.exists(lfd._lockfilepath())

else:

self.fail("expected OSError")

# END handle exceptions

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Uh oh!

FilesExpand file tree

test_util.py

Latest commit

History

test_util.py

File metadata and controls