khodowany
59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
import modulus.filesystem
|
|
|
|
class PublicationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = pathlib.Path(self.temporary.name)
|
|
self.source, self.target = self.root / 'new', self.root / 'installed'
|
|
for path, text in ((self.source, 'new'), (self.target, 'old')):
|
|
path.mkdir()
|
|
(path / 'content').write_text(text)
|
|
def test_publish_replaces_complete_directory(self):
|
|
with modulus.filesystem.publish_directories([(self.source, self.target)]):
|
|
self.assertEqual((self.target / 'content').read_text(), 'new')
|
|
self.assertFalse(self.source.exists())
|
|
self.assertEqual(list(self.root.iterdir()), [self.target])
|
|
def test_failure_restores_previous_directory(self):
|
|
with self.assertRaisesRegex(RuntimeError, 'failed'):
|
|
with modulus.filesystem.publish_directories([(self.source, self.target)]):
|
|
raise RuntimeError('failed')
|
|
self.assertEqual((self.target / 'content').read_text(), 'old')
|
|
self.assertFalse(list(self.root.glob('.publish_*')))
|
|
def test_failed_second_replacement_rolls_back_first(self):
|
|
second = self.root / 'second'
|
|
second.mkdir()
|
|
rename = pathlib.Path.rename
|
|
def fail(path, destination):
|
|
if path == second:
|
|
raise OSError('failed')
|
|
return rename(path, destination)
|
|
with unittest.mock.patch.object(pathlib.Path, 'rename', fail), self.assertRaises(OSError):
|
|
with modulus.filesystem.publish_directories([(self.source, self.target), (second, self.root / 'last')]):
|
|
pass
|
|
self.assertEqual((self.target / 'content').read_text(), 'old')
|
|
def test_overlaps_and_symlinks_rejected_before_replacement(self):
|
|
link = self.root / 'link'
|
|
link.symlink_to(self.target, target_is_directory=True)
|
|
for target in (self.source, self.source / 'child', self.source / '..' / 'new', self.root, link, link / 'child'):
|
|
with self.subTest(target=target), self.assertRaises(ValueError):
|
|
with modulus.filesystem.publish_directories([(self.source, target)]):
|
|
pass
|
|
self.assertEqual((self.target / 'content').read_text(), 'old')
|
|
def test_failed_rollback_retains_the_previous_directory_for_recovery(self):
|
|
rename = pathlib.Path.rename
|
|
def fail(path, destination):
|
|
if path.name == 'previous':
|
|
raise OSError('restore failed')
|
|
return rename(path, destination)
|
|
with unittest.mock.patch.object(pathlib.Path, 'rename', fail), self.assertRaisesRegex(RuntimeError, 'recovery directories retained'):
|
|
with modulus.filesystem.publish_directories([(self.source, self.target)]):
|
|
raise RuntimeError('publication failed')
|
|
backup = next(self.root.glob('.publish_*/previous/content'))
|
|
self.assertEqual(backup.read_text(), 'old')
|