Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Do not create snapshots if a recent one exists

  • Loading branch information...
commit 666388b79cfc910d51a0f60938d250bc4fcd72ba 1 parent 0bbe96d
@oranenj authored
Showing with 57 additions and 14 deletions.
  1. +32 −0 apt_btrfs_snapshot.py
  2. +25 −14 test/test_apt_btrfs_snapshot.py
View
32 apt_btrfs_snapshot.py
@@ -177,10 +177,41 @@ def _get_now_str(self):
return datetime.datetime.now().replace(microsecond=0).isoformat(
str('_'))
+ def _get_last_snapshot_time(self):
+ last_snapshot = datetime.datetime.fromtimestamp(0.0)
+ last_snapshot_file = '/run/apt_last_snapshot'
+
+ if os.path.exists(last_snapshot_file):
+ try:
+ t = open(last_snapshot_file)
+ last_snapshot = \
+ datetime.datetime.fromtimestamp(float(t.readline()))
+ except:
+ # If we fail to read the timestamp for some reason, just return
+ # the default value silently
+ pass
+ finally:
+ t.close()
+ return last_snapshot
+
+ def _save_last_snapshot_time(self):
+ f = open('/run/apt_last_snapshot', 'w')
+ f.write(str(time.time()))
+ f.close()
+
def create_btrfs_root_snapshot(self, additional_prefix=""):
if self.DISABLED:
print(_("apt-btrfs-snapshot: Disabled, skipping creation"))
return True
+ last = self._get_last_snapshot_time()
+
+ # If a prefix is supplied, the user most likely wants a snapshot even
+ # if there has been one made recently. Otherwise skip making a snapshot
+ # If one has been made recently.
+ if additional_prefix is "" \
+ and last > datetime.datetime.now() - datetime.timedelta(seconds=60):
+ print (_("A recent snapshot already exists: %s") % last)
+ return True
mp = self.mount_btrfs_root_volume()
snap_id = self._get_now_str()
@@ -190,6 +221,7 @@ def create_btrfs_root_snapshot(self, additional_prefix=""):
os.path.join(mp, self.ROOT),
os.path.join(mp, name))
self.umount_btrfs_root_volume()
+ self._save_last_snapshot_time()
return res
def get_btrfs_root_snapshots_list(self, older_than=0):
View
39 test/test_apt_btrfs_snapshot.py
@@ -8,6 +8,7 @@
import mock
import os
import sys
+import datetime
import time
import unittest
@@ -92,23 +93,33 @@ def test_btrfs_create_snapshot(self, mock_commands):
mock_commands.btrfs_subvolume_snapshot.return_value = True
mock_commands.mount.return_value = True
mock_commands.umount.return_value = True
+ f = lambda: datetime.datetime.fromtimestamp(0.0)
# do it
apt_btrfs = AptBtrfsSnapshot(
fstab=os.path.join(self.testdir, "data", "fstab"))
- res = apt_btrfs.create_btrfs_root_snapshot()
- # check results
- self.assertTrue(apt_btrfs.commands.mount.called)
- self.assertTrue(apt_btrfs.commands.umount.called)
- self.assertTrue(res)
- self.assertTrue(apt_btrfs.commands.btrfs_subvolume_snapshot.called)
- (args, kwargs) = apt_btrfs.commands.btrfs_subvolume_snapshot.call_args
- self.assertTrue(len(args), 2)
- self.assertTrue(args[0].endswith("@"))
- self.assertTrue("@apt-snapshot-" in args[1])
- # again with a additional prefix for the snapshot
- res = apt_btrfs.create_btrfs_root_snapshot("release-upgrade-natty-")
- (args, kwargs) = apt_btrfs.commands.btrfs_subvolume_snapshot.call_args
- self.assertTrue("@apt-snapshot-release-upgrade-natty-" in args[1])
+ with mock.patch.multiple(apt_btrfs,
+ _save_last_snapshot_time=nothing,
+ _get_last_snapshot_time=f):
+ res = apt_btrfs.create_btrfs_root_snapshot()
+ # check results
+ self.assertTrue(apt_btrfs.commands.mount.called)
+ self.assertTrue(apt_btrfs.commands.umount.called)
+ self.assertTrue(res)
+ self.assertTrue(apt_btrfs.commands.btrfs_subvolume_snapshot.called)
+ (args, kwargs) = \
+ apt_btrfs.commands.btrfs_subvolume_snapshot.call_args
+
+ self.assertTrue(len(args), 2)
+ self.assertTrue(args[0].endswith("@"))
+ self.assertTrue("@apt-snapshot-" in args[1])
+
+ # again with a additional prefix for the snapshot
+ prefix = "release-upgrade-natty-"
+ res = apt_btrfs.create_btrfs_root_snapshot(prefix)
+ (args, kwargs) = \
+ apt_btrfs.commands.btrfs_subvolume_snapshot.call_args
+
+ self.assertTrue("@apt-snapshot-release-upgrade-natty-" in args[1])
@mock.patch('apt_btrfs_snapshot.LowLevelCommands')
def test_btrfs_delete_snapshot(self, mock_commands):
Please sign in to comment.
Something went wrong with that request. Please try again.