src/osmo_gsm_tester/obj/android_apn.py

# osmo_gsm_tester: specifics for setting an APN on an AndroidUE modem
#
# Copyright (C) 2020 by Software Radio Systems Limited
#
# Author: Nils Fürste <nils.fuerste@softwareradiosystems.com>
# Author: Bedran Karakoc <bedran.karakoc@softwareradiosystems.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import re

from ..core import log
from ..core import schema
from .android_host import AndroidHost


class AndroidApn(AndroidHost):
##############
# PROTECTED
##############
    def __init__(self, apn, mcc, mnc, select=None):
        self.logger_name = 'apn_worker_'
        super().__init__(self.logger_name)
        self._apn_name = apn
        self._apn = apn
        self._mcc = mcc
        self._mnc = mnc
        self._select = select
        if not self._apn:
            raise log.Error('APN name not set')
        if not self._mcc:
            raise log.Error('MCC not set')
        if not self._mnc:
            raise log.Error('MNC not set')

        # optional parameters, set with set_additional_params()
        self.proxy = None
        self.port = None
        self.user = None
        self.password = None
        self.server = None
        self.mmsc = None
        self.mmsport = None
        self.mmsproxy = None
        self.auth = None
        self.type = None
        self.protocol = None
        self.mvnoval = None
        self.mvnotype = None

    def get_carrier_id(self, carrier_name):
        qry_carrier_cmd = "content query --uri \"content://telephony/carriers\""
        proc = self.run_androidue_cmd('get-carrier-id', [qry_carrier_cmd])
        proc.launch_sync()
        available_carriers = proc.get_stdout().split('\n')
        carr_id = -1
        for carr in available_carriers:
            if 'name=' + carrier_name in carr:  # found carrier
                carr_id = re.findall(r'_id=(\S+),', carr)[0]
                break
        return carr_id

    def set_new_carrier(self, apn_parameter, carr_id):
        # check if carrier was found, delete it if exists
        if carr_id != -1:
            self.delete_apn(apn_parameter['carrier'])

        set_carrier_cmd = "content insert --uri content://telephony/carriers" \
                          + " --bind name:s:\"" + apn_parameter["carrier"] + "\"" \
                          + " --bind numeric:s:\"" + apn_parameter["mcc"] + apn_parameter["mnc"] + "\"" \
                          + " --bind mcc:s:\"" + apn_parameter["mcc"] + "\"" \
                          + " --bind mnc:s:\"" + apn_parameter["mnc"] + "\""\
                          + " --bind apn:s:\"" + apn_parameter["apn"] + "\"" \
                          + " --bind user:s:\"" + apn_parameter["user"] + "\"" \
                          + " --bind password:s:\"" + apn_parameter["password"] + "\"" \
                          + " --bind mmsc:s:\"" + apn_parameter["mmsc"] + "\"" \
                          + " --bind mmsport:s:\"" + apn_parameter["mmsport"] + "\"" \
                          + " --bind mmsproxy:s:\"" + apn_parameter["mmsproxy"] + "\"" \
                          + " --bind authtype:s:\"" + apn_parameter["auth"] + "\"" \
                          + " --bind type:s:\"" + apn_parameter["type"] + "\"" \
                          + " --bind protocol:s:\"" + apn_parameter["protocol"] + "\"" \
                          + " --bind mvno_type:s:\"" + apn_parameter["mvnotype"] + "\"" \
                          + " --bind mvno_match_data:s:\"" + apn_parameter["mvnoval"] + "\""
        proc = self.run_androidue_cmd("set-new-carrier", [set_carrier_cmd])
        proc.launch_sync()
        return self.get_carrier_id(apn_parameter['carrier'])

    def set_preferred_apn(self, carr_id):
        if carr_id != -1:
            set_apn_cmd = "content insert --uri content://telephony/carriers/preferapn --bind apn_id:s:\"" + str(carr_id) + "\""
            proc = self.run_androidue_cmd('set-preferred-apn', [set_apn_cmd])
            proc.launch_sync()

    def select_apn(self, carr_name):
        carr_id = self.get_carrier_id(carr_name)
        if carr_id == 0:
            return False

        # select carrier by ID
        sel_apn_cmd = "content update --uri content://telephony/carriers/preferapn --bind apn_id:s:\"" + str(carr_id) + "\""
        proc = self.run_androidue_cmd('select-apn', [sel_apn_cmd])
        proc.launch_sync()
        return True

    def delete_apn(self, carr_name):
        set_apn_cmd = "content delete --uri content://telephony/carriers --where \'name=\"" + str(carr_name) + "\" \'"
        proc = self.run_androidue_cmd('delete-apn', [set_apn_cmd])
        proc.launch_sync()

########################
# PUBLIC - INTERNAL API
########################
    @classmethod
    def from_conf(cls, conf):
        return cls(conf.get('apn', None), conf.get('mcc', None),
                   conf.get('mnc', None), conf.get('select', None))

    @classmethod
    def schema(cls):
        resource_schema = {
            'apn': schema.STR,
            'mcc': schema.STR,
            'mnc': schema.STR,
            'select': schema.BOOL_STR,
            }
        return resource_schema

    def configure(self, testenv, run_dir, run_node, rem_host):
        self.testenv = testenv
        self.rem_host = rem_host
        self._run_node = run_node
        self.run_dir = run_dir
        self.logger_name += self._run_node.run_addr()
        self.set_name(self.logger_name)

    def set_additional_params(self, proxy=None, port=None, user=None, password=None, server=None, auth=None, apn_type=None,
                              mmsc=None, mmsport=None, mmsproxy=None,  protocol=None, mvnoval=None, mvnotype=None):
        self.proxy = proxy
        self.port = port
        self.user = user
        self.password = password
        self.server = server
        self.auth = auth
        self.type = apn_type
        self.mmsc = mmsc
        self.mmsport = mmsport
        self.mmsproxy = mmsproxy
        self.protocol = protocol
        self.mvnoval = mvnoval
        self.mvnotype = mvnotype

    def set_apn(self):
        apn_params = {
            'carrier': self._apn_name,
            'apn': self._apn,
            'proxy': self.proxy or '',
            'port': self.port or '',
            'user': self.user or '',
            'password': self.password or '',
            'server': self.server or '',
            'mmsc': self.mmsc or '',
            'mmsport': self.mmsport or '',
            'mmsproxy': self.mmsproxy or '',
            'mcc': self._mcc,
            'mnc': self._mnc,
            'auth': self.auth or '-1',
            'type': self.type or 'default',
            'protocol': self.protocol or '',
            'mvnotype': self.mvnotype or '',
            'mvnoval': self.mvnoval or '',
        }
        self.dbg('APN parameters: ' + str(apn_params))

        # search for carrier in database
        carrier_id = self.get_carrier_id(apn_params['carrier'])

        # add/update carrier
        carrier_id = self.set_new_carrier(apn_params, carrier_id)

        # select as preferred APN
        if self.select:
            self.set_preferred_apn(carrier_id)

    def __str__(self):
        return self.name()

    def apn(self):
        return self._apn

    def mcc(self):
        return self._mcc

    def mnc(self):
        return self._mnc

    def select(self):
        return self._select

# vim: expandtab tabstop=4 shiftwidth=4