[pytest/ntp]: add ntp test

ansible/roles/test/tasks/ntp.yml

@@ -1,25 +1,4 @@
-- name: Stop NTP service
-  become: true
-  service:
-    name: ntp
-    state: stopped
-
-- name: Force local clock to sync with an NTP clock
-  become: true
-  command: ntpd -gq
-
-- name: Start NTP service
-  become: true
-  service:
-    name: ntp
-    state: started
-
-- name: Check if NTP is synced
-  become: true
-  shell: ntpstat
-  register: ntpstat_result
-  until: ntpstat_result.rc == 0
-  retries: 10
-  delay: 30
-
-- debug: msg="NTP Status {{ ntpstat_result.stdout }}"
+- name: run test
+  include_tasks: roles/test/tasks/pytest_runner.yml
+  vars:
+    test_node: ntp/test_ntp.py

tests/ntp/test_ntp.py

@@ -0,0 +1,52 @@
+from common.utilities import wait_until
+import logging
+logger = logging.getLogger(__name__)
+
+import pytest
+
+pytestmark = [
+    pytest.mark.sanity_check(skip_sanity=True),
+    pytest.mark.disable_loganalyzer,
+]
+
+@pytest.fixture(scope="module")
+def setup_ntp(ptfhost, duthost):
+    """setup ntp client and server"""
+
+    # enable ntp server
+    ptfhost.service(name="ntp", state="started")
+
+    # setup ntp on dut to sync with ntp server
+    config_facts  = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
+    ntp_servers = config_facts.get('NTP_SERVER', {})
+    for ntp_server in ntp_servers:
+        duthost.command("sudo config ntp del %s" % ntp_server)
+
+    ptfip = ptfhost.host.options['inventory_manager'].get_host(ptfhost.hostname).vars['ansible_host']
+    duthost.command("sudo config ntp add %s" % ptfip)
+
+    wait_until(120, 5, check_ntp_status, ptfhost)
+
+    yield
+
+    # stop ntp server
+    ptfhost.service(name="ntp", state="stopped")
+
+    # reset ntp client configuration
+    duthost.command("sudo config ntp del %s" % ptfip)
+    for ntp_server in ntp_servers:
+        duthost.command("sudo config ntp add %s" % ntp_server)
+
+def check_ntp_status(host):
+    res = host.command("ntpstat")
+    if res['rc'] != 0:
+       return False
+    return True
+
+def test_ntp(testbed_devices, duthost, setup_ntp):
+    """ verify the LLDP message on DUT """
+
+    duthost.service(name='ntp', state='stopped')
+    duthost.command("ntpd -gq")
+    duthost.service(name='ntp', state='restarted')
+    assert wait_until(120, 5, check_ntp_status, duthost), "Ntp not in sync"