[Platform API][watchdog] Inherit from PlatformApiTestBase

tests/platform_tests/api/test_watchdog.py

@@ -3,7 +3,9 @@
 import logging
 import yaml
 import pytest
-from tests.common.helpers.platform_api import watchdog
+from common.helpers.platform_api import watchdog
+from common.helpers.assertions import pytest_assert
+from platform_api_test_base import PlatformApiTestBase
 
 pytestmark = [
     pytest.mark.topology('any')
@@ -15,7 +17,8 @@
 TEST_WAIT_TIME_SECONDS = 2
 TIMEOUT_DEVIATION = 2
 
-class TestWatchdogApi(object):
+
+class TestWatchdogApi(PlatformApiTestBase):
     ''' Hardware watchdog platform API test cases '''
 
     @pytest.fixture(scope='function', autouse=True)
@@ -47,57 +50,78 @@ def conf(self, request, duthost):
 
         if platform in test_config and 'default' in test_config[platform]:
             config.update(test_config[platform]['default'])
-
         if platform in test_config and hwsku in test_config[platform]:
             config.update(test_config[platform][hwsku])
 
-        assert 'valid_timeout' in config
+        pytest_assert('valid_timeout' in config, "valid_timeout is not defined in config")
         # make sure watchdog won't reboot the system when test sleeps for @TEST_WAIT_TIME_SECONDS
-        assert config['valid_timeout'] > TEST_WAIT_TIME_SECONDS * 2
-
+        pytest_assert(config['valid_timeout'] > TEST_WAIT_TIME_SECONDS * 2, "valid_timeout {} seconds is too short".format(config['valid_timeout']))
         return config
 
-
+    @pytest.mark.dependency()
     def test_arm_disarm_states(self, duthost, localhost, platform_api_conn, conf):
         ''' arm watchdog with a valid timeout value, verify it is in armed state,
         disarm watchdog and verify it is in disarmed state
         '''
         watchdog_timeout = conf['valid_timeout']
         actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
 
-        assert actual_timeout != -1
-        assert actual_timeout >= watchdog_timeout
-        assert watchdog.is_armed(platform_api_conn)
+        if self.expect(actual_timeout is not None, "Watchdog.arm is not supported"):
+            if self.expect(isinstance(actual_timeout, int), "actual_timeout appears incorrect"):
+                if self.expect(actual_timeout != -1, "Failed to arm the watchdog"):
+                    self.expect(actual_timeout >= watchdog_timeout, "Actual watchdog timeout {} seconds appears wrong, should be less than {} seconds".format(actual_timeout, watchdog_timeout))
 
-        assert watchdog.disarm(platform_api_conn)
-        assert not watchdog.is_armed(platform_api_conn)
+        watchdog_status = watchdog.is_armed(platform_api_conn)
+        if self.expect(watchdog_status is not None, "Failed to retrieve watchdog status"):
+            self.expect(watchdog_status is True, "Watchdog is not armed.")
+
+        remaining_time = watchdog.get_remaining_time(platform_api_conn)
 
-        res = localhost.wait_for(host=duthost.hostname,
-                port=22, state="stopped", delay=5,
-                timeout=watchdog_timeout + TIMEOUT_DEVIATION,
-                module_ignore_errors=True)
+        if self.expect(remaining_time is not None, "Failed to get the remaining time of watchdog"):
+            if self.expect(isinstance(remaining_time, int), "remaining_time appears incorrect"):
+                self.expect(remaining_time <= watchdog_timeout, "Watchdog remaining_time {} seconds appears wrong compared to watchdog timeout {} seocnds".format(remaining_time))
 
-        assert 'exception' in res
+        watchdog_status = watchdog.disarm(platform_api_conn)
+        if self.expect(watchdog_status is not None, "Watchdog.disarm is not supported"):
+            self.expect(watchdog_status is True, "Failed to disarm the watchdog")
 
+        watchdog_status = watchdog.is_armed(platform_api_conn)
+        if self.expect(watchdog_status is not None, "Failed to check the watchdog status"):
+            self.expect(watchdog_status is False, "Watchdog is not disarmed")
+
+        remaining_time = watchdog.get_remaining_time(platform_api_conn)
+        if self.expect(remaining_time is not None, "Failed to get the remaining time of watchdog"):
+            self.expect(remaining_time is -1, "Watchdog remaining_time {} seconds is wrong for disarmed state".format(remaining_time))
+
+        res = localhost.wait_for(host=duthost.hostname, port=22, state="stopped", delay=5, timeout=watchdog_timeout + TIMEOUT_DEVIATION, module_ignore_errors=True)
+
+        self.expect('exception' in res, "unexpected disconnection from dut")
+        self.assert_expectations()
+
+    @pytest.mark.dependency(depends=["test_arm_disarm_states"])
     def test_remaining_time(self, duthost, platform_api_conn, conf):
         ''' arm watchdog with a valid timeout and verify that remaining time API works correctly '''
 
         watchdog_timeout = conf['valid_timeout']
 
         # in the begginging of the test watchdog is not armed, so
         # get_remaining_time has to return -1
-        assert watchdog.get_remaining_time(platform_api_conn) == -1
-
-        actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
         remaining_time = watchdog.get_remaining_time(platform_api_conn)
+        if self.expect(remaining_time is not None and remaining_time is -1, "watchdog should be disabled in the initial state"):
+            actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
+            remaining_time = watchdog.get_remaining_time(platform_api_conn)
 
-        assert remaining_time > 0
-        assert remaining_time <= actual_timeout
+            if self.expect(actual_timeout >= watchdog_timeout, "watchdog arm with {} seconds failed".format(watchdog_timeout)):
+                if self.expect(remaining_time > 0, "Remaining_time {} seconds is not valid".format(remaining_time)):
+                    self.expect(remaining_time <= actual_timeout, "Remaining_time {} seconds should be less than watchdog armed timeout {} seconds".format(remaining_timeout, actual_timeout))
 
         remaining_time = watchdog.get_remaining_time(platform_api_conn)
         time.sleep(TEST_WAIT_TIME_SECONDS)
-        assert watchdog.get_remaining_time(platform_api_conn) < remaining_time
+        remaining_time_new = watchdog.get_remaining_time(platform_api_conn)
+        self.expect(remaining_time_new < remaining_time, "Remaining_time {} seconds should be decreased from previous remaining_time {} seconds".format(remaining_time_new, remaining_time))
+        self.assert_expectations()
 
+    @pytest.mark.dependency(depends=["test_arm_disarm_states"])
     def test_periodic_arm(self, duthost, platform_api_conn, conf):
         ''' arm watchdog several times as watchdog deamon would and verify API behaves correctly '''
 
@@ -106,10 +130,13 @@ def test_periodic_arm(self, duthost, platform_api_conn, conf):
         time.sleep(TEST_WAIT_TIME_SECONDS)
         remaining_time = watchdog.get_remaining_time(platform_api_conn)
         actual_timeout_new = watchdog.arm(platform_api_conn, watchdog_timeout)
+        remaining_time_new = watchdog.get_remaining_time(platform_api_conn)
 
-        assert actual_timeout == actual_timeout_new
-        assert watchdog.get_remaining_time(platform_api_conn) > remaining_time
+        self.expect(actual_timeout == actual_timeout_new, "{}: new watchdog timeout {} seconds setting should be same as the previous actual watchdog timeout {} seconds".format(test_periodic_arm.__name__, actual_timeout_new, actual_timeout))
+        self.expect(remaining_time_new > remaining_time, "{}: new remaining timeout {} seconds should be greater than the previous remaining timeout {} seconds by {} seconds".format(test_periodic_arm.__name__, remaining_time_new, remaining_time, TEST_WAIT_TIME_SECONDS))
+        self.assert_expectations()
 
+    @pytest.mark.dependency(depends=["test_arm_disarm_states"])
     def test_arm_different_timeout_greater(self, duthost, platform_api_conn, conf):
         ''' arm the watchdog with greater timeout value and verify new timeout was accepted;
         If platform accepts only single valid timeout value, @greater_timeout should be None.
@@ -119,13 +146,15 @@ def test_arm_different_timeout_greater(self, duthost, platform_api_conn, conf):
         watchdog_timeout_greater = conf['greater_timeout']
         if watchdog_timeout_greater is None:
             pytest.skip('"greater_timeout" parameter is required for this test case')
-        actual_timeout_second = watchdog.arm(platform_api_conn, watchdog_timeout)
+        actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
         remaining_time = watchdog.get_remaining_time(platform_api_conn)
-        actual_timeout_second_second = watchdog.arm(platform_api_conn, watchdog_timeout_greater)
-
-        assert actual_timeout_second < actual_timeout_second_second
-        assert watchdog.get_remaining_time(platform_api_conn) > remaining_time
+        actual_timeout_greater = watchdog.arm(platform_api_conn, watchdog_timeout_greater)
+        self.expect(actual_timeout < actual_timeout_greater, "{}: 1st timeout {} seconds should be less than 2nd timeout {} seconds".format(test_arm_different_timeout_greater.__name__, actual_timeout, actual_timeout_greater))
+        remaining_time_greater = watchdog.get_remaining_time(platform_api_conn)
+        self.expect(remaining_time_greater > remaining_time, "{}: 2nd remaining_timeout {} seconds should be greater than 1st remaining timeout {} seconds".format(test_arm_different_timeout_greater.__name__, remaining_time_greater, remaining_time))
+        self.assert_expectations()
 
+    @pytest.mark.dependency(depends=["test_arm_disarm_states"])
     def test_arm_different_timeout_smaller(self, duthost, platform_api_conn, conf):
         ''' arm the watchdog with smaller timeout value and verify new timeout was accepted;
         If platform accepts only single valid timeout value, @greater_timeout should be None.
@@ -139,9 +168,12 @@ def test_arm_different_timeout_smaller(self, duthost, platform_api_conn, conf):
         remaining_time = watchdog.get_remaining_time(platform_api_conn)
         actual_timeout_smaller = watchdog.arm(platform_api_conn, watchdog_timeout_smaller)
 
-        assert actual_timeout > actual_timeout_smaller
-        assert watchdog.get_remaining_time(platform_api_conn) < remaining_time
+        self.expect(actual_timeout > actual_timeout_smaller, "{}: 1st timeout {} seconds should be greater than 2nd timeout {} seconds".format(test_arm_different_timeout_smaller.__name__, actual_timeout, actual_timeout_smaller))
+        remaining_time_smaller = watchdog.get_remaining_time(platform_api_conn)
+        self.expect(remaining_time_smaller < remaining_time, "{}: 2nd remaining_timeout {} seconds should be less than 1st remaining timeout {} seconds".format(test_arm_different_timeout_smaller.__name__, remaining_time_smaller, remaining_time))
+        self.assert_expectations()
 
+    @pytest.mark.dependency(depends=["test_arm_disarm_states"])
     def test_arm_too_big_timeout(self, duthost, platform_api_conn, conf):
         ''' try to arm the watchdog with timeout that is too big for hardware watchdog;
         If no such limitation exist, @too_big_timeout should be None for such platform.
@@ -151,34 +183,14 @@ def test_arm_too_big_timeout(self, duthost, platform_api_conn, conf):
         if watchdog_timeout is None:
             pytest.skip('"too_big_timeout" parameter is required for this test case')
         actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
+        self.expect(actual_timeout == -1, "{}: Watchdog should be disarmed, but returned timeout of {} seconds".format(test_arm_too_big_timeout.__name__, watchdog_timeout))
+        self.assert_expectations()
 
-        assert actual_timeout == -1
-
+    @pytest.mark.dependency(depends=["test_arm_disarm_states"])
     def test_arm_negative_timeout(self, duthost, platform_api_conn):
         ''' try to arm the watchdog with negative value '''
 
         watchdog_timeout = -1
         actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
-
-        assert actual_timeout == -1
-
-    @pytest.mark.disable_loganalyzer
-    def test_reboot(self, duthost, localhost, platform_api_conn, conf):
-        ''' arm the watchdog and verify it did its job after timeout expiration '''
-
-        watchdog_timeout = conf['valid_timeout']
-        actual_timeout = watchdog.arm(platform_api_conn, watchdog_timeout)
-
-        assert actual_timeout != -1
-
-        res = localhost.wait_for(host=duthost.hostname, port=22, state="stopped", delay=2,
-                                 timeout=actual_timeout + TIMEOUT_DEVIATION,
-                                 module_ignore_errors=True)
-        assert 'exception' not in res
-
-        res = localhost.wait_for(host=duthost.hostname, port=22, state="started", delay=10, timeout=120,
-                                 module_ignore_errors=True)
-        assert 'exception' not in res
-
-        # wait for system to startup
-        time.sleep(120)
+        self.expect(actual_timeout == -1, "{}: Watchdog should be disarmed, but returned timeout of {} seconds".format(test_arm_negative_timeout.__name__, watchdog_timeout))
+        self.assert_expectations()