1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
import fcntl
import subprocess
import sys
import time
import unittest
from test_cli import Cli, WF
class LockTest(Cli):
def test_write_waits_for_lock(self):
lock_path = self.root / ".wf" / "lock"
lock_path.parent.mkdir(exist_ok=True)
with open(lock_path, "w") as held:
fcntl.flock(held, fcntl.LOCK_EX)
p = subprocess.Popen([sys.executable, str(WF), "--project", str(self.root), "note", "t-three", "first"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
time.sleep(0.5)
self.assertIsNone(p.poll(), "write ran while the lock was held")
self.ok("show", "t-three") # read commands never wait
out, err = p.communicate(timeout=10)
self.assertEqual((p.returncode, err), (0, ""))
self.ok("note", "t-three", "second")
body = self.item("t-three")
self.assertIn("first", body)
self.assertIn("second", body)
def test_lock_file_is_git_ignored_folder(self):
self.ok("note", "t-three", "x")
self.assertEqual((self.root / ".wf" / ".gitignore").read_text(), "*\n")
|
