112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
"""Cells for counterparty_sig.py (Q85): authorship-aware repository signature.
|
|
|
|
The module had no collected cell, so the property it exists for -- a signature
|
|
that names only counterparty-authored bytes -- was ungraded. Each cell builds
|
|
a throwaway git repository with two identities and reads the signature back, so
|
|
the authorship rule is measured against real commits rather than described.
|
|
"""
|
|
import hashlib
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
import counterparty_sig as S
|
|
|
|
SELF = ("Volt", "volt@troupe.local")
|
|
THEM = ("Merlin", "merlin@troupe.local")
|
|
|
|
|
|
def git(repo, *args, identity=None):
|
|
cmd = ["git", "-C", repo, "-c", "commit.gpgsign=false", "-c", "init.defaultBranch=main"]
|
|
if identity:
|
|
cmd += ["-c", "user.name=" + identity[0], "-c", "user.email=" + identity[1]]
|
|
return subprocess.run(cmd + list(args), check=True, capture_output=True, text=True)
|
|
|
|
|
|
def commit(repo, path, body, identity, message):
|
|
full = os.path.join(repo, path)
|
|
os.makedirs(os.path.dirname(full), exist_ok=True)
|
|
with open(full, "w", encoding="utf-8") as fh:
|
|
fh.write(body)
|
|
git(repo, "add", path)
|
|
git(repo, "commit", "-q", "-m", message, identity=identity)
|
|
return git(repo, "rev-parse", "HEAD").stdout.strip()
|
|
|
|
|
|
class SignatureFixture(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="counterparty-sig.")
|
|
self.repo = self.tmp.name
|
|
git(self.repo, "init", "-q")
|
|
self.mine = commit(self.repo, "inbox/mine.txt", "my note\n", SELF, "volt: note")
|
|
self.theirs = commit(self.repo, "inbox/theirs.md", "their note\n", THEM, "merlin: note")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
|
|
class TestCounterpartySignature(SignatureFixture):
|
|
def test_head_is_the_newest_counterparty_commit(self):
|
|
text, digest = S.signature(self.repo, SELF[0], ["inbox"])
|
|
self.assertEqual(text.splitlines()[0], "head %s" % self.theirs[:12])
|
|
self.assertNotIn(self.mine[:12], text)
|
|
|
|
def test_only_counterparty_authored_paths_are_listed(self):
|
|
text, _ = S.signature(self.repo, SELF[0], ["inbox"])
|
|
self.assertIn("inbox/theirs.md", text)
|
|
self.assertNotIn("inbox/mine.txt", text)
|
|
|
|
def test_blob_ids_are_the_twelve_character_prefixes_at_head(self):
|
|
text, _ = S.signature(self.repo, SELF[0], ["inbox"])
|
|
blob = git(self.repo, "rev-parse", "%s:inbox/theirs.md" % self.theirs).stdout.strip()
|
|
self.assertIn("inbox/theirs.md %s" % blob[:12], text)
|
|
|
|
def test_digest_is_the_hash_of_the_text_and_is_stable(self):
|
|
text1, digest1 = S.signature(self.repo, SELF[0], ["inbox"])
|
|
text2, digest2 = S.signature(self.repo, SELF[0], ["inbox"])
|
|
self.assertEqual(text1, text2)
|
|
self.assertEqual(digest1, digest2)
|
|
self.assertEqual(digest1, hashlib.sha256(text1.encode()).hexdigest())
|
|
self.assertEqual(len(digest1), 64)
|
|
|
|
def test_self_authored_later_edit_drops_the_path(self):
|
|
# The rule is authorship of the newest commit touching the path, not of
|
|
# the commit the file is read from: a self edit at head removes it.
|
|
commit(self.repo, "inbox/theirs.md", "their note\nplus mine\n", SELF, "volt: edit their note")
|
|
text, _ = S.signature(self.repo, SELF[0], ["inbox"])
|
|
self.assertNotIn("inbox/theirs.md", text)
|
|
self.assertEqual(len(text.splitlines()), 1, text)
|
|
|
|
def test_a_repository_with_no_counterparty_commit_raises(self):
|
|
with tempfile.TemporaryDirectory(prefix="counterparty-solo.") as solo:
|
|
git(solo, "init", "-q")
|
|
commit(solo, "a.txt", "only me\n", SELF, "volt: only")
|
|
with self.assertRaises(RuntimeError):
|
|
S.signature(solo, SELF[0], ["."])
|
|
|
|
def test_git_failure_is_raised_with_the_command_named(self):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
S.run(os.path.join(self.tmp.name, "no-such-repo"), "log")
|
|
self.assertIn("git", str(ctx.exception))
|
|
|
|
def test_commits_reads_author_triples(self):
|
|
rows = S.commits(self.repo)
|
|
self.assertEqual([row[1] for row in rows], ["Merlin", "Volt"])
|
|
self.assertEqual([row[2] for row in rows], [THEM[1], SELF[1]])
|
|
for sha, _name, _email in rows:
|
|
self.assertEqual(len(sha), 40)
|
|
|
|
def test_newest_commit_for_path_is_cached(self):
|
|
cache = {}
|
|
first = S.newest_commit_for_path(self.repo, "inbox/theirs.md", SELF[0], cache)
|
|
self.assertEqual(first[0], self.theirs)
|
|
self.assertTrue(first[3], "a counterparty commit must be marked counterparty")
|
|
before = dict(cache)
|
|
S.newest_commit_for_path(self.repo, "inbox/theirs.md", SELF[0], cache)
|
|
self.assertEqual(cache, before, "the second call must not re-enter git")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|