# osmo_gsm_tester: specifics for running an ip.access nanoBTS # # Copyright (C) 2018 by sysmocom - s.f.m.c. GmbH # # Author: Pau Espin Pedrol # # 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 . import os import pprint import tempfile import re from abc import ABCMeta, abstractmethod from . import log, config, util, template, process, pcap_recorder, bts, pcu from . import powersupply from .event_loop import MainLoop class NanoBts(bts.Bts): pwsup = None _pcu = None ############## # PROTECTED ############## def __init__(self, suite_run, conf): if conf.get('addr') is None: raise log.Error('No attribute addr provided in conf!') super().__init__(suite_run, conf, 'nanobts_%s' % conf.get('addr'), 'nanobts') def _configure(self): if self.bsc is None: raise log.Error('BTS needs to be added to a BSC or NITB before it can be configured') pwsup_opt = self.conf.get('power_supply', {}) if not pwsup_opt: raise log.Error('No power_supply attribute provided in conf!') pwsup_type = pwsup_opt.get('type') if not pwsup_type: raise log.Error('No type attribute provided in power_supply conf!') self.pwsup = powersupply.get_instance_by_type(pwsup_type, pwsup_opt) ######################## # PUBLIC - INTERNAL API ######################## def conf_for_bsc(self): values = self.conf_for_bsc_prepare() # Hack until we have proper ARFCN resource allocation support (OS#2230) band = values.get('band') trx_list = values.get('trx_list') if band == 'GSM-1900': config.overlay(trx_list[0], { 'arfcn' : '531' }) elif band == 'GSM-900': config.overlay(trx_list[0], { 'arfcn' : '50' }) config.overlay(values, { 'osmobsc_bts_type': 'nanobts' }) self.dbg(conf=values) return values def cleanup(self): if self.pwsup: self.dbg('Powering off NanoBTS') self.pwsup.power_set(False) ################### # PUBLIC (test API included) ################### def start(self): if self.conf.get('ipa_unit_id') is None: raise log.Error('No attribute %s provided in conf!' % attr) self.run_dir = util.Dir(self.suite_run.get_test_run_dir().new_dir(self.name())) self._configure() unitid = int(self.conf.get('ipa_unit_id')) bts_ip = self.remote_addr() # This fine for now, however concurrent tests using Nanobts may run into "address already in use" since dst is broadcast. # Once concurrency is needed, a new config attr should be added to have an extra static IP assigned on the main-unit to each Nanobts resource. local_bind_ip =util.dst_ip_get_local_bind(bts_ip) # Make sure nanoBTS is powered and in a clean state: self.pwsup.power_cycle(1.0) pcap_recorder.PcapRecorder(self.suite_run, self.run_dir.new_dir('pcap'), None, 'host %s and port not 22' % self.remote_addr()) self.log('Finding nanobts %s, binding on %s...' % (bts_ip, local_bind_ip)) ipfind = AbisIpFind(self.suite_run, self.run_dir, local_bind_ip, 'preconf') ipfind.start() ipfind.wait_bts_ready(bts_ip) running_unitid = ipfind.get_unitid_by_ip(bts_ip) self.log('Found nanobts %s with unit_id %d' % (bts_ip, running_unitid)) ipfind.stop() ipconfig = IpAccessConfig(self.suite_run, self.run_dir, bts_ip) if running_unitid != unitid: if not ipconfig.set_unit_id(unitid, False): raise log.Error('Failed configuring unit id %d' % unitid) # Apply OML IP and restart nanoBTS as it is required to apply the changes. if not ipconfig.set_oml_ip(self.bsc.addr(), True): raise log.Error('Failed configuring OML IP %s' % bts_ip) # Let some time for BTS to restart. It takes much more than 20 secs, and # this way we make sure we don't catch responses in abisip-find prior to # BTS restarting. MainLoop.sleep(self, 20) self.log('Starting to connect to', self.bsc) ipfind = AbisIpFind(self.suite_run, self.run_dir, local_bind_ip, 'postconf') ipfind.start() ipfind.wait_bts_ready(bts_ip) self.log('nanoBTS configured and running') ipfind.stop() MainLoop.wait(self, self.bsc.bts_is_connected, self, timeout=600) self.log('nanoBTS connected to BSC') #According to roh, it can be configured to use a static IP in a permanent way: # 1- use abisip-find to find the default address # 2- use ./ipaccess-config --ip-address IP/MASK # 3- use ./ipaccess-config --ip-gateway IP to set the IP of the main unit # 4- use ./ipaccess-config --restart to restart and apply the changes #Start must do the following: # 1- use abisip-find to find the default address # 2- use ./ipaccess-config --unit-id UNIT_ID # 3- use ./ipaccess-config --oml-ip --restart to set the IP of the BSC and apply+restart. # According to roh, using the 3 of them together was not reliable to work properly. def ready_for_pcu(self): """We don't really care as we use a Dummy PCU class.""" return True def pcu(self): if not self._pcu: self._pcu = pcu.PcuDummy(self.suite_run, self, self.conf) return self._pcu class AbisIpFind(log.Origin): suite_run = None parent_run_dir = None run_dir = None inst = None env = None bind_ip = None proc = None BIN_ABISIP_FIND = 'abisip-find' BTS_UNIT_ID_RE = re.compile("Unit_ID='(?P\d+)/\d+/\d+'") def __init__(self, suite_run, parent_run_dir, bind_ip, name_suffix): super().__init__(log.C_RUN, AbisIpFind.BIN_ABISIP_FIND + '-' + name_suffix) self.suite_run = suite_run self.parent_run_dir = parent_run_dir self.bind_ip = bind_ip self.env = {} def start(self): self.run_dir = util.Dir(self.parent_run_dir.new_dir(self.name())) self.inst = util.Dir(os.path.abspath(self.suite_run.trial.get_inst('osmo-bsc'))) lib = self.inst.child('lib') if not os.path.isdir(lib): raise log.Error('No lib/ in %r' % self.inst) ipfind_path = self.inst.child('bin', AbisIpFind.BIN_ABISIP_FIND) if not os.path.isfile(ipfind_path): raise RuntimeError('Binary missing: %r' % ipfind_path) env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) } self.proc = process.Process(self.name(), self.run_dir, (ipfind_path, '-i', '1', '-b', self.bind_ip), env=env) self.suite_run.remember_to_stop(self.proc) self.proc.launch() def stop(self): self.suite_run.stop_process(self.proc) def get_line_by_ip(self, ipaddr): """Get latest line (more up to date) from abisip-find based on ip address.""" token = "IP_Address='%s'" % ipaddr myline = None for line in (self.proc.get_stdout() or '').splitlines(): if token in line: myline = line return myline def get_unitid_by_ip(self, ipaddr): line = self.get_line_by_ip(ipaddr) if line is None: return None res = AbisIpFind.BTS_UNIT_ID_RE.search(line) if res: unit_id = int(res.group('unit_id')) return unit_id raise log.Error('abisip-find unit_id field for nanobts %s not found in %s' %(ipaddr, line)) def bts_ready(self, ipaddr): return self.get_line_by_ip(ipaddr) is not None def wait_bts_ready(self, ipaddr): MainLoop.wait(self, self.bts_ready, ipaddr) # There's a period of time after boot in which nanobts answers to # abisip-find but tcp RSTs ipacces-config conns. Let's wait in here a # bit more time to avoid failing after stating the BTS is ready. MainLoop.sleep(self, 2) class IpAccessConfig(log.Origin): suite_run = None parent_run_dir = None run_dir = None inst = None env = None bts_ip = None BIN_IPACCESS_CONFIG = 'ipaccess-config' def __init__(self, suite_run, parent_run_dir, bts_ip): super().__init__(log.C_RUN, IpAccessConfig.BIN_IPACCESS_CONFIG) self.suite_run = suite_run self.parent_run_dir = parent_run_dir self.bts_ip = bts_ip self.env = {} def launch_process(self, binary_name, *args): binary = os.path.abspath(self.inst.child('bin', binary_name)) run_dir = self.run_dir.new_dir(binary_name) if not os.path.isfile(binary): raise RuntimeError('Binary missing: %r' % binary) proc = process.Process(binary_name, run_dir, (binary,) + args, env=self.env) proc.launch() return proc def run(self, name_suffix, *args): self.run_dir = util.Dir(self.parent_run_dir.new_dir(self.name()+'-'+name_suffix)) self.inst = util.Dir(os.path.abspath(self.suite_run.trial.get_inst('osmo-bsc'))) lib = self.inst.child('lib') self.env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) } self.proc = self.launch_process(IpAccessConfig.BIN_IPACCESS_CONFIG, *args) try: MainLoop.wait(self, self.proc.terminated) except Exception as e: self.proc.terminate() raise e return self.proc.result def set_unit_id(self, unitid, restart=False): if restart: retcode = self.run('unitid', '--restart', '--unit-id', '%d/0/0' % unitid, self.bts_ip) else: retcode = self.run('unitid', '--unit-id', '%d/0/0' % unitid, self.bts_ip) if retcode != 0: log.err('ipaccess-config --unit-id %d/0/0 returned error code %d' % (unitid, retcode)) return retcode == 0 def set_oml_ip(self, omlip, restart=False): if restart: retcode = self.run('oml', '--restart', '--oml-ip', omlip, self.bts_ip) else: retcode = self.run('oml', '--oml-ip', omlip, self.bts_ip) if retcode != 0: self.error('ipaccess-config --oml-ip %s returned error code %d' % (omlip, retcode)) return retcode == 0 # vim: expandtab tabstop=4 shiftwidth=4