python: fix flake8 warnings and format with black
Most of the flake8 warnings in this repository were fairly trivial,
so fixed them. The "openbmctool" is 7000+ lines of pretty heavily
warned code, so just disabling that one. Format everything with
black.
Signed-off-by: Patrick Williams <patrick@stwcx.xyz>
Change-Id: Icb3f6ee9bf03dece58785f7af00617c87a84aa65
diff --git a/reboot-ping-pong/rpp b/reboot-ping-pong/rpp
index 251e250..78009c2 100755
--- a/reboot-ping-pong/rpp
+++ b/reboot-ping-pong/rpp
@@ -1,20 +1,20 @@
#!/usr/bin/python3
-import argparse
-import pexpect
import sys
import time
-
from collections import namedtuple
+import pexpect
+
Endpoint = namedtuple("Endpoint", "host, port")
Credentials = namedtuple("Credentials", "username, password")
Target = namedtuple("Target", "credentials, endpoint")
Entity = namedtuple("Entity", "console, ssh")
Machine = namedtuple("Machine", "bmc, host")
+
class Obmcutil(object):
- BMC_READY = "xyz.openbmc_project.State.BMC.BMCState.Ready"
+ BMC_READY = "xyz.openbmc_project.State.BMC.BMCState.Ready"
BMC_NOT_READY = "xyz.openbmc_project.State.BMC.BMCState.NotReady"
HOST_OFF = "xyz.openbmc_project.State.Host.HostState.Off"
@@ -37,10 +37,17 @@
return rc
def hoststate(self):
- return self._state("hoststate", "xyz\.openbmc_project\.State\.Host\.HostState\.(Off|Running|Quiesced)")
+ return self._state(
+ "hoststate",
+ "xyz\\.openbmc_project\\.State\\.Host\\.HostState\\."
+ + "(Off|Running|Quiesced)",
+ )
def bmcstate(self):
- return self._state("bmcstate", "xyz\.openbmc_project\.State\.BMC\.BMCState\.(Not)?Ready")
+ return self._state(
+ "bmcstate",
+ "xyz\\.openbmc_project\\.State\\.BMC\\.BMCState\\.(Not)?Ready",
+ )
def poweron(self):
self.session.sendline("obmcutil -w poweron")
@@ -50,6 +57,7 @@
self.session.sendline("obmcutil -w chassisoff")
self.session.expect(self.prompt)
+
class PexpectLogger(object):
def write(self, bstring):
try:
@@ -76,19 +84,29 @@
self.login()
def login(self):
- self.session.sendline(self.entity.console.credentials.username.encode())
+ self.session.sendline(
+ self.entity.console.credentials.username.encode()
+ )
self.session.expect("Password: ".encode())
- self.session.sendline(self.entity.console.credentials.password.encode())
+ self.session.sendline(
+ self.entity.console.credentials.password.encode()
+ )
self.session.expect(self.shell)
def reboot(self):
self.session.sendline("reboot")
- self.session.expect("Hit any key to stop autoboot:".encode(), timeout=None)
+ self.session.expect(
+ "Hit any key to stop autoboot:".encode(), timeout=None
+ )
self.session.expect(self.getty, timeout=None)
self.login()
state = self.obmcutil.bmcstate()
while state != self.obmcutil.BMC_READY:
- print("Wanted state '{}', got state '{}'".format(self.obmcutil.BMC_READY, state))
+ print(
+ "Wanted state '{}', got state '{}'".format(
+ self.obmcutil.BMC_READY, state
+ )
+ )
time.sleep(5)
state = self.obmcutil.bmcstate()
@@ -102,6 +120,7 @@
self.obmcutil.chassisoff()
self.obmcutil.poweron()
+
class Host(object):
def __init__(self, entity, bmc):
self.shell = "/? *#".encode()
@@ -112,35 +131,42 @@
self.connect()
def connect(self):
- fargs = (self.entity.console.endpoint.port,
- self.entity.console.credentials.username,
- self.entity.console.endpoint.host)
+ fargs = (
+ self.entity.console.endpoint.port,
+ self.entity.console.credentials.username,
+ self.entity.console.endpoint.host,
+ )
self.session = pexpect.spawn("ssh -p{} {}@{}".format(*fargs))
self.session.logfile = PexpectLogger()
self.session.expect("password:".encode())
- self.session.sendline(self.entity.console.credentials.password.encode())
+ self.session.sendline(
+ self.entity.console.credentials.password.encode()
+ )
def poweron(self):
self.bmc.chassisoff()
self.bmc.poweron()
- self.session.send('\f')
+ self.session.send("\f")
rc = self.session.expect([self.petitboot, self.shell], timeout=None)
if rc == 0:
self.session.sendline()
self.session.expect(self.shell)
def reboot(self):
- self.session.send('\f')
+ self.session.send("\f")
rc = self.session.expect([self.petitboot, self.shell], timeout=None)
if rc == 0:
self.session.sendline()
self.session.expect(self.shell)
self.session.sendline("reboot".encode())
- self.session.expect("INIT: Waiting for kernel...".encode(), timeout=None)
+ self.session.expect(
+ "INIT: Waiting for kernel...".encode(), timeout=None
+ )
self.session.expect("Petitboot".encode(), timeout=None)
self.session.sendline()
self.session.expect(self.shell)
+
def rpp(machine):
bmc = Bmc(machine.bmc)
host = Host(machine.host, bmc)
@@ -150,14 +176,22 @@
host.connect()
host.reboot()
+
def main():
bmccreds = Credentials("root", "0penBmc")
- b = Entity(Target(bmccreds, Endpoint("serial.concentrator.somewhere.com", 1234)),
- Target(bmccreds, Endpoint("bmc.somewhere.com", 22)))
- h = Entity(Target(bmccreds, Endpoint("bmc.somewhere.com", 2200)),
- Target(Credentials("user", "password"), Endpoint("host.somewhere.com", 22)))
+ b = Entity(
+ Target(bmccreds, Endpoint("serial.concentrator.somewhere.com", 1234)),
+ Target(bmccreds, Endpoint("bmc.somewhere.com", 22)),
+ )
+ h = Entity(
+ Target(bmccreds, Endpoint("bmc.somewhere.com", 2200)),
+ Target(
+ Credentials("user", "password"), Endpoint("host.somewhere.com", 22)
+ ),
+ )
m = Machine(b, h)
return rpp(m)
+
if __name__ == "__main__":
main()