diff --git a/jsonpatch_cli.py b/jsonpatch_cli.py index 93bbe83..34be2ca 100644 --- a/jsonpatch_cli.py +++ b/jsonpatch_cli.py @@ -126,12 +126,11 @@ def patch_files(): if args.backup: os.link(args.ORIGINAL.name, args.ORIGINAL.name + '.orig') os.chmod(pathname, os.stat(args.ORIGINAL.name).st_mode) - os.rename(pathname, args.ORIGINAL.name) + os.replace(pathname, args.ORIGINAL.name) except OSError: - # In the event we could not actually do the atomic - # replace, unlink the original to move it out of the - # way and finally move the temporary file into place. - - os.unlink(args.ORIGINAL.name) - os.rename(pathname, args.ORIGINAL.name) + # Preserve the original if backup, permissions, or replacement + # fails. replace() can overwrite an existing file on Windows + # without unlinking it first. + os.unlink(pathname) + raise diff --git a/tests.py b/tests.py index 57ad381..4f0c553 100755 --- a/tests.py +++ b/tests.py @@ -1902,21 +1902,21 @@ def test_jsonpatch_in_place(self): self.assertEqual(sorted(os.listdir(self.dir)), ['doc.json', 'patch.json']) # 2026-10-11: Windows cannot replace the original while its input is open. - def test_jsonpatch_in_place_closes_original_before_rename(self): + def test_jsonpatch_in_place_closes_original_before_replace(self): doc = self.write('doc.json', {'a': 1}) patch = self.write('patch.json', [{'op': 'replace', 'path': '/a', 'value': 2}]) parser = jsonpatch_cli.jsonpatch_parser args = parser.parse_args(['-i', doc, patch]) self.addCleanup(args.ORIGINAL.close) self.addCleanup(args.PATCH.close) - rename = os.rename + replace = os.replace - def rename_closed_original(source, target): + def replace_closed_original(source, target): self.assertTrue(args.ORIGINAL.closed) - return rename(source, target) + return replace(source, target) with mock.patch.object(parser, 'parse_args', return_value=args), \ - mock.patch('os.rename', side_effect=rename_closed_original): + mock.patch('os.replace', side_effect=replace_closed_original): jsonpatch_cli.patch_files() self.assertEqual(self.read(doc), {'a': 2}) @@ -1937,6 +1937,52 @@ def test_jsonpatch_in_place_backup(self): self.assertEqual(self.read(doc), {'a': 2}) self.assertEqual(self.read(doc + '.orig'), {'a': 1}) + def test_jsonpatch_in_place_preserves_original_on_replace_error(self): + doc = self.write('doc.json', {'a': 1}) + patch = self.write('patch.json', [{'op': 'replace', 'path': '/a', 'value': 2}]) + parser = jsonpatch_cli.jsonpatch_parser + args = parser.parse_args(['-i', doc, patch]) + self.addCleanup(args.ORIGINAL.close) + self.addCleanup(args.PATCH.close) + + with mock.patch.object(parser, 'parse_args', return_value=args), \ + mock.patch('os.rename', side_effect=PermissionError('cannot replace')), \ + mock.patch('os.replace', side_effect=PermissionError('cannot replace')): + with self.assertRaises(PermissionError): + jsonpatch_cli.patch_files() + + self.assertTrue(os.path.exists(doc)) + self.assertEqual(self.read(doc), {'a': 1}) + self.assertEqual(sorted(os.listdir(self.dir)), ['doc.json', 'patch.json']) + + def test_jsonpatch_in_place_preserves_original_on_chmod_error(self): + doc = self.write('doc.json', {'a': 1}) + patch = self.write('patch.json', [{'op': 'replace', 'path': '/a', 'value': 2}]) + parser = jsonpatch_cli.jsonpatch_parser + args = parser.parse_args(['-i', doc, patch]) + self.addCleanup(args.ORIGINAL.close) + self.addCleanup(args.PATCH.close) + + with mock.patch.object(parser, 'parse_args', return_value=args), \ + mock.patch('os.chmod', side_effect=PermissionError('cannot chmod')): + with self.assertRaises(PermissionError): + jsonpatch_cli.patch_files() + + self.assertEqual(self.read(doc), {'a': 1}) + self.assertEqual(sorted(os.listdir(self.dir)), ['doc.json', 'patch.json']) + + def test_jsonpatch_in_place_existing_backup(self): + doc = self.write('doc.json', {'a': 1}) + patch = self.write('patch.json', [{'op': 'replace', 'path': '/a', 'value': 2}]) + backup = self.write('doc.json.orig', {'previous': True}) + res = self.run_script('jsonpatch', '-i', '-b', doc, patch) + + self.assertNotEqual(res.returncode, 0) + self.assertEqual(self.read(doc), {'a': 1}) + self.assertEqual(self.read(backup), {'previous': True}) + self.assertEqual(sorted(os.listdir(self.dir)), + ['doc.json', 'doc.json.orig', 'patch.json']) + def test_jsonpatch_in_place_failing_patch(self): # the patch is applied before the file is replaced, so it is kept doc = self.write('doc.json', {'a': 1})