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

#1816 - Added the final pieces of the puzzle to get data up from NIC → session manager → software manager → service.

Browse Source 
- Implemented a basic sim DB that matches UC2 data manipulation DB in IY.
- Added a test that confirms DB queries can be sent over the network.
This commit is contained in:
Chris McCarthy
2023-09-06 22:26:23 +01:00
parent 6b41bec32a
commit 2f744af34e
11 changed files with 180 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,9 +2,9 @@ from ipaddress import IPv4Address
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.networks import arcd_uc2_network
from primaite.simulator.network.transmission.data_link_layer import Frame, EthernetHeader
from primaite.simulator.network.transmission.data_link_layer import EthernetHeader, Frame
from primaite.simulator.network.transmission.network_layer import IPPacket, Precedence
from primaite.simulator.network.transmission.transport_layer import TCPHeader, Port
from primaite.simulator.network.transmission.transport_layer import Port, TCPHeader
def test_database_query_across_the_network():
@@ -20,22 +20,15 @@ def test_database_query_across_the_network():
outbound_nic = client_1.arp.get_arp_cache_nic(IPv4Address("192.168.1.14"))
client_1.ping("192.168.1.14")
frame = Frame(
ethernet=EthernetHeader(
src_mac_addr=client_1.ethernet_port[1].mac_address,
dst_mac_addr=dst_mac_address
),
ethernet=EthernetHeader(src_mac_addr=client_1.ethernet_port[1].mac_address, dst_mac_addr=dst_mac_address),
ip=IPPacket(
src_ip_address=client_1.ethernet_port[1].ip_address,
dst_ip_address=IPv4Address("192.168.1.14"),
precedence=Precedence.FLASH
precedence=Precedence.FLASH,
),
tcp=TCPHeader(
src_port=Port.POSTGRES_SERVER,
dst_port=Port.POSTGRES_SERVER
),
payload="SELECT * FROM user;"
tcp=TCPHeader(src_port=Port.POSTGRES_SERVER, dst_port=Port.POSTGRES_SERVER),
payload="SELECT * FROM user;",
)
outbound_nic.send_frame(frame)
@@ -43,4 +36,4 @@ def test_database_query_across_the_network():
client_1_last_payload = outbound_nic.pcap.read()[-1]["payload"]
assert client_1_last_payload["status_code"] == 200
assert client_1_last_payload["data"]
assert client_1_last_payload["data"]