diff options
Diffstat (limited to 'openbsc')
39 files changed, 711 insertions, 555 deletions
diff --git a/openbsc/.gitignore b/openbsc/.gitignore index a1695c6ed..e99632d99 100644 --- a/openbsc/.gitignore +++ b/openbsc/.gitignore @@ -38,6 +38,7 @@ stamp-h1 libtool ltmain.sh m4/*.m4 +configure~ # git-version-gen magic .tarball-version diff --git a/openbsc/contrib/ipa.py b/openbsc/contrib/ipa.py index 71cbf45a4..cd1192c71 100755 --- a/openbsc/contrib/ipa.py +++ b/openbsc/contrib/ipa.py @@ -56,7 +56,11 @@ class IPA(object): """ Create TAG as TLV data """ - return struct.pack(">HB", len(v) + 1, t) + v + if isinstance(v, str): + v = v.encode() + if isinstance(t, str): + t = t.encode() + return struct.pack(">HB".encode(), len(v) + 1, t) + v def proto(self, p): """ @@ -96,9 +100,15 @@ class IPA(object): """ Add IPA header (with extension if necessary), data must be represented as bytes """ + if isinstance(data, str): + data = data.encode() + if isinstance(proto, str): + proto = proto.encode() + if isinstance(ext, str): + ext = ext.encode() if ext is None: - return struct.pack(">HB", len(data) + 1, proto) + data - return struct.pack(">HBB", len(data) + 1, proto, ext) + data + return struct.pack(">HB".encode(), len(data) + 1, proto) + data + return struct.pack(">HBB".encode(), len(data) + 1, proto, ext) + data def del_header(self, data): """ diff --git a/openbsc/doc/Makefile.am b/openbsc/doc/Makefile.am index 5a231074c..ac3c914cf 100644 --- a/openbsc/doc/Makefile.am +++ b/openbsc/doc/Makefile.am @@ -1,3 +1,5 @@ SUBDIRS = \ examples \ $(NULL) + +EXTRA_DIST = regen_doc.sh diff --git a/openbsc/doc/regen_doc.sh b/openbsc/doc/regen_doc.sh new file mode 100755 index 000000000..12bb878f8 --- /dev/null +++ b/openbsc/doc/regen_doc.sh @@ -0,0 +1,22 @@ +#!/bin/sh -x + +if [ -z "$DOCKER_PLAYGROUND" ]; then + echo "You need to set DOCKER_PLAYGROUND" + exit 1 +fi + +SCRIPT=$(realpath "$0") +MANUAL_DIR=$(dirname "$SCRIPT")/../../manuals + +COMMIT=${COMMIT:-$(git log -1 --format=format:%H)} + +cd "$DOCKER_PLAYGROUND/scripts" || exit 1 + +OSMO_NITB_BRANCH=$COMMIT ./regen_doc.sh osmo-nitb 4242 \ + "$MANUAL_DIR/OsmoNITB/chapters/counters_generated.adoc" \ + "$MANUAL_DIR/OsmoNITB/vty/nitb_vty_reference.xml" + +OSMO_BSCNAT_BRANCH=$COMMIT ./regen_doc.sh osmo-nitb 4244 \ + "$MANUAL_DIR/OsmoBSCNAT/chapters/counters_generated.adoc" \ + "$MANUAL_DIR/OsmoBSCNAT/vty/nat_vty_reference.xml" \ + "osmo-bsc_nat -c /data/osmo-bsc-nat.cfg" diff --git a/openbsc/include/openbsc/gsm_data_shared.h b/openbsc/include/openbsc/gsm_data_shared.h index e616c1260..23acf2f8b 100644 --- a/openbsc/include/openbsc/gsm_data_shared.h +++ b/openbsc/include/openbsc/gsm_data_shared.h @@ -379,13 +379,13 @@ struct gsm_lchan { #endif }; -enum { +enum gsm_bts_trx_ts_flags { TS_F_PDCH_ACTIVE = 0x1000, TS_F_PDCH_ACT_PENDING = 0x2000, TS_F_PDCH_DEACT_PENDING = 0x4000, TS_F_PDCH_PENDING_MASK = 0x6000 /*< TS_F_PDCH_ACT_PENDING | TS_F_PDCH_DEACT_PENDING */ -} gsm_bts_trx_ts_flags; +}; /* One Timeslot in a TRX */ struct gsm_bts_trx_ts { @@ -945,8 +945,8 @@ enum gsm_bts_type_variant str2btsvariant(const char *arg); const char *btsvariant2str(enum gsm_bts_type_variant v); extern const struct value_string gsm_chreq_descs[]; -const struct value_string gsm_pchant_names[13]; -const struct value_string gsm_pchant_descs[13]; +extern const struct value_string gsm_pchant_names[13]; +extern const struct value_string gsm_pchant_descs[13]; const char *gsm_pchan_name(enum gsm_phys_chan_config c); enum gsm_phys_chan_config gsm_pchan_parse(const char *name); const char *gsm_lchant_name(enum gsm_chan_t c); diff --git a/openbsc/osmoappdesc.py b/openbsc/osmoappdesc.py index 5b01e4c57..ba3ccf900 100644 --- a/openbsc/osmoappdesc.py +++ b/openbsc/osmoappdesc.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com> # This program is free software: you can redistribute it and/or modify @@ -39,10 +39,11 @@ app_configs = { apps = [(4242, "src/osmo-bsc/osmo-bsc-sccplite", "OsmoBSC", "osmo-bsc-sccplite"), (4244, "src/osmo-bsc_nat/osmo-bsc_nat", "OsmoBSCNAT", "nat"), (4243, "src/osmo-bsc_mgcp/osmo-bsc_mgcp", "OpenBSC MGCP", "mgcp"), - (4242, "src/osmo-nitb/osmo-nitb", "OpenBSC", "nitb"), + (4242, "src/osmo-nitb/osmo-nitb --yes-i-really-want-to-run-prehistoric-software", "OpenBSC", "nitb"), ] vty_command = ["./src/osmo-nitb/osmo-nitb", "-c", - "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg"] + "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg", + "--yes-i-really-want-to-run-prehistoric-software"] vty_app = apps[3] # reference apps[] entry for osmo-nitb diff --git a/openbsc/src/libbsc/abis_nm.c b/openbsc/src/libbsc/abis_nm.c index 922c2d54a..c1e2b5347 100644 --- a/openbsc/src/libbsc/abis_nm.c +++ b/openbsc/src/libbsc/abis_nm.c @@ -1993,6 +1993,8 @@ int abis_nm_perform_test(struct gsm_bts *bts, uint8_t obj_class, if (!msg) msg = nm_msgb_alloc(); + if (!msg) + return -ENOMEM; msgb_tv_push(msg, NM_ATT_AUTON_REPORT, auton_report); msgb_tv_push(msg, NM_ATT_TEST_NO, test_nr); diff --git a/openbsc/src/libbsc/abis_rsl.c b/openbsc/src/libbsc/abis_rsl.c index 5a508b207..b06f61d1a 100644 --- a/openbsc/src/libbsc/abis_rsl.c +++ b/openbsc/src/libbsc/abis_rsl.c @@ -686,6 +686,13 @@ int rsl_chan_activate_lchan(struct gsm_lchan *lchan, uint8_t act_type, msgb_tv_put(msg, RSL_IE_BS_POWER, lchan->bs_power); msgb_tv_put(msg, RSL_IE_MS_POWER, lchan->ms_power); msgb_tv_put(msg, RSL_IE_TIMING_ADVANCE, ta); + /* indicate MS power control to be performed by BTS: */ + if (lchan->ts->trx->bts->type == GSM_BTS_TYPE_OSMOBTS) + msgb_tl_put(msg, RSL_IE_MS_POWER_PARAM); + /* else: Since IE MS_POWER_PARAM content is operator dependent, it's not + known if non-osmocom BTS models will support an empty IE, so let's + better skip sending it unless we know for sure what each expects. */ + mr_config_for_bts(lchan, msg); msg->dst = lchan->ts->trx->rsl_link; diff --git a/openbsc/src/libbsc/bsc_msc.c b/openbsc/src/libbsc/bsc_msc.c index 82a572dbe..dac0e5203 100644 --- a/openbsc/src/libbsc/bsc_msc.c +++ b/openbsc/src/libbsc/bsc_msc.c @@ -81,7 +81,7 @@ static int msc_connection_connect(struct osmo_fd *fd, unsigned int what) queue = container_of(fd, struct osmo_wqueue, bfd); con = container_of(queue, struct bsc_msc_connection, write_queue); - if ((what & BSC_FD_WRITE) == 0) { + if ((what & OSMO_FD_WRITE) == 0) { LOGP(DMSC, LOGL_ERROR, "MSC(%s) Callback but not writable.\n", con->name); return -1; @@ -107,7 +107,7 @@ static int msc_connection_connect(struct osmo_fd *fd, unsigned int what) /* go to full operation */ fd->cb = osmo_wqueue_bfd_cb; - fd->when = BSC_FD_READ | BSC_FD_EXCEPT; + fd->when = OSMO_FD_READ | OSMO_FD_EXCEPT; con->is_connected = 1; LOGP(DMSC, LOGL_NOTICE, @@ -209,7 +209,7 @@ int bsc_msc_connect(struct bsc_msc_connection *con) if (ret == -1 && errno == EINPROGRESS) { LOGP(DMSC, LOGL_ERROR, "MSC(%s) Connection in progress\n", con->name); - fd->when = BSC_FD_WRITE; + fd->when = OSMO_FD_WRITE; fd->cb = msc_connection_connect; osmo_timer_setup(&con->timeout_timer, msc_con_timeout, con); osmo_timer_schedule(&con->timeout_timer, 20, 0); @@ -218,7 +218,7 @@ int bsc_msc_connect(struct bsc_msc_connection *con) connection_loss(con); return ret; } else { - fd->when = BSC_FD_READ | BSC_FD_EXCEPT; + fd->when = OSMO_FD_READ | OSMO_FD_EXCEPT; fd->cb = osmo_wqueue_bfd_cb; con->is_connected = 1; if (con->connected) diff --git a/openbsc/src/libbsc/bsc_rf_ctrl.c b/openbsc/src/libbsc/bsc_rf_ctrl.c index 0e28600b5..9159a8f76 100644 --- a/openbsc/src/libbsc/bsc_rf_ctrl.c +++ b/openbsc/src/libbsc/bsc_rf_ctrl.c @@ -373,7 +373,7 @@ static int rf_ctrl_accept(struct osmo_fd *bfd, unsigned int what) osmo_wqueue_init(&conn->queue, 10); conn->queue.bfd.data = conn; conn->queue.bfd.fd = fd; - conn->queue.bfd.when = BSC_FD_READ | BSC_FD_WRITE; + conn->queue.bfd.when = OSMO_FD_READ | OSMO_FD_WRITE; conn->queue.read_cb = rf_read_cmd; conn->queue.write_cb = rf_write_cmd; conn->rf = rf; @@ -479,7 +479,7 @@ static int rf_create_socket(struct osmo_bsc_rf *rf, const char *path) return -1; } - bfd->when = BSC_FD_READ; + bfd->when = OSMO_FD_READ; bfd->cb = rf_ctrl_accept; bfd->data = rf; diff --git a/openbsc/src/libbsc/bsc_subscriber.c b/openbsc/src/libbsc/bsc_subscriber.c index 73e61e801..f4300d760 100644 --- a/openbsc/src/libbsc/bsc_subscriber.c +++ b/openbsc/src/libbsc/bsc_subscriber.c @@ -91,6 +91,8 @@ struct bsc_subscr *bsc_subscr_find_or_create_by_imsi(struct llist_head *list, if (bsub) return bsub; bsub = bsc_subscr_alloc(list); + if (!bsub) + return NULL; bsc_subscr_set_imsi(bsub, imsi); return bsub; } @@ -103,6 +105,8 @@ struct bsc_subscr *bsc_subscr_find_or_create_by_tmsi(struct llist_head *list, if (bsub) return bsub; bsub = bsc_subscr_alloc(list); + if (!bsub) + return NULL; bsub->tmsi = tmsi; return bsub; } diff --git a/openbsc/src/libbsc/bsc_vty.c b/openbsc/src/libbsc/bsc_vty.c index c71ea4a3e..55a716b67 100644 --- a/openbsc/src/libbsc/bsc_vty.c +++ b/openbsc/src/libbsc/bsc_vty.c @@ -4335,7 +4335,7 @@ int bsc_vty_init(struct gsm_network *network) install_element_ve(&show_paging_cmd); install_element_ve(&show_paging_group_cmd); - logging_vty_add_cmds(NULL); + logging_vty_add_cmds(); osmo_stats_vty_add_cmds(); install_element(GSMNET_NODE, &cfg_net_neci_cmd); diff --git a/openbsc/src/libbsc/pcu_sock.c b/openbsc/src/libbsc/pcu_sock.c index e2066ba3e..4411df84a 100644 --- a/openbsc/src/libbsc/pcu_sock.c +++ b/openbsc/src/libbsc/pcu_sock.c @@ -484,7 +484,7 @@ static int pcu_sock_send(struct gsm_bts *bts, struct msgb *msg) return -EIO; } msgb_enqueue(&state->upqueue, msg); - conn_bfd->when |= BSC_FD_WRITE; + conn_bfd->when |= OSMO_FD_WRITE; return 0; } @@ -507,7 +507,7 @@ static void pcu_sock_close(struct pcu_sock_state *state) osmo_fd_unregister(bfd); /* re-enable the generation of ACCEPT for new connections */ - state->listen_bfd.when |= BSC_FD_READ; + state->listen_bfd.when |= OSMO_FD_READ; #if 0 /* remove si13, ... */ @@ -586,7 +586,7 @@ static int pcu_sock_write(struct osmo_fd *bfd) msg = llist_entry(state->upqueue.next, struct msgb, list); pcu_prim = (struct gsm_pcu_if *)msg->data; - bfd->when &= ~BSC_FD_WRITE; + bfd->when &= ~OSMO_FD_WRITE; /* bug hunter 8-): maybe someone forgot msgb_put(...) ? */ if (!msgb_length(msg)) { @@ -601,7 +601,7 @@ static int pcu_sock_write(struct osmo_fd *bfd) goto close; if (rc < 0) { if (errno == EAGAIN) { - bfd->when |= BSC_FD_WRITE; + bfd->when |= OSMO_FD_WRITE; break; } goto close; @@ -625,12 +625,12 @@ static int pcu_sock_cb(struct osmo_fd *bfd, unsigned int flags) { int rc = 0; - if (flags & BSC_FD_READ) + if (flags & OSMO_FD_READ) rc = pcu_sock_read(bfd); if (rc < 0) return rc; - if (flags & BSC_FD_WRITE) + if (flags & OSMO_FD_WRITE) rc = pcu_sock_write(bfd); return rc; @@ -656,13 +656,13 @@ static int pcu_sock_accept(struct osmo_fd *bfd, unsigned int flags) LOGP(DPCU, LOGL_NOTICE, "PCU connects but we already have " "another active connection ?!?\n"); /* We already have one PCU connected, this is all we support */ - state->listen_bfd.when &= ~BSC_FD_READ; + state->listen_bfd.when &= ~OSMO_FD_READ; close(rc); return 0; } conn_bfd->fd = rc; - conn_bfd->when = BSC_FD_READ; + conn_bfd->when = OSMO_FD_READ; conn_bfd->cb = pcu_sock_cb; conn_bfd->data = state; @@ -705,7 +705,7 @@ int pcu_sock_init(const char *path, struct gsm_bts *bts) return -1; } - bfd->when = BSC_FD_READ; + bfd->when = OSMO_FD_READ; bfd->cb = pcu_sock_accept; bfd->data = state; diff --git a/openbsc/src/libcommon/bsc_version.c b/openbsc/src/libcommon/bsc_version.c index f0369bffe..9111a0eb1 100644 --- a/openbsc/src/libcommon/bsc_version.c +++ b/openbsc/src/libcommon/bsc_version.c @@ -19,12 +19,19 @@ #include "../../bscconfig.h" +#define DIVIDER "======================================================================\r\n" const char *openbsc_copyright = "Copyright (C) 2008-2016 Harald Welte, Holger Freyther\r\n" "Contributions by Daniel Willmann, Jan Lübbe, Stefan Schmidt\r\n" "Dieter Spaar, Andreas Eversberg, Sylvain Munaut, Neels Hofmeyr\r\n\r\n" "License AGPLv3+: GNU AGPL version 3 or later <http://gnu.org/licenses/agpl-3.0.html>\r\n" "This is free software: you are free to change and redistribute it.\r\n" - "There is NO WARRANTY, to the extent permitted by law.\r\n"; + "There is NO WARRANTY, to the extent permitted by law.\r\n" DIVIDER + "WARNING: You should not run osmo-nitb except for archaeological purpose.\r\n" + "It is ancient, unmaintained and obsolete for several years. But you seem\r\n" + "to insist on using obsolete and known-broken software that ignores years \r\n" + "of hard development work. That's your choice, but you have been warned.\r\n" + "DON'T EVER BOTHER THE DEVELOPERS IF YOU ENCOUNTER PROBLEMS WITH IT!!!\r\n" + "See https://osmocom.org/projects/cellular-infrastructure/wiki/Osmocom_Network_In_The_Box\r\n" DIVIDER; diff --git a/openbsc/src/libcommon/gsm_data_shared.c b/openbsc/src/libcommon/gsm_data_shared.c index 75fe0b0ef..a97a2b9e6 100644 --- a/openbsc/src/libcommon/gsm_data_shared.c +++ b/openbsc/src/libcommon/gsm_data_shared.c @@ -87,7 +87,7 @@ const struct value_string osmo_bts_variant_names[_NUM_BTS_VARIANT + 1] = { { BTS_OSMO_LITECELL15, "osmo-bts-lc15" }, { BTS_OSMO_OCTPHY, "osmo-bts-octphy" }, { BTS_OSMO_SYSMO, "osmo-bts-sysmo" }, - { BTS_OSMO_TRX, "omso-bts-trx" }, + { BTS_OSMO_TRX, "osmo-bts-trx" }, { 0, NULL } }; diff --git a/openbsc/src/libcommon/socket.c b/openbsc/src/libcommon/socket.c index 2a64767f8..5697c3367 100644 --- a/openbsc/src/libcommon/socket.c +++ b/openbsc/src/libcommon/socket.c @@ -65,7 +65,7 @@ int make_sock(struct osmo_fd *bfd, int proto, bfd->fd = socket(AF_INET, type, proto); bfd->cb = cb; - bfd->when = BSC_FD_READ; + bfd->when = OSMO_FD_READ; bfd->data = data; bfd->priv_nr = priv_nr; diff --git a/openbsc/src/libmgcp/mgcp_network.c b/openbsc/src/libmgcp/mgcp_network.c index 49500d92c..e2686587d 100644 --- a/openbsc/src/libmgcp/mgcp_network.c +++ b/openbsc/src/libmgcp/mgcp_network.c @@ -972,14 +972,14 @@ static int bind_rtp(struct mgcp_config *cfg, const char *source_addr, mgcp_set_ip_tos(rtp_end->rtp.fd, cfg->endp_dscp); mgcp_set_ip_tos(rtp_end->rtcp.fd, cfg->endp_dscp); - rtp_end->rtp.when = BSC_FD_READ; + rtp_end->rtp.when = OSMO_FD_READ; if (osmo_fd_register(&rtp_end->rtp) != 0) { LOGP(DMGCP, LOGL_ERROR, "Failed to register RTP port %d on 0x%x\n", rtp_end->local_port, endpno); goto cleanup2; } - rtp_end->rtcp.when = BSC_FD_READ; + rtp_end->rtcp.when = OSMO_FD_READ; if (osmo_fd_register(&rtp_end->rtcp) != 0) { LOGP(DMGCP, LOGL_ERROR, "Failed to register RTCP port %d on 0x%x\n", rtp_end->local_port + 1, endpno); diff --git a/openbsc/src/libmgcp/mgcp_osmux.c b/openbsc/src/libmgcp/mgcp_osmux.c index ce344ca7b..232c23ad6 100644 --- a/openbsc/src/libmgcp/mgcp_osmux.c +++ b/openbsc/src/libmgcp/mgcp_osmux.c @@ -445,7 +445,7 @@ int osmux_init(int role, struct mgcp_config *cfg) return ret; } mgcp_set_ip_tos(osmux_fd.fd, cfg->endp_dscp); - osmux_fd.when |= BSC_FD_READ; + osmux_fd.when |= OSMO_FD_READ; ret = osmo_fd_register(&osmux_fd); if (ret < 0) { diff --git a/openbsc/src/libmsc/meas_feed.c b/openbsc/src/libmsc/meas_feed.c index 3ddcdc39c..64134934c 100644 --- a/openbsc/src/libmsc/meas_feed.c +++ b/openbsc/src/libmsc/meas_feed.c @@ -101,7 +101,7 @@ static int feed_read_cb(struct osmo_fd *ofd) char buf[256]; rc = read(ofd->fd, buf, sizeof(buf)); - ofd->fd &= ~BSC_FD_READ; + ofd->fd &= ~OSMO_FD_READ; return rc; } @@ -140,7 +140,7 @@ int meas_feed_cfg_set(const char *dst_host, uint16_t dst_port) if (rc < 0) return rc; - g_mfs.wqueue.bfd.when &= ~BSC_FD_READ; + g_mfs.wqueue.bfd.when &= ~OSMO_FD_READ; if (g_mfs.dst_host) talloc_free(g_mfs.dst_host); diff --git a/openbsc/src/libmsc/mncc_sock.c b/openbsc/src/libmsc/mncc_sock.c index 0efe3a156..98cafa16c 100644 --- a/openbsc/src/libmsc/mncc_sock.c +++ b/openbsc/src/libmsc/mncc_sock.c @@ -73,7 +73,7 @@ int mncc_sock_from_cc(struct gsm_network *net, struct msgb *msg) /* Actually enqueue the message and mark socket write need */ msgb_enqueue(&net->upqueue, msg); - net->mncc_state->conn_bfd.when |= BSC_FD_WRITE; + net->mncc_state->conn_bfd.when |= OSMO_FD_WRITE; return 0; } @@ -88,7 +88,7 @@ static void mncc_sock_close(struct mncc_sock_state *state) osmo_fd_unregister(bfd); /* re-enable the generation of ACCEPT for new connections */ - state->listen_bfd.when |= BSC_FD_READ; + state->listen_bfd.when |= OSMO_FD_READ; /* release all exisitng calls */ gsm0408_clear_all_trans(state->net, GSM48_PDISC_CC); @@ -151,7 +151,7 @@ static int mncc_sock_write(struct osmo_fd *bfd) msg = llist_entry(net->upqueue.next, struct msgb, list); mncc_prim = (struct gsm_mncc *)msg->data; - bfd->when &= ~BSC_FD_WRITE; + bfd->when &= ~OSMO_FD_WRITE; /* bug hunter 8-): maybe someone forgot msgb_put(...) ? */ if (!msgb_length(msg)) { @@ -166,7 +166,7 @@ static int mncc_sock_write(struct osmo_fd *bfd) goto close; if (rc < 0) { if (errno == EAGAIN) { - bfd->when |= BSC_FD_WRITE; + bfd->when |= OSMO_FD_WRITE; break; } goto close; @@ -190,12 +190,12 @@ static int mncc_sock_cb(struct osmo_fd *bfd, unsigned int flags) { int rc = 0; - if (flags & BSC_FD_READ) + if (flags & OSMO_FD_READ) rc = mncc_sock_read(bfd); if (rc < 0) return rc; - if (flags & BSC_FD_WRITE) + if (flags & OSMO_FD_WRITE) rc = mncc_sock_write(bfd); return rc; @@ -227,7 +227,7 @@ static void queue_hello(struct mncc_sock_state *mncc) hello->lchan_type_offset = offsetof(struct gsm_mncc, lchan_type); msgb_enqueue(&mncc->net->upqueue, msg); - mncc->conn_bfd.when |= BSC_FD_WRITE; + mncc->conn_bfd.when |= OSMO_FD_WRITE; } /* accept a new connection */ @@ -250,13 +250,13 @@ static int mncc_sock_accept(struct osmo_fd *bfd, unsigned int flags) LOGP(DMNCC, LOGL_NOTICE, "MNCC app connects but we already have " "another active connection ?!?\n"); /* We already have one MNCC app connected, this is all we support */ - state->listen_bfd.when &= ~BSC_FD_READ; + state->listen_bfd.when &= ~OSMO_FD_READ; close(rc); return 0; } conn_bfd->fd = rc; - conn_bfd->when = BSC_FD_READ; + conn_bfd->when = OSMO_FD_READ; conn_bfd->cb = mncc_sock_cb; conn_bfd->data = state; @@ -299,7 +299,7 @@ int mncc_sock_init(struct gsm_network *net, const char *sock_path) return -1; } - bfd->when = BSC_FD_READ; + bfd->when = OSMO_FD_READ; bfd->cb = mncc_sock_accept; bfd->data = state; diff --git a/openbsc/src/libmsc/smpp_smsc.c b/openbsc/src/libmsc/smpp_smsc.c index 1cee3fe81..3b76d5793 100644 --- a/openbsc/src/libmsc/smpp_smsc.c +++ b/openbsc/src/libmsc/smpp_smsc.c @@ -915,7 +915,7 @@ static int link_accept_cb(struct smsc *smsc, int fd, osmo_wqueue_init(&esme->wqueue, 10); esme->wqueue.bfd.fd = fd; esme->wqueue.bfd.data = esme; - esme->wqueue.bfd.when = BSC_FD_READ; + esme->wqueue.bfd.when = OSMO_FD_READ; if (osmo_fd_register(&esme->wqueue.bfd) != 0) { close(fd); diff --git a/openbsc/src/libtrau/rtp_proxy.c b/openbsc/src/libtrau/rtp_proxy.c index 6b38ee5fe..59fee12b8 100644 --- a/openbsc/src/libtrau/rtp_proxy.c +++ b/openbsc/src/libtrau/rtp_proxy.c @@ -286,7 +286,7 @@ int rtp_send_frame(struct rtp_socket *rs, struct gsm_data_frame *frame) else memcpy(payload, frame->data, payload_len); msgb_enqueue(&rss->tx_queue, msg); - rss->bfd.when |= BSC_FD_WRITE; + rss->bfd.when |= OSMO_FD_WRITE; return 0; } @@ -396,7 +396,7 @@ static int rtp_socket_read(struct rtp_socket *rs, struct rtp_sub_socket *rss) rc = read(rss->bfd.fd, msg->data, RTP_ALLOC_SIZE); if (rc == 0) { - rss->bfd.when &= ~BSC_FD_READ; + rss->bfd.when &= ~OSMO_FD_READ; goto out_free; } else if (rc < 0) { /* Ignore "connection refused". this happens, If we open the @@ -405,7 +405,7 @@ static int rtp_socket_read(struct rtp_socket *rs, struct rtp_sub_socket *rss) goto out_free; DEBUGPC(DLMUX, "Read of RTP socket (%p) failed (errno %d, " "%s)\n", rs, errno, strerror(errno)); - rss->bfd.when &= ~BSC_FD_READ; + rss->bfd.when &= ~OSMO_FD_READ; goto out_free; } @@ -430,7 +430,7 @@ static int rtp_socket_read(struct rtp_socket *rs, struct rtp_sub_socket *rss) goto out_free; } msgb_enqueue(&other_rss->tx_queue, msg); - other_rss->bfd.when |= BSC_FD_WRITE; + other_rss->bfd.when |= OSMO_FD_WRITE; break; case RTP_RECV_UPSTREAM: @@ -448,7 +448,7 @@ static int rtp_socket_read(struct rtp_socket *rs, struct rtp_sub_socket *rss) if (rc < 0) goto out_free; msgb_enqueue(&rss->tx_queue, msg); - rss->bfd.when |= BSC_FD_WRITE; + rss->bfd.when |= OSMO_FD_WRITE; break; } if (rss->bfd.priv_nr != RTP_PRIV_RTP) { @@ -482,7 +482,7 @@ static int rtp_socket_write(struct rtp_socket *rs, struct rtp_sub_socket *rss) msg = msgb_dequeue(&rss->tx_queue); if (!msg) { - rss->bfd.when &= ~BSC_FD_WRITE; + rss->bfd.when &= ~OSMO_FD_WRITE; return 0; } @@ -516,10 +516,10 @@ static int rtp_bfd_cb(struct osmo_fd *bfd, unsigned int flags) return -EINVAL; } - if (flags & BSC_FD_READ) + if (flags & OSMO_FD_READ) rtp_socket_read(rs, rss); - if (flags & BSC_FD_WRITE) + if (flags & OSMO_FD_WRITE) rtp_socket_write(rs, rss); return 0; @@ -600,7 +600,7 @@ static int rtp_sub_socket_bind(struct rtp_sub_socket *rss, uint32_t ip, rss->sin_local.sin_family = AF_INET; rss->sin_local.sin_addr.s_addr = htonl(ip); rss->sin_local.sin_port = htons(port); - rss->bfd.when |= BSC_FD_READ; + rss->bfd.when |= OSMO_FD_READ; rc = bind(rss->bfd.fd, (struct sockaddr *)&rss->sin_local, sizeof(rss->sin_local)); diff --git a/openbsc/src/osmo-bsc/osmo_bsc_msc.c b/openbsc/src/osmo-bsc/osmo_bsc_msc.c index badecc682..a4b0d0f01 100644 --- a/openbsc/src/osmo-bsc/osmo_bsc_msc.c +++ b/openbsc/src/osmo-bsc/osmo_bsc_msc.c @@ -153,7 +153,7 @@ static int mgcp_create_port(struct bsc_msc_data *data) } osmo_wqueue_init(&data->mgcp_agent, 10); - data->mgcp_agent.bfd.when = BSC_FD_READ; + data->mgcp_agent.bfd.when = OSMO_FD_READ; data->mgcp_agent.bfd.data = data; data->mgcp_agent.read_cb = mgcp_do_read; data->mgcp_agent.write_cb = mgcp_do_write; diff --git a/openbsc/src/osmo-bsc_mgcp/mgcp_main.c b/openbsc/src/osmo-bsc_mgcp/mgcp_main.c index 419f852d7..8a71f53f5 100644 --- a/openbsc/src/osmo-bsc_mgcp/mgcp_main.c +++ b/openbsc/src/osmo-bsc_mgcp/mgcp_main.c @@ -222,8 +222,8 @@ int main(int argc, char **argv) vty_info.copyright = openbsc_copyright; vty_init(&vty_info); - logging_vty_add_cmds(NULL); - osmo_stats_vty_add_cmds(&log_info); + logging_vty_add_cmds(); + osmo_stats_vty_add_cmds(); mgcp_vty_init(); handle_options(argc, argv); @@ -246,7 +246,7 @@ int main(int argc, char **argv) /* we need to bind a socket */ if (rc == 0) { - cfg->gw_fd.bfd.when = BSC_FD_READ; + cfg->gw_fd.bfd.when = OSMO_FD_READ; cfg->gw_fd.bfd.cb = read_call_agent; cfg->gw_fd.bfd.fd = socket(AF_INET, SOCK_DGRAM, 0); if (cfg->gw_fd.bfd.fd < 0) { diff --git a/openbsc/src/osmo-bsc_nat/bsc_mgcp_utils.c b/openbsc/src/osmo-bsc_nat/bsc_mgcp_utils.c index 17dc659bb..069c9646b 100644 --- a/openbsc/src/osmo-bsc_nat/bsc_mgcp_utils.c +++ b/openbsc/src/osmo-bsc_nat/bsc_mgcp_utils.c @@ -663,19 +663,38 @@ static void free_chan_downstream(struct mgcp_endpoint *endp, struct bsc_endpoint mgcp_release_endp(endp); } +/*! Check MGCP parameter line (string) for plausibility. + * \param[in] endp pointer to endpoint (only used for log output) + * \param[in] line single parameter line from the MGCP message + * \returns 1 when line seems plausible, 0 on error */ +static int bsc_mgcp_check_param(const struct mgcp_endpoint *endp, const char *line) +{ + const size_t line_len = strlen(line); + if (line[0] != '\0' && line_len < 2) { + LOGP(DLMGCP, LOGL_ERROR, + "Wrong MGCP option format: '%s' on 0x%x\n", + line, ENDPOINT_NUMBER(endp)); + return 0; + } + + /* FIXME: A couple more checks wouldn't hurt... */ + + return 1; +} + static void bsc_mgcp_osmux_confirm(struct mgcp_endpoint *endp, const char *str) { unsigned int osmux_cid; - char *res; + const char x_osmux_prefix[] = "X-Osmux: "; + const size_t x_osmux_prefix_len = strlen(x_osmux_prefix); - res = strstr(str, "X-Osmux: "); - if (!res) { + if (!str || strncasecmp(str, x_osmux_prefix, x_osmux_prefix_len)) { LOGP(DMGCP, LOGL_INFO, - "BSC doesn't want to use Osmux, failing back to RTP\n"); + "BSC doesn't want to use Osmux, failing back to RTP (str=%s)\n", str); goto err; } - if (sscanf(res, "X-Osmux: %u", &osmux_cid) != 1) { + if (sscanf(str + x_osmux_prefix_len, "%u", &osmux_cid) != 1) { LOGP(DMGCP, LOGL_ERROR, "Failed to parse Osmux CID '%s'\n", str); goto err; @@ -713,6 +732,10 @@ void bsc_mgcp_forward(struct bsc_connection *bsc, struct msgb *msg) struct mgcp_endpoint *endp = NULL; int i, code; char transaction_id[60]; + char *save; + char *line = NULL; + char *line_osmux = NULL; + char *line_ci = NULL; /* Some assumption that our buffer is big enough.. and null terminate */ if (msgb_l2len(msg) > 2000) { @@ -752,14 +775,42 @@ void bsc_mgcp_forward(struct bsc_connection *bsc, struct msgb *msg) return; } - endp->ci = bsc_mgcp_extract_ci((const char *) msg->l2h); + save = alloca(msgb_l2len(msg) + 1); /* +1 -> null char appended */ + memcpy(save, msg->l2h, msgb_l2len(msg) + 1); + for_each_line(line, save) { + + if (!bsc_mgcp_check_param(endp, line)) + continue; + + switch (line[0]) { + case 'I': + case 'i': + line_ci = line; + break; + case 'X': + case 'x': + if (strncasecmp(line, "X-Osmux: ", strlen("X-Osmux: ")) == 0) + line_osmux = line; + /* Ignore unknown X-headers */ + break; + case '\0': + /* Reached SDP section, we are done parsing header */ + goto mgcp_header_done; + break; + default: + break; + } + } +mgcp_header_done: + + endp->ci = bsc_mgcp_extract_ci((const char *) line_ci); if (endp->ci == CI_UNUSED) { free_chan_downstream(endp, bsc_endp, bsc); return; } if (endp->osmux.state == OSMUX_STATE_NEGOTIATING) - bsc_mgcp_osmux_confirm(endp, (const char *) msg->l2h); + bsc_mgcp_osmux_confirm(endp, line_osmux); /* If we require osmux and it is disabled.. fail */ if (nat_osmux_only(bsc->nat->mgcp_cfg, bsc->cfg) && @@ -806,14 +857,16 @@ int bsc_mgcp_parse_response(const char *str, int *code, char transaction[60]) uint32_t bsc_mgcp_extract_ci(const char *str) { unsigned int ci; - char *res = strstr(str, "I: "); - if (!res) { - LOGP(DMGCP, LOGL_ERROR, "No CI in msg '%s'\n", str); + const char ci_prefix[] = "I: "; + const size_t ci_prefix_len = strlen(ci_prefix); + + if (!str || strncasecmp(str, ci_prefix, ci_prefix_len)) { + LOGP(DMGCP, LOGL_ERROR, "No CI in line '%s'\n", str); return CI_UNUSED; } - if (sscanf(res, "I: %x", &ci) != 1) { - LOGP(DMGCP, LOGL_ERROR, "Failed to parse CI in msg '%s'\n", str); + if (sscanf(str + ci_prefix_len, "%x", &ci) != 1) { + LOGP(DMGCP, LOGL_ERROR, "Failed to parse CI in line '%s'\n", str); return CI_UNUSED; } @@ -1074,7 +1127,7 @@ static int init_mgcp_socket(struct bsc_nat *nat, struct mgcp_config *cfg) } osmo_wqueue_init(&cfg->gw_fd, 10); - cfg->gw_fd.bfd.when = BSC_FD_READ; + cfg->gw_fd.bfd.when = OSMO_FD_READ; cfg->gw_fd.bfd.data = nat; cfg->gw_fd.read_cb = mgcp_do_read; cfg->gw_fd.write_cb = mgcp_do_write; diff --git a/openbsc/src/osmo-bsc_nat/bsc_nat.c b/openbsc/src/osmo-bsc_nat/bsc_nat.c index 2e2eac04d..90af4d780 100644 --- a/openbsc/src/osmo-bsc_nat/bsc_nat.c +++ b/openbsc/src/osmo-bsc_nat/bsc_nat.c @@ -1381,7 +1381,7 @@ static int ipaccess_listen_bsc_cb(struct osmo_fd *bfd, unsigned int what) struct sockaddr_in sa; socklen_t sa_len = sizeof(sa); - if (!(what & BSC_FD_READ)) + if (!(what & OSMO_FD_READ)) return 0; fd = accept(bfd->fd, (struct sockaddr *) &sa, &sa_len); @@ -1431,7 +1431,7 @@ static int ipaccess_listen_bsc_cb(struct osmo_fd *bfd, unsigned int what) bsc->write_queue.bfd.fd = fd; bsc->write_queue.read_cb = ipaccess_bsc_read_cb; bsc->write_queue.write_cb = bsc_write_cb; - bsc->write_queue.bfd.when = BSC_FD_READ; + bsc->write_queue.bfd.when = OSMO_FD_READ; if (osmo_fd_register(&bsc->write_queue.bfd) < 0) { LOGP(DNAT, LOGL_ERROR, "Failed to register BSC fd.\n"); close(fd); @@ -1535,12 +1535,20 @@ static void handle_options(int argc, char **argv) } } -static void signal_handler(int signal) +static void signal_handler(int signum) { - switch (signal) { + switch (signum) { case SIGABRT: - /* in case of abort, we want to obtain a talloc report - * and then return to the caller, who will abort the process */ + /* in case of abort, we want to obtain a talloc report and + * then run default SIGABRT handler, who will generate coredump + * and abort the process. abort() should do this for us after we + * return, but program wouldn't exit if an external SIGABRT is + * received. + */ + talloc_report_full(tall_bsc_ctx, stderr); + signal(SIGABRT, SIG_DFL); + raise(SIGABRT); + break; case SIGUSR1: talloc_report_full(tall_bsc_ctx, stderr); break; @@ -1628,8 +1636,8 @@ int main(int argc, char **argv) vty_info.copyright = openbsc_copyright; vty_init(&vty_info); - logging_vty_add_cmds(NULL); - osmo_stats_vty_add_cmds(&log_info); + logging_vty_add_cmds(); + osmo_stats_vty_add_cmds(); bsc_nat_vty_init(nat); ctrl_vty_init(tall_bsc_ctx); diff --git a/openbsc/src/osmo-bsc_nat/bsc_nat_utils.c b/openbsc/src/osmo-bsc_nat/bsc_nat_utils.c index 223ef349e..c4fe0d399 100644 --- a/openbsc/src/osmo-bsc_nat/bsc_nat_utils.c +++ b/openbsc/src/osmo-bsc_nat/bsc_nat_utils.c @@ -46,19 +46,19 @@ #include <unistd.h> static const struct rate_ctr_desc bsc_cfg_ctr_description[] = { - [BCFG_CTR_SCCP_CONN] = { "sccp:conn", "SCCP Connections "}, - [BCFG_CTR_SCCP_CALLS] = { "sccp:calls", "SCCP Assignment Commands "}, - [BCFG_CTR_NET_RECONN] = { "net:reconnects", "Network reconnects "}, - [BCFG_CTR_DROPPED_SCCP] = { "dropped:sccp", "Dropped SCCP connections."}, - [BCFG_CTR_DROPPED_CALLS] = { "dropped:calls", "Dropped active calls. "}, - [BCFG_CTR_REJECTED_CR] = { "rejected:cr", "Rejected CR due filter "}, - [BCFG_CTR_REJECTED_MSG] = { "rejected:msg", "Rejected MSG due filter "}, - [BCFG_CTR_ILL_PACKET] = { "rejected:ill", "Rejected due parse error "}, - [BCFG_CTR_CON_TYPE_LU] = { "conn:lu", "Conn Location Update "}, - [BCFG_CTR_CON_CMSERV_RQ] = { "conn:rq", "Conn CM Service Req "}, - [BCFG_CTR_CON_PAG_RESP] = { "conn:pag", "Conn Paging Response "}, - [BCFG_CTR_CON_SSA] = { "conn:ssa", "Conn USSD "}, - [BCFG_CTR_CON_OTHER] = { "conn:other", "Conn Other "}, + [BCFG_CTR_SCCP_CONN] = { "sccp:conn", "SCCP Connections" }, + [BCFG_CTR_SCCP_CALLS] = { "sccp:calls", "SCCP Assignment Commands"}, + [BCFG_CTR_NET_RECONN] = { "net:reconnects", "Network reconnects" }, + [BCFG_CTR_DROPPED_SCCP] = { "dropped:sccp", "Dropped SCCP connections"}, + [BCFG_CTR_DROPPED_CALLS] = { "dropped:calls", "Dropped active calls." }, + [BCFG_CTR_REJECTED_CR] = { "rejected:cr", "Rejected CR due filter" }, + [BCFG_CTR_REJECTED_MSG] = { "rejected:msg", "Rejected MSG due filter" }, + [BCFG_CTR_ILL_PACKET] = { "rejected:ill", "Rejected due parse error"}, + [BCFG_CTR_CON_TYPE_LU] = { "conn:lu", "Conn Location Update" }, + [BCFG_CTR_CON_CMSERV_RQ] = { "conn:rq", "Conn CM Service Req" }, + [BCFG_CTR_CON_PAG_RESP] = { "conn:pag", "Conn Paging Response" }, + [BCFG_CTR_CON_SSA] = { "conn:ssa", "Conn USSD" }, + [BCFG_CTR_CON_OTHER] = { "conn:other", "Conn Other" }, }; static const struct rate_ctr_group_desc bsc_cfg_ctrg_desc = { diff --git a/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c b/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c index bc1050dc9..a90dc154b 100644 --- a/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c +++ b/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c @@ -1429,11 +1429,3 @@ int bsc_nat_vty_init(struct bsc_nat *nat) return 0; } - - -/* called by the telnet interface... we have our own init above */ -int bsc_vty_init(struct gsm_network *network) -{ - logging_vty_add_cmds(NULL); - return 0; -} diff --git a/openbsc/src/osmo-bsc_nat/bsc_ussd.c b/openbsc/src/osmo-bsc_nat/bsc_ussd.c index d44b1b2fe..471743ac9 100644 --- a/openbsc/src/osmo-bsc_nat/bsc_ussd.c +++ b/openbsc/src/osmo-bsc_nat/bsc_ussd.c @@ -237,7 +237,7 @@ static int ussd_listen_cb(struct osmo_fd *bfd, unsigned int what) socklen_t sa_len = sizeof(sa); int fd; - if (!(what & BSC_FD_READ)) + if (!(what & OSMO_FD_READ)) return 0; fd = accept(bfd->fd, (struct sockaddr *) &sa, &sa_len); @@ -259,7 +259,7 @@ static int ussd_listen_cb(struct osmo_fd *bfd, unsigned int what) osmo_wqueue_init(&conn->queue, 10); conn->queue.bfd.data = conn; conn->queue.bfd.fd = fd; - conn->queue.bfd.when = BSC_FD_READ; + conn->queue.bfd.when = OSMO_FD_READ; conn->queue.read_cb = ussd_read_cb; conn->queue.write_cb = bsc_write_cb; diff --git a/openbsc/src/osmo-nitb/bsc_hack.c b/openbsc/src/osmo-nitb/bsc_hack.c index 2aeaa8fef..9ce5e124a 100644 --- a/openbsc/src/osmo-nitb/bsc_hack.c +++ b/openbsc/src/osmo-nitb/bsc_hack.c @@ -65,6 +65,7 @@ extern const char *openbsc_copyright; static int daemonize = 0; static const char *mncc_sock_path = NULL; static int use_db_counter = 1; +static bool yes_i_really_want = false; /* timer to store statistics */ #define DB_SYNC_INTERVAL 60, 0 @@ -115,6 +116,7 @@ static void handle_options(int argc, char **argv) while (1) { int option_index = 0, c; static struct option long_options[] = { + {"yes-i-really-want-to-run-prehistoric-software", 0, 0, 0}, {"help", 0, 0, 'h'}, {"debug", 1, 0, 'd'}, {"daemonize", 0, 0, 'D'}, @@ -139,6 +141,10 @@ static void handle_options(int argc, char **argv) break; switch (c) { + case 0: + if (option_index == 0) + yes_i_really_want = true; + break; case 'h': print_usage(); print_help(); @@ -196,11 +202,11 @@ static void handle_options(int argc, char **argv) } extern void *tall_vty_ctx; -static void signal_handler(int signal) +static void signal_handler(int signum) { - fprintf(stdout, "signal %u received\n", signal); + fprintf(stdout, "signal %u received\n", signum); - switch (signal) { + switch (signum) { case SIGINT: case SIGTERM: bsc_shutdown_net(bsc_gsmnet); @@ -210,8 +216,17 @@ static void signal_handler(int signal) break; case SIGABRT: osmo_generate_backtrace(); - /* in case of abort, we want to obtain a talloc report - * and then return to the caller, who will abort the process */ + /* in case of abort, we want to obtain a talloc report and + * then run default SIGABRT handler, who will generate coredump + * and abort the process. abort() should do this for us after we + * return, but program wouldn't exit if an external SIGABRT is + * received. + */ + talloc_report(tall_vty_ctx, stderr); + talloc_report_full(tall_bsc_ctx, stderr); + signal(SIGABRT, SIG_DFL); + raise(SIGABRT); + break; case SIGUSR1: talloc_report(tall_vty_ctx, stderr); talloc_report_full(tall_bsc_ctx, stderr); @@ -273,6 +288,28 @@ int main(int argc, char **argv) /* Parse options */ handle_options(argc, argv); +#define DIVIDER "======================================================================\n" + if (!yes_i_really_want) { + fprintf(stderr, DIVIDER + "ERROR: You should not run osmo-nitb. It is ancient, unmaintained and\n" + "obsolete for several years. If you really want to ignore years of hard work on\n" + "the new post-NITB stack (osmo-bsc, osmo-mgw, osmo-msc, osmo-hlr, osmo-stp),\n" + "you may ues the '--yes-i-really-want-to-run-prehistoric-software' command line\n" + "argument. DON'T EVER BOTHER THE DEVELOPERS IF YOU ENCOUNTER PROBLEMS WITH IT!!!\n" + "Use https://osmocom.org/projects/cellular-infrastructure/wiki/Osmocom_Network_In_The_Box\n" + DIVIDER); + exit(2); + } else { + fprintf(stderr, DIVIDER + "WARNING: You should not run osmo-nitb except for archaeological purpose.\n" + "It is ancient, unmaintained and obsolete for several years. But you seem\n" + "to insist on using obsolete and known-broken software that ignores years \n" + "of hard development work. That's your choice, but you have been warned.\n" + "DON'T EVER BOTHER THE DEVELOPERS IF YOU ENCOUNTER PROBLEMS WITH IT!!!\n" + "Use https://osmocom.org/projects/cellular-infrastructure/wiki/Osmocom_Network_In_The_Box\n" + DIVIDER); + } + /* Allocate global gsm_network struct; choose socket/internal MNCC */ rc = bsc_network_alloc(mncc_sock_path? mncc_sock_from_cc : int_mncc_recv); diff --git a/openbsc/tests/bsc-nat/bsc_nat_test.c b/openbsc/tests/bsc-nat/bsc_nat_test.c index 4bee60d8c..f6cf8b081 100644 --- a/openbsc/tests/bsc-nat/bsc_nat_test.c +++ b/openbsc/tests/bsc-nat/bsc_nat_test.c @@ -675,7 +675,7 @@ static void test_mgcp_parse(void) abort(); } - ci = bsc_mgcp_extract_ci(crcx_resp); + ci = bsc_mgcp_extract_ci(strstr(crcx_resp, "I: ")); if (ci != 0x0F) { printf("Failed to parse the CI. Got: %d\n", ci); abort(); diff --git a/openbsc/tests/ctrl_test_runner.py b/openbsc/tests/ctrl_test_runner.py index ad0a16b2e..7724f77fd 100755 --- a/openbsc/tests/ctrl_test_runner.py +++ b/openbsc/tests/ctrl_test_runner.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python2 +#!/usr/bin/env python3 # (C) 2013 by Jacob Erlbeck <jerlbeck@sysmocom.de> # (C) 2014 by Holger Hans Peter Freyther @@ -57,8 +57,8 @@ class TestCtrlBase(unittest.TestCase): try: self.proc = osmoutil.popen_devnull(osmo_ctrl_cmd) except OSError: - print >> sys.stderr, "Current directory: %s" % os.getcwd() - print >> sys.stderr, "Consider setting -b" + print("Current directory: %s" % os.getcwd(), file=sys.stderr) + print("Consider setting -b", file=sys.stderr) time.sleep(2) appstring = self.ctrl_app()[2] @@ -76,7 +76,7 @@ class TestCtrlBase(unittest.TestCase): def connect(self, host, port): if verbose: - print "Connecting to host %s:%i" % (host, port) + print("Connecting to host %s:%i" % (host, port)) retries = 30 while True: @@ -96,7 +96,7 @@ class TestCtrlBase(unittest.TestCase): def send(self, data): if verbose: - print "Sending \"%s\"" %(data) + print("Sending \"%s\"" %(data)) data = Ctrl().add_header(data) return self.sock.send(data) == len(data) @@ -125,9 +125,9 @@ class TestCtrlBase(unittest.TestCase): data = self.sock.recv(4096) while (len(data)>0): (head, data) = IPA().split_combined(data) - answer = Ctrl().rem_header(head) + answer = Ctrl().rem_header(head).decode() if verbose: - print "Got message:", answer + print("Got message:", answer) (mtype, id, msg) = answer.split(None, 2) id = int(id) rsp = {'mtype': mtype, 'id': id} @@ -143,7 +143,7 @@ class TestCtrlBase(unittest.TestCase): responses[id] = rsp if verbose: - print "Decoded replies: ", responses + print("Decoded replies: ", responses) return responses @@ -163,80 +163,80 @@ class TestCtrlBSC(TestCtrlBase): def testCtrlErrs(self): r = self.do_get('invalid') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Command not found') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Command not found') r = self.do_set('rf_locked', '999') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Value failed verification.') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Value failed verification.') r = self.do_get('bts') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Error while parsing the index.') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Error while parsing the index.') r = self.do_get('bts.999') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Error while resolving object') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Error while resolving object') def testBtsLac(self): r = self.do_get('bts.0.location-area-code') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.location-area-code') - self.assertEquals(r['value'], '1') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.location-area-code') + self.assertEqual(r['value'], '1') r = self.do_set('bts.0.location-area-code', '23') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'bts.0.location-area-code') - self.assertEquals(r['value'], '23') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'bts.0.location-area-code') + self.assertEqual(r['value'], '23') r = self.do_get('bts.0.location-area-code') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.location-area-code') - self.assertEquals(r['value'], '23') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.location-area-code') + self.assertEqual(r['value'], '23') r = self.do_set('bts.0.location-area-code', '-1') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Input not within the range') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Input not within the range') def testBtsCi(self): r = self.do_get('bts.0.cell-identity') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.cell-identity') - self.assertEquals(r['value'], '0') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.cell-identity') + self.assertEqual(r['value'], '0') r = self.do_set('bts.0.cell-identity', '23') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'bts.0.cell-identity') - self.assertEquals(r['value'], '23') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'bts.0.cell-identity') + self.assertEqual(r['value'], '23') r = self.do_get('bts.0.cell-identity') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.cell-identity') - self.assertEquals(r['value'], '23') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.cell-identity') + self.assertEqual(r['value'], '23') r = self.do_set('bts.0.cell-identity', '-1') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Input not within the range') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Input not within the range') def testBtsGenerateSystemInformation(self): r = self.do_get('bts.0.send-new-system-informations') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Write Only attribute') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Write Only attribute') # No RSL links so it will fail r = self.do_set('bts.0.send-new-system-informations', '1') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Failed to generate SI') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Failed to generate SI') def testBtsChannelLoad(self): r = self.do_set('bts.0.channel-load', '1') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Read Only attribute') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Read Only attribute') # No RSL link so everything is 0 r = self.do_get('bts.0.channel-load') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['value'], + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['value'], 'CCCH+SDCCH4,0,0 TCH/F,0,0 TCH/H,0,0 SDCCH8,0,0' + ' TCH/F_PDCH,0,0 CCCH+SDCCH4+CBCH,0,0' + ' SDCCH8+CBCH,0,0 TCH/F_TCH/H_PDCH,0,0') @@ -244,221 +244,221 @@ class TestCtrlBSC(TestCtrlBase): def testBtsOmlConnectionState(self): """Check OML state. It will not be connected""" r = self.do_set('bts.0.oml-connection-state', '1') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Read Only attribute') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Read Only attribute') # No RSL link so everything is 0 r = self.do_get('bts.0.oml-connection-state') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['value'], 'disconnected') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['value'], 'disconnected') def testTrxPowerRed(self): r = self.do_get('bts.0.trx.0.max-power-reduction') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.trx.0.max-power-reduction') - self.assertEquals(r['value'], '20') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.trx.0.max-power-reduction') + self.assertEqual(r['value'], '20') r = self.do_set('bts.0.trx.0.max-power-reduction', '22') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'bts.0.trx.0.max-power-reduction') - self.assertEquals(r['value'], '22') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'bts.0.trx.0.max-power-reduction') + self.assertEqual(r['value'], '22') r = self.do_get('bts.0.trx.0.max-power-reduction') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.trx.0.max-power-reduction') - self.assertEquals(r['value'], '22') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.trx.0.max-power-reduction') + self.assertEqual(r['value'], '22') r = self.do_set('bts.0.trx.0.max-power-reduction', '1') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Value must be even') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Value must be even') def testTrxArfcn(self): r = self.do_get('bts.0.trx.0.arfcn') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.trx.0.arfcn') - self.assertEquals(r['value'], '871') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.trx.0.arfcn') + self.assertEqual(r['value'], '871') r = self.do_set('bts.0.trx.0.arfcn', '873') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'bts.0.trx.0.arfcn') - self.assertEquals(r['value'], '873') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'bts.0.trx.0.arfcn') + self.assertEqual(r['value'], '873') r = self.do_get('bts.0.trx.0.arfcn') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.trx.0.arfcn') - self.assertEquals(r['value'], '873') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.trx.0.arfcn') + self.assertEqual(r['value'], '873') r = self.do_set('bts.0.trx.0.arfcn', '2000') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Input not within the range') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Input not within the range') def testRfLock(self): r = self.do_get('bts.0.rf_state') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.rf_state') - self.assertEquals(r['value'], 'inoperational,unlocked,on') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.rf_state') + self.assertEqual(r['value'], 'inoperational,unlocked,on') r = self.do_set('rf_locked', '1') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'rf_locked') - self.assertEquals(r['value'], '1') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'rf_locked') + self.assertEqual(r['value'], '1') time.sleep(1.5) r = self.do_get('bts.0.rf_state') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.rf_state') - self.assertEquals(r['value'], 'inoperational,locked,off') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.rf_state') + self.assertEqual(r['value'], 'inoperational,locked,off') r = self.do_get('rf_locked') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'rf_locked') - self.assertEquals(r['value'], 'state=off,policy=off') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'rf_locked') + self.assertEqual(r['value'], 'state=off,policy=off') r = self.do_set('rf_locked', '0') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'rf_locked') - self.assertEquals(r['value'], '0') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'rf_locked') + self.assertEqual(r['value'], '0') time.sleep(1.5) r = self.do_get('bts.0.rf_state') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.rf_state') - self.assertEquals(r['value'], 'inoperational,unlocked,on') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.rf_state') + self.assertEqual(r['value'], 'inoperational,unlocked,on') r = self.do_get('rf_locked') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'rf_locked') - self.assertEquals(r['value'], 'state=off,policy=on') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'rf_locked') + self.assertEqual(r['value'], 'state=off,policy=on') def testTimezone(self): r = self.do_get('timezone') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'timezone') - self.assertEquals(r['value'], 'off') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'timezone') + self.assertEqual(r['value'], 'off') r = self.do_set('timezone', '-2,15,2') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'timezone') - self.assertEquals(r['value'], '-2,15,2') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'timezone') + self.assertEqual(r['value'], '-2,15,2') r = self.do_get('timezone') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'timezone') - self.assertEquals(r['value'], '-2,15,2') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'timezone') + self.assertEqual(r['value'], '-2,15,2') # Test invalid input r = self.do_set('timezone', '-2,15,2,5,6,7') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'timezone') - self.assertEquals(r['value'], '-2,15,2') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'timezone') + self.assertEqual(r['value'], '-2,15,2') r = self.do_set('timezone', '-2,15') - self.assertEquals(r['mtype'], 'ERROR') + self.assertEqual(r['mtype'], 'ERROR') r = self.do_set('timezone', '-2') - self.assertEquals(r['mtype'], 'ERROR') + self.assertEqual(r['mtype'], 'ERROR') r = self.do_set('timezone', '1') r = self.do_set('timezone', 'off') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'timezone') - self.assertEquals(r['value'], 'off') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'timezone') + self.assertEqual(r['value'], 'off') r = self.do_get('timezone') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'timezone') - self.assertEquals(r['value'], 'off') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'timezone') + self.assertEqual(r['value'], 'off') def testMcc(self): r = self.do_set('mcc', '23') r = self.do_get('mcc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mcc') - self.assertEquals(r['value'], '023') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mcc') + self.assertEqual(r['value'], '023') r = self.do_set('mcc', '023') r = self.do_get('mcc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mcc') - self.assertEquals(r['value'], '023') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mcc') + self.assertEqual(r['value'], '023') def testMnc(self): r = self.do_set('mnc', '9') r = self.do_get('mnc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mnc') - self.assertEquals(r['value'], '09') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mnc') + self.assertEqual(r['value'], '09') r = self.do_set('mnc', '09') r = self.do_get('mnc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mnc') - self.assertEquals(r['value'], '09') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mnc') + self.assertEqual(r['value'], '09') r = self.do_set('mnc', '009') r = self.do_get('mnc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mnc') - self.assertEquals(r['value'], '009') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mnc') + self.assertEqual(r['value'], '009') def testMccMncApply(self): # Test some invalid input r = self.do_set('mcc-mnc-apply', 'WRONG') - self.assertEquals(r['mtype'], 'ERROR') + self.assertEqual(r['mtype'], 'ERROR') r = self.do_set('mcc-mnc-apply', '1,') - self.assertEquals(r['mtype'], 'ERROR') + self.assertEqual(r['mtype'], 'ERROR') r = self.do_set('mcc-mnc-apply', '200,3') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'mcc-mnc-apply') - self.assertEquals(r['value'], 'Tried to drop the BTS') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'mcc-mnc-apply') + self.assertEqual(r['value'], 'Tried to drop the BTS') # Set it again r = self.do_set('mcc-mnc-apply', '200,3') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'mcc-mnc-apply') - self.assertEquals(r['value'], 'Nothing changed') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'mcc-mnc-apply') + self.assertEqual(r['value'], 'Nothing changed') # Change it r = self.do_set('mcc-mnc-apply', '200,4') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'mcc-mnc-apply') - self.assertEquals(r['value'], 'Tried to drop the BTS') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'mcc-mnc-apply') + self.assertEqual(r['value'], 'Tried to drop the BTS') # Change it r = self.do_set('mcc-mnc-apply', '201,4') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'mcc-mnc-apply') - self.assertEquals(r['value'], 'Tried to drop the BTS') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'mcc-mnc-apply') + self.assertEqual(r['value'], 'Tried to drop the BTS') # Verify r = self.do_get('mnc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mnc') - self.assertEquals(r['value'], '04') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mnc') + self.assertEqual(r['value'], '04') r = self.do_get('mcc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mcc') - self.assertEquals(r['value'], '201') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mcc') + self.assertEqual(r['value'], '201') # Change it r = self.do_set('mcc-mnc-apply', '202,03') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'mcc-mnc-apply') - self.assertEquals(r['value'], 'Tried to drop the BTS') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'mcc-mnc-apply') + self.assertEqual(r['value'], 'Tried to drop the BTS') r = self.do_get('mnc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mnc') - self.assertEquals(r['value'], '03') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mnc') + self.assertEqual(r['value'], '03') r = self.do_get('mcc') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'mcc') - self.assertEquals(r['value'], '202') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'mcc') + self.assertEqual(r['value'], '202') class TestCtrlNITB(TestCtrlBase): @@ -468,110 +468,111 @@ class TestCtrlNITB(TestCtrlBase): def ctrl_command(self): return ["./src/osmo-nitb/osmo-nitb", "-c", - "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg", "-l", "test_hlr.sqlite3"] + "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg", "-l", "test_hlr.sqlite3", + "--yes-i-really-want-to-run-prehistoric-software"] def ctrl_app(self): return (4249, "./src/osmo-nitb/osmo-nitb", "OsmoBSC", "nitb") def testNumberOfBTS(self): r = self.do_get('number-of-bts') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'number-of-bts') - self.assertEquals(r['value'], '1') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'number-of-bts') + self.assertEqual(r['value'], '1') def testSubscriberAddWithKi(self): """Test that we can set the algorithm to none, xor, comp128v1""" r = self.do_set('subscriber-modify-v1', '2620345,445566') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') r = self.do_set('subscriber-modify-v1', '2620345,445566,none') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') r = self.do_set('subscriber-modify-v1', '2620345,445566,xor') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Value failed verification.') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Value failed verification.') r = self.do_set('subscriber-modify-v1', '2620345,445566,comp128v1,00112233445566778899AABBCCDDEEFF') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') r = self.do_set('subscriber-modify-v1', '2620345,445566,comp128v2,00112233445566778899AABBCCDDEEFF') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') r = self.do_set('subscriber-modify-v1', '2620345,445566,comp128v3,00112233445566778899AABBCCDDEEFF') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') r = self.do_set('subscriber-modify-v1', '2620345,445566,none') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') def testSubscriberAddRemove(self): r = self.do_set('subscriber-modify-v1', '2620345,445566') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') r = self.do_set('subscriber-modify-v1', '2620345,445567') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'subscriber-modify-v1') - self.assertEquals(r['value'], 'OK') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'subscriber-modify-v1') + self.assertEqual(r['value'], 'OK') # TODO. verify that the entry has been created and modified? Invoke # the sqlite3 CLI or do it through the DB libraries? r = self.do_set('subscriber-delete-v1', '2620345') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['value'], 'Removed') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['value'], 'Removed') r = self.do_set('subscriber-delete-v1', '2620345') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Failed to find subscriber') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Failed to find subscriber') def testSubscriberList(self): # TODO. Add command to mark a subscriber as active r = self.do_get('subscriber-list-active-v1') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'subscriber-list-active-v1') - self.assertEquals(r['value'], None) + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'subscriber-list-active-v1') + self.assertEqual(r['value'], None) def testApplyConfiguration(self): r = self.do_get('bts.0.apply-configuration') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Write Only attribute') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Write Only attribute') r = self.do_set('bts.0.apply-configuration', '1') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['value'], 'Tried to drop the BTS') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['value'], 'Tried to drop the BTS') def testGprsMode(self): r = self.do_get('bts.0.gprs-mode') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.gprs-mode') - self.assertEquals(r['value'], 'none') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.gprs-mode') + self.assertEqual(r['value'], 'none') r = self.do_set('bts.0.gprs-mode', 'bla') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Mode is not known') + self.assertEqual(r['mtype'], 'ERROR') + self.assertEqual(r['error'], 'Mode is not known') r = self.do_set('bts.0.gprs-mode', 'egprs') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['value'], 'egprs') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['value'], 'egprs') r = self.do_get('bts.0.gprs-mode') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'bts.0.gprs-mode') - self.assertEquals(r['value'], 'egprs') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'bts.0.gprs-mode') + self.assertEqual(r['value'], 'egprs') class TestCtrlNAT(TestCtrlBase): @@ -584,38 +585,38 @@ class TestCtrlNAT(TestCtrlBase): def testAccessList(self): r = self.do_get('net.0.bsc_cfg.0.access-list-name') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'net') - self.assertEquals(r['value'], None) + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'net') + self.assertEqual(r['value'], None) r = self.do_set('net.0.bsc_cfg.0.access-list-name', 'bla') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'net') - self.assertEquals(r['value'], 'bla') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'net') + self.assertEqual(r['value'], 'bla') r = self.do_get('net.0.bsc_cfg.0.access-list-name') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'net') - self.assertEquals(r['value'], 'bla') + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'net') + self.assertEqual(r['value'], 'bla') r = self.do_set('net.0.bsc_cfg.0.no-access-list-name', '1') - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'net') - self.assertEquals(r['value'], None) + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'net') + self.assertEqual(r['value'], None) r = self.do_get('net.0.bsc_cfg.0.access-list-name') - self.assertEquals(r['mtype'], 'GET_REPLY') - self.assertEquals(r['var'], 'net') - self.assertEquals(r['value'], None) + self.assertEqual(r['mtype'], 'GET_REPLY') + self.assertEqual(r['var'], 'net') + self.assertEqual(r['value'], None) def testAccessListManagement(self): r = self.do_set("net.0.add.allow.access-list.404", "abc") - self.assertEquals(r['mtype'], 'ERROR') + self.assertEqual(r['mtype'], 'ERROR') r = self.do_set("net.0.add.allow.access-list.bla", "^234$") - self.assertEquals(r['mtype'], 'SET_REPLY') - self.assertEquals(r['var'], 'net.0.add.allow.access-list.bla') - self.assertEquals(r['value'], 'IMSI allow added to access list') + self.assertEqual(r['mtype'], 'SET_REPLY') + self.assertEqual(r['var'], 'net.0.add.allow.access-list.bla') + self.assertEqual(r['value'], 'IMSI allow added to access list') # TODO.. find a way to actually see if this rule has been # added. e.g. by implementing a get for the list. @@ -664,9 +665,9 @@ if __name__ == '__main__': if args.p: confpath = args.p - print "confpath %s, workdir %s" % (confpath, workdir) + print("confpath %s, workdir %s" % (confpath, workdir)) os.chdir(workdir) - print "Running tests for specific control commands" + print("Running tests for specific control commands") suite = unittest.TestSuite() add_bsc_test(suite, workdir) add_nitb_test(suite, workdir) diff --git a/openbsc/tests/db/db_test.c b/openbsc/tests/db/db_test.c index 516718522..ad417295a 100644 --- a/openbsc/tests/db/db_test.c +++ b/openbsc/tests/db/db_test.c @@ -205,7 +205,10 @@ int main() { printf("Testing subscriber database code.\n"); osmo_init_logging(&log_info); - log_set_print_filename(osmo_stderr_target, 0); + log_set_print_filename2(osmo_stderr_target, LOG_FILENAME_NONE); + log_set_use_color(osmo_stderr_target, 0); + log_set_print_category(osmo_stderr_target, 0); + log_set_print_category_hex(osmo_stderr_target, 0); dummy_net.subscr_group = &dummy_sgrp; dummy_sgrp.net = &dummy_net; diff --git a/openbsc/tests/db/db_test.err b/openbsc/tests/db/db_test.err index 27e570372..4be5b10c7 100644 --- a/openbsc/tests/db/db_test.err +++ b/openbsc/tests/db/db_test.err @@ -1,3 +1,2 @@ Going to migrate from revision 3 -[0;mGoing to migrate from revision 4 -[0;m
\ No newline at end of file +Going to migrate from revision 4 diff --git a/openbsc/tests/smpp/smpp_test.c b/openbsc/tests/smpp/smpp_test.c index 62fa9d2e9..99c268365 100644 --- a/openbsc/tests/smpp/smpp_test.c +++ b/openbsc/tests/smpp/smpp_test.c @@ -66,7 +66,9 @@ int main(int argc, char **argv) { osmo_init_logging(&log_info); log_set_use_color(osmo_stderr_target, 0); - log_set_print_filename(osmo_stderr_target, 0); + log_set_print_filename2(osmo_stderr_target, LOG_FILENAME_NONE); + log_set_print_category(osmo_stderr_target, 0); + log_set_print_category_hex(osmo_stderr_target, 0); test_coding_scheme(); return EXIT_SUCCESS; diff --git a/openbsc/tests/smpp_test_runner.py b/openbsc/tests/smpp_test_runner.py index eccfa6b86..b518ebbdd 100755 --- a/openbsc/tests/smpp_test_runner.py +++ b/openbsc/tests/smpp_test_runner.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python2 +#!/usr/bin/env python3 # (C) 2014 by Holger Hans Peter Freyther # based on vty_test_runner.py: @@ -46,8 +46,8 @@ class TestVTYBase(unittest.TestCase): try: self.proc = osmoutil.popen_devnull(osmo_vty_cmd) except OSError: - print >> sys.stderr, "Current directory: %s" % os.getcwd() - print >> sys.stderr, "Consider setting -b" + print("Current directory: %s" % os.getcwd(), file=sys.stderr) + print("Consider setting -b", file=sys.stderr) appstring = self.vty_app()[2] appport = self.vty_app()[0] @@ -64,7 +64,8 @@ class TestSMPPNITB(TestVTYBase): def vty_command(self): return ["./src/osmo-nitb/osmo-nitb", "-c", - "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg"] + "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg", + "--yes-i-really-want-to-run-prehistoric-software"] def vty_app(self): return (4242, "./src/osmo-nitb/osmo-nitb", "OpenBSC", "nitb") @@ -73,14 +74,14 @@ class TestSMPPNITB(TestVTYBase): # Enable the configuration self.vty.enable() self.assertTrue(self.vty.verify("configure terminal", [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify('smpp', [''])) - self.assertEquals(self.vty.node(), 'config-smpp') + self.assertEqual(self.vty.node(), 'config-smpp') self.assertTrue(self.vty.verify('system-id test', [''])) self.assertTrue(self.vty.verify('local-tcp-port 2775', [''])) self.assertTrue(self.vty.verify('esme test', [''])) - self.assertEquals(self.vty.node(), 'config-smpp-esme') + self.assertEqual(self.vty.node(), 'config-smpp-esme') self.assertTrue(self.vty.verify('default-route', [''])) self.assertTrue(self.vty.verify('end', [''])) @@ -88,7 +89,7 @@ class TestSMPPNITB(TestVTYBase): sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sck.setblocking(1) sck.connect(('0.0.0.0', 2775)) - sck.sendall('\x00\x00\x00\x02\x00') + sck.sendall(b'\x00\x00\x00\x02\x00') sck.close() # Check if the VTY is still there @@ -98,7 +99,7 @@ class TestSMPPNITB(TestVTYBase): sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sck.setblocking(1) sck.connect(('0.0.0.0', 2775)) - sck.sendall('\x00\x01\x00\x01\x01') + sck.sendall(b'\x00\x01\x00\x01\x01') sck.close() self.vty.verify('enable',['']) @@ -128,9 +129,9 @@ if __name__ == '__main__': if args.p: confpath = args.p - print "confpath %s, workdir %s" % (confpath, workdir) + print("confpath %s, workdir %s" % (confpath, workdir)) os.chdir(workdir) - print "Running tests for specific SMPP" + print("Running tests for specific SMPP") suite = unittest.TestSuite() suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestSMPPNITB)) res = unittest.TextTestRunner(verbosity=verbose_level).run(suite) diff --git a/openbsc/tests/subscr/bsc_subscr_test.c b/openbsc/tests/subscr/bsc_subscr_test.c index 60d687d58..cbe635e08 100644 --- a/openbsc/tests/subscr/bsc_subscr_test.c +++ b/openbsc/tests/subscr/bsc_subscr_test.c @@ -114,10 +114,11 @@ int main() { printf("Testing BSC subscriber core code.\n"); osmo_init_logging(&log_info); - log_set_print_filename(osmo_stderr_target, 0); + log_set_print_filename2(osmo_stderr_target, LOG_FILENAME_NONE); log_set_print_timestamp(osmo_stderr_target, 0); log_set_use_color(osmo_stderr_target, 0); log_set_print_category(osmo_stderr_target, 1); + log_set_print_category_hex(osmo_stderr_target, 0); log_set_category_filter(osmo_stderr_target, DREF, 1, LOGL_DEBUG); bsc_subscribers = talloc_zero(NULL, struct llist_head); diff --git a/openbsc/tests/subscr/subscr_test.c b/openbsc/tests/subscr/subscr_test.c index 2a5d0e1c2..947c2c3ef 100644 --- a/openbsc/tests/subscr/subscr_test.c +++ b/openbsc/tests/subscr/subscr_test.c @@ -105,7 +105,9 @@ int main() { printf("Testing subscriber core code.\n"); osmo_init_logging(&log_info); - log_set_print_filename(osmo_stderr_target, 0); + log_set_print_filename2(osmo_stderr_target, LOG_FILENAME_NONE); + log_set_print_category(osmo_stderr_target, 0); + log_set_print_category_hex(osmo_stderr_target, 0); dummy_net.subscr_group = &dummy_sgrp; dummy_sgrp.net = &dummy_net; diff --git a/openbsc/tests/vty_test_runner.py b/openbsc/tests/vty_test_runner.py index 67adb71fd..a1293be4b 100755 --- a/openbsc/tests/vty_test_runner.py +++ b/openbsc/tests/vty_test_runner.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python2 +#!/usr/bin/env python3 # (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com> # (C) 2013 by Holger Hans Peter Freyther @@ -51,8 +51,8 @@ class TestVTYBase(unittest.TestCase): try: self.proc = osmoutil.popen_devnull(osmo_vty_cmd) except OSError: - print >> sys.stderr, "Current directory: %s" % os.getcwd() - print >> sys.stderr, "Consider setting -b" + print("Current directory: %s" % os.getcwd(), file=sys.stderr) + print("Consider setting -b", file=sys.stderr) appstring = self.vty_app()[2] appport = self.vty_app()[0] @@ -75,28 +75,28 @@ class TestVTYMGCP(TestVTYBase): def testForcePtime(self): self.vty.enable() res = self.vty.command("show running-config") - self.assert_(res.find(' rtp force-ptime 20\r') > 0) - self.assertEquals(res.find(' no rtp force-ptime\r'), -1) + self.assertTrue(res.find(' rtp force-ptime 20\r') > 0) + self.assertEqual(res.find(' no rtp force-ptime\r'), -1) self.vty.command("configure terminal") self.vty.command("mgcp") self.vty.command("no rtp force-ptime") res = self.vty.command("show running-config") - self.assertEquals(res.find(' rtp force-ptime 20\r'), -1) - self.assertEquals(res.find(' no rtp force-ptime\r'), -1) + self.assertEqual(res.find(' rtp force-ptime 20\r'), -1) + self.assertEqual(res.find(' no rtp force-ptime\r'), -1) def testOmitAudio(self): self.vty.enable() res = self.vty.command("show running-config") - self.assert_(res.find(' sdp audio-payload send-name\r') > 0) - self.assertEquals(res.find(' no sdp audio-payload send-name\r'), -1) + self.assertTrue(res.find(' sdp audio-payload send-name\r') > 0) + self.assertEqual(res.find(' no sdp audio-payload send-name\r'), -1) self.vty.command("configure terminal") self.vty.command("mgcp") self.vty.command("no sdp audio-payload send-name") res = self.vty.command("show running-config") - self.assertEquals(res.find(' rtp sdp audio-payload send-name\r'), -1) - self.assert_(res.find(' no sdp audio-payload send-name\r') > 0) + self.assertEqual(res.find(' rtp sdp audio-payload send-name\r'), -1) + self.assertTrue(res.find(' no sdp audio-payload send-name\r') > 0) # TODO: test it for the trunk! @@ -109,18 +109,18 @@ class TestVTYMGCP(TestVTYBase): # enable.. disable bts-bind-ip self.vty.command("rtp bts-bind-ip 254.253.252.250") res = self.vty.command("show running-config") - self.assert_(res.find('rtp bts-bind-ip 254.253.252.250') > 0) + self.assertTrue(res.find('rtp bts-bind-ip 254.253.252.250') > 0) self.vty.command("no rtp bts-bind-ip") res = self.vty.command("show running-config") - self.assertEquals(res.find(' rtp bts-bind-ip'), -1) + self.assertEqual(res.find(' rtp bts-bind-ip'), -1) # enable.. disable net-bind-ip self.vty.command("rtp net-bind-ip 254.253.252.250") res = self.vty.command("show running-config") - self.assert_(res.find('rtp net-bind-ip 254.253.252.250') > 0) + self.assertTrue(res.find('rtp net-bind-ip 254.253.252.250') > 0) self.vty.command("no rtp net-bind-ip") res = self.vty.command("show running-config") - self.assertEquals(res.find(' rtp net-bind-ip'), -1) + self.assertEqual(res.find(' rtp net-bind-ip'), -1) class TestVTYGenericBSC(TestVTYBase): @@ -128,42 +128,42 @@ class TestVTYGenericBSC(TestVTYBase): def checkForEndAndExit(self): res = self.vty.command("list") #print ('looking for "exit"\n') - self.assert_(res.find(' exit\r') > 0) + self.assertTrue(res.find(' exit\r') > 0) #print 'found "exit"\nlooking for "end"\n' - self.assert_(res.find(' end\r') > 0) + self.assertTrue(res.find(' end\r') > 0) #print 'found "end"\n' def _testConfigNetworkTree(self): self.vty.enable() self.assertTrue(self.vty.verify("configure terminal",[''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.checkForEndAndExit() self.assertTrue(self.vty.verify("network",[''])) - self.assertEquals(self.vty.node(), 'config-net') + self.assertEqual(self.vty.node(), 'config-net') self.checkForEndAndExit() self.assertTrue(self.vty.verify("bts 0",[''])) - self.assertEquals(self.vty.node(), 'config-net-bts') + self.assertEqual(self.vty.node(), 'config-net-bts') self.checkForEndAndExit() self.assertTrue(self.vty.verify("trx 0",[''])) - self.assertEquals(self.vty.node(), 'config-net-bts-trx') + self.assertEqual(self.vty.node(), 'config-net-bts-trx') self.checkForEndAndExit() self.vty.command("write terminal") self.assertTrue(self.vty.verify("exit",[''])) - self.assertEquals(self.vty.node(), 'config-net-bts') + self.assertEqual(self.vty.node(), 'config-net-bts') self.assertTrue(self.vty.verify("exit",[''])) self.assertTrue(self.vty.verify("bts 1",[''])) - self.assertEquals(self.vty.node(), 'config-net-bts') + self.assertEqual(self.vty.node(), 'config-net-bts') self.checkForEndAndExit() self.assertTrue(self.vty.verify("trx 1",[''])) - self.assertEquals(self.vty.node(), 'config-net-bts-trx') + self.assertEqual(self.vty.node(), 'config-net-bts-trx') self.checkForEndAndExit() self.vty.command("write terminal") self.assertTrue(self.vty.verify("exit",[''])) - self.assertEquals(self.vty.node(), 'config-net-bts') + self.assertEqual(self.vty.node(), 'config-net-bts') self.assertTrue(self.vty.verify("exit",[''])) - self.assertEquals(self.vty.node(), 'config-net') + self.assertEqual(self.vty.node(), 'config-net') self.assertTrue(self.vty.verify("exit",[''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify("exit",[''])) self.assertTrue(self.vty.node() is None) @@ -171,7 +171,8 @@ class TestVTYNITB(TestVTYGenericBSC): def vty_command(self): return ["./src/osmo-nitb/osmo-nitb", "-c", - "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg"] + "doc/examples/osmo-nitb/nanobts/osmo-nitb.cfg", + "--yes-i-really-want-to-run-prehistoric-software"] def vty_app(self): return (4242, "./src/osmo-nitb/osmo-nitb", "OpenBSC", "nitb") @@ -196,35 +197,35 @@ class TestVTYNITB(TestVTYGenericBSC): # check the default res = self.vty.command("write terminal") - self.assert_(res.find(' no smpp-first') > 0) + self.assertTrue(res.find(' no smpp-first') > 0) self.vty.verify("smpp-first", ['']) res = self.vty.command("write terminal") - self.assert_(res.find(' smpp-first') > 0) - self.assertEquals(res.find('no smpp-first'), -1) + self.assertTrue(res.find(' smpp-first') > 0) + self.assertEqual(res.find('no smpp-first'), -1) self.vty.verify("no smpp-first", ['']) res = self.vty.command("write terminal") - self.assert_(res.find('no smpp-first') > 0) + self.assertTrue(res.find('no smpp-first') > 0) def testVtyTree(self): self.vty.enable() self.assertTrue(self.vty.verify("configure terminal", [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.checkForEndAndExit() self.assertTrue(self.vty.verify('mncc-int', [''])) - self.assertEquals(self.vty.node(), 'config-mncc-int') + self.assertEqual(self.vty.node(), 'config-mncc-int') self.checkForEndAndExit() self.assertTrue(self.vty.verify('exit', [''])) if self.checkForSmpp(): - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify('smpp', [''])) - self.assertEquals(self.vty.node(), 'config-smpp') + self.assertEqual(self.vty.node(), 'config-smpp') self.checkForEndAndExit() self.assertTrue(self.vty.verify("exit", [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify("exit", [''])) self.assertTrue(self.vty.node() is None) @@ -234,10 +235,10 @@ class TestVTYNITB(TestVTYGenericBSC): if self.checkForSmpp(): self.vty.command('smpp') - self.assertEquals(self.vty.node(), 'config-smpp') + self.assertEqual(self.vty.node(), 'config-smpp') self.vty.command('mncc-int') - self.assertEquals(self.vty.node(), 'config-mncc-int') + self.assertEqual(self.vty.node(), 'config-mncc-int') def testVtyAuthorization(self): self.vty.enable() @@ -267,7 +268,7 @@ class TestVTYNITB(TestVTYGenericBSC): self.vty.command("si2quater neighbor-list del earfcn 1911") self.vty.command("si2quater neighbor-list del earfcn 1924") self.vty.command("si2quater neighbor-list del earfcn 2111") - self.assertEquals(before, self.vty.command("show running-config")) + self.assertEqual(before, self.vty.command("show running-config")) self.vty.command("si2quater neighbor-list add uarfcn 1976 13 1") self.vty.command("si2quater neighbor-list add uarfcn 1976 38 1") self.vty.command("si2quater neighbor-list add uarfcn 1976 44 1") @@ -290,7 +291,7 @@ class TestVTYNITB(TestVTYGenericBSC): self.vty.command("si2quater neighbor-list del uarfcn 1976 224") self.vty.command("si2quater neighbor-list del uarfcn 1976 225") self.vty.command("si2quater neighbor-list del uarfcn 1976 226") - self.assertEquals(before, self.vty.command("show running-config")) + self.assertEqual(before, self.vty.command("show running-config")) def testEnableDisablePeriodicLU(self): self.vty.enable() @@ -306,14 +307,14 @@ class TestVTYNITB(TestVTYGenericBSC): # Enable periodic lu.. self.vty.verify("periodic location update 60", ['']) res = self.vty.command("write terminal") - self.assert_(res.find('periodic location update 60') > 0) - self.assertEquals(res.find('no periodic location update'), -1) + self.assertTrue(res.find('periodic location update 60') > 0) + self.assertEqual(res.find('no periodic location update'), -1) # Now disable it.. self.vty.verify("no periodic location update", ['']) res = self.vty.command("write terminal") - self.assertEquals(res.find('periodic location update 60'), -1) - self.assert_(res.find('no periodic location update') > 0) + self.assertEqual(res.find('periodic location update 60'), -1) + self.assertTrue(res.find('no periodic location update') > 0) def testEnableDisableSiHacks(self): self.vty.enable() @@ -324,14 +325,14 @@ class TestVTYNITB(TestVTYGenericBSC): # Enable periodic lu.. self.vty.verify("force-combined-si", ['']) res = self.vty.command("write terminal") - self.assert_(res.find(' force-combined-si') > 0) - self.assertEquals(res.find('no force-combined-si'), -1) + self.assertTrue(res.find(' force-combined-si') > 0) + self.assertEqual(res.find('no force-combined-si'), -1) # Now disable it.. self.vty.verify("no force-combined-si", ['']) res = self.vty.command("write terminal") - self.assertEquals(res.find(' force-combined-si'), -1) - self.assert_(res.find('no force-combined-si') > 0) + self.assertEqual(res.find(' force-combined-si'), -1) + self.assertTrue(res.find('no force-combined-si') > 0) def testRachAccessControlClass(self): self.vty.enable() @@ -355,7 +356,7 @@ class TestVTYNITB(TestVTYGenericBSC): res = self.vty.command("write terminal") for classNum in range(16): if classNum != 10: - self.assert_(res.find("rach access-control-class " + str(classNum) + " barred") > 0) + self.assertTrue(res.find("rach access-control-class " + str(classNum) + " barred") > 0) # Allowed rach access control classes for classNum in range(16): @@ -366,7 +367,7 @@ class TestVTYNITB(TestVTYGenericBSC): res = self.vty.command("write terminal") for classNum in range(16): if classNum != 10: - self.assertEquals(res.find("rach access-control-class " + str(classNum) + " barred"), -1) + self.assertEqual(res.find("rach access-control-class " + str(classNum) + " barred"), -1) def testSubscriberCreateDeleteTwice(self): """ @@ -385,23 +386,23 @@ class TestVTYNITB(TestVTYGenericBSC): # Lets create one res = self.vty.command('subscriber create imsi '+imsi) - self.assert_(res.find(" IMSI: "+imsi) > 0) + self.assertTrue(res.find(" IMSI: "+imsi) > 0) # And now create one again. res2 = self.vty.command('subscriber create imsi '+imsi) - self.assert_(res2.find(" IMSI: "+imsi) > 0) + self.assertTrue(res2.find(" IMSI: "+imsi) > 0) self.assertEqual(res, res2) # Verify it has been created res = self.vty.command('show subscriber imsi '+imsi) - self.assert_(res.find(" IMSI: "+imsi) > 0) + self.assertTrue(res.find(" IMSI: "+imsi) > 0) # Delete it res = self.vty.command('subscriber imsi ' + imsi + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) # Now it should not be there anymore res = self.vty.command('show subscriber imsi '+imsi) - self.assert_(('% No subscriber found for imsi ' + imsi) == res) + self.assertTrue(('% No subscriber found for imsi ' + imsi) == res) def testSubscriberCreateDelete(self): @@ -417,12 +418,12 @@ class TestVTYNITB(TestVTYGenericBSC): # Lets create one res = self.vty.command('subscriber create imsi '+imsi) - self.assert_(res.find(" IMSI: "+imsi) > 0) - self.assert_(res.find("Extension") > 0) + self.assertTrue(res.find(" IMSI: "+imsi) > 0) + self.assertTrue(res.find("Extension") > 0) # Now we have it res = self.vty.command('show subscriber imsi '+imsi) - self.assert_(res.find(" IMSI: "+imsi) > 0) + self.assertTrue(res.find(" IMSI: "+imsi) > 0) # With narrow random interval self.vty.command("configure terminal") @@ -431,16 +432,16 @@ class TestVTYNITB(TestVTYGenericBSC): # wrong interval res = self.vty.command("subscriber-create-on-demand random 221 122") # error string will contain arguments - self.assert_(res.find("122") > 0) - self.assert_(res.find("221") > 0) + self.assertTrue(res.find("122") > 0) + self.assertTrue(res.find("221") > 0) # correct interval - silent ok self.assertTrue(self.vty.verify("subscriber-create-on-demand random 221 222", [''])) self.vty.command("end") res = self.vty.command('subscriber create imsi ' + imsi2) - self.assert_(res.find(" IMSI: " + imsi2) > 0) - self.assert_(res.find("221") > 0 or res.find("222") > 0) - self.assert_(res.find(" Extension: ") > 0) + self.assertTrue(res.find(" IMSI: " + imsi2) > 0) + self.assertTrue(res.find("221") > 0 or res.find("222") > 0) + self.assertTrue(res.find(" Extension: ") > 0) # Without extension self.vty.command("configure terminal") @@ -448,8 +449,8 @@ class TestVTYNITB(TestVTYGenericBSC): self.assertTrue(self.vty.verify("subscriber-create-on-demand no-extension", [''])) self.vty.command("end") res = self.vty.command('subscriber create imsi ' + imsi3) - self.assert_(res.find(" IMSI: " + imsi3) > 0) - self.assertEquals(res.find("Extension"), -1) + self.assertTrue(res.find(" IMSI: " + imsi3) > 0) + self.assertEqual(res.find("Extension"), -1) # With extension again self.vty.command("configure terminal") @@ -460,22 +461,22 @@ class TestVTYNITB(TestVTYGenericBSC): self.vty.command("end") res = self.vty.command('subscriber create imsi ' + imsi4) - self.assert_(res.find(" IMSI: " + imsi4) > 0) - self.assert_(res.find(" Extension: ") > 0) + self.assertTrue(res.find(" IMSI: " + imsi4) > 0) + self.assertTrue(res.find(" Extension: ") > 0) # Delete it res = self.vty.command('subscriber imsi ' + imsi + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) res = self.vty.command('subscriber imsi ' + imsi2 + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) res = self.vty.command('subscriber imsi ' + imsi3 + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) res = self.vty.command('subscriber imsi ' + imsi4 + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) # Now it should not be there anymore res = self.vty.command('show subscriber imsi '+imsi) - self.assert_(('% No subscriber found for imsi ' + imsi) == res) + self.assertTrue(('% No subscriber found for imsi ' + imsi) == res) # range self.vty.command("end") @@ -483,20 +484,20 @@ class TestVTYNITB(TestVTYGenericBSC): self.vty.command("nitb") self.assertTrue(self.vty.verify("subscriber-create-on-demand random 9999999998 9999999999", [''])) res = self.vty.command("show running-config") - self.assert_(res.find("subscriber-create-on-demand random 9999999998 9999999999")) + self.assertTrue(res.find("subscriber-create-on-demand random 9999999998 9999999999")) self.vty.command("end") res = self.vty.command('subscriber create imsi ' + imsi) print(res) - self.assert_(res.find(" IMSI: " + imsi) > 0) - self.assert_(res.find("9999999998") > 0 or res.find("9999999999") > 0) - self.assert_(res.find(" Extension: ") > 0) + self.assertTrue(res.find(" IMSI: " + imsi) > 0) + self.assertTrue(res.find("9999999998") > 0 or res.find("9999999999") > 0) + self.assertTrue(res.find(" Extension: ") > 0) res = self.vty.command('subscriber imsi ' + imsi + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) res = self.vty.command('show subscriber imsi '+imsi) - self.assert_(('% No subscriber found for imsi ' + imsi) == res) + self.assertTrue(('% No subscriber found for imsi ' + imsi) == res) def testSubscriberSettings(self): @@ -508,18 +509,18 @@ class TestVTYNITB(TestVTYGenericBSC): # Lets create one res = self.vty.command('subscriber create imsi '+imsi) - self.assert_(res.find(" IMSI: "+imsi) > 0) - self.assert_(res.find("Extension") > 0) + self.assertTrue(res.find(" IMSI: "+imsi) > 0) + self.assertTrue(res.find("Extension") > 0) self.vty.verify('subscriber imsi '+wrong_imsi+' name wrong', ['% No subscriber found for imsi '+wrong_imsi]) res = self.vty.command('subscriber imsi '+imsi+' name '+('X' * 160)) - self.assert_(res.find("NAME is too long") > 0) + self.assertTrue(res.find("NAME is too long") > 0) self.vty.verify('subscriber imsi '+imsi+' name '+('G' * 159), ['']) self.vty.verify('subscriber imsi '+wrong_imsi+' extension 840', ['% No subscriber found for imsi '+wrong_imsi]) res = self.vty.command('subscriber imsi '+imsi+' extension '+('9' * 15)) - self.assert_(res.find("EXTENSION is too long") > 0) + self.assertTrue(res.find("EXTENSION is too long") > 0) self.vty.verify('subscriber imsi '+imsi+' extension '+('1' * 14), ['']) @@ -529,34 +530,34 @@ class TestVTYNITB(TestVTYGenericBSC): self.assertTrue(self.vty.verify("subscriber-create-on-demand", [''])) # wrong interval res = self.vty.command("subscriber-create-on-demand random 221 122") - self.assert_(res.find("122") > 0) - self.assert_(res.find("221") > 0) + self.assertTrue(res.find("122") > 0) + self.assertTrue(res.find("221") > 0) # correct interval self.assertTrue(self.vty.verify("subscriber-create-on-demand random 221 222", [''])) self.vty.command("end") # create subscriber with extension in a configured interval res = self.vty.command('subscriber create imsi ' + imsi2) - self.assert_(res.find(" IMSI: " + imsi2) > 0) - self.assert_(res.find("221") > 0 or res.find("222") > 0) - self.assert_(res.find(" Extension: ") > 0) + self.assertTrue(res.find(" IMSI: " + imsi2) > 0) + self.assertTrue(res.find("221") > 0 or res.find("222") > 0) + self.assertTrue(res.find(" Extension: ") > 0) # Delete it res = self.vty.command('subscriber imsi ' + imsi + ' delete') - self.assert_(res != "") + self.assertTrue(res != "") # imsi2 is inactive so deletion should succeed res = self.vty.command('subscriber imsi ' + imsi2 + ' delete') - self.assert_("" == res) + self.assertTrue("" == res) def testShowPagingGroup(self): res = self.vty.command("show paging-group 255 1234567") self.assertEqual(res, "% can't find BTS 255") res = self.vty.command("show paging-group 0 1234567") - self.assertEquals(res, "%Paging group for IMSI 1234567 on BTS #0 is 7") + self.assertEqual(res, "%Paging group for IMSI 1234567 on BTS #0 is 7") def testShowNetwork(self): res = self.vty.command("show network") - self.assert_(res.startswith('BSC is on Country Code') >= 0) + self.assertTrue(res.startswith('BSC is on Country Code') >= 0) def testMeasurementFeed(self): self.vty.enable() @@ -564,17 +565,17 @@ class TestVTYNITB(TestVTYGenericBSC): self.vty.command("mncc-int") res = self.vty.command("write terminal") - self.assertEquals(res.find('meas-feed scenario'), -1) + self.assertEqual(res.find('meas-feed scenario'), -1) self.vty.command("meas-feed scenario bla") res = self.vty.command("write terminal") - self.assert_(res.find('meas-feed scenario bla') > 0) + self.assertTrue(res.find('meas-feed scenario bla') > 0) self.vty.command("meas-feed scenario abcdefghijklmnopqrstuvwxyz01234567890") res = self.vty.command("write terminal") - self.assertEquals(res.find('meas-feed scenario abcdefghijklmnopqrstuvwxyz01234567890'), -1) - self.assertEquals(res.find('meas-feed scenario abcdefghijklmnopqrstuvwxyz012345'), -1) - self.assert_(res.find('meas-feed scenario abcdefghijklmnopqrstuvwxyz01234') > 0) + self.assertEqual(res.find('meas-feed scenario abcdefghijklmnopqrstuvwxyz01234567890'), -1) + self.assertEqual(res.find('meas-feed scenario abcdefghijklmnopqrstuvwxyz012345'), -1) + self.assertTrue(res.find('meas-feed scenario abcdefghijklmnopqrstuvwxyz01234') > 0) class TestVTYBSC(TestVTYGenericBSC): @@ -592,18 +593,18 @@ class TestVTYBSC(TestVTYGenericBSC): def testVtyTree(self): self.vty.enable() self.assertTrue(self.vty.verify("configure terminal", [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.checkForEndAndExit() self.assertTrue(self.vty.verify("msc 0", [''])) - self.assertEquals(self.vty.node(), 'config-msc') + self.assertEqual(self.vty.node(), 'config-msc') self.checkForEndAndExit() self.assertTrue(self.vty.verify("exit", [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify("bsc", [''])) - self.assertEquals(self.vty.node(), 'config-bsc') + self.assertEqual(self.vty.node(), 'config-bsc') self.checkForEndAndExit() self.assertTrue(self.vty.verify("exit", [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify("exit", [''])) self.assertTrue(self.vty.node() is None) @@ -624,12 +625,12 @@ class TestVTYBSC(TestVTYGenericBSC): # Verify settings res = self.vty.command("write terminal") - self.assert_(res.find('bsc-msc-lost-text MSC disconnected') > 0) - self.assertEquals(res.find('no bsc-msc-lost-text'), -1) - self.assert_(res.find('bsc-welcome-text Hello MS') > 0) - self.assertEquals(res.find('no bsc-welcome-text'), -1) - self.assert_(res.find('bsc-grace-text In grace period') > 0) - self.assertEquals(res.find('no bsc-grace-text'), -1) + self.assertTrue(res.find('bsc-msc-lost-text MSC disconnected') > 0) + self.assertEqual(res.find('no bsc-msc-lost-text'), -1) + self.assertTrue(res.find('bsc-welcome-text Hello MS') > 0) + self.assertEqual(res.find('no bsc-welcome-text'), -1) + self.assertTrue(res.find('bsc-grace-text In grace period') > 0) + self.assertEqual(res.find('no bsc-grace-text'), -1) # Now disable it.. self.vty.verify("no bsc-msc-lost-text", ['']) @@ -638,12 +639,12 @@ class TestVTYBSC(TestVTYGenericBSC): # Verify settings res = self.vty.command("write terminal") - self.assertEquals(res.find('bsc-msc-lost-text MSC disconnected'), -1) - self.assert_(res.find('no bsc-msc-lost-text') > 0) - self.assertEquals(res.find('bsc-welcome-text Hello MS'), -1) - self.assert_(res.find('no bsc-welcome-text') > 0) - self.assertEquals(res.find('bsc-grace-text In grace period'), -1) - self.assert_(res.find('no bsc-grace-text') > 0) + self.assertEqual(res.find('bsc-msc-lost-text MSC disconnected'), -1) + self.assertTrue(res.find('no bsc-msc-lost-text') > 0) + self.assertEqual(res.find('bsc-welcome-text Hello MS'), -1) + self.assertTrue(res.find('no bsc-welcome-text') > 0) + self.assertEqual(res.find('bsc-grace-text In grace period'), -1) + self.assertTrue(res.find('no bsc-grace-text') > 0) def testUssdNotificationsBsc(self): self.vty.enable() @@ -658,16 +659,16 @@ class TestVTYBSC(TestVTYGenericBSC): # Verify settings res = self.vty.command("write terminal") - self.assert_(res.find('missing-msc-text No MSC found') > 0) - self.assertEquals(res.find('no missing-msc-text'), -1) + self.assertTrue(res.find('missing-msc-text No MSC found') > 0) + self.assertEqual(res.find('no missing-msc-text'), -1) # Now disable it.. self.vty.verify("no missing-msc-text", ['']) # Verify settings res = self.vty.command("write terminal") - self.assertEquals(res.find('missing-msc-text No MSC found'), -1) - self.assert_(res.find('no missing-msc-text') > 0) + self.assertEqual(res.find('missing-msc-text No MSC found'), -1) + self.assertTrue(res.find('no missing-msc-text') > 0) def testNetworkTimezone(self): self.vty.enable() @@ -685,26 +686,26 @@ class TestVTYBSC(TestVTYGenericBSC): # Verify settings res = self.vty.command("write terminal") - self.assert_(res.find('timezone 2 30') > 0) - self.assertEquals(res.find('timezone 2 30 '), -1) + self.assertTrue(res.find('timezone 2 30') > 0) + self.assertEqual(res.find('timezone 2 30 '), -1) # Set time zone with DST self.vty.verify("timezone 2 30 1", ['']) # Verify settings res = self.vty.command("write terminal") - self.assert_(res.find('timezone 2 30 1') > 0) + self.assertTrue(res.find('timezone 2 30 1') > 0) # Now disable it.. self.vty.verify("no timezone", ['']) # Verify settings res = self.vty.command("write terminal") - self.assertEquals(res.find(' timezone'), -1) + self.assertEqual(res.find(' timezone'), -1) def testShowNetwork(self): res = self.vty.command("show network") - self.assert_(res.startswith('BSC is on Country Code') >= 0) + self.assertTrue(res.startswith('BSC is on Country Code') >= 0) def testPingPongConfiguration(self): self.vty.enable() @@ -714,28 +715,28 @@ class TestVTYBSC(TestVTYGenericBSC): self.vty.verify("timeout-ping 12", ['']) self.vty.verify("timeout-pong 14", ['']) res = self.vty.command("show running-config") - self.assert_(res.find(" timeout-ping 12") > 0) - self.assert_(res.find(" timeout-pong 14") > 0) - self.assert_(res.find(" no timeout-ping advanced") > 0) + self.assertTrue(res.find(" timeout-ping 12") > 0) + self.assertTrue(res.find(" timeout-pong 14") > 0) + self.assertTrue(res.find(" no timeout-ping advanced") > 0) self.vty.verify("timeout-ping advanced", ['']) res = self.vty.command("show running-config") - self.assert_(res.find(" timeout-ping 12") > 0) - self.assert_(res.find(" timeout-pong 14") > 0) - self.assert_(res.find(" timeout-ping advanced") > 0) + self.assertTrue(res.find(" timeout-ping 12") > 0) + self.assertTrue(res.find(" timeout-pong 14") > 0) + self.assertTrue(res.find(" timeout-ping advanced") > 0) self.vty.verify("no timeout-ping advanced", ['']) res = self.vty.command("show running-config") - self.assert_(res.find(" timeout-ping 12") > 0) - self.assert_(res.find(" timeout-pong 14") > 0) - self.assert_(res.find(" no timeout-ping advanced") > 0) + self.assertTrue(res.find(" timeout-ping 12") > 0) + self.assertTrue(res.find(" timeout-pong 14") > 0) + self.assertTrue(res.find(" no timeout-ping advanced") > 0) self.vty.verify("no timeout-ping", ['']) res = self.vty.command("show running-config") - self.assertEquals(res.find(" timeout-ping 12"), -1) - self.assertEquals(res.find(" timeout-pong 14"), -1) - self.assertEquals(res.find(" no timeout-ping advanced"), -1) - self.assert_(res.find(" no timeout-ping") > 0) + self.assertEqual(res.find(" timeout-ping 12"), -1) + self.assertEqual(res.find(" timeout-pong 14"), -1) + self.assertEqual(res.find(" no timeout-ping advanced"), -1) + self.assertTrue(res.find(" no timeout-ping") > 0) self.vty.verify("timeout-ping advanced", ['%ping handling is disabled. Enable it first.']) @@ -743,15 +744,15 @@ class TestVTYBSC(TestVTYGenericBSC): self.vty.verify("timeout-ping 12", ['']) self.vty.verify("timeout-pong 14", ['']) res = self.vty.command("show running-config") - self.assert_(res.find(" timeout-ping 12") > 0) - self.assert_(res.find(" timeout-pong 14") > 0) - self.assert_(res.find(" timeout-ping advanced") > 0) + self.assertTrue(res.find(" timeout-ping 12") > 0) + self.assertTrue(res.find(" timeout-pong 14") > 0) + self.assertTrue(res.find(" timeout-ping advanced") > 0) def testMscDataCoreLACCI(self): self.vty.enable() res = self.vty.command("show running-config") - self.assertEquals(res.find("core-location-area-code"), -1) - self.assertEquals(res.find("core-cell-identity"), -1) + self.assertEqual(res.find("core-location-area-code"), -1) + self.assertEqual(res.find("core-cell-identity"), -1) self.vty.command("configure terminal") self.vty.command("msc 0") @@ -759,8 +760,8 @@ class TestVTYBSC(TestVTYGenericBSC): self.vty.command("core-cell-identity 333") res = self.vty.command("show running-config") - self.assert_(res.find("core-location-area-code 666") > 0) - self.assert_(res.find("core-cell-identity 333") > 0) + self.assertTrue(res.find("core-location-area-code 666") > 0) + self.assertTrue(res.find("core-cell-identity 333") > 0) class TestVTYNAT(TestVTYGenericBSC): @@ -781,7 +782,7 @@ class TestVTYNAT(TestVTYGenericBSC): nat_bsc_reload(self) bscs2 = self.vty.command("show bscs-config") # check that multiple calls to bscs-config-file give the same result - self.assertEquals(bscs1, bscs2) + self.assertEqual(bscs1, bscs2) # add new bsc self.vty.command("configure terminal") @@ -805,23 +806,23 @@ class TestVTYNAT(TestVTYGenericBSC): b1 = nat_bsc_sock_test(1, "xyu", verbose=True, proc=self.proc) b2 = nat_bsc_sock_test(5, "key", verbose=True, proc=self.proc) - self.assertEquals("3 BSCs configured", self.vty.command("show nat num-bscs-configured")) + self.assertEqual("3 BSCs configured", self.vty.command("show nat num-bscs-configured")) self.assertTrue(3 == nat_bsc_num_con(self)) - self.assertEquals("MSC is connected: 1", self.vty.command("show msc connection")) + self.assertEqual("MSC is connected: 1", self.vty.command("show msc connection")) nat_bsc_reload(self) bscs2 = self.vty.command("show bscs-config") # check that the reset to initial config succeeded - self.assertEquals(bscs1, bscs2) + self.assertEqual(bscs1, bscs2) - self.assertEquals("2 BSCs configured", self.vty.command("show nat num-bscs-configured")) + self.assertEqual("2 BSCs configured", self.vty.command("show nat num-bscs-configured")) self.assertTrue(1 == nat_bsc_num_con(self)) rem = self.vty.command("show bsc connections").split(' ') # remaining connection is for BSC0 - self.assertEquals('0', rem[2]) + self.assertEqual('0', rem[2]) # remaining connection is authorized - self.assertEquals('1', rem[4]) - self.assertEquals("MSC is connected: 1", self.vty.command("show msc connection")) + self.assertEqual('1', rem[4]) + self.assertEqual("MSC is connected: 1", self.vty.command("show msc connection")) finally: msc.close() msc_socket.close() @@ -829,23 +830,23 @@ class TestVTYNAT(TestVTYGenericBSC): def testVtyTree(self): self.vty.enable() self.assertTrue(self.vty.verify('configure terminal', [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.checkForEndAndExit() self.assertTrue(self.vty.verify('mgcp', [''])) - self.assertEquals(self.vty.node(), 'config-mgcp') + self.assertEqual(self.vty.node(), 'config-mgcp') self.checkForEndAndExit() self.assertTrue(self.vty.verify('exit', [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify('nat', [''])) - self.assertEquals(self.vty.node(), 'config-nat') + self.assertEqual(self.vty.node(), 'config-nat') self.checkForEndAndExit() self.assertTrue(self.vty.verify('bsc 0', [''])) - self.assertEquals(self.vty.node(), 'config-nat-bsc') + self.assertEqual(self.vty.node(), 'config-nat-bsc') self.checkForEndAndExit() self.assertTrue(self.vty.verify('exit', [''])) - self.assertEquals(self.vty.node(), 'config-nat') + self.assertEqual(self.vty.node(), 'config-nat') self.assertTrue(self.vty.verify('exit', [''])) - self.assertEquals(self.vty.node(), 'config') + self.assertEqual(self.vty.node(), 'config') self.assertTrue(self.vty.verify('exit', [''])) self.assertTrue(self.vty.node() is None) @@ -863,15 +864,15 @@ class TestVTYNAT(TestVTYGenericBSC): # Ensure the default res = self.vty.command("show running-config") - self.assert_(res.find('\n sdp-ensure-amr-mode-set') > 0) + self.assertTrue(res.find('\n sdp-ensure-amr-mode-set') > 0) self.vty.command("sdp-ensure-amr-mode-set") res = self.vty.command("show running-config") - self.assert_(res.find('\n sdp-ensure-amr-mode-set') > 0) + self.assertTrue(res.find('\n sdp-ensure-amr-mode-set') > 0) self.vty.command("no sdp-ensure-amr-mode-set") res = self.vty.command("show running-config") - self.assert_(res.find('\n no sdp-ensure-amr-mode-set') > 0) + self.assertTrue(res.find('\n no sdp-ensure-amr-mode-set') > 0) def testRewritePostNoRewrite(self): self.vty.enable() @@ -916,35 +917,35 @@ class TestVTYNAT(TestVTYGenericBSC): ussdSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) ussdSocket.connect(('127.0.0.1', 5001)) ussdSocket.settimeout(2.0) - print "Connected to %s:%d" % ussdSocket.getpeername() + print("Connected to %s:%d" % ussdSocket.getpeername()) - print "Expecting ID_GET request" + print("Expecting ID_GET request") data = ussdSocket.recv(4) - self.assertEqual(data, "\x00\x01\xfe\x04") + self.assertEqual(data, b"\x00\x01\xfe\x04") - print "Going to send ID_RESP response" + print("Going to send ID_RESP response") res = ussdSocket.send(IPA().id_resp(IPA().tag_name('key'+'\0'))) self.assertEqual(res, 11) # initiating PING/PONG cycle to know, that the ID_RESP message has been processed - print "Going to send PING request" + print("Going to send PING request") res = ussdSocket.send(IPA().ping()) self.assertEqual(res, 4) - print "Expecting PONG response" + print("Expecting PONG response") data = ussdSocket.recv(4) - self.assertEqual(data, "\x00\x01\xfe\x01") + self.assertEqual(data, b"\x00\x01\xfe\x01") res = self.vty.verify("show ussd-connection", ['The USSD side channel provider is connected and authorized.']) self.assertTrue(res) - print "Going to shut down connection" + print("Going to shut down connection") ussdSocket.shutdown(socket.SHUT_WR) - print "Expecting EOF" + print("Expecting EOF") data = ussdSocket.recv(4) - self.assertEqual(data, "") + self.assertEqual(data, b"") ussdSocket.close() @@ -967,7 +968,7 @@ class TestVTYNAT(TestVTYGenericBSC): if line.startswith(" access-list test-default"): self.assertEqual(line, " access-list test-default imsi-deny ^123[0-9]*$ 11 11") asserted = True - self.assert_(asserted) + self.assertTrue(asserted) # Check the optional CM Service Reject Cause self.vty.command("access-list test-cm-deny imsi-deny ^123[0-9]*$ 42").split("\r\n") @@ -977,7 +978,7 @@ class TestVTYNAT(TestVTYGenericBSC): if line.startswith(" access-list test-cm"): self.assertEqual(line, " access-list test-cm-deny imsi-deny ^123[0-9]*$ 42 11") asserted = True - self.assert_(asserted) + self.assertTrue(asserted) # Check the optional LU Reject Cause self.vty.command("access-list test-lu-deny imsi-deny ^123[0-9]*$ 23 42").split("\r\n") @@ -987,7 +988,7 @@ class TestVTYNAT(TestVTYGenericBSC): if line.startswith(" access-list test-lu"): self.assertEqual(line, " access-list test-lu-deny imsi-deny ^123[0-9]*$ 23 42") asserted = True - self.assert_(asserted) + self.assertTrue(asserted) def add_nat_test(suite, workdir): if not os.path.isfile(os.path.join(workdir, "src/osmo-bsc_nat/osmo-bsc_nat")): @@ -1010,7 +1011,7 @@ def nat_msc_ip(x, ip, port): x.vty.command("end") def data2str(d): - return d.encode('hex').lower() + return d.hex() def nat_msc_test(x, ip, port, verbose = False): msc = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -1019,11 +1020,11 @@ def nat_msc_test(x, ip, port, verbose = False): msc.bind((ip, port)) msc.listen(5) if (verbose): - print "MSC is ready at " + ip + print("MSC is ready at " + ip) conn = None while True: vty_response = x.vty.command("show msc connection") - print "'show msc connection' says: %r" % vty_response + print("'show msc connection' says: %r" % vty_response) if vty_response == "MSC is connected: 1": # success break; @@ -1035,10 +1036,10 @@ def nat_msc_test(x, ip, port, verbose = False): while timeout_retries > 0: try: conn, addr = msc.accept() - print "MSC got connection from ", addr + print("MSC got connection from ", addr) break except socket.timeout: - print "socket timed out." + print("socket timed out.") timeout_retries -= 1 continue @@ -1053,52 +1054,52 @@ def ipa_handle_small(x, verbose = False): raise Exception("expected to receive 4 bytes, but got %d (%r)" % (len(s)/2, s)) if "0001fe00" == s: if (verbose): - print "\tBSC <- NAT: PING?" + print("\tBSC <- NAT: PING?") x.send(IPA().pong()) elif "0001fe06" == s: if (verbose): - print "\tBSC <- NAT: IPA ID ACK" + print("\tBSC <- NAT: IPA ID ACK") x.send(IPA().id_ack()) elif "0001fe00" == s: if (verbose): - print "\tBSC <- NAT: PONG!" + print("\tBSC <- NAT: PONG!") else: if (verbose): - print "\tBSC <- NAT: ", s + print("\tBSC <- NAT: ", s) def ipa_handle_resp(x, tk, verbose = False, proc=None): s = data2str(x.recv(38)) if "0023fe040108010701020103010401050101010011" in s: retries = 3 while True: - print "\tsending IPA identity(%s) at %s" % (tk, time.strftime("%T")) + print("\tsending IPA identity(%s) at %s" % (tk, time.strftime("%T"))) try: x.send(IPA().id_resp(IPA().identity(name = (tk+'\0').encode('utf-8')))) - print "\tdone sending IPA identity(%s) at %s" % (tk, - time.strftime("%T")) + print("\tdone sending IPA identity(%s) at %s" % (tk, + time.strftime("%T"))) break except: - print "\tfailed sending IPA identity at", time.strftime("%T") + print("\tfailed sending IPA identity at", time.strftime("%T")) if proc: - print "\tproc.poll() = %r" % proc.poll() + print("\tproc.poll() = %r" % proc.poll()) if retries < 1: - print "\tgiving up" + print("\tgiving up") raise - print "\tretrying (%d attempts left)" % retries + print("\tretrying (%d attempts left)" % retries) retries -= 1 else: if (verbose): - print "\tBSC <- NAT: ", s + print("\tBSC <- NAT: ", s) def ipa_handle_mgcp(x, verbose = False): data = x.recv(3) s = data2str(data) if s[4:] != "fc": - print "expected IPA(MGCP) but received %r instead" % (s[4:]) + print("expected IPA(MGCP) but received %r instead" % (s[4:])) ipa_len, = struct.unpack('>H', data[:2]) mgcp_msg = x.recv(ipa_len) # MGCP msg if (verbose): - print "\tBSC <- NAT (MGCP[%d]): %s" % (ipa_len, mgcp_msg) + print("\tBSC <- NAT (MGCP[%d]): %s" % (ipa_len, mgcp_msg)) def nat_bsc_num_con(x): return len(x.vty.command("show bsc connections").split('\n')) @@ -1108,18 +1109,18 @@ def nat_bsc_sock_test(nr, tk, verbose = False, proc=None): bsc.bind(('127.0.0.1', 0)) bsc.connect(('127.0.0.1', 5000)) if (verbose): - print "BSC%d " %nr - print "\tconnected to %s:%d" % bsc.getpeername() + print("BSC%d " %nr) + print("\tconnected to %s:%d" % bsc.getpeername()) if proc: - print "\tproc.poll() = %r" % proc.poll() - print "\tproc.pid = %r" % proc.pid + print("\tproc.poll() = %r" % proc.poll()) + print("\tproc.pid = %r" % proc.pid) ipa_handle_small(bsc, verbose) ipa_handle_resp(bsc, tk, verbose, proc=proc) if proc: - print "\tproc.poll() = %r" % proc.poll() + print("\tproc.poll() = %r" % proc.poll()) ipa_handle_mgcp(bsc, verbose) if proc: - print "\tproc.poll() = %r" % proc.poll() + print("\tproc.poll() = %r" % proc.poll()) ipa_handle_small(bsc, verbose) return bsc @@ -1156,9 +1157,9 @@ if __name__ == '__main__': if args.p: confpath = args.p - print "confpath %s, workdir %s" % (confpath, workdir) + print("confpath %s, workdir %s" % (confpath, workdir)) os.chdir(workdir) - print "Running tests for specific VTY commands" + print("Running tests for specific VTY commands") suite = unittest.TestSuite() suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestVTYMGCP)) suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestVTYNITB)) |