Add initial fixture and fix all lint issues

.codespell-whitelist

@@ -4,3 +4,4 @@ SOM
 som
 sinc
 ser
+hsi

.pre-commit-config.yaml

@@ -1,15 +1,15 @@
 repos:
 -   repo: https://github.com/codespell-project/codespell
-    rev: v1.15.0
+    rev: v2.2.6
     hooks:
     -   id: codespell
         args: [--ignore-words=.codespell-whitelist]
--   repo: https://github.com/pre-commit/mirrors-isort
-    rev: v4.3.4
+-   repo: https://github.com/pycqa/isort
+    rev: 5.13.2
     hooks:
-    -   id: isort
+      - id: isort
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.720
+    rev: v1.7.1
     hooks:
     -   id: mypy
         args: [--no-strict-optional, --ignore-missing-imports]

nebula/__init__.py

@@ -3,22 +3,20 @@
 # print(f.renderText("Nebula"))
 import os
 
-from nebula.netconsole import netconsole
-from nebula.uart import uart
-from nebula.tftpboot import tftpboot
-from nebula.pdu import pdu
-from nebula.manager import manager
-from nebula.network import network
-from nebula.driver import driver
 from nebula.builder import builder
-from nebula.common import utils
-from nebula.helper import helper
-from nebula.downloader import downloader
+from nebula.common import LINUX_DEFAULT_PATH, utils
 from nebula.coverage import coverage
+from nebula.downloader import downloader
+from nebula.driver import driver
+from nebula.helper import helper
 from nebula.jtag import jtag
+from nebula.manager import manager
 from nebula.netbox import netbox
-
-from nebula.common import LINUX_DEFAULT_PATH
+from nebula.netconsole import netconsole
+from nebula.network import network
+from nebula.pdu import pdu
+from nebula.tftpboot import tftpboot
+from nebula.uart import uart
 
 if os.name in ["nt", "posix"] and os.path.exists(LINUX_DEFAULT_PATH):
     from nebula.usbmux import usbmux

nebula/common.py

@@ -1,9 +1,10 @@
 import logging
 import os
 
-import nebula.errors as ne
 import yaml
 
+import nebula.errors as ne
+
 LINUX_DEFAULT_PATH = "/etc/default/nebula"
 WINDOWS_DEFAULT_PATH = "C:\\nebula\\nebula.yaml"
 

nebula/downloader.py

@@ -16,11 +16,12 @@
 from artifactory import ArtifactoryPath
 from bs4 import BeautifulSoup
 from github import Github
-from nebula.common import utils
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
 from tqdm import tqdm
 
+from nebula.common import utils
+
 log = logging.getLogger(__name__)
 
 

nebula/fixtures.py

@@ -0,0 +1,94 @@
+import logging
+import os
+
+import pytest
+
+import nebula
+
+logging.getLogger().setLevel(logging.INFO)
+
+
+class MyFilter(logging.Filter):
+    def filter(self, record):
+        return "nebula" in record.name
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("nebula")
+    group.addoption(
+        "--enable-update",
+        action="store_true",
+        dest="enable_update",
+        default=False,
+        help="Update boot files and reboot board(s) with nebula",
+    )
+    group.addoption(
+        "--nb-log-level",
+        action="store",
+        dest="nb_log_level",
+        default="ERROR",
+        help="Set log level for nebula",
+    )
+
+
+@pytest.fixture(scope="function")
+def sd_card_update_boot(request):
+    """pytest fixture to update SD card and reboot board(s) with nebula"""
+    enable_update = request.config.getoption("--enable-update")
+    if not enable_update:
+        yield
+        return
+
+    marker = request.node.get_closest_marker("nebula_update_boot")
+
+    if not marker:
+        yield
+        return
+
+    board_name = marker.args[0]
+    if not hasattr(pytest, "nebula_boards_booted"):
+        pytest.nebula_boards_booted = []
+    if board_name not in pytest.nebula_boards_booted:
+        pytest.nebula_boards_booted.append(board_name)
+
+        print("Running boot test for board: " + board_name)
+
+        # board_name = "zynq-adrv9364"
+        yamlfilename = "/tmp/hw_test/test.yaml"
+        folder = "/tmp/hw_test/boot_files"
+
+        level = request.config.getoption("--nb-log-level")
+        if level not in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
+            raise ValueError("Invalid nebula log level: " + level)
+
+        log = logging.getLogger("nebula")
+        log.setLevel(getattr(logging, level))
+        log = logging.getLogger()
+        root_handler = log.handlers[0]
+        root_handler.addFilter(MyFilter())
+        root_handler.setFormatter(
+            logging.Formatter("%(levelname)s | %(name)s : %(message)s")
+        )
+
+        # Update SD card over networking
+        m = nebula.manager(configfilename=yamlfilename, board_name=board_name)
+
+        m.board_reboot_auto_folder(folder, design_name=board_name)
+
+    else:
+        print("Board already booted: " + board_name)
+
+    yield
+
+    print("Generated log files:")
+
+
+# # Example use
+
+# @pytest.mark.nebula_update_boot("zynq-adrv9364")
+# def test_boot_hw(sd_card_update_boot):
+
+#     import iio
+#     ctx = iio.Context("ip:analog.local")
+#     for dev in ctx.devices:
+#         print(dev.name)

nebula/helper.py

@@ -4,9 +4,10 @@
 import pathlib
 
 import click
-import nebula.errors as ne
 import netifaces
 import yaml
+
+import nebula.errors as ne
 from nebula.common import multi_device_check
 from nebula.netbox import NetboxDevice, NetboxDevices, netbox
 

nebula/main.py

@@ -1,4 +1,5 @@
 from invoke import Collection, Program
+
 from nebula import tasks
 
 program = Program(namespace=Collection.from_module(tasks), version="0.0.1")

nebula/manager.py

@@ -4,10 +4,11 @@
 import tarfile
 import time
 
+import yaml
+
 import nebula.common as common
 import nebula.errors as ne
 import nebula.helper as helper
-import yaml
 from nebula.driver import driver
 from nebula.jtag import jtag
 from nebula.netconsole import netconsole
@@ -196,7 +197,7 @@ def network_check(self):
             self.monitor[0].stop_log()
             raise ne.SSHNotFunctionalAfterBootFileUpdate
 
-    @_release_thread_lock
+    @_release_thread_lock  # type: ignore
     def recover_board(
         self, system_top_bit_path, bootbinpath, uimagepath, devtreepath, sdcard=False
     ):
@@ -309,7 +310,7 @@ def recover_board(
                     bootbinpath, uimagepath, devtreepath, sdcard
                 )
 
-    @_release_thread_lock
+    @_release_thread_lock  # type: ignore
     def board_reboot_jtag_uart(
         self, bootbinpath, uimagepath, devtreepath, sdcard=False
     ):
@@ -322,7 +323,7 @@ def board_reboot_jtag_uart(
         # Check if u-boot loads first
         # log.info("Resetting with JTAG and checking if u-boot is reachable")
         # self.jtag.restart_board()
-        # do a power cylcle rather than jtag reboot to make sure jtag devices are working
+        # do a power cycle rather than jtag reboot to make sure jtag devices are working
         log.info("Resetting with JTAG and checking if u-boot is reachable")
         self.jtag.restart_board()
         if self.monitor[0]._enter_uboot_menu_from_power_cycle():
@@ -362,7 +363,7 @@ def board_reboot_jtag_uart(
 
         self.monitor[0].stop_log()
 
-    @_release_thread_lock
+    @_release_thread_lock  # type: ignore
     def board_reboot_uart_net_pdu(
         self, system_top_bit_path, bootbinpath, uimagepath, devtreepath
     ):
@@ -642,7 +643,9 @@ def board_reboot_auto_folder(
         else:
             log.info("SD-Card/microblaze based device selected")
             (bootbin, kernel, dt, bit) = self._find_boot_files(folder)
-            print(bootbin, kernel, dt, bit)
+            log.info("Found boot files:")
+            for file in [bootbin, kernel, dt, bit]:
+                log.info(file)
             if jtag_mode:
                 self.board_reboot_jtag_uart(
                     system_top_bit_path=bit,

nebula/netbox.py

@@ -7,9 +7,10 @@
 
 import pynetbox
 import yaml
-from nebula.common import utils
 from numpy import isin
 
+from nebula.common import utils
+
 log = logging.getLogger(__name__)
 
 

nebula/network.py

@@ -8,8 +8,9 @@
 import time
 
 import fabric
-import nebula.errors as ne
 from fabric import Connection
+
+import nebula.errors as ne
 from nebula.common import utils
 
 log = logging.getLogger(__name__)
@@ -85,7 +86,13 @@ def check_ssh(self):
                 result = fabric.Connection(
                     self.dutusername + "@" + self.dutip,
                     connect_kwargs={"password": self.dutpassword},
-                ).run("uname -a", hide=True, timeout=self.ssh_timeout)
+                ).run(
+                    "uname -a",
+                    hide=True,
+                    timeout=self.ssh_timeout,
+                    pty=True,
+                    in_stream=False,
+                )
                 break
             except Exception as ex:
                 log.warning("Exception raised: " + str(ex))
@@ -146,7 +153,13 @@ def run_ssh_command(
                 result = fabric.Connection(
                     self.dutusername + "@" + self.dutip,
                     connect_kwargs={"password": self.dutpassword},
-                ).run(command, hide=True, timeout=self.ssh_timeout)
+                ).run(
+                    command,
+                    hide=True,
+                    timeout=self.ssh_timeout,
+                    pty=True,
+                    in_stream=False,
+                )
                 if result.failed:
                     raise Exception("Failed to run command:", command)
 

nebula/pdu.py

@@ -1,9 +1,10 @@
 import logging
 import time
 
+from pyvesync_v2 import VeSync
+
 from nebula import cyberpower as cpdu
 from nebula.common import utils
-from pyvesync_v2 import VeSync
 
 log = logging.getLogger(__name__)
 

nebula/resources/template_gen.yaml

@@ -26,7 +26,7 @@ board-config:
   field_3:
     name: daughter
     default: FMCOMMS2-3
-    help: "Daugther board name"
+    help: "Daughter board name"
     optional: False
     netbox_field: devices.custom_fields.device_daughter
   field_4:

nebula/resources/template_rpi_gen.yaml

@@ -25,7 +25,7 @@ board-config:
     netbox_field: devices.custom_fields.device_carrier
   field_3:
     name: daughter
-    help: "Daugther board name"
+    help: "Daughter board name"
     optional: True
     netbox_field: devices.custom_fields.device_daughter
   field_4:

nebula/tasks.py

@@ -3,10 +3,11 @@
 import time
 from operator import truediv
 
-import nebula
 import yaml
 from invoke import Collection, task
 
+import nebula
+
 logging.getLogger().setLevel(logging.WARNING)
 
 

nebula/uart.py

@@ -9,9 +9,10 @@
 
 import serial
 import xmodem
-from nebula.common import utils
 from tqdm import tqdm
 
+from nebula.common import utils
+
 log = logging.getLogger(__name__)
 
 LINUX_SERIAL_FOLDER = "/dev/serial"
@@ -132,7 +133,6 @@ def start_log(self, logappend=False, force=False):
         """Trigger monitoring with UART interface"""
         if not self.listen_thread_run or force:
             self.listen_thread_run = True
-            print("STARTING UART LOG")
             log.info("Launching UART listening thread")
             if not self.print_to_console:
                 log.info("UART console saving to file: " + self.logfilename)

nebula/usbmux.py

@@ -9,9 +9,10 @@
 from pathlib import Path
 
 import pyudev
-from nebula.common import utils
 from usbsdmux import usbsdmux
 
+from nebula.common import utils
+
 log = logging.getLogger(__name__)