fix(marketplace): Publish plugin sync atomically
This commit is contained in:
@@ -2,6 +2,7 @@ import importlib.util
|
||||
import errno
|
||||
import pathlib
|
||||
import sys
|
||||
import tempfile
|
||||
from unittest import mock
|
||||
import unittest
|
||||
|
||||
@@ -160,6 +161,31 @@ class EditorialBundlesTests(unittest.TestCase):
|
||||
self.assertEqual(calls["count"], 2)
|
||||
sleep_mock.assert_called_once()
|
||||
|
||||
def test_replace_directory_atomically_swaps_only_after_staging_is_ready(self):
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
temp_root = pathlib.Path(temp_dir)
|
||||
destination = temp_root / "plugin"
|
||||
old_file = destination / "skills" / "old.txt"
|
||||
old_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
old_file.write_text("old", encoding="utf-8")
|
||||
|
||||
observed = {}
|
||||
|
||||
def populate(staging_root):
|
||||
new_file = staging_root / "skills" / "new.txt"
|
||||
new_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
new_file.write_text("new", encoding="utf-8")
|
||||
|
||||
observed["old_visible_during_populate"] = old_file.is_file()
|
||||
observed["new_hidden_during_populate"] = not (destination / "skills" / "new.txt").exists()
|
||||
|
||||
editorial_bundles._replace_directory_atomically(destination, populate)
|
||||
|
||||
self.assertTrue(observed["old_visible_during_populate"])
|
||||
self.assertTrue(observed["new_hidden_during_populate"])
|
||||
self.assertFalse(old_file.exists())
|
||||
self.assertTrue((destination / "skills" / "new.txt").is_file())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user