Add support for SMPP testing
As defined in [1], the different related actors are implemented in this
commit: ESME and SMSC.
SMSC: In Osmocom, the SMSC is currently implemented inside the NITB or
the MSC. A new Smsc abstract class is created to shared code between the
NITB and the MSC, and also makes it easier for later when the SMSC is
splitted. ESMEs can be dynamically added to its configuration in a
similar way to how the BTSs are added.
ESME: A new class Esme is created which can be used by tests to control
an ESME to interact with the SMSC. The ESME functionalities are
implemented using python-smpplib. Required version of this library is at
least 43cc6f819ec76b2c0a9d36d1d439308634716227, which contains support
for python 3 and some required features to poll the socket.
This commit already contains a few tests which checks different
features and tests the API. Extending tested features or scenarios can be
later done quite easily.
The tests are not enabled by default right now, because there are several
of them in a suite and the ip_address resources are not freed after every
tests which ends up in the suite failing due to missing reserved
resources. All the tests run alone work though. When the issue is fixed
they can then be added to the default list of tests to be run.
[1] http://opensmpp.org/specs/SMPP_v3_4_Issue1_2.pdf
Change-Id: I14ca3cb009d6d646a449ca99b0200da12085c0da
diff --git a/src/osmo_gsm_tester/esme.py b/src/osmo_gsm_tester/esme.py
new file mode 100644
index 0000000..f92863d
--- /dev/null
+++ b/src/osmo_gsm_tester/esme.py
@@ -0,0 +1,138 @@
+# osmo_gsm_tester: SMPP ESME to talk to SMSC
+#
+# Copyright (C) 2017 by sysmocom - s.f.m.c. GmbH
+#
+# Author: Pau Espin Pedrol <pespin@sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import smpplib.gsm
+import smpplib.client
+import smpplib.consts
+import smpplib.exceptions
+
+from . import log, util, event_loop, sms
+
+# if you want to know what's happening inside python-smpplib
+#import logging
+#logging.basicConfig(level='DEBUG')
+
+MAX_SYS_ID_LEN = 16
+MAX_PASSWD_LEN = 16
+
+class Esme(log.Origin):
+ client = None
+ smsc = None
+
+ def __init__(self, msisdn):
+ self.msisdn = msisdn
+ # Get last characters of msisdn to stay inside MAX_SYS_ID_LEN. Similar to modulus operator.
+ self.set_system_id('esme-' + self.msisdn[-11:])
+ super().__init__(log.C_TST, self.system_id)
+ self.set_password('esme-pwd')
+ self.connected = False
+ self.bound = False
+ self.listening = False
+
+ def __del__(self):
+ try:
+ self.disconnect()
+ except smpplib.exceptions.ConnectionError:
+ pass
+
+ def set_smsc(self, smsc):
+ self.smsc = smsc
+
+ def set_system_id(self, name):
+ if len(name) > MAX_SYS_ID_LEN:
+ raise log.Error('Esme system_id too long! %d vs %d', len(name), MAX_SYS_ID_LEN)
+ self.system_id = name
+
+ def set_password(self, password):
+ if len(password) > MAX_PASSWD_LEN:
+ raise log.Error('Esme password too long! %d vs %d', len(password), MAX_PASSWD_LEN)
+ self.password = password
+
+ def conf_for_smsc(self):
+ config = { 'system_id': self.system_id, 'password': self.password }
+ return config
+
+ def poll(self):
+ self.client.poll()
+
+ def start_listening(self):
+ self.listening = True
+ event_loop.register_poll_func(self.poll)
+
+ def stop_listening(self):
+ if not self.listening:
+ return
+ self.listening = False
+ # Empty the queue before processing the unbind + disconnect PDUs
+ event_loop.unregister_poll_func(self.poll)
+ self.poll()
+
+ def connect(self):
+ host, port = self.smsc.addr_port
+ if self.client:
+ self.disconnect()
+ self.client = smpplib.client.Client(host, port, timeout=None)
+ self.client.set_message_sent_handler(
+ lambda pdu: self.dbg('message sent:', repr(pdu)) )
+ self.client.set_message_received_handler(
+ lambda pdu: self.dbg('message received:', repr(pdu)) )
+ self.client.connect()
+ self.connected = True
+ self.client.bind_transceiver(system_id=self.system_id, password=self.password)
+ self.bound = True
+ self.log('Connected and bound successfully. Starting to listen')
+ self.start_listening()
+
+ def disconnect(self):
+ self.stop_listening()
+ if self.bound:
+ self.client.unbind()
+ self.bound = False
+ if self.connected:
+ self.client.disconnect()
+ self.connected = False
+
+ def run_method_expect_failure(self, errcode, method, *args):
+ try:
+ method(*args)
+ #it should not succeed, raise an exception:
+ raise log.Error('SMPP Failure: %s should have failed with SMPP error %d (%s) but succeeded.' % (method, errcode, smpplib.consts.DESCRIPTIONS[errcode]))
+ except smpplib.exceptions.PDUError as e:
+ if e.args[1] != errcode:
+ raise e
+
+ def sms_send(self, sms_obj):
+ parts, encoding_flag, msg_type_flag = smpplib.gsm.make_parts(str(sms_obj))
+
+ self.log('Sending SMS "%s" to %s' % (str(sms_obj), sms_obj.dst_msisdn()))
+ for part in parts:
+ pdu = self.client.send_message(
+ source_addr_ton=smpplib.consts.SMPP_TON_INTL,
+ source_addr_npi=smpplib.consts.SMPP_NPI_ISDN,
+ source_addr=sms_obj.src_msisdn(),
+ dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
+ dest_addr_npi=smpplib.consts.SMPP_NPI_ISDN,
+ destination_addr=sms_obj.dst_msisdn(),
+ short_message=part,
+ data_coding=encoding_flag,
+ esm_class=smpplib.consts.SMPP_MSGMODE_FORWARD,
+ registered_delivery=False,
+ )
+
+# vim: expandtab tabstop=4 shiftwidth=4