diff --git a/bromelia/setup.py b/bromelia/setup.py index 10a14e8..def89b0 100644 --- a/bromelia/setup.py +++ b/bromelia/setup.py @@ -2,9 +2,9 @@ """ bromelia.setup ~~~~~~~~~~~~~~ - + This module implements the central Diameter application object. It works - as per Peer State Machine defined in RFC 6733 in order to establish a + as per Peer State Machine defined in RFC 6733 in order to establish a Diameter association with another Peer Node. :copyright: (c) 2020-present Henrique Marques Ribeiro. @@ -16,7 +16,7 @@ import logging import platform import queue -import socket +import socket import sys import threading import time @@ -101,7 +101,7 @@ def __init__(self, connection: Connection, base: BaseMessages) -> None: self._recv_messages = queue.Queue() self._send_messages = queue.Queue() - self.postprocess_recv_messages = queue.Queue() + self.postprocess_recv_messages = queue.Queue() self.postprocess_recv_messages_ready = threading.Event() self.postprocess_recv_messages_lock = threading.Lock() self.lock = threading.Lock() @@ -110,7 +110,7 @@ def __init__(self, connection: Connection, base: BaseMessages) -> None: def is_connected(self) -> bool: if self.transport: return self.transport.is_connected - + return False @@ -125,21 +125,25 @@ def start(self) -> None: if self.connection.mode == DIAMETER_AGENT_CLIENT_MODE: if self.connection.transport_type == DIAMETER_AGENT_TRANSPORT_TYPE_TCP: - self.transport = TcpClient(self.connection.peer_node.ip_address, + self.transport = TcpClient(self.connection.local_node.ip_address, + self.connection.local_node.port or 0, # will mean that any free port will be used for the client + self.connection.peer_node.ip_address, self.connection.peer_node.port) elif self.connection.transport_type == DIAMETER_AGENT_TRANSPORT_TYPE_SCTP: - self.transport = SctpClient(self.connection.peer_node.ip_address, - self.connection.peer_node.port) + self.transport = SctpClient(self.connection.local_node.ip_address, + self.connection.local_node.port or 0, # will mean that any free port will be used for the client + self.connection.peer_node.ip_address, + self.connection.peer_node.port) else: raise DiameterAssociationError("Invalid Diameter Agent transport type.") elif self.connection.mode == DIAMETER_AGENT_SERVER_MODE: if self.connection.transport_type == DIAMETER_AGENT_TRANSPORT_TYPE_TCP: self.transport = TcpServer(self.connection.local_node.ip_address, - self.connection.local_node.port) + self.connection.local_node.port) elif self.connection.transport_type == DIAMETER_AGENT_TRANSPORT_TYPE_SCTP: self.transport = SctpServer(self.connection.local_node.ip_address, - self.connection.local_node.port) + self.connection.local_node.port) else: raise DiameterAssociationError("Invalid Diameter Agent transport type.") @@ -183,7 +187,7 @@ def recv_message_from_queue(self) -> None: for msg in msgs: make_logging(msg, disable_else=True) self._recv_messages.put(msg) - + diameter_conn_logger.debug(f"Found {len(msgs)} Diameter "\ f"Message(s).") except AVPParsingError: @@ -228,7 +232,7 @@ def put_message_into_send_queue(self, msg: Type[DiameterMessage]) -> None: diameter_conn_logger.debug(f"[{hop_by_hop.hex()}] Diameter "\ f"Message (Answer) have been put "\ f"into _send_messages Queue.") - + self.lock.release() @@ -259,7 +263,7 @@ def send_message_from_queue(self) -> None: diameter_conn_logger.debug(f"[{msg.header.hop_by_hop.hex()}] "\ f"Diameter Request have been "\ f"put into Pending Request Queue.") - + stream += msg.dump() if self.transport: @@ -279,9 +283,9 @@ def send_message_from_queue(self) -> None: diameter_conn_logger.debug("Transport Layer is not in "\ "WRITE mode again, so we "\ "can send data stream.") - + self.transport._set_selector_events_mask("rw", stream) - + # maybe include a verification here before the "break" if a given message has been sent from transport layer. break @@ -317,7 +321,7 @@ def get_message(self) -> Type[DiameterMessage]: else: diameter_conn_logger.debug("No need to wait for go ahead for "\ "postprocess_recv_messages_ready") - + return self.get_postprocess_recv_message() @@ -328,7 +332,7 @@ def tracking_events(self) -> None: diameter_conn_logger.debug("Generating a DWR message.") self.transport.tracking_events_count = 0 - + except AttributeError as e: if e.args[0] == "'TcpServer' object has no attribute 'events'": pass @@ -358,9 +362,9 @@ def __init__(self, debug: bool = False, is_logging: bool = False, app_name: str = None) -> None: - + self.logging = DiameterLogging(debug, is_logging, app_name) - + self.config = self.make_config(config) self._connection = _convert_config_to_connection_obj(self.config) self._base = self.get_base_messages() @@ -418,7 +422,7 @@ def reset(self) -> None: def close(self) -> None: current_state = self.get_current_state() - + if current_state == CLOSED: raise DiameterApplicationError("Cannot stop the application. "\ "Peer State Machine is already "\ @@ -477,7 +481,7 @@ def send_message(self, msg: Type[DiameterMessage], avoid: bool = True) -> Any: if is_base_answer(msg): raise DiameterApplicationError("Cannot send a Base protocol "\ "answer") - + self._association.put_message_into_send_queue(msg) @@ -510,7 +514,7 @@ def context(self) -> None: if self.is_open(): break - + if self.is_closed() and (stop - start).seconds >= 5: start = datetime.datetime.utcnow() self.start() diff --git a/bromelia/transport.py b/bromelia/transport.py index 7e783fc..0369e8c 100644 --- a/bromelia/transport.py +++ b/bromelia/transport.py @@ -2,7 +2,7 @@ """ bromelia.transport ~~~~~~~~~~~~~~~~~~ - + This module defines the TCP transport layer connections that are used by the Diameter application protocol underlying. @@ -26,7 +26,7 @@ class TcpConnection(): - def __init__(self, ip_address: str, port: str) -> None: + def __init__(self, local_ip_address: str, local_port: str = 0, peer_ip_address: str = '', peer_port: str = 0) -> None: self._recv_buffer = b"" self._send_buffer = b"" self.send_data_stream_queued = False @@ -40,9 +40,11 @@ def __init__(self, ip_address: str, port: str) -> None: self.lock = threading.Lock() self.recv_data_consumed = False - - self.ip_address = ip_address - self.port = port + + self.local_ip_address = local_ip_address + self.local_port = local_port + self.peer_ip_address = peer_ip_address + self.peer_port = peer_port self.sock = None self.is_connected = False @@ -57,14 +59,14 @@ def __init__(self, ip_address: str, port: str) -> None: self.connection_attempts = 3 self.events_mask = selectors.EVENT_READ - + def is_write_mode(self) -> bool: if self.events_mask & selectors.EVENT_WRITE: return True return False - + def is_read_mode(self) -> bool: if self.events_mask & selectors.EVENT_READ: return True @@ -74,7 +76,7 @@ def is_read_write_mode(self) -> bool: if self.events_mask & (selectors.EVENT_READ | selectors.EVENT_WRITE): return True return False - + def close(self) -> None: if not self.is_connected: @@ -87,7 +89,7 @@ def close(self) -> None: tcp_connection.debug(f"[Socket-{self.sock_id}] De-registering "\ f"Socket from Selector address: "\ f"{self.selector.get_map()}") - + self.sock.close() tcp_connection.debug(f"[Socket-{self.sock_id}] Shutting "\ f"down Socket") @@ -103,7 +105,7 @@ def run(self) -> None: if not self.is_connected: raise ConnectionError(f"[Socket-{self.sock_id}] There is no "\ f"transport connection up for this Peer") - threading.Thread(name="transport_layer_thread", + threading.Thread(name="transport_layer_thread", target=self._run).start() @@ -135,7 +137,7 @@ def _set_selector_events_mask(self, mode: Literal["r", "w", "rw"], msg: Any = No self.selector.modify(self.sock, self.events_mask) self.write_mode_on.clear() self.read_mode_on.set() - + elif mode == "w": tcp_connection.debug(f"[Socket-{self.sock_id}] Updating "\ f"selector events mask [WRITE]") @@ -167,13 +169,13 @@ def _write(self) -> None: sent = self.sock.send(self._send_buffer) tcp_connection.debug(f"[Socket-{self.sock_id}] Just sent "\ f"{sent} bytes in _send_buffer") - + except BlockingIOError: tcp_connection.exception(f"[Socket-{self.sock_id}] An error "\ f"has occurred") self._stop_threads = True - + else: self._send_buffer = self._send_buffer[sent:] @@ -252,14 +254,14 @@ def test_connection(self) -> bool: import importlib class SctpConnection(TcpConnection): - def __init__(self, ip_address, port): + def __init__(self, local_ip_address: str, local_port: str = 0, peer_ip_address: str = '', peer_port: str = 0): try: self.sctp = importlib.import_module("sctp") self._sctp = importlib.import_module("_sctp") except ModuleNotFoundError as ex: tcp_connection.error(f"Python 'pysctp' module is required. Cannot initialize SctpConnection.") raise ex - super().__init__(ip_address, port) + super().__init__(local_ip_address, local_port, peer_ip_address, peer_port) def _write(self): if self._send_buffer: @@ -312,8 +314,8 @@ def test_connection(self): class TcpClient(TcpConnection): - def __init__(self, ip_address: str, port: str) -> None: - super().__init__(ip_address, port) + def __init__(self, local_ip_address: str, local_port: str = 0, peer_ip_address: str = '', peer_port: str = 0) -> None: + super().__init__(local_ip_address, local_port, peer_ip_address, peer_port) def start(self) -> None: @@ -326,7 +328,11 @@ def start(self) -> None: tcp_client.debug(f"[Socket-{self.sock_id}] Setting as "\ f"Non-Blocking") - self.sock.connect_ex((self.ip_address, self.port)) + tcp_client.debug(f"[Socket-{self.sock_id}] Binding to the "\ + f"Local Endpoint") + self.sock.bind((self.local_ip_address, self.local_port)) + + self.sock.connect_ex((self.peer_ip_address, self.peer_port)) tcp_client.debug(f"[Socket-{self.sock_id}] Connecting to the "\ f"Remote Peer") self.is_connected = True @@ -340,8 +346,8 @@ def start(self) -> None: class SctpClient(TcpClient,SctpConnection): - def __init__(self, ip_address, port): - SctpConnection.__init__(self, ip_address, port) + def __init__(self, local_ip_address: str, local_port: str = 0, peer_ip_address: str = '', peer_port: str = 0): + SctpConnection.__init__(self, local_ip_address, local_port, peer_ip_address, peer_port) def test_connection(self): return SctpConnection.test_connection(self) @@ -358,10 +364,13 @@ def start(self): tcp_client.debug(f"[Socket-{self.sock_id}] Client-side Socket: "\ f"{self.sock}") + tcp_client.debug(f"[Socket-{self.sock_id}] Binding to the "\ + f"Local Endpoint") + self.sock.bind((self.local_ip_address, self.local_port)) tcp_client.debug(f"[Socket-{self.sock_id}] Connecting to the "\ f"Remote Peer") - self.sock.connect((self.ip_address, self.port)) + self.sock.connect((self.peer_ip_address, self.peer_port)) self.is_connected = True tcp_client.debug(f"[Socket-{self.sock_id}] Setting as "\ @@ -377,9 +386,9 @@ def start(self): class TcpServer(TcpConnection): - def __init__(self, ip_address: str, port: str) -> None: - super().__init__(ip_address, port) - + def __init__(self, local_ip_address: str, local_port: str) -> None: + super().__init__(local_ip_address, local_port) + def start(self) -> None: try: @@ -390,10 +399,10 @@ def start(self) -> None: self.server_selector = selectors.DefaultSelector() self.server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 4096*64) - self.server_sock.bind((self.ip_address, self.port)) + self.server_sock.bind((self.local_ip_address, self.local_port)) self.server_sock.listen() tcp_server.debug(f"[Socket-{self.sock_id}] Listening on "\ - f"{self.ip_address}:{self.port}") + f"{self.local_ip_address}:{self.local_port}") self.server_sock.setblocking(False) tcp_server.debug(f"[Socket-{self.sock_id}] Setting as "\ @@ -426,7 +435,7 @@ def run(self) -> None: tcp_server.debug(f"[Socket-{self.sock_id}] Registering "\ f"New Socket into Selector address: "\ f"{self.selector.get_map()}") - + super().run() @@ -437,7 +446,7 @@ def close(self) -> None: self.server_selector.unregister(self.server_sock) tcp_server.debug(f"De-registering Main Socket from Selector "\ f"address: {self.server_selector.get_map()}") - + self.server_sock.close() tcp_server.debug("Shutting down Main Socket") @@ -446,8 +455,8 @@ def close(self) -> None: class SctpServer(TcpServer,SctpConnection): - def __init__(self, ip_address, port): - SctpConnection.__init__(self, ip_address, port) + def __init__(self, local_ip_address: str, local_port: str): + SctpConnection.__init__(self, local_ip_address, local_port) def test_connection(self): return SctpConnection.test_connection(self) @@ -474,10 +483,10 @@ def start(self): self.server_selector = selectors.DefaultSelector() self.server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 4096*64) - self.server_sock.bind((self.ip_address, self.port)) + self.server_sock.bind((self.local_ip_address, self.local_port)) self.server_sock.listen() tcp_server.debug(f"[Socket-{self.sock_id}] Listening on "\ - f"{self.ip_address}:{self.port}") + f"{self.local_ip_address}:{self.local_port}") self.server_sock.setblocking(False) tcp_server.debug(f"[Socket-{self.sock_id}] Setting as "\ diff --git a/tests/test_base.py b/tests/test_base.py index f76a13d..0817f34 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -4,7 +4,7 @@ ~~~~~~~~~~~~~~ This module contains the Diameter protocol base unittests. - + :copyright: (c) 2020 Henrique Marques Ribeiro. :license: MIT, see LICENSE for more details. """ @@ -46,9 +46,9 @@ def test_diameter_avp__vendor_id_bit__default(self): def test_diameter_avp__vendor_id_bit__unset_when_is_unset(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.set_vendor_id_bit(False) - + self.assertEqual(cm.exception.args[0], "V-bit was already unset") def test_diameter_avp__vendor_id_bit__set_when_is_set(self): @@ -58,9 +58,9 @@ def test_diameter_avp__vendor_id_bit__set_when_is_set(self): avp.set_vendor_id_bit(True) self.assertTrue(avp.is_vendor_id()) - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.set_vendor_id_bit(True) - + self.assertEqual(cm.exception.args[0], "V-bit was already set") def test_diameter_avp__mandatory_bit__default(self): @@ -82,9 +82,9 @@ def test_diameter_avp__mandatory_bit__default(self): def test_diameter_avp__mandatory_bit__unset_when_is_unset(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.set_mandatory_bit(False) - + self.assertEqual(cm.exception.args[0], "M-bit was already unset") def test_diameter_avp__mandatory_bit__set_when_is_set(self): @@ -94,9 +94,9 @@ def test_diameter_avp__mandatory_bit__set_when_is_set(self): avp.set_mandatory_bit(True) self.assertTrue(avp.is_mandatory()) - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.set_mandatory_bit(True) - + self.assertEqual(cm.exception.args[0], "M-bit was already set") def test_diameter_avp__protected_bit__default(self): @@ -118,9 +118,9 @@ def test_diameter_avp__protected_bit__default(self): def test_diameter_avp__protected_bit__unset_when_is_unset(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.set_protected_bit(False) - + self.assertEqual(cm.exception.args[0], "P-bit was already unset") def test_diameter_avp__protected_bit__set_when_is_set(self): @@ -130,14 +130,14 @@ def test_diameter_avp__protected_bit__set_when_is_set(self): avp.set_protected_bit(True) self.assertTrue(avp.is_protected()) - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.set_protected_bit(True) - + self.assertEqual(cm.exception.args[0], "P-bit was already set") def test_diameter_avp__default_object(self): avp = DiameterAVP() - + self.assertEqual(avp.code.hex(), "00000000") self.assertEqual(avp.flags.hex(), "00") self.assertIsNone(avp.vendor_id) @@ -195,113 +195,113 @@ def test_diameter_avp__custom_object_by_instance_attributes(self): def test_diameter_avp__custom_object__invalid_code_value__string(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = "4294967296" - + self.assertEqual(cm.exception.args[0], "invalid code attribute value") def test_diameter_avp__custom_object__invalid_code_value__list(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = ["4294967296"] - + self.assertEqual(cm.exception.args[0], "invalid code attribute value") def test_diameter_avp__custom_object__invalid_code_value__integer_1(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = 4294967296 - + self.assertEqual(cm.exception.args[0], "code attribute has 4-bytes length long") def test_diameter_avp__custom_object__invalid_code_value__integer_2(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = -1 - + self.assertEqual(cm.exception.args[0], "code attribute has 4-bytes length long") def test_diameter_avp__custom_object__invalid_code_value__bytes_1(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = bytes.fromhex("00") - + self.assertEqual(cm.exception.args[0], "code attribute has 4-bytes length long") def test_diameter_avp__custom_object__invalid_code_value__bytes_2(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = bytes.fromhex("0000") - + self.assertEqual(cm.exception.args[0], "code attribute has 4-bytes length long") def test_diameter_avp__custom_object__invalid_code_value__bytes_3(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = bytes.fromhex("000000") - + self.assertEqual(cm.exception.args[0], "code attribute has 4-bytes length long") def test_diameter_avp__custom_object__invalid_code_value__bytes_4(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.code = bytes.fromhex("0000000001") - + self.assertEqual(cm.exception.args[0], "code attribute has 4-bytes length long") def test_diameter_avp__custom_object__invalid_flags_value__string(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.flags = "256" - + self.assertEqual(cm.exception.args[0], "invalid flags attribute value") def test_diameter_avp__custom_object__invalid_flags_value__list(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.flags = ["256"] - + self.assertEqual(cm.exception.args[0], "invalid flags attribute value") def test_diameter_avp__custom_object__invalid_flags_value__integer_1(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.flags = 256 - + self.assertEqual(cm.exception.args[0], "flags attribute has 1-byte length long") def test_diameter_avp__custom_object__invalid_flags_value__integer_2(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.flags = -1 - + self.assertEqual(cm.exception.args[0], "flags attribute has 1-byte length long") def test_diameter_avp__custom_object__invalid_flags_value__bytes_1(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.flags = bytes.fromhex("0000") - + self.assertEqual(cm.exception.args[0], "flags attribute has 1-byte length long") def test_diameter_avp__custom_object__invalid_flags_value__bytes_2(self): avp = DiameterAVP() - with self.assertRaises(AVPAttributeValueError) as cm: + with self.assertRaises(AVPAttributeValueError) as cm: avp.flags = bytes.fromhex("000000") - + self.assertEqual(cm.exception.args[0], "flags attribute has 1-byte length long") def test_diameter_avp__custom_object__flags_by_setting_class_attributes(self): @@ -373,7 +373,7 @@ def test_diameter_avp__custom_object__length_by_using_len_builtin_function(self) def test_diameter_avp__custom_object__length_changing_based_on_data(self): avp = DiameterAVP() - + avp.data = "srcrary" self.assertEqual(avp.get_length(), 15) self.assertEqual(avp.get_padding_length(), 1) @@ -393,14 +393,14 @@ def test_diameter_avp__custom_object__length_changing_based_on_data(self): avp.data += b"B" self.assertEqual(avp.get_length(), 23) self.assertEqual(avp.get_padding_length(), 1) - + avp.data += b"=" self.assertEqual(avp.get_length(), 24) self.assertIsNone(avp.get_padding_length()) def test_diameter_avp__custom_object__length_by_adding_vendor_id(self): avp = DiameterAVP() - + avp.data = "srcrary" self.assertEqual(avp.get_length(), 15) self.assertEqual(avp.get_padding_length(), 1) @@ -527,9 +527,9 @@ def test_diameter_avp__eq_dunder(self): def test_diameter_avp__custom_object__padding_not_allowed_to_set(self): avp = DiameterAVP() - with self.assertRaises(AttributeError) as cm: + with self.assertRaises(AttributeError) as cm: avp.padding = bytes.fromhex("0000") - self.assertEqual(cm.exception.args[0], "can't set attribute") + self.assertEqual(cm.exception.args[0], "can't set attribute 'padding'") class TestDiameterHeader(unittest.TestCase): @@ -724,9 +724,9 @@ def test_diameter_header_request_bit__default(self): def test_diameter_header_request_bit__unset_when_is_unset(self): header = DiameterHeader() - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_request_bit(False) - + self.assertEqual(cm.exception.args[0], "R-bit was already unset") def test_diameter_header_request_bit__set_when_is_set(self): @@ -736,9 +736,9 @@ def test_diameter_header_request_bit__set_when_is_set(self): header.set_request_bit(True) self.assertTrue(header.is_request()) - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_request_bit(True) - + self.assertEqual(cm.exception.args[0], "R-bit was already set") def test_diameter_header_request_bit__set_when_error_bit_is_set(self): @@ -747,10 +747,10 @@ def test_diameter_header_request_bit__set_when_error_bit_is_set(self): header.set_error_bit(True) self.assertTrue(header.is_error()) - - with self.assertRaises(DiameterHeaderError) as cm: + + with self.assertRaises(DiameterHeaderError) as cm: header.set_request_bit(True) - + self.assertEqual(cm.exception.args[0], "R-bit MUST NOT be set when E-bit is set") header.set_error_bit(False) @@ -778,9 +778,9 @@ def test_diameter_header_proxiable_bit__default(self): def test_diameter_header_proxiable_bit__unset_when_is_unset(self): header = DiameterHeader() - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_proxiable_bit(False) - + self.assertEqual(cm.exception.args[0], "P-bit was already unset") def test_diameter_header_proxiable_bit__set_when_is_set(self): @@ -790,9 +790,9 @@ def test_diameter_header_proxiable_bit__set_when_is_set(self): header.set_proxiable_bit(True) self.assertTrue(header.is_proxiable()) - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_proxiable_bit(True) - + self.assertEqual(cm.exception.args[0], "P-bit was already set") def test_diameter_header_error_bit__default(self): @@ -814,9 +814,9 @@ def test_diameter_header_error_bit__default(self): def test_diameter_header_error_bit__unset_when_is_unset(self): header = DiameterHeader() - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_error_bit(False) - + self.assertEqual(cm.exception.args[0], "E-bit was already unset") def test_diameter_header_error_bit__set_when_is_set(self): @@ -826,9 +826,9 @@ def test_diameter_header_error_bit__set_when_is_set(self): header.set_error_bit(True) self.assertTrue(header.is_error()) - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_error_bit(True) - + self.assertEqual(cm.exception.args[0], "E-bit was already set") def test_diameter_header_error_bit__set_when_request_bit_is_set(self): @@ -837,10 +837,10 @@ def test_diameter_header_error_bit__set_when_request_bit_is_set(self): header.set_request_bit(True) self.assertTrue(header.is_request()) - - with self.assertRaises(DiameterHeaderError) as cm: + + with self.assertRaises(DiameterHeaderError) as cm: header.set_error_bit(True) - + self.assertEqual(cm.exception.args[0], "E-bit MUST NOT be set when R-bit is set") header.set_request_bit(False) @@ -868,9 +868,9 @@ def test_diameter_header_retransmitted_bit__default(self): def test_diameter_header_retransmitted_bit__unset_when_is_unset(self): header = DiameterHeader() - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_retransmitted_bit(False) - + self.assertEqual(cm.exception.args[0], "T-bit was already unset") def test_diameter_header_retransmitted_bit__set_when_is_set(self): @@ -880,9 +880,9 @@ def test_diameter_header_retransmitted_bit__set_when_is_set(self): header.set_retransmitted_bit(True) self.assertTrue(header.is_retransmitted()) - with self.assertRaises(DiameterHeaderError) as cm: + with self.assertRaises(DiameterHeaderError) as cm: header.set_retransmitted_bit(True) - + self.assertEqual(cm.exception.args[0], "T-bit was already set") @@ -901,18 +901,18 @@ def test_diameter_message__invalid_way_of_passing_avp_input_argument__1(self): header = DiameterHeader() avp = DiameterAVP() - with self.assertRaises(DiameterMessageError) as cm: + with self.assertRaises(DiameterMessageError) as cm: message = DiameterMessage(header, avp) - + self.assertEqual(cm.exception.args[0], "invalid input argument: 'avps'. It MUST be a list of DiameterAVP objects") def test_diameter_message__invalid_way_of_passing_avp_input_argument__2(self): header = DiameterHeader() avps = [DiameterAVP(), 1] - with self.assertRaises(DiameterMessageError) as cm: + with self.assertRaises(DiameterMessageError) as cm: message = DiameterMessage(header, avps) - + self.assertEqual(cm.exception.args[0], "invalid element found in list argument: 'avps'. The element 1 in position 1 does not represent a DiameterAVP object") def test_diameter_message__invalid_way_of_passing_avp_input_argument__3(self): @@ -921,16 +921,16 @@ def test_diameter_message__invalid_way_of_passing_avp_input_argument__3(self): with self.assertRaises(DiameterMessageError) as cm: message = DiameterMessage(header, avps) - + self.assertEqual(cm.exception.args[0], "invalid element found in list argument: 'avps'. The element DiameterAVP() in position 2 does not represent a DiameterAVP object") def test_diameter_message__invalid_way_of_passing_avp_input_argument__4(self): header = DiameterHeader() avps = [DiameterAVP(), OriginHostAVP("host.example.com"), "DiameterAVP()"] - with self.assertRaises(DiameterMessageError) as cm: + with self.assertRaises(DiameterMessageError) as cm: message = DiameterMessage(header, avps) - + self.assertEqual(cm.exception.args[0], "invalid element found in list argument: 'avps'. The element DiameterAVP() in position 2 does not represent a DiameterAVP object") def test_diameter_message__valid_way_of_passing_avp_input_argument(self): @@ -1080,7 +1080,7 @@ def test_diameter_message__append_method__valid_way_to_call__creating_cer_like_m message.header.command_code = CAPABILITIES_EXCHANGE_MESSAGE self.assertEqual(message.__str__(), "") - + message.header.application_id = DIAMETER_APPLICATION_DEFAULT self.assertEqual(message.__str__(), "") @@ -1118,12 +1118,12 @@ def test_diameter_message__pop_method__valid_way_to_call(self): with self.assertRaises(DiameterMessageError) as cm: message.pop("origin_realm_avp") - + self.assertEqual(cm.exception.args[0], "`avps` attribute is empty. There is no DiameterAVP object to be removed") def test_diameter_message__cleanup_method_after_appending_2_avps(self): origin_host_avp = OriginHostAVP("host.example.com") - origin_realm_avp = OriginRealmAVP("example.com") + origin_realm_avp = OriginRealmAVP("example.com") avps = [origin_host_avp, origin_realm_avp] message = DiameterMessage(avps=avps) @@ -1198,7 +1198,7 @@ def test_diameter_message__extend_method__valid_way_to_call__5_avps(self): origin_state_id_avp = OriginStateIdAVP(1524733202) auth_session_state_avp = AuthSessionStateAVP(STATE_MAINTAINED) auth_request_type_avp = AuthRequestTypeAVP(AUTH_REQUEST_TYPE_AUTHENTICATE_ONLY) - + avps = [origin_host_avp, origin_realm_avp, origin_state_id_avp, auth_session_state_avp, auth_request_type_avp] message = DiameterMessage() @@ -1251,7 +1251,7 @@ def test_diameter_message__has_avp_method__valid_way_to_call__2(self): self.assertEqual(message.auth_request_type_avp.__str__(), "") self.assertTrue(message.has_avp("auth_session_state_avp")) self.assertTrue(message.has_avp("auth_request_type_avp")) - + self.assertFalse(message.has_avp("origin_host_avp")) self.assertFalse(message.has_avp("origin_realm_avp")) self.assertFalse(message.has_avp("origin_state_id_avp")) @@ -1420,7 +1420,7 @@ def test_diameter_message__update_key_method__invalid_way_to_call(self): with self.assertRaises(DiameterMessageError) as cm: message.update_key("firmware1", "firmware0") - + self.assertEqual(cm.exception.args[0], "`firmware0` key already defined") def test_diameter_message__refresh_method__1(self): @@ -1716,7 +1716,7 @@ def test_diameter_message__refresh_method__3(self): self.assertEqual(message.avps[2].data, b"mme.epc.3gppnetwork.org") self.assertEqual(message.avps[2].get_padding_length(), 1) - #: See the needed to refresh it. Before call it the length is wrong if + #: See the needed to refresh it. Before call it the length is wrong if #: you try to access through the Header object self.assertEqual(message.header.length.hex(), "0001a0") self.assertEqual(message.header.get_length(), 416) @@ -1744,7 +1744,7 @@ def test_diameter_message__refresh_method__3(self): self.assertEqual(message.avps[3].data, b"epc.3gppnetwork.org") self.assertEqual(message.avps[3].get_padding_length(), 1) - #: See the needed to refresh it. Before call it the length is wrong if + #: See the needed to refresh it. Before call it the length is wrong if #: you try to access through the Header object self.assertEqual(message.header.length.hex(), "000190") self.assertEqual(message.header.get_length(), 400) @@ -1772,7 +1772,7 @@ def test_diameter_message__refresh_method__3(self): self.assertEqual(message.avps[0].data, b"mme.epc.3gppnetwork.org;1559529822;356549175;2.17;940463984") self.assertEqual(message.avps[0].get_padding_length(), 1) - #: See the needed to refresh it. Before call it the length is wrong if + #: See the needed to refresh it. Before call it the length is wrong if #: you try to access through the Header object self.assertEqual(message.header.length.hex(), "000180") self.assertEqual(message.header.get_length(), 384) @@ -1800,7 +1800,7 @@ def test_diameter_message__refresh_method__3(self): self.assertEqual(message.avps[4].data, b"epc.3gppnetwork.org") self.assertEqual(message.avps[4].get_padding_length(), 1) - #: See the needed to refresh it. Before call it the length is wrong if + #: See the needed to refresh it. Before call it the length is wrong if #: you try to access through the Header object self.assertEqual(message.header.length.hex(), "000170") self.assertEqual(message.header.get_length(), 368) @@ -1828,7 +1828,7 @@ def test_diameter_message__refresh_method__3(self): self.assertEqual(message.avps[5].data, b"frodo") self.assertEqual(message.avps[5].get_padding_length(), 3) - #: See the needed to refresh it. Before call it the length is wrong if + #: See the needed to refresh it. Before call it the length is wrong if #: you try to access through the Header object self.assertEqual(message.header.length.hex(), "000160") self.assertEqual(message.header.get_length(), 352) @@ -1846,9 +1846,10 @@ def test_diameter_message__update_avps_method__1(self): #: Initial Setup config = { "MODE": "CLIENT", + "TRANSPORT_TYPE": "TCP", "APPLICATIONS": [ { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx }, { @@ -2022,7 +2023,7 @@ def test_diameter_message__update_avps_method__1(self): self.assertEqual(cea.avps[8].avps[1].length.hex(), "00000c") self.assertEqual(cea.avps[8].avps[1].get_length(), 12) self.assertEqual(cea.avps[8].avps[1].data, DIAMETER_APPLICATION_Rx) - self.assertIsNone(cea.avps[8].avps[1].get_padding_length()) + self.assertIsNone(cea.avps[8].avps[1].get_padding_length()) #: Update all AVPs @@ -2479,7 +2480,7 @@ def test_diameter_message__update_avps_method__2(self): self.assertEqual(cea.avps[8].avps[1].length.hex(), "00000c") self.assertEqual(cea.avps[8].avps[1].get_length(), 12) self.assertEqual(cea.avps[8].avps[1].data, DIAMETER_APPLICATION_Rx) - self.assertIsNone(cea.avps[8].avps[1].get_padding_length()) + self.assertIsNone(cea.avps[8].avps[1].get_padding_length()) #: Update all AVPs avps = { @@ -2491,11 +2492,11 @@ def test_diameter_message__update_avps_method__2(self): "product_name": f"Bromelia-AAA", "auth_application_id": DIAMETER_APPLICATION_S6b, "vendor_specific_application_id": [ - VendorIdAVP(VENDOR_ID_ETSI), + VendorIdAVP(VENDOR_ID_ETSI), AuthApplicationIdAVP(DIAMETER_APPLICATION_SWm) ], "vendor_specific_application_id__1": [ - VendorIdAVP(VENDOR_ID_ETSI), + VendorIdAVP(VENDOR_ID_ETSI), AuthApplicationIdAVP(DIAMETER_APPLICATION_SWx) ], "origin_state_id": 42, @@ -2796,4 +2797,4 @@ def test_diameter_message__update_avps_method__2(self): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_proxy.py b/tests/test_proxy.py index b87146e..5d7fdf3 100644 --- a/tests/test_proxy.py +++ b/tests/test_proxy.py @@ -4,7 +4,7 @@ ~~~~~~~~~~~~~~~ This module contains the Bromelia proxy unittests. - + :copyright: (c) 2021 Henrique Marques Ribeiro. :license: MIT, see LICENSE for more details. """ @@ -35,6 +35,7 @@ class TestDiameterBaseProxy(unittest.TestCase): def setUp(self): config = { "MODE": "CLIENT", + "TRANSPORT_TYPE": "TCP", "APPLICATIONS": [], "LOCAL_NODE_HOSTNAME": "client.network", "LOCAL_NODE_REALM": "network", @@ -47,7 +48,7 @@ def setUp(self): "WATCHDOG_TIMEOUT": 30 } self.connection = _convert_config_to_connection_obj(config) - + def test__load_cea(self): cea = DiameterBaseProxy.load_cea(self.connection) @@ -389,7 +390,7 @@ def test__diameter_base_proxy__get_default_messages(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -478,7 +479,7 @@ def test__diameter_base_proxy__get_default_messages(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -728,7 +729,7 @@ def test__diameter_base_proxy__get_custom_messages__no_list(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -817,7 +818,7 @@ def test__diameter_base_proxy__get_custom_messages__no_list(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -1073,7 +1074,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_cea(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -1195,7 +1196,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_cea(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -1451,7 +1452,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_cer(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -1540,7 +1541,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_cer(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -1830,7 +1831,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dwa(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -1919,7 +1920,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dwa(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -2209,7 +2210,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dwr(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -2298,7 +2299,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dwr(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -2588,7 +2589,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dpa(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -2677,7 +2678,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dpa(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -2967,7 +2968,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dpr(self): self.assertTrue(isinstance(base.dpr, DisconnectPeerRequest)) ##: Check Capabilities-Exchange-Answer - + #: Check Diameter Header self.assertEqual(base.cea.header.version, DIAMETER_VERSION) self.assertEqual(base.cea.header.flags, FLAG_RESPONSE) @@ -3056,7 +3057,7 @@ def test__diameter_base_proxy__get_custom_messages__custom_dpr(self): ##: Check Capabilities-Exchange-Request - + #: Check Diameter Header self.assertEqual(base.cer.header.version, DIAMETER_VERSION) self.assertEqual(base.cer.header.flags, FLAG_REQUEST) @@ -3331,8 +3332,9 @@ class TestDiameterBaseProxyWithOneApplicationId(unittest.TestCase): def setUp(self): config = { "MODE": "CLIENT", + "TRANSPORT_TYPE": "TCP", "APPLICATIONS": [{ - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx }], "LOCAL_NODE_HOSTNAME": "client.network", @@ -3588,9 +3590,10 @@ class TestDiameterBaseProxyWithTwoApplicationsId(unittest.TestCase): def setUp(self): config = { "MODE": "CLIENT", + "TRANSPORT_TYPE": "TCP", "APPLICATIONS": [ { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx }, { @@ -3913,4 +3916,4 @@ def test__load_cer(self): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_setup.py b/tests/test_setup.py index e341b18..ccea8b6 100644 --- a/tests/test_setup.py +++ b/tests/test_setup.py @@ -4,7 +4,7 @@ ~~~~~~~~~~~~~~~ This module contains the Bromelia setup unittests. - + :copyright: (c) 2021 Henrique Marques Ribeiro. :license: MIT, see LICENSE for more details. """ @@ -68,7 +68,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client( s_config["LOCAL_NODE_PORT"] = 3868 c_config["PEER_NODE_PORT"] = 3868 - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -86,7 +86,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client( self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -110,12 +110,12 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client( time.sleep(5) # It may be better - #: Check the Diameter Server state now is running. + #: Check the Diameter Server state now is closed. self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - #: Check the Diameter Client state now is running. + #: Check the Diameter Client state now is closed. self.assertFalse(c.is_open()) self.assertTrue(c.is_closed()) self.assertEqual(c.get_current_state(), CLOSED) @@ -128,7 +128,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server( s_config["LOCAL_NODE_PORT"] = 3869 c_config["PEER_NODE_PORT"] = 3869 - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -146,7 +146,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server( self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -247,7 +247,7 @@ def test__get_base_message__cea__from_client(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -323,7 +323,7 @@ def test__get_base_message__dwa__from_client(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -387,7 +387,7 @@ def test__get_base_message__dpa__from_client(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -463,7 +463,7 @@ def test__get_base_message__cea__from_server(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "00000108400000167365727665722e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"server.network") @@ -539,7 +539,7 @@ def test__get_base_message__dwa__from_server(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "00000108400000167365727665722e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"server.network") @@ -603,7 +603,7 @@ def test__get_base_message__dpa__from_server(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "00000108400000167365727665722e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"server.network") @@ -653,14 +653,14 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client( c_config["PEER_NODE_PORT"] = 3868 app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } s_config["APPLICATIONS"].append(app) c_config["APPLICATIONS"].append(app) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -678,7 +678,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client( self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -721,14 +721,14 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server( c_config["PEER_NODE_PORT"] = 3869 app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } s_config["APPLICATIONS"].append(app) c_config["APPLICATIONS"].append(app) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -746,7 +746,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server( self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -783,7 +783,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server( def test__get_base_message__cer__from_client__app_Cx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } @@ -833,11 +833,11 @@ def test__get_base_message__cer__from_client__app_Cx(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000000") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_Cx) - + def test__get_base_message__cea__from_client__app_Cx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } @@ -862,7 +862,7 @@ def test__get_base_message__cea__from_client__app_Cx(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -895,7 +895,7 @@ def test__get_base_message__cea__from_client__app_Cx(self): def test__get_base_message__cer__from_client__app_Sh(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Sh } @@ -945,11 +945,11 @@ def test__get_base_message__cer__from_client__app_Sh(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000001") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_Sh) - + def test__get_base_message__cea__from_client__app_Sh(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Sh } @@ -974,7 +974,7 @@ def test__get_base_message__cea__from_client__app_Sh(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1007,7 +1007,7 @@ def test__get_base_message__cea__from_client__app_Sh(self): def test__get_base_message__cer__from_client__app_Zh(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } @@ -1057,11 +1057,11 @@ def test__get_base_message__cer__from_client__app_Zh(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000005") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_Zh) - + def test__get_base_message__cea__from_client__app_Zh(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } @@ -1086,7 +1086,7 @@ def test__get_base_message__cea__from_client__app_Zh(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1119,7 +1119,7 @@ def test__get_base_message__cea__from_client__app_Zh(self): def test__get_base_message__cer__from_client__app_Rx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } @@ -1169,11 +1169,11 @@ def test__get_base_message__cer__from_client__app_Rx(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000014") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_Rx) - + def test__get_base_message__cea__from_client__app_Rx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } @@ -1198,7 +1198,7 @@ def test__get_base_message__cea__from_client__app_Rx(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1231,7 +1231,7 @@ def test__get_base_message__cea__from_client__app_Rx(self): def test__get_base_message__cer__from_client__app_Gx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } @@ -1281,11 +1281,11 @@ def test__get_base_message__cer__from_client__app_Gx(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000016") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_Gx) - + def test__get_base_message__cea__from_client__app_Gx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } @@ -1310,7 +1310,7 @@ def test__get_base_message__cea__from_client__app_Gx(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1343,7 +1343,7 @@ def test__get_base_message__cea__from_client__app_Gx(self): def test__get_base_message__cer__from_client__app_S6a(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S6a } @@ -1393,11 +1393,11 @@ def test__get_base_message__cer__from_client__app_S6a(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000023") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_S6a) - + def test__get_base_message__cea__from_client__app_S6a(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S6a } @@ -1422,7 +1422,7 @@ def test__get_base_message__cea__from_client__app_S6a(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1455,7 +1455,7 @@ def test__get_base_message__cea__from_client__app_S6a(self): def test__get_base_message__cer__from_client__app_S13(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S13 } @@ -1505,11 +1505,11 @@ def test__get_base_message__cer__from_client__app_S13(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000024") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_S13) - + def test__get_base_message__cea__from_client__app_S13(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S13 } @@ -1534,7 +1534,7 @@ def test__get_base_message__cea__from_client__app_S13(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1567,7 +1567,7 @@ def test__get_base_message__cea__from_client__app_S13(self): def test__get_base_message__cer__from_client__app_SWm(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_SWm } @@ -1617,11 +1617,11 @@ def test__get_base_message__cer__from_client__app_SWm(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000030") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_SWm) - + def test__get_base_message__cea__from_client__app_SWm(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_SWm } @@ -1646,7 +1646,7 @@ def test__get_base_message__cea__from_client__app_SWm(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1679,7 +1679,7 @@ def test__get_base_message__cea__from_client__app_SWm(self): def test__get_base_message__cer__from_client__app_SWx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_SWx } @@ -1729,11 +1729,11 @@ def test__get_base_message__cer__from_client__app_SWx(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000031") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_SWx) - + def test__get_base_message__cea__from_client__app_SWx(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_SWx } @@ -1758,7 +1758,7 @@ def test__get_base_message__cea__from_client__app_SWx(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1791,7 +1791,7 @@ def test__get_base_message__cea__from_client__app_SWx(self): def test__get_base_message__cer__from_client__app_S6b(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S6b } @@ -1841,11 +1841,11 @@ def test__get_base_message__cer__from_client__app_S6b(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c01000038") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_S6b) - + def test__get_base_message__cea__from_client__app_S6b(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S6b } @@ -1870,7 +1870,7 @@ def test__get_base_message__cea__from_client__app_S6b(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -1903,7 +1903,7 @@ def test__get_base_message__cea__from_client__app_S6b(self): def test__get_base_message__cer__from_client__app_SLh(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_SLh } @@ -1953,11 +1953,11 @@ def test__get_base_message__cer__from_client__app_SLh(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c000028af000001024000000c0100004b") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_SLh) - + def test__get_base_message__cea__from_client__app_SLh(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_SLh } @@ -1982,7 +1982,7 @@ def test__get_base_message__cea__from_client__app_SLh(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -2015,7 +2015,7 @@ def test__get_base_message__cea__from_client__app_SLh(self): def test__get_base_message__cer__from_client__app_UNKNOWN(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_DEFAULT, + "vendor_id": VENDOR_ID_DEFAULT, "app_id": DIAMETER_APPLICATION_UNKNOWN } @@ -2065,11 +2065,11 @@ def test__get_base_message__cer__from_client__app_UNKNOWN(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c00000000000001024000000cf5c6f515") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_DEFAULT) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_UNKNOWN) - + def test__get_base_message__cea__from_client__app_UNKNOWN(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_DEFAULT, + "vendor_id": VENDOR_ID_DEFAULT, "app_id": DIAMETER_APPLICATION_UNKNOWN } @@ -2094,7 +2094,7 @@ def test__get_base_message__cea__from_client__app_UNKNOWN(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -2127,7 +2127,7 @@ def test__get_base_message__cea__from_client__app_UNKNOWN(self): def test__get_base_message__cer__from_client__app_RELAY(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_DEFAULT, + "vendor_id": VENDOR_ID_DEFAULT, "app_id": DIAMETER_APPLICATION_RELAY } @@ -2177,11 +2177,11 @@ def test__get_base_message__cer__from_client__app_RELAY(self): self.assertEqual(msg.avps[6].dump().hex(), "00000104400000200000010a4000000c00000000000001024000000cffffffff") self.assertEqual(msg.avps[6].vendor_id_avp.data, VENDOR_ID_DEFAULT) self.assertEqual(msg.avps[6].auth_application_id_avp.data, DIAMETER_APPLICATION_RELAY) - + def test__get_base_message__cea__from_client__app_RELAY(self): #: Preconditions. app = { - "vendor_id": VENDOR_ID_DEFAULT, + "vendor_id": VENDOR_ID_DEFAULT, "app_id": DIAMETER_APPLICATION_RELAY } @@ -2206,7 +2206,7 @@ def test__get_base_message__cea__from_client__app_RELAY(self): #: Result-Code AVP self.assertEqual(msg.avps[0].dump().hex(), "0000010c4000000c000007d1") self.assertEqual(msg.avps[0].data, DIAMETER_SUCCESS) - + #: Origin-Host AVP self.assertEqual(msg.avps[1].dump().hex(), "0000010840000016636c69656e742e6e6574776f726b0000") self.assertEqual(msg.avps[1].data, b"client.network") @@ -2277,11 +2277,11 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["PEER_NODE_PORT"] = 3868 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } @@ -2291,7 +2291,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["APPLICATIONS"].append(app1) c_config["APPLICATIONS"].append(app2) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2309,7 +2309,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2352,11 +2352,11 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["PEER_NODE_PORT"] = 3869 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } @@ -2366,7 +2366,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["APPLICATIONS"].append(app1) c_config["APPLICATIONS"].append(app2) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2384,7 +2384,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2427,15 +2427,15 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["PEER_NODE_PORT"] = 3868 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } @@ -2447,7 +2447,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["APPLICATIONS"].append(app2) c_config["APPLICATIONS"].append(app3) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2465,7 +2465,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2508,15 +2508,15 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["PEER_NODE_PORT"] = 3869 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } @@ -2528,7 +2528,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["APPLICATIONS"].append(app2) c_config["APPLICATIONS"].append(app3) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2546,7 +2546,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2589,19 +2589,19 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["PEER_NODE_PORT"] = 3868 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } app4 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } @@ -2615,7 +2615,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["APPLICATIONS"].append(app3) c_config["APPLICATIONS"].append(app4) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2633,7 +2633,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2676,19 +2676,19 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["PEER_NODE_PORT"] = 3869 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } app4 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } @@ -2702,7 +2702,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["APPLICATIONS"].append(app3) c_config["APPLICATIONS"].append(app4) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2720,7 +2720,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2763,23 +2763,23 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["PEER_NODE_PORT"] = 3868 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } app4 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } app5 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S6b } @@ -2795,7 +2795,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ c_config["APPLICATIONS"].append(app4) c_config["APPLICATIONS"].append(app5) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2813,7 +2813,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_client_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2856,23 +2856,23 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["PEER_NODE_PORT"] = 3869 app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } app4 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } app5 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_S6b } @@ -2888,7 +2888,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ c_config["APPLICATIONS"].append(app4) c_config["APPLICATIONS"].append(app5) - #: Instantiate Diameter objects Diameter Server and Client, + #: Instantiate Diameter objects Diameter Server and Client, #: respectivetly. s = Diameter(config=s_config) c = Diameter(config=c_config) @@ -2906,7 +2906,7 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ self.assertFalse(s.is_open()) self.assertTrue(s.is_closed()) self.assertEqual(s.get_current_state(), CLOSED) - + time.sleep(1) #: Start the Diameter Client. @@ -2943,11 +2943,11 @@ def test__capability_exchange_procedure_and_disconnection_procedure_from_server_ def test__get_base_message__cer__from_client__2_apps(self): #: Preconditions. app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Sh } @@ -3007,15 +3007,15 @@ def test__get_base_message__cer__from_client__2_apps(self): def test__get_base_message__cer__from_client__3_apps(self): #: Preconditions. app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Sh } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } @@ -3081,19 +3081,19 @@ def test__get_base_message__cer__from_client__3_apps(self): def test__get_base_message__cer__from_client__4_apps(self): #: Preconditions. app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Sh } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } app4 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } @@ -3165,23 +3165,23 @@ def test__get_base_message__cer__from_client__4_apps(self): def test__get_base_message__cer__from_client__5_apps(self): #: Preconditions. app1 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Cx } app2 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Sh } app3 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Zh } app4 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Rx } app5 = { - "vendor_id": VENDOR_ID_3GPP, + "vendor_id": VENDOR_ID_3GPP, "app_id": DIAMETER_APPLICATION_Gx } @@ -3256,6 +3256,165 @@ def test__get_base_message__cer__from_client__5_apps(self): self.assertEqual(msg.avps[10].vendor_id_avp.data, VENDOR_ID_3GPP) self.assertEqual(msg.avps[10].auth_application_id_avp.data, DIAMETER_APPLICATION_Gx) +# @unittest.SkipTest +class TestDiameterBaseSpecificClientPort(unittest.TestCase): + def setUp(self): + self.server_config = { + "MODE": "SERVER", + "APPLICATIONS": [], + "LOCAL_NODE_HOSTNAME": "server.network", + "LOCAL_NODE_REALM": "network", + "LOCAL_NODE_IP_ADDRESS": "127.0.0.1", + "LOCAL_NODE_PORT": None, + "PEER_NODE_HOSTNAME": "client.network", + "PEER_NODE_REALM": "network", + "PEER_NODE_IP_ADDRESS": "127.0.0.1", + "PEER_NODE_PORT": None, + "WATCHDOG_TIMEOUT": 30 + } + + self.client_config = { + "MODE": "CLIENT", + "APPLICATIONS": [], + "LOCAL_NODE_HOSTNAME": "client.network", + "LOCAL_NODE_REALM": "network", + "LOCAL_NODE_IP_ADDRESS": "127.0.0.1", + "LOCAL_NODE_PORT": None, + "PEER_NODE_HOSTNAME": "server.network", + "PEER_NODE_REALM": "network", + "PEER_NODE_IP_ADDRESS": "127.0.0.1", + "PEER_NODE_PORT": None, + "WATCHDOG_TIMEOUT": 30 + } + + def test__capability_exchange_procedure_and_disconnection_procedure_from_client_with_specific_client_port(self): + #: Preconditions. + s_config = copy(self.server_config) + c_config = copy(self.client_config) + + s_config["LOCAL_NODE_PORT"] = 3868 + c_config["LOCAL_NODE_PORT"] = 3869 + c_config["PEER_NODE_PORT"] = 3868 + + #: Instantiate Diameter objects Diameter Server and Client, + #: respectivetly. + s = Diameter(config=s_config) + c = Diameter(config=c_config) + + #: It worth note Diameter object is unique across processes. There is + #: no reason to start two or more in a single process. However, in order + #: to allow unittests in a single process, two Diameter objects will be + #: run by using the threading module. That's because the `start` method + #: is blocking for Diameter objects who play role as Diameter Server. + thr_s = threading.Thread(name="server", target=s.start) + thr_s.start() + + #: Before run the Diameter Client, let's verify the Diameter Server + #: state. + self.assertFalse(s.is_open()) + self.assertTrue(s.is_closed()) + self.assertEqual(s.get_current_state(), CLOSED) + + time.sleep(1) + + #: Start the Diameter Client. + thr_c = threading.Thread(name="client", target=c.start) + thr_c.start() + + time.sleep(1) + + #: Check the Diameter Server state now is running. + self.assertTrue(s.is_open()) + self.assertFalse(s.is_closed()) + self.assertEqual(s.get_current_state(), R_OPEN) + + #: Check the Diameter Client state now is running. + self.assertTrue(c.is_open()) + self.assertFalse(c.is_closed()) + self.assertEqual(c.get_current_state(), I_OPEN) + + #: Check that client is bound to expected local port + self.assertEqual(c._association.transport.sock.getsockname()[1], c_config["LOCAL_NODE_PORT"]) + + #: Close the Diameter connection from Client. + c.close() + + time.sleep(5) # It may be better + + #: Check the Diameter Server state now is closed. + self.assertFalse(s.is_open()) + self.assertTrue(s.is_closed()) + self.assertEqual(s.get_current_state(), CLOSED) + + #: Check the Diameter Client state now is closed. + self.assertFalse(c.is_open()) + self.assertTrue(c.is_closed()) + self.assertEqual(c.get_current_state(), CLOSED) + + def test__capability_exchange_procedure_and_disconnection_procedure_from_server_with_specific_client_port(self): + #: Preconditions. + s_config = copy(self.server_config) + c_config = copy(self.client_config) + + s_config["LOCAL_NODE_PORT"] = 3870 + c_config["LOCAL_NODE_PORT"] = 3871 + c_config["PEER_NODE_PORT"] = 3870 + + #: Instantiate Diameter objects Diameter Server and Client, + #: respectivetly. + s = Diameter(config=s_config) + c = Diameter(config=c_config) + + #: It worth note Diameter object is unique across processes. There is + #: no reason to start two or more in a single process. However, in order + #: to allow unittests in a single process, two Diameter objects will be + #: run by using the threading module. That's because the `start` method + #: is blocking for Diameter objects who play role as Diameter Server. + thr_s = threading.Thread(name="server", target=s.start) + thr_s.start() + + #: Before run the Diameter Client, let's verify the Diameter Server + #: state. + self.assertFalse(s.is_open()) + self.assertTrue(s.is_closed()) + self.assertEqual(s.get_current_state(), CLOSED) + + time.sleep(1) + + #: Start the Diameter Client. + thr_c = threading.Thread(name="client", target=c.start) + thr_c.start() + + time.sleep(1) + + #: Check the Diameter Server state now is running. + self.assertTrue(s.is_open()) + self.assertFalse(s.is_closed()) + self.assertEqual(s.get_current_state(), R_OPEN) + + #: Check the Diameter Client state now is running. + self.assertTrue(c.is_open()) + self.assertFalse(c.is_closed()) + self.assertEqual(c.get_current_state(), I_OPEN) + + #: Check that client is bound to expected local port + self.assertEqual(c._association.transport.sock.getsockname()[1], c_config["LOCAL_NODE_PORT"]) + + #: Close the Diameter connection from Server. + s.close() + + time.sleep(5) # It may be better + + #: Check the Diameter Server state now is running. + self.assertFalse(s.is_open()) + self.assertTrue(s.is_closed()) + self.assertEqual(s.get_current_state(), CLOSED) + + #: Check the Diameter Client state now is running. + self.assertFalse(c.is_open()) + self.assertTrue(c.is_closed()) + self.assertEqual(c.get_current_state(), CLOSED) + if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()