mirror of
https://gitlab.com/d3tn/ud3tn.git
synced 2026-08-13 12:33:27 +02:00
84de01f9ab
- test LocalNode EIDs - test 3-element ipn EIDs - also check that the EIDs are present in received bundles Signed-off-by: Felix Walter <felix.walter@d3tn.com>
127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
# SPDX-License-Identifier: BSD-3-Clause OR Apache-2.0
|
|
import os
|
|
import time
|
|
|
|
from ud3tn_utils.config import LegacyConfigMessage, RouterCommand
|
|
from pyd3tn.bundle7 import Bundle, CRCType
|
|
|
|
USER_SELECTED_CLA = os.environ.get("CLA", None)
|
|
TESTED_CLAS = [USER_SELECTED_CLA] if USER_SELECTED_CLA else [
|
|
"tcpclv3",
|
|
"smtcp",
|
|
]
|
|
|
|
UD3TN_HOST = "localhost"
|
|
TCPSPP_PORT = 4223
|
|
TCPCL_PORT = 4556
|
|
SMTCP_PORT = 4222
|
|
# MTCP_PORT = 4224
|
|
|
|
AAP_USE_TCP = os.environ.get("AAP_USE_TCP", "0") == "1"
|
|
AAP_SOCKET = os.environ.get(
|
|
"AAP_SOCKET",
|
|
("localhost", 4242) if AAP_USE_TCP else "ud3tn.socket"
|
|
)
|
|
AAP_AGENT_ID = "testagent"
|
|
TEST_AAP = os.environ.get("TEST_AAP", "1") == "1"
|
|
|
|
AAP2_SOCKET = os.environ.get(
|
|
"AAP2_SOCKET",
|
|
"ud3tn.aap2.socket"
|
|
)
|
|
AAP2_AGENT_ID = "testagentaap2"
|
|
AAP2_SECRET = "testsecret"
|
|
TEST_AAP2 = os.environ.get("TEST_AAP2", "1") == "1"
|
|
TEST_AAP2_ASYNC = TEST_AAP2 and os.environ.get("TEST_AAP2_ASYNC", "1") == "1"
|
|
AAP2_BDM_SECRET = os.environ.get("TEST_AAP2_BDM_SECRET", None)
|
|
|
|
TEST_JSON_CONFIG = os.environ.get("TEST_JSON_CONFIG", "0") == "1"
|
|
TEST_QUERY = os.environ.get("TEST_QUERY", "0") == "1"
|
|
|
|
UD3TN_EID = "dtn://ud3tn.dtn/"
|
|
UD3TN_CONFIG_EP = UD3TN_EID + "config"
|
|
|
|
TEST_SCRIPT_EID = "dtn://manager.dtn/"
|
|
|
|
SPP_USE_CRC = os.environ.get("TCPSPP_CRC_ENABLED", "1") == "1"
|
|
TCP_TIMEOUT = 1.
|
|
|
|
|
|
def validate_bundle7(bindata, expected_payload=None,
|
|
expected_source=None, expected_dest=None):
|
|
b = Bundle.parse(bindata)
|
|
print("Received bundle: {}".format(repr(b)))
|
|
crc_valid = True
|
|
for block in b:
|
|
print((
|
|
"Block {} {} CRC: provided={}, calculated={}"
|
|
).format(
|
|
block.block_number,
|
|
repr(block.block_type),
|
|
(
|
|
"0x{:08x}".format(block.crc_provided)
|
|
if block.crc_provided else "None"
|
|
),
|
|
(
|
|
"0x{:08x}".format(block.calculate_crc())
|
|
if block.crc_type != CRCType.NONE else "None"
|
|
),
|
|
))
|
|
if block.crc_type != CRCType.NONE:
|
|
crc_valid &= block.crc_provided == block.calculate_crc()
|
|
assert crc_valid
|
|
if expected_payload:
|
|
assert b.payload_block.data == expected_payload
|
|
if expected_source:
|
|
assert str(b.primary_block.source) == expected_source
|
|
if expected_dest:
|
|
assert str(b.primary_block.destination) == expected_dest
|
|
return b
|
|
|
|
|
|
def _check_eid_ssp_in_bindata(bindata, eid):
|
|
if eid[0:4] == "ipn:":
|
|
parts = eid[4:].split(".")
|
|
if len(parts) == 2:
|
|
assert eid[4:].encode("ascii") in bindata
|
|
elif len(parts) == 3:
|
|
two_elem_ssp = (
|
|
str(int(parts[0]) << 32 | int(parts[1])) +
|
|
"." + parts[2]
|
|
)
|
|
assert two_elem_ssp.encode("ascii") in bindata
|
|
elif eid[0:4] == "dtn:":
|
|
assert eid[4:].encode("ascii") in bindata
|
|
else:
|
|
raise ValueError
|
|
|
|
|
|
def validate_bundle6(bindata, expected_payload=None,
|
|
expected_source=None, expected_dest=None):
|
|
assert bindata[0] == 0x06
|
|
print("Received RFC 5050 binary data: {}".format(repr(bindata)))
|
|
# TODO: Add a proper validation
|
|
if expected_payload:
|
|
assert expected_payload in bindata
|
|
if expected_source:
|
|
_check_eid_ssp_in_bindata(bindata, expected_source)
|
|
if expected_dest:
|
|
_check_eid_ssp_in_bindata(bindata, expected_dest)
|
|
return bindata
|
|
|
|
|
|
def send_delete_gs(conn, serialize_func, gs_iterable):
|
|
for eid in gs_iterable:
|
|
print("Sending DELETE command for {}".format(eid))
|
|
conn.send_bundle(serialize_func(
|
|
TEST_SCRIPT_EID,
|
|
UD3TN_CONFIG_EP,
|
|
bytes(LegacyConfigMessage(
|
|
eid,
|
|
"NULL",
|
|
type=RouterCommand.DELETE,
|
|
))
|
|
))
|
|
# Wait for uD3TN to properly process the deletion request to make sure
|
|
# it is not processed after the next test's configuration is received.
|
|
time.sleep(1)
|