import multiprocessing
from pathlib import Path
import tempfile
import unittest

from network_runtime import NetworkFailure, allocation_lock


def hold_lock(path, ready):
    with allocation_lock(path=path):
        ready.set()
        ready.wait(30)
        import time
        time.sleep(30)


class LockTests(unittest.TestCase):
    def test_busy_lock_refuses_and_process_death_releases_only_its_lock(self):
        with tempfile.TemporaryDirectory() as root:
            path = Path(root) / 'allocation.lock'
            context = multiprocessing.get_context('spawn')
            ready = context.Event()
            child = context.Process(target=hold_lock, args=(path, ready))
            child.start()
            try:
                self.assertTrue(ready.wait(5))
                with self.assertRaisesRegex(NetworkFailure, 'allocation_lock_busy'):
                    with allocation_lock(path=path, seconds=0.1):
                        self.fail('A second writer entered the allocation window')
            finally:
                child.terminate()
                child.join(5)
            with allocation_lock(path=path, seconds=0.1):
                self.assertTrue(path.is_file())

    def test_symlink_and_publicly_writable_lock_refuse(self):
        with tempfile.TemporaryDirectory() as root:
            target, link = Path(root) / 'target', Path(root) / 'link'
            target.write_text('retained')
            link.symlink_to(target)
            for path in (target, link):
                with self.assertRaises(NetworkFailure):
                    with allocation_lock(path=path):
                        self.fail('An unsafe lock was accepted')
            self.assertEqual(target.read_text(), 'retained')


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