ud3tn/test/integration/test_bundle_send_receive.py

# SPDX-License-Identifier: BSD-3-Clause OR Apache-2.0
import os
import time
import math

import pytest

from ud3tn_utils.config import (
    LegacyConfigMessage as ConfigMessage,
    make_contact,
)

from pyd3tn.bundle7 import serialize_bundle7
from pyd3tn.bundle6 import serialize_bundle6
from pyd3tn.mtcp import MTCPConnection
from pyd3tn.spp import TCPSPPConnection
from pyd3tn.tcpcl import TCPCLConnection

from .helpers import (
    TESTED_CLAS,
    UD3TN_CONFIG_EP,
    UD3TN_HOST,
    TCPSPP_PORT,
    TCPCL_PORT,
    SMTCP_PORT,
    SPP_USE_CRC,
    TCP_TIMEOUT,
    validate_bundle6,
    validate_bundle7,
    send_delete_gs,
)

SENDING_CONTACT = (1, 1, 1000)
RECEIVING_CONTACT = (3, 1, 1000)
BUNDLE_SIZE = 200


def _wait_for_even_timestamp():
    # Wait until TARGET_OFFSET before the next second starts (for syncing time
    # using second granularity)
    TARGET_OFFSET = 0.9
    cur_time = time.time()
    next_second = math.ceil(cur_time)
    if next_second - cur_time < TARGET_OFFSET:
        next_second += 1
    time.sleep(next_second - cur_time - TARGET_OFFSET)
    return int(next_second)


def perform_basic_test(connection_obj, serialize_func, validate_func,
                       cla_str):
    SENDER_EID = "dtn://sender.dtn/"
    RECEIVER_EIDS = [
        "dtn://receiver.dtn/",
        "ipn:1.1",
        "ipn:5.42.75",
    ]
    with connection_obj as conn:
        # Configure contact during which we send a bundle
        conn.send_bundle(serialize_func(
            SENDER_EID,
            UD3TN_CONFIG_EP,
            bytes(ConfigMessage(
                SENDER_EID,
                cla_str + ":",
                contacts=[
                    make_contact(*SENDING_CONTACT),
                ],
            )),
        ))
        # Configure contact during which we want to receive the bundle
        conn.send_bundle(serialize_func(
            SENDER_EID,
            UD3TN_CONFIG_EP,
            bytes(ConfigMessage(
                RECEIVER_EIDS[0],
                cla_str + ":",
                contacts=[
                    make_contact(
                        *RECEIVING_CONTACT,
                        reachable_eids=RECEIVER_EIDS,
                    ),
                ],
            )),
        ))
        try:
            # Wait until first contact starts and send bundle
            time.sleep(SENDING_CONTACT[0])
            payload_data = os.urandom(BUNDLE_SIZE)
            for receiver_eid in RECEIVER_EIDS:
                conn.send_bundle(serialize_func(
                    SENDER_EID,
                    receiver_eid,
                    payload_data,
                ))
            # Wait until second contact starts and try to receive bundle
            time.sleep(RECEIVING_CONTACT[0] - SENDING_CONTACT[0])
            for receiver_eid in RECEIVER_EIDS:
                bdl = conn.recv_bundle()
                validate_func(bdl, payload_data, SENDER_EID, receiver_eid)
            time.sleep(RECEIVING_CONTACT[1])
        finally:
            send_delete_gs(conn, serialize_func, (SENDER_EID,
                                                  RECEIVER_EIDS[0]))


@pytest.mark.skipif("tcpspp" not in TESTED_CLAS, reason="not selected")
def test_send_receive_spp_bundle6():
    perform_basic_test(
        TCPSPPConnection(
            UD3TN_HOST,
            TCPSPP_PORT,
            SPP_USE_CRC,
            timeout=TCP_TIMEOUT,
        ),
        serialize_bundle6,
        validate_bundle6,
        "tcpspp",
    )


@pytest.mark.skipif("tcpspp" not in TESTED_CLAS, reason="not selected")
def test_send_receive_spp_bundle7():
    perform_basic_test(
        TCPSPPConnection(
            UD3TN_HOST,
            TCPSPP_PORT,
            SPP_USE_CRC,
            timeout=TCP_TIMEOUT,
        ),
        serialize_bundle7,
        validate_bundle7,
        "tcpspp",
    )


@pytest.mark.skipif("smtcp" not in TESTED_CLAS, reason="not selected")
def test_send_receive_smtcp_bundle6():
    perform_basic_test(
        MTCPConnection(UD3TN_HOST, SMTCP_PORT, timeout=TCP_TIMEOUT),
        serialize_bundle6,
        validate_bundle6,
        "smtcp",
    )


@pytest.mark.skipif("smtcp" not in TESTED_CLAS, reason="not selected")
def test_send_receive_smtcp_bundle7():
    perform_basic_test(
        MTCPConnection(UD3TN_HOST, SMTCP_PORT, timeout=TCP_TIMEOUT),
        serialize_bundle7,
        validate_bundle7,
        "smtcp",
    )


@pytest.mark.skipif("tcpclv3" not in TESTED_CLAS, reason="not selected")
def test_send_receive_tcpcl_bundle6():
    perform_basic_test(
        TCPCLConnection(
            "dtn://receiver.dtn/",
            UD3TN_HOST,
            TCPCL_PORT,
            timeout=TCP_TIMEOUT,
        ),
        serialize_bundle6,
        validate_bundle6,
        "tcpclv3",
    )


@pytest.mark.skipif("tcpclv3" not in TESTED_CLAS, reason="not selected")
def test_send_receive_tcpcl_bundle7():
    perform_basic_test(
        TCPCLConnection(
            "dtn://receiver.dtn/",
            UD3TN_HOST,
            TCPCL_PORT,
            timeout=TCP_TIMEOUT,
        ),
        serialize_bundle7,
        validate_bundle7,
        "tcpclv3",
    )