Merge cd0b9f362b into fdbbe28ac7

without monitor_dont_warn

tests/3_monitor_snapshots/run.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+set -x
+
+# this test will create pools in a number of states
+# and check the output text and return code of 
+# sanoid --monitor-snapshots
+
+. ../common/lib.sh
+
+# prepare
+setup
+checkEnvironment
+disableTimeSync
+
+# set timezone
+ln -sf /usr/share/zoneinfo/Europe/Vienna /etc/localtime
+
+python3 test_monitoring.py

tests/3_monitor_snapshots/sanoid.conf

@@ -0,0 +1,37 @@
+[sanoid-test-1]
+	use_template = production
+
+[sanoid-test-2]
+	use_template = demo
+
+[template_production]
+	hourly = 36
+	daily = 30
+	monthly = 3
+	yearly = 0
+	autosnap = yes
+	autoprune = no
+	hourly_warn = 90m
+	hourly_crit = 360m
+	daily_warn = 28h
+	daily_crit = 32h
+	weekly_warn = 0
+	weekly_crit = 0
+	monthly_warn = 32d
+	monthly_crit = 40d
+	yearly_warn = 0
+	yearly_crit = 0
+	
+
+[template_demo]
+	daily = 60
+	hourly_warn = 290m
+	hourly_crit = 360m
+	daily_warn = 28h
+	daily_crit = 48h
+	weekly_warn = 0
+	weekly_crit = 0
+	monthly_warn = 32d
+	monthly_crit = 40d
+	yearly_warn = 0
+	yearly_crit = 0

tests/3_monitor_snapshots/test_monitoring.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+
+# this software is licensed for use under the Free Software Foundation's GPL v3.0 license, as retrieved
+# from http://www.gnu.org/licenses/gpl-3.0.html on 2014-11-17.  A copy should also be available in this
+# project's Git repository at https://github.com/jimsalterjrs/sanoid/blob/master/LICENSE.
+
+
+import os
+import subprocess
+import time
+import unittest
+
+
+sanoid_cmd = os.environ.get("SANOID")
+pool_disk_image1 = "/zpool1.img"
+pool_name1 = "sanoid-test-1"
+pool_disk_image2 = "/zpool2.img"
+pool_name2 = "sanoid-test-2"
+
+clk_id = time.CLOCK_REALTIME
+starting_time = time.clock_gettime(clk_id)
+
+
+def monitor_snapshots_command():
+    """Runs sanoid --monitor-snapshots and returns a CompletedProcess instance"""
+    return_info = subprocess.run([sanoid_cmd,  "--monitor-snapshots"], capture_output=True)
+    return return_info
+    
+def run_sanoid_cron_command():
+    """Runs sanoid and returns a CompletedProcess instance"""
+    return_info = subprocess.run([sanoid_cmd,  "--cron", "--verbose"], capture_output=True, check=True)
+    return return_info
+
+def advance_time(seconds):
+    """Advances the system clock by seconds"""
+    
+    # Get the current time
+    clk_id = time.CLOCK_REALTIME
+    time_seconds = time.clock_gettime(clk_id)
+    # print("Current unix time is", time_seconds, "or", time.asctime(time.gmtime(time_seconds)), "in GMT")
+    
+    # Set the clock to the current time plus seconds
+    time.clock_settime(clk_id, time_seconds + seconds)
+
+    # Print the new time
+    time_seconds = time.clock_gettime(clk_id)
+    # print("Current unix time is", time_seconds, "or", time.asctime(time.gmtime(time_seconds)), "in GMT")
+    return time_seconds
+
+
+class TestMonitoringOutput(unittest.TestCase):
+    def test_no_zpool(self):
+        """Test what happens if there is no zpool at all"""
+
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"CRIT: sanoid-test-1 has no daily snapshots at all!, CRIT: sanoid-test-1 has no hourly snapshots at all!, CRIT: sanoid-test-1 has no monthly snapshots at all!, CRIT: sanoid-test-2 has no daily snapshots at all!, CRIT: sanoid-test-2 has no hourly snapshots at all!, CRIT: sanoid-test-2 has no monthly snapshots at all!\n")
+        self.assertEqual(return_info.returncode, 2)
+
+class TestsWithZpool(unittest.TestCase):
+    """Tests that require a test zpool"""
+
+    def setUp(self):
+        """Set up the zpool"""
+        subprocess.run(["truncate", "-s", "512M", pool_disk_image1], check=True)
+        subprocess.run(["zpool", "create", "-f", pool_name1, pool_disk_image1], check=True)
+
+        subprocess.run(["truncate", "-s", "512M", pool_disk_image2], check=True)
+        subprocess.run(["zpool", "create", "-f", pool_name2, pool_disk_image2], check=True)
+
+        # Clear the snapshot cache in between
+        subprocess.run([sanoid_cmd, "--force-update"])
+
+    def tearDown(self):
+        """Clean up on either passed or failed tests"""
+        subprocess.run(["zpool", "export", pool_name1])
+        subprocess.run(["rm", "-f", pool_disk_image1])
+        subprocess.run(["zpool", "export", pool_name2])
+        subprocess.run(["rm", "-f", pool_disk_image2])
+        time.clock_settime(clk_id, starting_time)
+
+    def test_with_zpool_no_snapshots(self):
+        """Test what happens if there is a zpool, but with no snapshots"""
+
+        # Run sanoid --monitor-snapshots before doing anything else
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"CRIT: sanoid-test-1 has no daily snapshots at all!, CRIT: sanoid-test-1 has no hourly snapshots at all!, CRIT: sanoid-test-1 has no monthly snapshots at all!, CRIT: sanoid-test-2 has no daily snapshots at all!, CRIT: sanoid-test-2 has no hourly snapshots at all!, CRIT: sanoid-test-2 has no monthly snapshots at all!\n")
+        self.assertEqual(return_info.returncode, 2)
+
+    def test_immediately_after_running_sanoid(self):
+        """Test immediately after running sanoid --cron"""
+
+        run_sanoid_cron_command()
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"OK: all monitored datasets (sanoid-test-1, sanoid-test-2) have fresh snapshots\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_one_warning_hourly(self):
+        """Test one warning on hourly snapshots, no critical warnings, to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance 100 mins to trigger the hourly warning on sanoid-test-1 but nothing else
+        advance_time(100 * 60)
+        return_info = monitor_snapshots_command()
+        # Output should be something like: 
+        # WARN: sanoid-test-1 newest hourly snapshot is 1h 40m 0s old (should be < 1h 30m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        self.assertEqual(return_info.stdout[:49], b"WARN: sanoid-test-1 newest hourly snapshot is 1h ")
+        self.assertEqual(return_info.stdout[-30:], b"s old (should be < 1h 30m 0s)\n")
+        self.assertEqual(return_info.returncode, 1)
+
+    def test_two_criticals_hourly(self):
+        """Test two criticals (hourly), to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance 390 mins to trigger the hourly critical on both sanoid-test-1 and sanoid-test-2
+        advance_time(390 * 60)
+        return_info = monitor_snapshots_command()
+
+        # Output should be something like: 
+        # CRIT: sanoid-test-1 newest hourly snapshot is 6h 30m 1s old (should be < 6h 0m 0s), CRIT: sanoid-test-2 newest hourly snapshot is 6h 30m 1s old (should be < 6h 0m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        comma_location = return_info.stdout.find(b",")
+        self.assertEqual(return_info.stdout[:49], b"CRIT: sanoid-test-1 newest hourly snapshot is 6h ")
+        self.assertEqual(return_info.stdout[comma_location - 28:comma_location], b"s old (should be < 6h 0m 0s)")
+        self.assertEqual(return_info.stdout[comma_location:comma_location + 51], b", CRIT: sanoid-test-2 newest hourly snapshot is 6h ")
+        self.assertEqual(return_info.stdout[-29:], b"s old (should be < 6h 0m 0s)\n")
+        self.assertEqual(return_info.returncode, 2)
+
+    def test_two_criticals_hourly_two_warnings_daily(self):
+        """Test two criticals (hourly) and two warnings (daily), to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance more than 28 hours to trigger the daily warning on both sanoid-test-1 and sanoid-test-2
+        advance_time(29 * 60 * 60)
+        return_info = monitor_snapshots_command()
+
+        # Output should be something like: 
+        # CRIT: sanoid-test-1 newest hourly snapshot is 1d 5h 0m 0s old (should be < 6h 0m 0s), CRIT: sanoid-test-2 newest hourly snapshot is 1d 5h 0m 0s old (should be < 6h 0m 0s), WARN: sanoid-test-1 newest daily snapshot is 1d 5h 0m 0s old (should be < 1d 4h 0m 0s), WARN: sanoid-test-2 newest daily snapshot is 1d 5h 0m 0s old (should be < 1d 4h 0m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        print(return_info.stdout)
+        output_list = return_info.stdout.split(b", ")
+
+        self.assertEqual(output_list[0][:49], b"CRIT: sanoid-test-1 newest hourly snapshot is 1d ")
+        self.assertEqual(output_list[0][-28:], b"s old (should be < 6h 0m 0s)")
+        self.assertEqual(output_list[1][:49], b"CRIT: sanoid-test-2 newest hourly snapshot is 1d ")
+        self.assertEqual(output_list[1][-28:], b"s old (should be < 6h 0m 0s)")
+        
+        self.assertEqual(output_list[2][:48], b"WARN: sanoid-test-1 newest daily snapshot is 1d ")
+        self.assertEqual(output_list[2][-31:], b"s old (should be < 1d 4h 0m 0s)")
+        self.assertEqual(output_list[3][:48], b"WARN: sanoid-test-2 newest daily snapshot is 1d ")
+        self.assertEqual(output_list[3][-32:], b"s old (should be < 1d 4h 0m 0s)\n")
+
+        self.assertEqual(return_info.returncode, 2)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/4_monitor_snapshots_dont_warncrit/run.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+set -x
+
+# this test will create pools in a number of states
+# and check the output text and return code of 
+# sanoid --monitor-snapshots
+
+. ../common/lib.sh
+
+# prepare
+setup
+checkEnvironment
+disableTimeSync
+
+# set timezone
+ln -sf /usr/share/zoneinfo/Europe/Vienna /etc/localtime
+
+python3 test_monitoring.py

tests/4_monitor_snapshots_dont_warncrit/sanoid.conf

@@ -0,0 +1,41 @@
+[sanoid-test-1]
+	use_template = production
+
+[sanoid-test-2]
+	use_template = demo
+
+[template_production]
+	hourly = 36
+	daily = 30
+	monthly = 3
+	yearly = 0
+	autosnap = yes
+	autoprune = no
+	hourly_warn = 90m
+	hourly_crit = 360m
+	daily_warn = 28h
+	daily_crit = 32h
+	weekly_warn = 0
+	weekly_crit = 0
+	monthly_warn = 32d
+	monthly_crit = 40d
+	yearly_warn = 0
+	yearly_crit = 0
+	monitor_dont_warn = true
+	monitor_dont_crit = Yes
+	
+
+[template_demo]
+	daily = 60
+	hourly_warn = 290m
+	hourly_crit = 360m
+	daily_warn = 28h
+	daily_crit = 48h
+	weekly_warn = 0
+	weekly_crit = 0
+	monthly_warn = 32d
+	monthly_crit = 40d
+	yearly_warn = 0
+	yearly_crit = 0
+	monitor_dont_warn = ON
+	monitor_dont_crit = 1

tests/4_monitor_snapshots_dont_warncrit/test_monitoring.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+
+# this software is licensed for use under the Free Software Foundation's GPL v3.0 license, as retrieved
+# from http://www.gnu.org/licenses/gpl-3.0.html on 2014-11-17.  A copy should also be available in this
+# project's Git repository at https://github.com/jimsalterjrs/sanoid/blob/master/LICENSE.
+
+
+import os
+import subprocess
+import time
+import unittest
+
+
+sanoid_cmd = os.environ.get("SANOID")
+pool_disk_image1 = "/zpool1.img"
+pool_name1 = "sanoid-test-1"
+pool_disk_image2 = "/zpool2.img"
+pool_name2 = "sanoid-test-2"
+
+clk_id = time.CLOCK_REALTIME
+starting_time = time.clock_gettime(clk_id)
+
+
+def monitor_snapshots_command():
+    """Runs sanoid --monitor-snapshots and returns a CompletedProcess instance"""
+    return_info = subprocess.run([sanoid_cmd,  "--monitor-snapshots"], capture_output=True)
+    return return_info
+    
+def run_sanoid_cron_command():
+    """Runs sanoid and returns a CompletedProcess instance"""
+    return_info = subprocess.run([sanoid_cmd,  "--cron", "--verbose"], capture_output=True, check=True)
+    return return_info
+
+def advance_time(seconds):
+    """Advances the system clock by seconds"""
+    
+    # Get the current time
+    clk_id = time.CLOCK_REALTIME
+    time_seconds = time.clock_gettime(clk_id)
+    # print("Current unix time is", time_seconds, "or", time.asctime(time.gmtime(time_seconds)), "in GMT")
+    
+    # Set the clock to the current time plus seconds
+    time.clock_settime(clk_id, time_seconds + seconds)
+
+    # Print the new time
+    time_seconds = time.clock_gettime(clk_id)
+    # print("Current unix time is", time_seconds, "or", time.asctime(time.gmtime(time_seconds)), "in GMT")
+    return time_seconds
+
+
+class TestMonitoringOutput(unittest.TestCase):
+    def test_no_zpool(self):
+        """Test what happens if there is no zpool at all"""
+
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"CRIT: sanoid-test-1 has no daily snapshots at all!, CRIT: sanoid-test-1 has no hourly snapshots at all!, CRIT: sanoid-test-1 has no monthly snapshots at all!, CRIT: sanoid-test-2 has no daily snapshots at all!, CRIT: sanoid-test-2 has no hourly snapshots at all!, CRIT: sanoid-test-2 has no monthly snapshots at all!\n")
+        self.assertEqual(return_info.returncode, 0)
+
+class TestsWithZpool(unittest.TestCase):
+    """Tests that require a test zpool"""
+
+    def setUp(self):
+        """Set up the zpool"""
+        subprocess.run(["truncate", "-s", "512M", pool_disk_image1], check=True)
+        subprocess.run(["zpool", "create", "-f", pool_name1, pool_disk_image1], check=True)
+
+        subprocess.run(["truncate", "-s", "512M", pool_disk_image2], check=True)
+        subprocess.run(["zpool", "create", "-f", pool_name2, pool_disk_image2], check=True)
+
+        # Clear the snapshot cache in between
+        subprocess.run([sanoid_cmd, "--force-update"])
+
+    def tearDown(self):
+        """Clean up on either passed or failed tests"""
+        subprocess.run(["zpool", "export", pool_name1])
+        subprocess.run(["rm", "-f", pool_disk_image1])
+        subprocess.run(["zpool", "export", pool_name2])
+        subprocess.run(["rm", "-f", pool_disk_image2])
+        time.clock_settime(clk_id, starting_time)
+
+    def test_with_zpool_no_snapshots(self):
+        """Test what happens if there is a zpool, but with no snapshots"""
+
+        # Run sanoid --monitor-snapshots before doing anything else
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"CRIT: sanoid-test-1 has no daily snapshots at all!, CRIT: sanoid-test-1 has no hourly snapshots at all!, CRIT: sanoid-test-1 has no monthly snapshots at all!, CRIT: sanoid-test-2 has no daily snapshots at all!, CRIT: sanoid-test-2 has no hourly snapshots at all!, CRIT: sanoid-test-2 has no monthly snapshots at all!\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_immediately_after_running_sanoid(self):
+        """Test immediately after running sanoid --cron"""
+
+        run_sanoid_cron_command()
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"OK: all monitored datasets (sanoid-test-1, sanoid-test-2) have fresh snapshots\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_one_warning_hourly(self):
+        """Test one warning on hourly snapshots, no critical warnings, to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance 100 mins to trigger the hourly warning on sanoid-test-1 but nothing else
+        advance_time(100 * 60)
+        return_info = monitor_snapshots_command()
+        # Output should be something like: 
+        # WARN: sanoid-test-1 newest hourly snapshot is 1h 40m 0s old (should be < 1h 30m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        self.assertEqual(return_info.stdout[:49], b"WARN: sanoid-test-1 newest hourly snapshot is 1h ")
+        self.assertEqual(return_info.stdout[-30:], b"s old (should be < 1h 30m 0s)\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_two_criticals_hourly(self):
+        """Test two criticals (hourly), to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance 390 mins to trigger the hourly critical on both sanoid-test-1 and sanoid-test-2
+        advance_time(390 * 60)
+        return_info = monitor_snapshots_command()
+
+        # Output should be something like: 
+        # CRIT: sanoid-test-1 newest hourly snapshot is 6h 30m 1s old (should be < 6h 0m 0s), CRIT: sanoid-test-2 newest hourly snapshot is 6h 30m 1s old (should be < 6h 0m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        comma_location = return_info.stdout.find(b",")
+        self.assertEqual(return_info.stdout[:49], b"CRIT: sanoid-test-1 newest hourly snapshot is 6h ")
+        self.assertEqual(return_info.stdout[comma_location - 28:comma_location], b"s old (should be < 6h 0m 0s)")
+        self.assertEqual(return_info.stdout[comma_location:comma_location + 51], b", CRIT: sanoid-test-2 newest hourly snapshot is 6h ")
+        self.assertEqual(return_info.stdout[-29:], b"s old (should be < 6h 0m 0s)\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_two_criticals_hourly_two_warnings_daily(self):
+        """Test two criticals (hourly) and two warnings (daily), to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance more than 28 hours to trigger the daily warning on both sanoid-test-1 and sanoid-test-2
+        advance_time(29 * 60 * 60)
+        return_info = monitor_snapshots_command()
+
+        # Output should be something like: 
+        # CRIT: sanoid-test-1 newest hourly snapshot is 1d 5h 0m 0s old (should be < 6h 0m 0s), CRIT: sanoid-test-2 newest hourly snapshot is 1d 5h 0m 0s old (should be < 6h 0m 0s), WARN: sanoid-test-1 newest daily snapshot is 1d 5h 0m 0s old (should be < 1d 4h 0m 0s), WARN: sanoid-test-2 newest daily snapshot is 1d 5h 0m 0s old (should be < 1d 4h 0m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        print(return_info.stdout)
+        output_list = return_info.stdout.split(b", ")
+
+        self.assertEqual(output_list[0][:49], b"CRIT: sanoid-test-1 newest hourly snapshot is 1d ")
+        self.assertEqual(output_list[0][-28:], b"s old (should be < 6h 0m 0s)")
+        self.assertEqual(output_list[1][:49], b"CRIT: sanoid-test-2 newest hourly snapshot is 1d ")
+        self.assertEqual(output_list[1][-28:], b"s old (should be < 6h 0m 0s)")
+        
+        self.assertEqual(output_list[2][:48], b"WARN: sanoid-test-1 newest daily snapshot is 1d ")
+        self.assertEqual(output_list[2][-31:], b"s old (should be < 1d 4h 0m 0s)")
+        self.assertEqual(output_list[3][:48], b"WARN: sanoid-test-2 newest daily snapshot is 1d ")
+        self.assertEqual(output_list[3][-32:], b"s old (should be < 1d 4h 0m 0s)\n")
+
+        self.assertEqual(return_info.returncode, 0)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/5_monitor_snapshots_warn_dont_crit/run.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+set -x
+
+# this test will create pools in a number of states
+# and check the output text and return code of 
+# sanoid --monitor-snapshots
+
+. ../common/lib.sh
+
+# prepare
+setup
+checkEnvironment
+disableTimeSync
+
+# set timezone
+ln -sf /usr/share/zoneinfo/Europe/Vienna /etc/localtime
+
+python3 test_monitoring.py

tests/5_monitor_snapshots_warn_dont_crit/sanoid.conf

@@ -0,0 +1,41 @@
+[sanoid-test-1]
+	use_template = production
+
+[sanoid-test-2]
+	use_template = demo
+
+[template_production]
+	hourly = 36
+	daily = 30
+	monthly = 3
+	yearly = 0
+	autosnap = yes
+	autoprune = no
+	hourly_warn = 90m
+	hourly_crit = 360m
+	daily_warn = 28h
+	daily_crit = 32h
+	weekly_warn = 0
+	weekly_crit = 0
+	monthly_warn = 32d
+	monthly_crit = 40d
+	yearly_warn = 0
+	yearly_crit = 0
+	monitor_dont_warn = FALSE
+	monitor_dont_crit = yes
+	
+
+[template_demo]
+	daily = 60
+	hourly_warn = 290m
+	hourly_crit = 360m
+	daily_warn = 28h
+	daily_crit = 48h
+	weekly_warn = 0
+	weekly_crit = 0
+	monthly_warn = 32d
+	monthly_crit = 40d
+	yearly_warn = 0
+	yearly_crit = 0
+	monitor_dont_warn = Off
+	monitor_dont_crit = On

tests/5_monitor_snapshots_warn_dont_crit/test_monitoring.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+
+# this software is licensed for use under the Free Software Foundation's GPL v3.0 license, as retrieved
+# from http://www.gnu.org/licenses/gpl-3.0.html on 2014-11-17.  A copy should also be available in this
+# project's Git repository at https://github.com/jimsalterjrs/sanoid/blob/master/LICENSE.
+
+
+import os
+import subprocess
+import time
+import unittest
+
+
+sanoid_cmd = os.environ.get("SANOID")
+pool_disk_image1 = "/zpool1.img"
+pool_name1 = "sanoid-test-1"
+pool_disk_image2 = "/zpool2.img"
+pool_name2 = "sanoid-test-2"
+
+clk_id = time.CLOCK_REALTIME
+starting_time = time.clock_gettime(clk_id)
+
+
+def monitor_snapshots_command():
+    """Runs sanoid --monitor-snapshots and returns a CompletedProcess instance"""
+    return_info = subprocess.run([sanoid_cmd,  "--monitor-snapshots"], capture_output=True)
+    return return_info
+    
+def run_sanoid_cron_command():
+    """Runs sanoid and returns a CompletedProcess instance"""
+    return_info = subprocess.run([sanoid_cmd,  "--cron", "--verbose"], capture_output=True, check=True)
+    return return_info
+
+def advance_time(seconds):
+    """Advances the system clock by seconds"""
+    
+    # Get the current time
+    clk_id = time.CLOCK_REALTIME
+    time_seconds = time.clock_gettime(clk_id)
+    # print("Current unix time is", time_seconds, "or", time.asctime(time.gmtime(time_seconds)), "in GMT")
+    
+    # Set the clock to the current time plus seconds
+    time.clock_settime(clk_id, time_seconds + seconds)
+
+    # Print the new time
+    time_seconds = time.clock_gettime(clk_id)
+    # print("Current unix time is", time_seconds, "or", time.asctime(time.gmtime(time_seconds)), "in GMT")
+    return time_seconds
+
+
+class TestMonitoringOutput(unittest.TestCase):
+    def test_no_zpool(self):
+        """Test what happens if there is no zpool at all"""
+
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"CRIT: sanoid-test-1 has no daily snapshots at all!, CRIT: sanoid-test-1 has no hourly snapshots at all!, CRIT: sanoid-test-1 has no monthly snapshots at all!, CRIT: sanoid-test-2 has no daily snapshots at all!, CRIT: sanoid-test-2 has no hourly snapshots at all!, CRIT: sanoid-test-2 has no monthly snapshots at all!\n")
+        self.assertEqual(return_info.returncode, 0)
+
+class TestsWithZpool(unittest.TestCase):
+    """Tests that require a test zpool"""
+
+    def setUp(self):
+        """Set up the zpool"""
+        subprocess.run(["truncate", "-s", "512M", pool_disk_image1], check=True)
+        subprocess.run(["zpool", "create", "-f", pool_name1, pool_disk_image1], check=True)
+
+        subprocess.run(["truncate", "-s", "512M", pool_disk_image2], check=True)
+        subprocess.run(["zpool", "create", "-f", pool_name2, pool_disk_image2], check=True)
+
+        # Clear the snapshot cache in between
+        subprocess.run([sanoid_cmd, "--force-update"])
+
+    def tearDown(self):
+        """Clean up on either passed or failed tests"""
+        subprocess.run(["zpool", "export", pool_name1])
+        subprocess.run(["rm", "-f", pool_disk_image1])
+        subprocess.run(["zpool", "export", pool_name2])
+        subprocess.run(["rm", "-f", pool_disk_image2])
+        time.clock_settime(clk_id, starting_time)
+
+    def test_with_zpool_no_snapshots(self):
+        """Test what happens if there is a zpool, but with no snapshots"""
+
+        # Run sanoid --monitor-snapshots before doing anything else
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"CRIT: sanoid-test-1 has no daily snapshots at all!, CRIT: sanoid-test-1 has no hourly snapshots at all!, CRIT: sanoid-test-1 has no monthly snapshots at all!, CRIT: sanoid-test-2 has no daily snapshots at all!, CRIT: sanoid-test-2 has no hourly snapshots at all!, CRIT: sanoid-test-2 has no monthly snapshots at all!\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_immediately_after_running_sanoid(self):
+        """Test immediately after running sanoid --cron"""
+
+        run_sanoid_cron_command()
+        return_info = monitor_snapshots_command()
+        self.assertEqual(return_info.stdout, b"OK: all monitored datasets (sanoid-test-1, sanoid-test-2) have fresh snapshots\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_one_warning_hourly(self):
+        """Test one warning on hourly snapshots, no critical warnings, to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance 100 mins to trigger the hourly warning on sanoid-test-1 but nothing else
+        advance_time(100 * 60)
+        return_info = monitor_snapshots_command()
+        # Output should be something like: 
+        # WARN: sanoid-test-1 newest hourly snapshot is 1h 40m 0s old (should be < 1h 30m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        self.assertEqual(return_info.stdout[:49], b"WARN: sanoid-test-1 newest hourly snapshot is 1h ")
+        self.assertEqual(return_info.stdout[-30:], b"s old (should be < 1h 30m 0s)\n")
+        self.assertEqual(return_info.returncode, 1)
+
+    def test_two_criticals_hourly(self):
+        """Test two criticals (hourly), to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance 390 mins to trigger the hourly critical on both sanoid-test-1 and sanoid-test-2
+        advance_time(390 * 60)
+        return_info = monitor_snapshots_command()
+
+        # Output should be something like: 
+        # CRIT: sanoid-test-1 newest hourly snapshot is 6h 30m 1s old (should be < 6h 0m 0s), CRIT: sanoid-test-2 newest hourly snapshot is 6h 30m 1s old (should be < 6h 0m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        comma_location = return_info.stdout.find(b",")
+        self.assertEqual(return_info.stdout[:49], b"CRIT: sanoid-test-1 newest hourly snapshot is 6h ")
+        self.assertEqual(return_info.stdout[comma_location - 28:comma_location], b"s old (should be < 6h 0m 0s)")
+        self.assertEqual(return_info.stdout[comma_location:comma_location + 51], b", CRIT: sanoid-test-2 newest hourly snapshot is 6h ")
+        self.assertEqual(return_info.stdout[-29:], b"s old (should be < 6h 0m 0s)\n")
+        self.assertEqual(return_info.returncode, 0)
+
+    def test_two_criticals_hourly_two_warnings_daily(self):
+        """Test two criticals (hourly) and two warnings (daily), to check output and error status"""
+
+        run_sanoid_cron_command()
+        
+        # Advance more than 28 hours to trigger the daily warning on both sanoid-test-1 and sanoid-test-2
+        advance_time(29 * 60 * 60)
+        return_info = monitor_snapshots_command()
+
+        # Output should be something like: 
+        # CRIT: sanoid-test-1 newest hourly snapshot is 1d 5h 0m 0s old (should be < 6h 0m 0s), CRIT: sanoid-test-2 newest hourly snapshot is 1d 5h 0m 0s old (should be < 6h 0m 0s), WARN: sanoid-test-1 newest daily snapshot is 1d 5h 0m 0s old (should be < 1d 4h 0m 0s), WARN: sanoid-test-2 newest daily snapshot is 1d 5h 0m 0s old (should be < 1d 4h 0m 0s)\n
+        # But we cannot be sure about the exact time as test execution could be different on
+        # different machines, so ignore the bits that may be different.
+        print(return_info.stdout)
+        output_list = return_info.stdout.split(b", ")
+
+        self.assertEqual(output_list[0][:49], b"CRIT: sanoid-test-1 newest hourly snapshot is 1d ")
+        self.assertEqual(output_list[0][-28:], b"s old (should be < 6h 0m 0s)")
+        self.assertEqual(output_list[1][:49], b"CRIT: sanoid-test-2 newest hourly snapshot is 1d ")
+        self.assertEqual(output_list[1][-28:], b"s old (should be < 6h 0m 0s)")
+        
+        self.assertEqual(output_list[2][:48], b"WARN: sanoid-test-1 newest daily snapshot is 1d ")
+        self.assertEqual(output_list[2][-31:], b"s old (should be < 1d 4h 0m 0s)")
+        self.assertEqual(output_list[3][:48], b"WARN: sanoid-test-2 newest daily snapshot is 1d ")
+        self.assertEqual(output_list[3][-32:], b"s old (should be < 1d 4h 0m 0s)\n")
+
+        self.assertEqual(return_info.returncode, 1)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/README.md

@@ -0,0 +1,31 @@
+### Requirements ###
+Tests must be run inside a virtual machine. This is for your own safety, as the tests may create and destroy zpools etc.
+
+A VM with 35GB of storage and 8 cores running Ubuntu 20.04 completes the tests in about 5 hours.  
+
+#### Packages ####
+The tests require the following packages to be installed in the VM (Ubuntu 20.04 package names are used, translate as appropriate):
+```
+zfsutils-linux
+libconfig-inifiles-perl
+libcapture-tiny-perl
+```
+```
+apt install zfsutils-linux libconfig-inifiles-perl libcapture-tiny-perl
+```
+
+#### Install sanoid within the VM ####
+Install sanoid within the VM, for example
+```
+apt install git
+git clone https://github.com/jimsalterjrs/sanoid.git
+mkdir /etc/sanoid/
+cp sanoid/sanoid.defaults.conf /etc/sanoid/
+```
+
+### Run the tests ##
+This requires root/sudo privileges.
+```
+cd sanoid/tests/
+./run-tests.sh
+```

tests/common/lib.sh

@@ -10,7 +10,7 @@ function setup {
     export SANOID="../../sanoid"
 
     # make sure that there is no cache file
-    rm -f /var/cache/sanoidsnapshots.txt
+    rm -f /var/cache/sanoid/snapshots.txt
 
     # install needed sanoid configuration files
     [ -f sanoid.conf ] && cp sanoid.conf /etc/sanoid/sanoid.conf

tests/run-tests.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-# run's all the available tests
+# runs all the available tests
 
 for test in $(find . -mindepth 1 -maxdepth 1 -type d -printf "%P\n" | sort -g); do
     if [ ! -x "${test}/run.sh" ]; then