import concurrent.futures
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from import_once import initialize, apply_event


class ReplayTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.database = Path(self.directory.name) / "demo.sqlite"
        initialize(self.database)

    def counts(self):
        with closing(sqlite3.connect(self.database)) as connection:
            return (connection.execute("SELECT COUNT(*) FROM events").fetchone()[0],
                    connection.execute("SELECT COALESCE(SUM(units),0) FROM inventory").fetchone()[0])

    def test_repeat_has_one_effect(self):
        self.assertEqual(apply_event(self.database, "batch:1", "00123", 7), "applied")
        self.assertEqual(apply_event(self.database, "batch:1", "00123", 7), "already_applied")
        self.assertEqual(self.counts(), (1, 7))

    def test_failure_before_commit_rolls_back_both_tables(self):
        with self.assertRaises(RuntimeError):
            apply_event(self.database, "batch:1", "00123", 7, fail_at="before_commit")
        self.assertEqual(self.counts(), (0, 0))
        apply_event(self.database, "batch:1", "00123", 7)
        self.assertEqual(self.counts(), (1, 7))

    def test_lost_acknowledgement_after_commit_does_not_duplicate(self):
        with self.assertRaises(RuntimeError):
            apply_event(self.database, "batch:1", "00123", 7, fail_at="after_commit")
        self.assertEqual(self.counts(), (1, 7))
        self.assertEqual(apply_event(self.database, "batch:1", "00123", 7), "already_applied")
        self.assertEqual(self.counts(), (1, 7))

    def test_changed_payload_for_same_event_is_rejected(self):
        apply_event(self.database, "batch:1", "00123", 7)
        with self.assertRaisesRegex(ValueError, "different data"):
            apply_event(self.database, "batch:1", "00123", 8)
        self.assertEqual(self.counts(), (1, 7))

    def test_distinct_events_add(self):
        apply_event(self.database, "batch:1", "00123", 7)
        apply_event(self.database, "batch:2", "00123", 3)
        self.assertEqual(self.counts(), (2, 10))

    def test_concurrent_replays_apply_once(self):
        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
            results = list(pool.map(lambda _: apply_event(self.database, "batch:1", "00123", 7), range(8)))
        self.assertEqual(results.count("applied"), 1)
        self.assertEqual(results.count("already_applied"), 7)
        self.assertEqual(self.counts(), (1, 7))

    def test_bad_input_does_not_write(self):
        for value in [True, -1, 1.5, "7", 1_000_001]:
            with self.subTest(value=value), self.assertRaises(ValueError):
                apply_event(self.database, "batch:1", "00123", value)
        self.assertEqual(self.counts(), (0, 0))


if __name__ == "__main__":
    unittest.main()