test_acidfs_git_segfault.py

import os.path
import unittest
import transaction
import threading
import shutil
import subprocess
import acidfs

cwd = os.path.dirname(os.path.realpath(__file__))
fs_path = os.path.join(cwd, "acidfs_testdata")


def threader(value, conflict, other=None):
    threadlocal_fs = acidfs.AcidFS(fs_path)
    with threadlocal_fs.open("b", "w") as fh:
        fh.write(value)

    if other:
        other.start()
        other.join()

    try:
        transaction.commit()
    except subprocess.CalledProcessError:
        conflict.git_error = True
    except acidfs.ConflictError:
        conflict.happened = True


def first_commit():
    fs = acidfs.AcidFS(fs_path)
    with fs.open("a", "w") as fh:
        fh.write(u"b")
    transaction.commit()


def run_threads(conflict, a, b):
    thread_b = threading.Thread(target=threader, args=(a, conflict))
    thread_a = threading.Thread(target=threader, args=(b, conflict, thread_b))

    thread_a.start()
    thread_a.join()


def remove_testdir():
    if os.path.isdir(fs_path):
        shutil.rmtree(fs_path)
        assert not os.path.isdir(fs_path)


class Conflict(object):
    def __init__(self):
        self.happened = False
        self.git_error = False


class AcidFsLeadingNewlineProblemTests(unittest.TestCase):
    def setUp(self):
        pass

    def test_git_error_leading_newline_with_initial_commit(self):
        first_commit()
        conflict = Conflict()
        run_threads(conflict, u"\na", u"a")

        self.assertTrue(conflict.git_error)
        self.assertFalse(conflict.happened)

    def test_no_git_error_leading_newline_without_initial_commit(self):
        conflict = Conflict()
        run_threads(conflict, u"\na", u"a")

        self.assertFalse(conflict.git_error)
        self.assertTrue(conflict.happened)

    def tearDown(self):
        remove_testdir()

if __name__ == "__main__":
    unittest.main()