asterisk/Asterisk_Tests.ttcn

module Asterisk_Tests {

/* Asterisk test suite in TTCN-3
 * (C) 2024 by sysmocom - s.f.m.c. GmbH <info@sysmocom.de>
 * All rights reserved.
 * Author: Pau Espin Pedrol <pespin@sysmocom.de>
 *
 * Released under the terms of GNU General Public License, Version 2 or
 * (at your option) any later version.
 *
 * SPDX-License-Identifier: GPL-2.0-or-later
 */

import from TCCOpenSecurity_Functions all;
import from General_Types all;
import from Osmocom_Types all;
import from Native_Functions all;
import from Misc_Helpers all;
import from TELNETasp_PortType all;
import from AMI_Functions all;

import from SDP_Types all;
import from SDP_Templates all;

import from SIP_Emulation all;
import from SIPmsg_Types all;
import from SIP_Templates all;

import from SIP_ConnectionHandler all;

modulepar {
	charstring mp_local_sip_host := "127.0.0.2";
	integer mp_local_sip_port := 5060;
	charstring mp_remote_sip_host := "127.0.0.1";
	integer mp_remote_sip_port := 5060;

	/* Asterisk AMI: */
	charstring mp_ami_user := "test_user";
	charstring mp_ami_secret := "1234";
}

type component test_CT {
	var SIP_Emulation_CT vc_SIP;
	port TELNETasp_PT AMI;
	port Coord_PT COORD;
}

const charstring broadcast_sip_extension := "0500";

function f_init_ConnHdlrPars(integer idx := 1) runs on test_CT return SIPConnHdlrPars {
	var template (value) CallPars cp := t_CallPars(mp_local_sip_host, 1234 + 2*idx);
	var template (value) SIPConnHdlrPars pars := t_Pars(mp_local_sip_host,
							    mp_local_sip_port,
							    mp_remote_sip_host,
							    mp_remote_sip_port,
							    "0" & int2str(str2int(broadcast_sip_extension) + idx),
							    cp := cp);
	return valueof(pars);
}

/* Initialize connection towards Asterisk AMI */
private function f_init_ami() runs on test_CT {
	map(self:AMI, system:AMI);
	f_ami_action_login(AMI, mp_ami_user, mp_ami_secret);
}

function f_init() runs on test_CT {
	f_init_ami();
	f_init_sip(vc_SIP, "Asterisk_Test");
	log("end of f_init");
}

function f_start_handler(void_fn fn, SIPConnHdlrPars pars)
runs on test_CT return SIPConnHdlr {
	var SIPConnHdlr vc_conn;
	var charstring id := testcasename() & "-ConnHdlr-" & pars.user;

	vc_conn := SIPConnHdlr.create(id) alive;

	connect(vc_conn:SIP, vc_SIP:CLIENT);
	connect(vc_conn:SIP_PROC, vc_SIP:CLIENT_PROC);

	connect(vc_conn:COORD, self:COORD);

	vc_conn.start(f_handler_init(fn, id, pars));
	return vc_conn;
}

/* Test SIP registration of local clients */
private function f_TC_internal_registration(charstring id) runs on SIPConnHdlr {

	f_SIP_register();
	// f_SIP_deregister();
	setverdict(pass);
}
testcase TC_internal_registration() runs on test_CT {
	var SIPConnHdlrPars pars;
	var SIPConnHdlr vc_conn;
	f_init();
	pars := f_init_ConnHdlrPars();
	vc_conn := f_start_handler(refers(f_TC_internal_registration), pars);
	vc_conn.done;
}

/* Successful SIP MO-MT Call between local clients: */
private function f_TC_internal_call_mo(charstring id) runs on SIPConnHdlr {

	f_SIP_register();
	COORD.send(COORD_CMD_REGISTERED);

	COORD.receive(COORD_CMD_START);
	f_SIP_mo_call_setup();
	COORD.send(COORD_CMD_CALL_ESTABLISHED);

	COORD.receive(COORD_CMD_HANGUP);
	f_SIP_do_call_hangup();

	setverdict(pass);
}
private function f_TC_internal_call_mt(charstring id) runs on SIPConnHdlr {

	f_create_sip_expect(valueof(ts_SipUrl_from_Addr_Union(g_pars.cp.called.addr)));

	f_SIP_register();
	COORD.send(COORD_CMD_REGISTERED);

	if (g_pars.cp.mt.exp_cancel) {
		as_SIP_mt_call_cancelled();
		COORD.send(COORD_CMD_CALL_CANCELLED);
		setverdict(pass);
		return;
	}

	as_SIP_mt_call_accept();
	COORD.send(COORD_CMD_CALL_ESTABLISHED);

	/* Once MO hangs up, Asterisk updates us to point RTP to it: */
	as_SIP_exp_call_update(g_pars.cp.sip_seq_nr + 1);
	as_SIP_exp_call_hangup(g_pars.cp.sip_seq_nr + 1);

	setverdict(pass);
}
testcase TC_internal_call_momt() runs on test_CT {
	var SIPConnHdlrPars pars[2];
	var SIPConnHdlr vc_conn[2];

	f_init();

	pars[0] := f_init_ConnHdlrPars(idx := 1);
	pars[1] := f_init_ConnHdlrPars(idx := 2);

	pars[0].cp.calling := pars[0].registrar_sip_record;
	pars[0].cp.called := pars[1].registrar_sip_record;

	pars[1].cp.calling := pars[0].registrar_sip_record;
	pars[1].cp.called := pars[1].local_sip_record;

	vc_conn[0] := f_start_handler(refers(f_TC_internal_call_mo), pars[0]);
	vc_conn[1] := f_start_handler(refers(f_TC_internal_call_mt), pars[1]);

	interleave {
	[] COORD.receive(COORD_CMD_REGISTERED) from vc_conn[0];
	[] COORD.receive(COORD_CMD_REGISTERED) from vc_conn[1];
	}

	COORD.send(COORD_CMD_START) to vc_conn[0];

	interleave {
	[] COORD.receive(COORD_CMD_CALL_ESTABLISHED) from vc_conn[0];
	[] COORD.receive(COORD_CMD_CALL_ESTABLISHED) from vc_conn[1];
	}

	/* Call on-going */
	f_sleep(1.0);

	COORD.send(COORD_CMD_HANGUP) to vc_conn[0];


	vc_conn[0].done;
	vc_conn[1].done;
}

/* One of the users calls (INVITE) shared extension, which makes all other user
 * equipments ring (INVITE). The first one to pick up the call (OK 200) gets the
 * call established (ACK), others get a CANCEL event. */
private function TC_internal_call_all_Nregistered(integer num_conns := 2) runs on test_CT {
	var SIPConnHdlrList vc_conn_list := {};
	const integer vc_conn_mo_idx := 0; /* Index of MO leg in vc_conn_list */
	const integer vc_conn_mt_idx := 1; /* Index of MT leg in vc_conn_list, peer picking up first the call */
	var SipAddr broadcast_sip_record;
	var SIPConnHdlrPars pars_mo;

	f_init();

	broadcast_sip_record := valueof(ts_SipAddr(ts_HostPort(mp_local_sip_host),
					ts_UserInfo(broadcast_sip_extension)));

	for (var integer i := 0; i < num_conns; i := i + 1) {
		var SIPConnHdlrPars pars;
		var SIPConnHdlr vc_conn;
		pars := f_init_ConnHdlrPars(idx := i + 1);
		if (i == vc_conn_mo_idx) { /* MO */
			pars.cp.calling := pars.registrar_sip_record;
			pars.cp.called := broadcast_sip_record;
			vc_conn := f_start_handler(refers(f_TC_internal_call_mo), pars);
			pars_mo := pars;
		} else { /* MT */
			pars.cp.calling := pars_mo.registrar_sip_record;
			pars.cp.called := pars.local_sip_record;
			pars.cp.mt.wait_coord_cmd_pickup := true;
			if (i != vc_conn_mt_idx) {
				/* Only first MT picking up (OK 200 INVITE) will be ACKed, others CANCELed: */
				pars.cp.mt.exp_cancel := true;
			}
			vc_conn := f_start_handler(refers(f_TC_internal_call_mt), pars);
		}
		vc_conn_list := vc_conn_list & { vc_conn };
	}

	/* Wait all users are registered: */
	for (var integer i := 0; i < num_conns; i := i + 1) {
		/* Note: "from vc_conn_list[i]" can't be used since they may arrive from components in any order: */
		COORD.receive(COORD_CMD_REGISTERED);
	}

	/* Ask MO user to start the call: */
	COORD.send(COORD_CMD_START) to vc_conn_list[vc_conn_mo_idx];

	/* Make sure the desired MT is the one picking up first the call: */
	COORD.send(COORD_CMD_PICKUP) to vc_conn_list[vc_conn_mt_idx];
	interleave {
	[] COORD.receive(COORD_CMD_CALL_ESTABLISHED) from vc_conn_list[vc_conn_mo_idx];
	[] COORD.receive(COORD_CMD_CALL_ESTABLISHED) from vc_conn_list[vc_conn_mt_idx];
	}

	/* Pick up from other phone calls and expect CANCEL: */
	for (var integer i := 0; i < num_conns; i := i + 1) {
		if (i != vc_conn_mo_idx and i != vc_conn_mt_idx) {
			COORD.send(COORD_CMD_PICKUP) to vc_conn_list[i];
			COORD.receive(COORD_CMD_CALL_CANCELLED) from vc_conn_list[i];
		}
	}

	/* Call on-going */
	f_sleep(1.0);

	COORD.send(COORD_CMD_HANGUP) to vc_conn_list[vc_conn_mo_idx];

	for (var integer i := 0; i < num_conns; i := i + 1) {
		vc_conn_list[i].done;
	}
}
testcase TC_internal_call_all_2registered() runs on test_CT {
	TC_internal_call_all_Nregistered(2);
}
testcase TC_internal_call_all_3registered() runs on test_CT {
	TC_internal_call_all_Nregistered(3);
}
testcase TC_internal_call_all_4registered() runs on test_CT {
	TC_internal_call_all_Nregistered(4);
}

testcase TC_selftest() runs on test_CT {
	f_sip_digest_selftest();
	setverdict(pass);
}

control {
	execute( TC_internal_registration() );
	execute( TC_internal_call_momt() );
	execute( TC_internal_call_all_2registered() );
	execute( TC_internal_call_all_3registered() );
	execute( TC_internal_call_all_4registered() );
}

}