blob: 22e3102f29d2cf1969e955de1d71c953b3b5cde0 [file] [log] [blame]
#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import importlib
import logging
from antlion import tracelogger
MOBLY_CONTROLLER_CONFIG_NAME = "PduDevice"
ACTS_CONTROLLER_REFERENCE_NAME = "pdu_devices"
def create(configs):
"""Creates a PduDevice for each config in configs.
Args:
configs: List of configs from PduDevice field.
Fields:
device: a string "<brand>.<model>" that corresponds to module
in pdu_lib/
host: a string of the device ip address
username (optional): a string of the username for device sign-in
password (optional): a string of the password for device sign-in
Return:
A list of PduDevice objects.
"""
if configs:
pdus = []
for config in configs:
device = config.get("device")
if not device:
raise PduError("Config must provide a device")
host = config.get("host")
if not device:
raise PduError("Config must provide a host ip address")
username = config.get("username")
password = config.get("password")
pdu = _create_device(device, host, username, password)
pdus.append(pdu)
return pdus
def destroy(pdu_list):
"""Ensure any connections to devices are closed.
Args:
pdu_list: A list of PduDevice objects.
"""
for pdu in pdu_list:
pdu.close()
def get_info(pdu_list):
"""Retrieves info from a list of PduDevice objects.
Args:
pdu_list: A list of PduDevice objects.
Return:
A list containing a dictionary for each PduDevice, with keys:
'host': a string of the device ip address
'username': a string of the username
'password': a string of the password
"""
info = []
for pdu in pdu_list:
info.append(
{"host": pdu.host, "username": pdu.username, "password": pdu.password}
)
return info
def _create_device(device, host, username, password):
"""Factory method that returns an instance of PduDevice implementation
based on the device string.
"""
module_name = f"antlion.controllers.pdu_lib.{device}"
module = importlib.import_module(module_name)
return module.PduDevice(host, username, password)
def get_pdu_port_for_device(device_pdu_config, pdus):
"""Retrieves the pdu object and port of that PDU powering a given device.
This is especially necessary when there are multilpe devices on a single PDU
or multiple PDUs registered.
Args:
device_pdu_config: a dict, representing the config of the device.
pdus: a list of registered PduDevice objects.
Returns:
A tuple: (PduObject for the device, string port number on that PDU).
Raises:
ValueError, if there is no PDU matching the given host in the config.
Example ACTS config:
...
"testbed": [
...
"FuchsiaDevice": [
{
"ip": "<device_ip>",
"ssh_config": "/path/to/sshconfig",
"PduDevice": {
"host": "192.168.42.185",
"port": 2
}
}
],
"AccessPoint": [
{
"ssh_config": {
...
},
"PduDevice": {
"host": "192.168.42.185",
"port" 1
}
}
],
"PduDevice": [
{
"device": "synaccess.np02b",
"host": "192.168.42.185"
}
]
],
...
"""
pdu_ip = device_pdu_config["host"]
port = device_pdu_config["port"]
for pdu in pdus:
if pdu.host == pdu_ip:
return pdu, port
raise ValueError(f"No PduDevice with host: {pdu_ip}")
class PduDevice(object):
"""An object that defines the basic Pdu functionality and abstracts
the actual hardware.
This is a pure abstract class. Implementations should be of the same
class name (eg. class PduDevice(pdu.PduDevice)) and exist in
pdu_lib/<brand>/<device_name>.py. PduDevice objects should not be
instantiated by users directly.
"""
def __init__(self, host, username, password):
if type(self) is PduDevice:
raise NotImplementedError("Base class: cannot be instantiated directly")
self.host = host
self.username = username
self.password = password
self.log = tracelogger.TraceLogger(logging.getLogger())
def on_all(self):
"""Turns on all outlets on the device."""
raise NotImplementedError("Base class: cannot be called directly")
def off_all(self):
"""Turns off all outlets on the device."""
raise NotImplementedError("Base class: cannot be called directly")
def on(self, outlet):
"""Turns on specific outlet on the device.
Args:
outlet: a string of the outlet to turn on.
"""
raise NotImplementedError("Base class: cannot be called directly")
def off(self, outlet):
"""Turns off specific outlet on the device.
Args:
outlet: a string of the outlet to turn off.
"""
raise NotImplementedError("Base class: cannot be called directly")
def reboot(self, outlet):
"""Toggles a specific outlet on the device to off, then to on.
Args:
outlet: a string of the outlet to reboot.
"""
raise NotImplementedError("Base class: cannot be called directly")
def status(self):
"""Retrieves the status of the outlets on the device.
Return:
A dictionary matching outlet string to:
True: if outlet is On
False: if outlet is Off
"""
raise NotImplementedError("Base class: cannot be called directly")
def close(self):
"""Closes connection to the device."""
raise NotImplementedError("Base class: cannot be called directly")
class PduError(Exception):
"""An exception for use within PduDevice implementations"""