diff --git a/jsonpatch_cli.py b/jsonpatch_cli.py index bd64115..178e2f3 100644 --- a/jsonpatch_cli.py +++ b/jsonpatch_cli.py @@ -73,6 +73,23 @@ def diff_files(): help='Output Unicode character as-is without using Code Point') +def _link_backup(original, temporary): + """ Links original to its backup name, replacing an older backup """ + # link() does not overwrite an existing file, so the link is created + # under a temporary name, and then moved over an older backup + link = temporary + '.orig' + os.link(original, link) + try: + os.replace(link, original + '.orig') + finally: + # It is still there if replacing failed, and if the older backup is + # a link to original already, as replace() does nothing then + try: + os.unlink(link) + except OSError: + pass + + def jsonpatch_main(): try: patch_files() @@ -121,7 +138,8 @@ def patch_files(): # the original (if desired) and writing a new file. if args.backup: - os.rename(args.ORIGINAL.name, args.ORIGINAL.name + '.orig') + # like in atomic mode, an older backup is replaced + os.replace(args.ORIGINAL.name, args.ORIGINAL.name + '.orig') fp = open(args.ORIGINAL.name, 'w', encoding='utf-8') atomic = False @@ -157,11 +175,12 @@ def patch_files(): if atomic: try: # Complete the atomic replace by linking the original - # to a backup (if desired), fixing up the permissions + # to a backup (if desired), which replaces an older + # backup, fixing up the permissions # on the temporary file, and moving it into place. if args.backup: - os.link(args.ORIGINAL.name, args.ORIGINAL.name + '.orig') + _link_backup(args.ORIGINAL.name, pathname) os.chmod(pathname, os.stat(args.ORIGINAL.name).st_mode) os.replace(pathname, args.ORIGINAL.name) diff --git a/tests.py b/tests.py index f0d7581..65f95dd 100644 --- a/tests.py +++ b/tests.py @@ -3595,12 +3595,73 @@ def test_jsonpatch_in_place_preserves_original_on_chmod_error(self): self.assertEqual(sorted(os.listdir(self.dir)), ['doc.json', 'patch.json']) def test_jsonpatch_in_place_existing_backup(self): + # an older backup is replaced, also when the backup is repeated 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(res.returncode, 0, res.stderr) + self.assertEqual(self.read(doc), {'a': 2}) + self.assertEqual(self.read(backup), {'a': 1}) + + patch = self.write('patch.json', [{'op': 'replace', 'path': '/a', 'value': 3}]) + res = self.run_script('jsonpatch', '-i', '-b', doc, patch) + self.assertEqual(res.returncode, 0, res.stderr) + self.assertEqual(self.read(doc), {'a': 3}) + self.assertEqual(self.read(backup), {'a': 2}) + self.assertEqual(sorted(os.listdir(self.dir)), + ['doc.json', 'doc.json.orig', 'patch.json']) + + def test_jsonpatch_in_place_existing_backup_without_temporary_file(self): + # the fallback that writes ORIGINAL directly replaces it as well + 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}) + parser = jsonpatch_cli.jsonpatch_parser + args = parser.parse_args(['-i', '-b', doc, patch]) + self.addCleanup(args.ORIGINAL.close) + self.addCleanup(args.PATCH.close) + + with mock.patch.object(parser, 'parse_args', return_value=args), \ + mock.patch('tempfile.mkstemp', side_effect=OSError('read-only')): + jsonpatch_cli.patch_files() + + self.assertEqual(self.read(doc), {'a': 2}) + self.assertEqual(self.read(backup), {'a': 1}) + self.assertEqual(sorted(os.listdir(self.dir)), + ['doc.json', 'doc.json.orig', 'patch.json']) + + @unittest.skipUnless(hasattr(os, 'link'), 'needs hard links') + def test_jsonpatch_in_place_backup_that_links_to_original(self): + # left behind if an earlier run failed after linking the backup; + # replacing a link to a file by another one does nothing + doc = self.write('doc.json', {'a': 1}) + patch = self.write('patch.json', [{'op': 'replace', 'path': '/a', 'value': 2}]) + os.link(doc, doc + '.orig') + res = self.run_script('jsonpatch', '-i', '-b', doc, patch) + + self.assertEqual(res.returncode, 0, res.stderr) + self.assertEqual(self.read(doc), {'a': 2}) + self.assertEqual(self.read(doc + '.orig'), {'a': 1}) + self.assertEqual(sorted(os.listdir(self.dir)), + ['doc.json', 'doc.json.orig', 'patch.json']) + + def test_jsonpatch_in_place_keeps_backup_on_backup_error(self): + # neither the original nor an older backup is lost + 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}) + parser = jsonpatch_cli.jsonpatch_parser + args = parser.parse_args(['-i', '-b', 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.replace', side_effect=PermissionError('cannot replace')): + with self.assertRaises(PermissionError): + jsonpatch_cli.patch_files() + self.assertEqual(self.read(doc), {'a': 1}) self.assertEqual(self.read(backup), {'previous': True}) self.assertEqual(sorted(os.listdir(self.dir)),