51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
import tempfile
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
import app
|
|
|
|
|
|
class StoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.store = app.Store(Path(self.temp.name) / "state.sqlite3")
|
|
self.store.initialize()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_empty_store(self):
|
|
self.assertEqual(self.store.snapshot()["checks"], {})
|
|
self.assertEqual(self.store.snapshot()["revision"], 0)
|
|
|
|
def test_shared_check_round_trip(self):
|
|
result = self.store.set_check("crop-parsnip", True, "fred")
|
|
self.assertEqual(result["revision"], 1)
|
|
snapshot = self.store.snapshot()
|
|
self.assertTrue(snapshot["checks"]["crop-parsnip"])
|
|
self.assertEqual(snapshot["updatedBy"], "fred")
|
|
|
|
self.store.set_check("crop-parsnip", False, "kayla")
|
|
snapshot = self.store.snapshot()
|
|
self.assertFalse(snapshot["checks"]["crop-parsnip"])
|
|
self.assertEqual(snapshot["revision"], 2)
|
|
|
|
def test_two_clients_keep_distinct_concurrent_changes(self):
|
|
second_client = app.Store(self.store.path)
|
|
|
|
def write(index):
|
|
store = self.store if index % 2 else second_client
|
|
store.set_check(f"concurrent-item-{index}", True, f"player-{index % 2}")
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
list(pool.map(write, range(20)))
|
|
|
|
snapshot = self.store.snapshot()
|
|
self.assertEqual(snapshot["revision"], 20)
|
|
self.assertEqual(len(snapshot["checks"]), 20)
|
|
self.assertTrue(all(snapshot["checks"].values()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|