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

2041: Remove IP address from NTP client (review comment)

Browse Source
This commit is contained in:
Nick Todd
2023-12-12 17:08:11 +00:00
parent c3015b7f0e
commit e620771c8d
3 changed files with 15 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,8 +14,6 @@ _LOGGER = getLogger(__name__)
class NTPClient(Service): class NTPClient(Service):
"""Represents a NTP client as a service.""" """Represents a NTP client as a service."""
ip_addr: Optional[IPv4Address] = None
"The IP address of the NTP client"
ntp_server: Optional[IPv4Address] = None ntp_server: Optional[IPv4Address] = None
"The NTP server the client sends requests to." "The NTP server the client sends requests to."
time: Optional[datetime] = None time: Optional[datetime] = None
@@ -27,16 +25,15 @@ class NTPClient(Service):
super().__init__(**kwargs) super().__init__(**kwargs)
self.start() self.start()
def configure(self, ntp_server_ip_address: IPv4Address, ntp_client_ip_address: IPv4Address) -> None: def configure(self, ntp_server_ip_address: IPv4Address) -> None:
""" """
Set the IP address for the NTP server. Set the IP address for the NTP server.
:param ntp_server_ip_address: IPv4 address of NTP server. :param ntp_server_ip_address: IPv4 address of NTP server.
:param ntp_client_ip_Address: IPv4 address of NTP client. :param ntp_client_ip_Address: IPv4 address of NTP client.
""" """
self.ip_addr = ntp_client_ip_address
self.ntp_server = ntp_server_ip_address self.ntp_server = ntp_server_ip_address
self.sys_log.info(f"{self.name}: ip_addr: {self.ip_addr}, ntp_server: {self.ntp_server}") self.sys_log.info(f"{self.name}: ntp_server: {self.ntp_server}")
def describe_state(self) -> Dict: def describe_state(self) -> Dict:
""" """
@@ -78,7 +75,6 @@ class NTPClient(Service):
:return: True if successful, False otherwise. :return: True if successful, False otherwise.
""" """
self.ip_addr = payload.ntp_request.ntp_client
return super().send( return super().send(
payload=payload, payload=payload,
dest_ip_address=dest_ip_address, dest_ip_address=dest_ip_address,
@@ -99,9 +95,7 @@ class NTPClient(Service):
:param session_id: The Session ID the payload is to originate from. Optional. :param session_id: The Session ID the payload is to originate from. Optional.
:return: True if successful, False otherwise. :return: True if successful, False otherwise.
""" """
self.sys_log.info(f"{self.name}: Receiving NTP request from {payload.ntp_request.ntp_client}") if not isinstance(payload, NTPPacket):
if not (isinstance(payload, NTPPacket) and payload.ntp_request.ntp_client):
_LOGGER.debug(f"{payload} is not a NTPPacket") _LOGGER.debug(f"{payload} is not a NTPPacket")
return False return False
if payload.ntp_reply.ntp_datetime: if payload.ntp_reply.ntp_datetime:
@@ -114,7 +108,7 @@ class NTPClient(Service):
def request_time(self) -> None: def request_time(self) -> None:
"""Send request to ntp_server.""" """Send request to ntp_server."""
ntp_request = NTPRequest(ntp_client=self.ip_addr) ntp_request = NTPRequest()
ntp_server_packet = NTPPacket(ntp_request=ntp_request) ntp_server_packet = NTPPacket(ntp_request=ntp_request)
self.send(payload=ntp_server_packet, dest_ip_address=self.ntp_server) self.send(payload=ntp_server_packet, dest_ip_address=self.ntp_server)
@@ -129,7 +123,7 @@ class NTPClient(Service):
:param timestep: The current timestep number. (Amount of time since simulation episode began) :param timestep: The current timestep number. (Amount of time since simulation episode began)
:type timestep: int :type timestep: int
""" """
self.sys_log.info(f"{self.name} apply_timestep: IP address: {self.ip_addr}") self.sys_log.info(f"{self.name} apply_timestep")
super().apply_timestep(timestep) super().apply_timestep(timestep)
if self.operating_state == ServiceOperatingState.RUNNING: if self.operating_state == ServiceOperatingState.RUNNING:
# request time from server # request time from server

25
src/primaite/simulator/system/services/ntp/ntp_server.py
View File

@@ -60,23 +60,14 @@ class NTPServer(Service):
:return: True if valid NTP request else False. :return: True if valid NTP request else False.
""" """
self.sys_log.info(f"{self.name} received request from {payload.ntp_request.ntp_client}") if not (isinstance(payload, NTPPacket)):
if not (isinstance(payload, NTPPacket) and payload.ntp_request.ntp_client):
_LOGGER.debug(f"{payload} is not a NTPPacket") _LOGGER.debug(f"{payload} is not a NTPPacket")
return False return False
payload: NTPPacket = payload payload: NTPPacket = payload
if payload.ntp_request.ntp_client:
self.sys_log.info( # generate a reply with the current time
f"{self.name}: Received request for {payload.ntp_request.ntp_client} \ time = datetime.now()
from session {session_id}" payload = payload.generate_reply(time)
) # send reply
# generate a reply with the current time self.send(payload, session_id)
time = datetime.now() return True
payload = payload.generate_reply(time)
self.sys_log.info(
f"{self.name}: Responding to NTP request for {payload.ntp_request.ntp_client} "
f"with current time: {time}"
)
# send reply
self.send(payload, session_id)
return True

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

@@ -51,9 +51,7 @@ def test_ntp_client_server(create_ntp_network):
assert ntp_server.operating_state == ServiceOperatingState.RUNNING assert ntp_server.operating_state == ServiceOperatingState.RUNNING
assert ntp_client.operating_state == ServiceOperatingState.RUNNING assert ntp_client.operating_state == ServiceOperatingState.RUNNING
ntp_client.configure( ntp_client.configure(ntp_server_ip_address=IPv4Address("192.168.0.2"))
ntp_server_ip_address=IPv4Address("192.168.0.2"), ntp_client_ip_address=IPv4Address("192.168.0.1")
)
assert ntp_client.time is None assert ntp_client.time is None
ntp_client.request_time() ntp_client.request_time()
@@ -74,9 +72,7 @@ def test_ntp_server_failure(create_ntp_network):
assert ntp_client.operating_state == ServiceOperatingState.RUNNING assert ntp_client.operating_state == ServiceOperatingState.RUNNING
assert ntp_client.operating_state == ServiceOperatingState.RUNNING assert ntp_client.operating_state == ServiceOperatingState.RUNNING
ntp_client.configure( ntp_client.configure(ntp_server_ip_address=IPv4Address("192.168.0.2"))
ntp_server_ip_address=IPv4Address("192.168.0.2"), ntp_client_ip_address=IPv4Address("192.168.0.1")
)
# Turn off ntp server. # Turn off ntp server.
ntp_server.stop() ntp_server.stop()