oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#2041: ntp_server test fixes.

Browse Source
This commit is contained in:
Nick Todd
2023-11-20 18:08:55 +00:00
parent b0b37f9da5
commit f7215847d4
3 changed files with 21 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
tests/integration_tests/system/test_ntp_client_server.py
View File

@@ -5,7 +5,7 @@ import pytest
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.network.protocols.ntp import NTPPacket
from primaite.simulator.network.protocols.ntp import NTPPacket, NTPRequest
from primaite.simulator.system.services.ntp.ntp_client import NTPClient
from primaite.simulator.system.services.ntp.ntp_server import NTPServer
from primaite.simulator.system.services.service import ServiceOperatingState
@@ -58,9 +58,11 @@ def test_ntp_client_server():
assert ntp_server.operating_state == ServiceOperatingState.RUNNING
assert ntp_client.operating_state == ServiceOperatingState.RUNNING
ntp_client.send(payload=NTPPacket())
assert ntp_server.receive() is True
assert ntp_client.receive() is True
ntp_request = NTPRequest(ntp_client="192.168.1.3")
ntp_packet = NTPPacket(ntp_request=ntp_request)
ntp_client.send(payload=ntp_packet)
assert ntp_server.receive(payload=ntp_packet) is True
assert ntp_client.receive(payload=ntp_packet) is True
assert ntp_client.apply_timestep(1) is True
@@ -71,15 +73,20 @@ def test_ntp_server_failure():
server: Server = network.get_node_by_hostname("ntp_server")
client: Computer = network.get_node_by_hostname("ntp_client")
ntp_server: NTPServer = server.software_manager.software["NTP_Server"]
ntp_client: NTPClient = client.software_manager.software["NTP_Client"]
ntp_server: NTPServer = server.software_manager.software["NTPServer"]
ntp_client: NTPClient = client.software_manager.software["NTPClient"]
assert ntp_client.operating_state == ServiceOperatingState.RUNNING
# Turn off ntp server.
ntp_server.stop()
assert ntp_server.operating_state == ServiceOperatingState.STOPPED
assert ntp_client.receive() is False
# And request a time update.
ntp_request = NTPRequest(ntp_client="192.168.1.3")
ntp_packet = NTPPacket(ntp_request=ntp_request)
ntp_client.send(payload=ntp_packet)
assert ntp_server.receive(payload=ntp_packet) is False
assert ntp_client.receive(payload=ntp_packet) is False
# Restart ntp server.
ntp_server.start()