conman/test/fake_python/subprocess/cwmp.py

#!/usr/bin/python

"""Fake catawampus implementation."""

import logging
import os

import connection_check


logger = logging.getLogger('subprocess.cwmp')

CONMAN_CONFIG_PATH = None
CWMP_PATH = None
CONFIG = {}
ACCESS_POINT = {}
ACS_SESSION_FAILS = False


def call(command, env=None):
  if command == 'wakeup':
    if not CONMAN_CONFIG_PATH:
      raise ValueError('Call subprocess.set_conman_paths before calling '
                       '"cwmp wakeup".')

    write_acscontact()

    if ACS_SESSION_FAILS:
      return 0, ''

    if ((env and 'write_now_testonly' in env) or
        [result for result in connection_check.RESULTS.itervalues()
         if result in ('restricted', 'succeed')]):
      for band in ('2.4', '5'):
        if CONFIG.get(band, None):
          write_wlan_config(band)
        else:
          delete_wlan_config(band)
          disable_access_point(band)

        if ACCESS_POINT.get(band, False):
          enable_access_point(band)
        else:
          disable_access_point(band)

      logger.debug('Fake ACS session completing')
      write_acsconnected()
    else:
      logger.debug('ACS session failed due to no working connections')

    return 0, ''

  raise ValueError('Fake cwmp only supports "wakeup" command.')


def wlan_config_filename(band):
  return os.path.join(CONMAN_CONFIG_PATH, 'command.%s' % band)


def access_point_filename(band):
  return os.path.join(CONMAN_CONFIG_PATH, 'access_point.%s' % band)


def write_wlan_config(band):
  final_filename = wlan_config_filename(band)
  logger.debug('Writing config for band %s: %s', band, final_filename)
  # We don't care which writes are atomic, as long as some but not all are.
  # Making it depend on band achieves this.
  atomic = band == '2.4'
  filename = final_filename + ('.tmp' if atomic else '')
  with open(filename, 'w') as f:
    f.write('\n'.join(['env', 'WIFI_PSK=%s' % CONFIG[band]['psk'],
                       'wifi', 'set', '--band', band,
                       '--ssid', CONFIG[band]['ssid']]))
    logger.debug(  'wrote to filename %s', filename)
  if atomic:
    logger.debug(  'moving from %s to %s', filename, final_filename)
    os.rename(filename, final_filename)


def enable_access_point(band):
  logger.debug('Enabling AP for band %s', band)
  open(access_point_filename(band), 'w')


def delete_wlan_config(band):
  config_filename = wlan_config_filename(band)
  if os.path.exists(config_filename):
    logger.debug('Deleting config for band %s', band)
    os.unlink(config_filename)


def disable_access_point(band):
  ap_filename = access_point_filename(band)
  if os.path.isfile(ap_filename):
    logger.debug('Disabling AP for band %s', band)
    os.unlink(ap_filename)


def write_acscontact():
  logger.debug('ACS session started')
  open(os.path.join(CWMP_PATH, 'acscontact'), 'w')


def write_acsconnected():
  logger.debug('ACS session completed')
  open(os.path.join(CWMP_PATH, 'acsconnected'), 'w')


def mock(band, access_point=None, delete_config=False, ssid=None, psk=None,
         write_now=False, acs_session_fails=None):
  """Mock the config written by catawampus.

  Args:
    band:  The band for which things are being mocked.
    access_point:  Set to True or False to enable/disable the AP.
    delete_config:  Set to True to delete the config.
    ssid:  If updating config, the ssid to use.  psk must also be set.
    psk:  If updating config, the psk to use.  ssid must also be set.
    write_now:  If updating config, write it immediately.

  Raises:
    ValueError:  If invalid values are specified.
  """
  if acs_session_fails is not None:
    global ACS_SESSION_FAILS
    ACS_SESSION_FAILS = acs_session_fails

  if access_point is not None:
    if access_point not in (True, False):
      raise ValueError('access_point should only be mocked as True/False')
    ACCESS_POINT[band] = access_point
    logger.debug('AP mocked %s', access_point)

  if delete_config:
    logger.debug('Config mock removed for band %s', band)
    CONFIG[band] = None
  elif ssid and psk:
    logger.debug('Config mock updated for band %s', band)
    CONFIG[band] = {'ssid': ssid, 'psk': psk}
  elif ssid or psk:
    raise ValueError('Cannot set only one of ssid (%r) and psk (%r).',
                     ssid, psk)

  if write_now:
    call('wakeup', env={'write_now_testonly': True})