37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from grm import store # noqa: E402
|
|
|
|
|
|
class LedgerTest(unittest.TestCase):
|
|
def test_parallel_appends_give_whole_lines(self):
|
|
"""B3: two threads appending at once leave only whole, parsable lines."""
|
|
with tempfile.TemporaryDirectory() as d:
|
|
run = str(Path(d) / "run.md")
|
|
store.init_state(run)
|
|
|
|
def append(job):
|
|
for i in range(50):
|
|
store.ledger_append(run, {"job": job, "i": i, "pad": "x" * 500})
|
|
threads = [threading.Thread(target=append, args=(j,)) for j in ("a", "b")]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
raw = (store.state_dir(run) / "ledger.jsonl").read_bytes().splitlines()
|
|
lines = store.ledger_lines(run)
|
|
self.assertEqual(len(raw), 100)
|
|
self.assertEqual(len(lines), 100)
|
|
self.assertEqual(sorted((x["job"], x["i"]) for x in lines),
|
|
sorted((j, i) for j in ("a", "b") for i in range(50)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|