Skip to content

Commit aac323e

Browse files
author
LittleCoinCoin
committed
test: add atomic file operations and backup-aware operation tests
Add comprehensive test suite for atomic file operations and backup-aware operation patterns. AtomicFileOperations test coverage: - Atomic write with automatic backup creation - Atomic write with backup skipping - Atomic copy operations with integrity verification - Failure handling with automatic rollback - Temporary file cleanup on operation failure BackupAwareOperation test coverage: - Explicit backup preparation workflow - No-backup mode operation - Backup failure exception handling - Rollback functionality on operation failure - Integration with MCPHostConfigBackupManager All tests use host-agnostic configurations and validate the simplified API design that forces consumers to explicitly acknowledge backup creation. Uses wobble decorators (@regression_test) and follows CrackingShells testing standards. Test results: 11/11 tests passing (100% success rate)
1 parent 0bfeecf commit aac323e

File tree

1 file changed

+276
-0
lines changed

1 file changed

+276
-0
lines changed
Lines changed: 276 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,276 @@
1+
"""Tests for MCP atomic file operations.
2+
3+
This module contains tests for atomic file operations and backup-aware
4+
operations with host-agnostic design.
5+
"""
6+
7+
import unittest
8+
import tempfile
9+
import shutil
10+
import json
11+
from pathlib import Path
12+
from unittest.mock import patch, mock_open
13+
14+
from wobble.decorators import regression_test
15+
from test_data_utils import MCPBackupTestDataLoader
16+
17+
from hatch.mcp.backup import (
18+
AtomicFileOperations,
19+
MCPHostConfigBackupManager,
20+
BackupAwareOperation,
21+
BackupError
22+
)
23+
24+
25+
class TestAtomicFileOperations(unittest.TestCase):
26+
"""Test atomic file operations with host-agnostic design."""
27+
28+
def setUp(self):
29+
"""Set up test environment."""
30+
self.temp_dir = Path(tempfile.mkdtemp(prefix="test_atomic_"))
31+
self.test_file = self.temp_dir / "test_config.json"
32+
self.backup_manager = MCPHostConfigBackupManager(backup_root=self.temp_dir / "backups")
33+
self.atomic_ops = AtomicFileOperations()
34+
self.test_data = MCPBackupTestDataLoader()
35+
36+
def tearDown(self):
37+
"""Clean up test environment."""
38+
shutil.rmtree(self.temp_dir, ignore_errors=True)
39+
40+
@regression_test
41+
def test_atomic_write_success_host_agnostic(self):
42+
"""Test successful atomic write with any JSON configuration format."""
43+
test_data = self.test_data.load_host_agnostic_config("complex_server")
44+
45+
result = self.atomic_ops.atomic_write_with_backup(
46+
self.test_file, test_data, self.backup_manager, "claude-desktop"
47+
)
48+
49+
self.assertTrue(result)
50+
self.assertTrue(self.test_file.exists())
51+
52+
# Verify content (host-agnostic)
53+
with open(self.test_file) as f:
54+
written_data = json.load(f)
55+
self.assertEqual(written_data, test_data)
56+
57+
@regression_test
58+
def test_atomic_write_with_existing_file(self):
59+
"""Test atomic write with existing file creates backup."""
60+
# Create initial file
61+
initial_data = self.test_data.load_host_agnostic_config("simple_server")
62+
with open(self.test_file, 'w') as f:
63+
json.dump(initial_data, f)
64+
65+
# Update with atomic write
66+
new_data = self.test_data.load_host_agnostic_config("complex_server")
67+
result = self.atomic_ops.atomic_write_with_backup(
68+
self.test_file, new_data, self.backup_manager, "vscode"
69+
)
70+
71+
self.assertTrue(result)
72+
73+
# Verify backup was created
74+
backups = self.backup_manager.list_backups("vscode")
75+
self.assertEqual(len(backups), 1)
76+
77+
# Verify backup contains original data
78+
with open(backups[0].file_path) as f:
79+
backup_data = json.load(f)
80+
self.assertEqual(backup_data, initial_data)
81+
82+
# Verify file contains new data
83+
with open(self.test_file) as f:
84+
current_data = json.load(f)
85+
self.assertEqual(current_data, new_data)
86+
87+
@regression_test
88+
def test_atomic_write_skip_backup(self):
89+
"""Test atomic write with backup skipped."""
90+
# Create initial file
91+
initial_data = self.test_data.load_host_agnostic_config("simple_server")
92+
with open(self.test_file, 'w') as f:
93+
json.dump(initial_data, f)
94+
95+
# Update with atomic write, skipping backup
96+
new_data = self.test_data.load_host_agnostic_config("complex_server")
97+
result = self.atomic_ops.atomic_write_with_backup(
98+
self.test_file, new_data, self.backup_manager, "cursor", skip_backup=True
99+
)
100+
101+
self.assertTrue(result)
102+
103+
# Verify no backup was created
104+
backups = self.backup_manager.list_backups("cursor")
105+
self.assertEqual(len(backups), 0)
106+
107+
# Verify file contains new data
108+
with open(self.test_file) as f:
109+
current_data = json.load(f)
110+
self.assertEqual(current_data, new_data)
111+
112+
@regression_test
113+
def test_atomic_write_failure_rollback(self):
114+
"""Test atomic write failure triggers rollback."""
115+
# Create initial file
116+
initial_data = self.test_data.load_host_agnostic_config("simple_server")
117+
with open(self.test_file, 'w') as f:
118+
json.dump(initial_data, f)
119+
120+
# Mock file write failure after backup creation
121+
with patch('builtins.open', side_effect=[
122+
# First call succeeds (backup creation)
123+
open(self.test_file, 'r'),
124+
# Second call fails (atomic write)
125+
PermissionError("Access denied")
126+
]):
127+
with self.assertRaises(BackupError):
128+
self.atomic_ops.atomic_write_with_backup(
129+
self.test_file, {"new": "data"}, self.backup_manager, "lmstudio"
130+
)
131+
132+
# Verify original file is unchanged
133+
with open(self.test_file) as f:
134+
current_data = json.load(f)
135+
self.assertEqual(current_data, initial_data)
136+
137+
@regression_test
138+
def test_atomic_copy_success(self):
139+
"""Test successful atomic copy operation."""
140+
source_file = self.temp_dir / "source.json"
141+
target_file = self.temp_dir / "target.json"
142+
143+
test_data = self.test_data.load_host_agnostic_config("simple_server")
144+
with open(source_file, 'w') as f:
145+
json.dump(test_data, f)
146+
147+
result = self.atomic_ops.atomic_copy(source_file, target_file)
148+
149+
self.assertTrue(result)
150+
self.assertTrue(target_file.exists())
151+
152+
# Verify content integrity
153+
with open(target_file) as f:
154+
copied_data = json.load(f)
155+
self.assertEqual(copied_data, test_data)
156+
157+
@regression_test
158+
def test_atomic_copy_failure_cleanup(self):
159+
"""Test atomic copy failure cleans up temporary files."""
160+
source_file = self.temp_dir / "source.json"
161+
target_file = self.temp_dir / "target.json"
162+
163+
test_data = self.test_data.load_host_agnostic_config("simple_server")
164+
with open(source_file, 'w') as f:
165+
json.dump(test_data, f)
166+
167+
# Mock copy failure
168+
with patch('shutil.copy2', side_effect=PermissionError("Access denied")):
169+
result = self.atomic_ops.atomic_copy(source_file, target_file)
170+
171+
self.assertFalse(result)
172+
self.assertFalse(target_file.exists())
173+
174+
# Verify no temporary files left behind
175+
temp_files = list(self.temp_dir.glob("*.tmp"))
176+
self.assertEqual(len(temp_files), 0)
177+
178+
179+
class TestBackupAwareOperation(unittest.TestCase):
180+
"""Test backup-aware operation API."""
181+
182+
def setUp(self):
183+
"""Set up test environment."""
184+
self.temp_dir = Path(tempfile.mkdtemp(prefix="test_backup_aware_"))
185+
self.test_file = self.temp_dir / "test_config.json"
186+
self.backup_manager = MCPHostConfigBackupManager(backup_root=self.temp_dir / "backups")
187+
self.test_data = MCPBackupTestDataLoader()
188+
189+
def tearDown(self):
190+
"""Clean up test environment."""
191+
shutil.rmtree(self.temp_dir, ignore_errors=True)
192+
193+
@regression_test
194+
def test_prepare_backup_success(self):
195+
"""Test explicit backup preparation."""
196+
# Create initial configuration
197+
initial_data = self.test_data.load_host_agnostic_config("simple_server")
198+
with open(self.test_file, 'w') as f:
199+
json.dump(initial_data, f)
200+
201+
# Test backup-aware operation
202+
operation = BackupAwareOperation(self.backup_manager)
203+
204+
# Test explicit backup preparation
205+
backup_result = operation.prepare_backup(self.test_file, "gemini", no_backup=False)
206+
self.assertIsNotNone(backup_result)
207+
self.assertTrue(backup_result.success)
208+
209+
# Verify backup was created
210+
backups = self.backup_manager.list_backups("gemini")
211+
self.assertEqual(len(backups), 1)
212+
213+
@regression_test
214+
def test_prepare_backup_no_backup_mode(self):
215+
"""Test no-backup mode."""
216+
# Create initial configuration
217+
initial_data = self.test_data.load_host_agnostic_config("simple_server")
218+
with open(self.test_file, 'w') as f:
219+
json.dump(initial_data, f)
220+
221+
operation = BackupAwareOperation(self.backup_manager)
222+
223+
# Test no-backup mode
224+
no_backup_result = operation.prepare_backup(self.test_file, "claude-code", no_backup=True)
225+
self.assertIsNone(no_backup_result)
226+
227+
# Verify no backup was created
228+
backups = self.backup_manager.list_backups("claude-code")
229+
self.assertEqual(len(backups), 0)
230+
231+
@regression_test
232+
def test_prepare_backup_failure_raises_exception(self):
233+
"""Test backup preparation failure raises BackupError."""
234+
# Test with nonexistent file
235+
nonexistent_file = self.temp_dir / "nonexistent.json"
236+
237+
operation = BackupAwareOperation(self.backup_manager)
238+
239+
with self.assertRaises(BackupError):
240+
operation.prepare_backup(nonexistent_file, "vscode", no_backup=False)
241+
242+
@regression_test
243+
def test_rollback_on_failure_success(self):
244+
"""Test successful rollback functionality."""
245+
# Create initial configuration
246+
initial_data = self.test_data.load_host_agnostic_config("simple_server")
247+
with open(self.test_file, 'w') as f:
248+
json.dump(initial_data, f)
249+
250+
operation = BackupAwareOperation(self.backup_manager)
251+
252+
# Create backup
253+
backup_result = operation.prepare_backup(self.test_file, "cursor", no_backup=False)
254+
self.assertTrue(backup_result.success)
255+
256+
# Modify file (simulate failed operation)
257+
modified_data = self.test_data.load_host_agnostic_config("complex_server")
258+
with open(self.test_file, 'w') as f:
259+
json.dump(modified_data, f)
260+
261+
# Test rollback functionality
262+
rollback_success = operation.rollback_on_failure(backup_result, self.test_file, "cursor")
263+
self.assertTrue(rollback_success)
264+
265+
@regression_test
266+
def test_rollback_on_failure_no_backup(self):
267+
"""Test rollback with no backup result."""
268+
operation = BackupAwareOperation(self.backup_manager)
269+
270+
# Test rollback with None backup result
271+
rollback_success = operation.rollback_on_failure(None, self.test_file, "lmstudio")
272+
self.assertFalse(rollback_success)
273+
274+
275+
if __name__ == '__main__':
276+
unittest.main()

0 commit comments

Comments
 (0)