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

#2042: Add time attribute

Browse Source
This commit is contained in:
Nick Todd
2023-11-21 17:24:24 +00:00
parent 243f2dd938
commit eb2e37429a
2 changed files with 25 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/primaite/simulator/system/services/ntp/ntp_client.py
View File

@@ -1,3 +1,4 @@
from datetime import datetime
from ipaddress import IPv4Address
from typing import Dict, Optional
@@ -16,6 +17,7 @@ class NTPClient(Service):
ip_addr: Optional[IPv4Address] = None
ntp_server: Optional[IPv4Address] = None
"The NTP server the client sends requests to."
time: Optional[datetime] = None
def __init__(self, **kwargs):
kwargs["name"] = "NTPClient"
@@ -94,6 +96,7 @@ class NTPClient(Service):
f"{self.name}: Received time \
update from NTP server{payload.ntp_reply.ntp_datetime}"
)
self.time = payload.ntp_reply.ntp_datetime
return True
def apply_timestep(self, timestep: int) -> None:

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

@@ -53,38 +53,39 @@ def test_ntp_client_server():
assert ntp_server.operating_state == ServiceOperatingState.RUNNING
assert ntp_client.operating_state == ServiceOperatingState.RUNNING
assert ntp_client.time is None
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.time is not None
assert ntp_client.apply_timestep(1) is True
# TODO: Disabled until a service such as NTP can introspect to see if it's running.
# Test ntp client behaviour when ntp server is unavailable.
# def test_ntp_server_failure():
# network = create_ntp_network()
# server: Server = network.get_node_by_hostname("ntp_server")
# client: Computer = network.get_node_by_hostname("ntp_client")
@pytest.mark.skip(reason="NTP needs to know if underly node is RUNNING")
def test_ntp_server_failure():
network = create_ntp_network()
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["NTPServer"]
# ntp_client: NTPClient = client.software_manager.software["NTPClient"]
ntp_server: NTPServer = server.software_manager.software["NTPServer"]
ntp_client: NTPClient = client.software_manager.software["NTPClient"]
# assert ntp_client.operating_state == ServiceOperatingState.RUNNING
assert ntp_client.operating_state == ServiceOperatingState.RUNNING
# # Turn off ntp server.
# ntp_server.stop()
# assert ntp_server.operating_state == ServiceOperatingState.STOPPED
# # 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
# Turn off ntp server.
ntp_server.stop()
assert ntp_server.operating_state == ServiceOperatingState.STOPPED
# 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()
# assert ntp_server.operating_state == ServiceOperatingState.RUNNING
# Restart ntp server.
ntp_server.start()
assert ntp_server.operating_state == ServiceOperatingState.RUNNING