77 lines
2.4 KiB
Python
Executable file
77 lines
2.4 KiB
Python
Executable file
#!/usr/bin/python3
|
|
|
|
import newfol.database
|
|
|
|
import os
|
|
import pwd
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
class TestGitTransactions(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
with open(self.tempdir.name + "/schema", "w") as fp:
|
|
fp.write("fmt:0:newfol schema file:\ntxn:git\n")
|
|
self.db = newfol.database.Database.load(self.tempdir.name)
|
|
|
|
def tearDown(self):
|
|
self.tempdir.cleanup()
|
|
|
|
def test_all_checked_in(self):
|
|
db = newfol.database.Database.load(self.tempdir.name)
|
|
db.store()
|
|
curdir = os.getcwd()
|
|
os.chdir(self.tempdir.name)
|
|
output = ""
|
|
with subprocess.Popen(["git", "status", "--porcelain"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL) as proc:
|
|
output = proc.stdout.read()
|
|
os.chdir(curdir)
|
|
self.assertEqual(output, b"")
|
|
|
|
def test_files_checked_in(self):
|
|
db = newfol.database.Database.load(self.tempdir.name)
|
|
db.store()
|
|
curdir = os.getcwd()
|
|
os.chdir(self.tempdir.name)
|
|
output = ""
|
|
with subprocess.Popen(["git", "ls-files"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL) as proc:
|
|
output = proc.stdout.read().decode()
|
|
os.chdir(curdir)
|
|
expected = "".join(map(lambda x: "%s\n" % x,
|
|
['dtb', 'dtb.checksum', 'schema']))
|
|
self.assertEqual(output, expected)
|
|
|
|
def test_has_git_dir(self):
|
|
self.db.store()
|
|
self.assertTrue(os.path.isdir(self.tempdir.name + "/.git"))
|
|
|
|
def get_format_data(self, fmt):
|
|
cwd = os.getcwd()
|
|
data = None
|
|
try:
|
|
os.chdir(self.tempdir.name)
|
|
with os.popen("git log --pretty=format:" + fmt + " -n1") as fp:
|
|
data = fp.read()
|
|
finally:
|
|
os.chdir(cwd)
|
|
return data
|
|
|
|
def test_correct_committer_name(self):
|
|
self.db.store()
|
|
self.assertEqual(self.get_format_data("%cn"), "newfol")
|
|
|
|
def test_correct_committer_email(self):
|
|
self.db.store()
|
|
user = pwd.getpwuid(os.getuid())[0]
|
|
fqdn = socket.getfqdn()
|
|
self.assertEqual(self.get_format_data("%ce"), user + "@" + fqdn)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|