diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 78d9d21..4e52f53 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -9,7 +9,7 @@ ci: repos: - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v5.0.0 + rev: v6.0.0 hooks: - id: check-added-large-files - id: check-merge-conflict @@ -20,53 +20,53 @@ repos: - id: mixed-line-ending - repo: https://github.com/talos-systems/conform - rev: v0.1.0-alpha.30 + rev: v0.1.0-alpha.31 hooks: - id: conform stages: - commit-msg - repo: https://github.com/codespell-project/codespell - rev: v2.3.0 + rev: v2.4.3 hooks: - id: codespell - repo: https://github.com/adrienverge/yamllint - rev: v1.35.1 + rev: v1.38.0 hooks: - id: yamllint args: [-c=.yamllint] - repo: https://github.com/igorshubovych/markdownlint-cli - rev: v0.43.0 + rev: v0.49.1 hooks: - id: markdownlint - id: markdownlint-fix - repo: https://github.com/koalaman/shellcheck-precommit - rev: v0.10.0 + rev: v0.11.0 hooks: - id: shellcheck args: ["--severity=warning"] - repo: https://github.com/pycqa/isort - rev: 5.13.2 + rev: 9.0.0b1 hooks: - id: isort name: isort (python) - repo: https://github.com/psf/black - rev: 24.10.0 + rev: 26.5.1 hooks: - id: black args: ["--line-length", "79"] - - repo: https://github.com/MarketSquare/robotframework-tidy - rev: 4.15.0 - hooks: - - id: robotidy - - repo: https://github.com/MarketSquare/robotframework-robocop - rev: 5.7.0 + rev: v8.4.1 hooks: - id: robocop + additional_dependencies: + - typing_extensions + - id: robocop-format + additional_dependencies: + - typing_extensions diff --git a/osfv_cli/poetry.lock b/osfv_cli/poetry.lock index 91b3fbf..84f1cf8 100644 --- a/osfv_cli/poetry.lock +++ b/osfv_cli/poetry.lock @@ -1,4 +1,16 @@ -# This file is automatically @generated by Poetry 1.8.5 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand. + +[[package]] +name = "annotated-doc" +version = "0.0.4" +description = "Document parameters, class attributes, return types, and variables inline, with Annotated." +optional = false +python-versions = ">=3.8" +groups = ["main"] +files = [ + {file = "annotated_doc-0.0.4-py3-none-any.whl", hash = "sha256:571ac1dc6991c450b25a9c2d84a3705e2ae7a53467b5d111c24fa8baabbed320"}, + {file = "annotated_doc-0.0.4.tar.gz", hash = "sha256:fbcda96e87e9c92ad167c2e53839e57503ecfda18804ea28102353485033faa4"}, +] [[package]] name = "bcrypt" @@ -6,6 +18,7 @@ version = "4.3.0" description = "Modern password hashing for your software and your servers" optional = false python-versions = ">=3.8" +groups = ["main"] files = [ {file = "bcrypt-4.3.0-cp313-cp313t-macosx_10_12_universal2.whl", hash = "sha256:f01e060f14b6b57bbb72fc5b4a83ac21c443c9a2ee708e04a10e9192f90a6281"}, {file = "bcrypt-4.3.0-cp313-cp313t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c5eeac541cefd0bb887a371ef73c62c3cd78535e4887b310626036a7c0a817bb"}, @@ -70,6 +83,7 @@ version = "2025.7.14" description = "Python package for providing Mozilla's CA Bundle." optional = false python-versions = ">=3.7" +groups = ["main"] files = [ {file = "certifi-2025.7.14-py3-none-any.whl", hash = "sha256:6b31f564a415d79ee77df69d757bb49a5bb53bd9f756cbbe24394ffd6fc1f4b2"}, {file = "certifi-2025.7.14.tar.gz", hash = "sha256:8ea99dbdfaaf2ba2f9bac77b9249ef62ec5218e7c2b2e903378ed5fccf765995"}, @@ -81,6 +95,7 @@ version = "1.17.1" description = "Foreign Function Interface for Python calling C code." optional = false python-versions = ">=3.8" +groups = ["main"] files = [ {file = "cffi-1.17.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:df8b1c11f177bc2313ec4b2d46baec87a5f3e71fc8b45dab2ee7cae86d9aba14"}, {file = "cffi-1.17.1-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8f2cdc858323644ab277e9bb925ad72ae0e67f69e804f4898c070998d50b1a67"}, @@ -160,6 +175,7 @@ version = "3.4.2" description = "The Real First Universal Charset Detector. Open, modern and actively maintained alternative to Chardet." optional = false python-versions = ">=3.7" +groups = ["main"] files = [ {file = "charset_normalizer-3.4.2-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:7c48ed483eb946e6c04ccbe02c6b4d1d48e51944b6db70f697e089c193404941"}, {file = "charset_normalizer-3.4.2-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b2d318c11350e10662026ad0eb71bb51c7812fc8590825304ae0bdd4ac283acd"}, @@ -255,12 +271,26 @@ files = [ {file = "charset_normalizer-3.4.2.tar.gz", hash = "sha256:5baececa9ecba31eff645232d59845c07aa030f0c81ee70184a90d35099a0e63"}, ] +[[package]] +name = "colorama" +version = "0.4.6" +description = "Cross-platform colored terminal text." +optional = false +python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7" +groups = ["main"] +markers = "platform_system == \"Windows\"" +files = [ + {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"}, + {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, +] + [[package]] name = "cryptography" version = "45.0.5" description = "cryptography is a package which provides cryptographic recipes and primitives to Python developers." optional = false python-versions = "!=3.9.0,!=3.9.1,>=3.7" +groups = ["main"] files = [ {file = "cryptography-45.0.5-cp311-abi3-macosx_10_9_universal2.whl", hash = "sha256:101ee65078f6dd3e5a028d4f19c07ffa4dd22cce6a20eaa160f8b5219911e7d8"}, {file = "cryptography-45.0.5-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:3a264aae5f7fbb089dbc01e0242d3b67dffe3e6292e1f5182122bdf58e65215d"}, @@ -305,10 +335,10 @@ files = [ cffi = {version = ">=1.14", markers = "platform_python_implementation != \"PyPy\""} [package.extras] -docs = ["sphinx (>=5.3.0)", "sphinx-inline-tabs", "sphinx-rtd-theme (>=3.0.0)"] +docs = ["sphinx (>=5.3.0)", "sphinx-inline-tabs ; python_full_version >= \"3.8.0\"", "sphinx-rtd-theme (>=3.0.0) ; python_full_version >= \"3.8.0\""] docstest = ["pyenchant (>=3)", "readme-renderer (>=30.0)", "sphinxcontrib-spelling (>=7.3.1)"] -nox = ["nox (>=2024.4.15)", "nox[uv] (>=2024.3.2)"] -pep8test = ["check-sdist", "click (>=8.0.1)", "mypy (>=1.4)", "ruff (>=0.3.6)"] +nox = ["nox (>=2024.4.15)", "nox[uv] (>=2024.3.2) ; python_full_version >= \"3.8.0\""] +pep8test = ["check-sdist ; python_full_version >= \"3.8.0\"", "click (>=8.0.1)", "mypy (>=1.4)", "ruff (>=0.3.6)"] sdist = ["build (>=1.0.0)"] ssh = ["bcrypt (>=3.1.5)"] test = ["certifi (>=2024)", "cryptography-vectors (==45.0.5)", "pretend (>=0.7)", "pytest (>=7.4.0)", "pytest-benchmark (>=4.0)", "pytest-cov (>=2.10.1)", "pytest-xdist (>=3.5.0)"] @@ -320,6 +350,7 @@ version = "3.10" description = "Internationalized Domain Names in Applications (IDNA)" optional = false python-versions = ">=3.6" +groups = ["main"] files = [ {file = "idna-3.10-py3-none-any.whl", hash = "sha256:946d195a0d259cbba61165e88e65941f16e9b36ea6ddb97f00452bae8b1287d3"}, {file = "idna-3.10.tar.gz", hash = "sha256:12f65c9b470abda6dc35cf8e63cc574b1c52b11df2c86030af0ac09b01b13ea9"}, @@ -334,25 +365,63 @@ version = "6.5.2" description = "Read resources from Python packages" optional = false python-versions = ">=3.9" +groups = ["main"] files = [ {file = "importlib_resources-6.5.2-py3-none-any.whl", hash = "sha256:789cfdc3ed28c78b67a06acb8126751ced69a3d5f79c095a98298cd8a760ccec"}, {file = "importlib_resources-6.5.2.tar.gz", hash = "sha256:185f87adef5bcc288449d98fb4fba07cea78bc036455dd44c5fc4a2fe78fed2c"}, ] [package.extras] -check = ["pytest-checkdocs (>=2.4)", "pytest-ruff (>=0.2.1)"] +check = ["pytest-checkdocs (>=2.4)", "pytest-ruff (>=0.2.1) ; sys_platform != \"cygwin\""] cover = ["pytest-cov"] doc = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-lint"] enabler = ["pytest-enabler (>=2.2)"] test = ["jaraco.test (>=5.4)", "pytest (>=6,!=8.1.*)", "zipp (>=3.17)"] type = ["pytest-mypy"] +[[package]] +name = "markdown-it-py" +version = "4.2.0" +description = "Python port of markdown-it. Markdown parsing, done right!" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a"}, + {file = "markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49"}, +] + +[package.dependencies] +mdurl = ">=0.1,<1.0" + +[package.extras] +benchmarking = ["psutil", "pytest", "pytest-benchmark"] +compare = ["commonmark (>=0.9,<1.0)", "markdown (>=3.4,<4.0)", "markdown-it-pyrs", "mistletoe (>=1.0,<2.0)", "mistune (>=3.0,<4.0)", "panflute (>=2.3,<3.0)"] +linkify = ["linkify-it-py (>=1,<3)"] +plugins = ["mdit-py-plugins (>=0.5.0)"] +profiling = ["gprof2dot"] +rtd = ["ipykernel", "jupyter_sphinx", "mdit-py-plugins (>=0.5.0)", "myst-parser", "pyyaml", "sphinx", "sphinx-book-theme (>=1.0,<2.0)", "sphinx-copybutton", "sphinx-design"] +testing = ["coverage", "pytest", "pytest-cov", "pytest-regressions", "pytest-timeout", "requests"] + +[[package]] +name = "mdurl" +version = "0.1.2" +description = "Markdown URL utilities" +optional = false +python-versions = ">=3.7" +groups = ["main"] +files = [ + {file = "mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8"}, + {file = "mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba"}, +] + [[package]] name = "paramiko" version = "3.5.1" description = "SSH2 protocol library" optional = false python-versions = ">=3.6" +groups = ["main"] files = [ {file = "paramiko-3.5.1-py3-none-any.whl", hash = "sha256:43b9a0501fc2b5e70680388d9346cf252cfb7d00b0667c39e80eb43a408b8f61"}, {file = "paramiko-3.5.1.tar.gz", hash = "sha256:b2c665bc45b2b215bd7d7f039901b14b067da00f3a11e6640995fd58f2664822"}, @@ -364,8 +433,8 @@ cryptography = ">=3.3" pynacl = ">=1.5" [package.extras] -all = ["gssapi (>=1.4.1)", "invoke (>=2.0)", "pyasn1 (>=0.1.7)", "pywin32 (>=2.1.8)"] -gssapi = ["gssapi (>=1.4.1)", "pyasn1 (>=0.1.7)", "pywin32 (>=2.1.8)"] +all = ["gssapi (>=1.4.1) ; platform_system != \"Windows\"", "invoke (>=2.0)", "pyasn1 (>=0.1.7)", "pywin32 (>=2.1.8) ; platform_system == \"Windows\""] +gssapi = ["gssapi (>=1.4.1) ; platform_system != \"Windows\"", "pyasn1 (>=0.1.7)", "pywin32 (>=2.1.8) ; platform_system == \"Windows\""] invoke = ["invoke (>=2.0)"] [[package]] @@ -374,6 +443,7 @@ version = "4.9.0" description = "Pexpect allows easy control of interactive console applications." optional = false python-versions = "*" +groups = ["main"] files = [ {file = "pexpect-4.9.0-py2.py3-none-any.whl", hash = "sha256:7236d1e080e4936be2dc3e326cec0af72acf9212a7e1d060210e70a47e253523"}, {file = "pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f"}, @@ -388,6 +458,7 @@ version = "0.7.0" description = "Run a subprocess in a pseudo terminal" optional = false python-versions = "*" +groups = ["main"] files = [ {file = "ptyprocess-0.7.0-py2.py3-none-any.whl", hash = "sha256:4b41f3967fce3af57cc7e94b888626c18bf37a083e3651ca8feeb66d492fef35"}, {file = "ptyprocess-0.7.0.tar.gz", hash = "sha256:5c5d0a3b48ceee0b48485e0c26037c0acd7d29765ca3fbb5cb3831d347423220"}, @@ -399,17 +470,34 @@ version = "2.22" description = "C parser in Python" optional = false python-versions = ">=3.8" +groups = ["main"] files = [ {file = "pycparser-2.22-py3-none-any.whl", hash = "sha256:c3702b6d3dd8c7abc1afa565d7e63d53a1d0bd86cdc24edd75470f4de499cfcc"}, {file = "pycparser-2.22.tar.gz", hash = "sha256:491c8be9c040f5390f5bf44a5b07752bd07f56edf992381b05c701439eec10f6"}, ] +[[package]] +name = "pygments" +version = "2.20.0" +description = "Pygments is a syntax highlighting package written in Python." +optional = false +python-versions = ">=3.9" +groups = ["main"] +files = [ + {file = "pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176"}, + {file = "pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f"}, +] + +[package.extras] +windows-terminal = ["colorama (>=0.4.6)"] + [[package]] name = "pynacl" version = "1.5.0" description = "Python binding to the Networking and Cryptography (NaCl) library" optional = false python-versions = ">=3.6" +groups = ["main"] files = [ {file = "PyNaCl-1.5.0-cp36-abi3-macosx_10_10_universal2.whl", hash = "sha256:401002a4aaa07c9414132aaed7f6836ff98f59277a234704ff66878c2ee4a0d1"}, {file = "PyNaCl-1.5.0-cp36-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.manylinux_2_24_aarch64.whl", hash = "sha256:52cb72a79269189d4e0dc537556f4740f7f0a9ec41c1322598799b0bdad4ef92"}, @@ -436,6 +524,7 @@ version = "6.0.2" description = "YAML parser and emitter for Python" optional = false python-versions = ">=3.8" +groups = ["main"] files = [ {file = "PyYAML-6.0.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:0a9a2848a5b7feac301353437eb7d5957887edbf81d56e903999a75a3d743086"}, {file = "PyYAML-6.0.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:29717114e51c84ddfba879543fb232a6ed60086602313ca38cce623c1d62cfbf"}, @@ -498,6 +587,7 @@ version = "2.32.4" description = "Python HTTP for Humans." optional = false python-versions = ">=3.8" +groups = ["main"] files = [ {file = "requests-2.32.4-py3-none-any.whl", hash = "sha256:27babd3cda2a6d50b30443204ee89830707d396671944c998b5975b031ac2b2c"}, {file = "requests-2.32.4.tar.gz", hash = "sha256:27d0316682c8a29834d3264820024b62a36942083d52caf2f14c0591336d3422"}, @@ -513,23 +603,74 @@ urllib3 = ">=1.21.1,<3" socks = ["PySocks (>=1.5.6,!=1.5.7)"] use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"] +[[package]] +name = "rich" +version = "15.0.0" +description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal" +optional = false +python-versions = ">=3.9.0" +groups = ["main"] +files = [ + {file = "rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb"}, + {file = "rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36"}, +] + +[package.dependencies] +markdown-it-py = ">=2.2.0" +pygments = ">=2.13.0,<3.0.0" + +[package.extras] +jupyter = ["ipywidgets (>=7.5.1,<9)"] + [[package]] name = "robotframework" version = "7.3.2" description = "Generic automation framework for acceptance testing and robotic process automation (RPA)" optional = false python-versions = ">=3.8" +groups = ["test"] files = [ {file = "robotframework-7.3.2-py3-none-any.whl", hash = "sha256:14ef2afa905285cc073df6ce06d0cd3af4a113df6f815532718079e00c98cca4"}, {file = "robotframework-7.3.2.tar.gz", hash = "sha256:3bb3e299831ecb1664f3d5082f6ff9f08ba82d61a745bef2227328ef3049e93a"}, ] +[[package]] +name = "shellingham" +version = "1.5.4" +description = "Tool to Detect Surrounding Shell" +optional = false +python-versions = ">=3.7" +groups = ["main"] +files = [ + {file = "shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686"}, + {file = "shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de"}, +] + +[[package]] +name = "typer" +version = "0.27.0" +description = "Typer, build great CLIs. Easy to code. Based on Python type hints." +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "typer-0.27.0-py3-none-any.whl", hash = "sha256:6f4b27631e47f077871b7dc30e933ec0131c1390fbe0e387ea5574b5bac9ccf1"}, + {file = "typer-0.27.0.tar.gz", hash = "sha256:629bd12ea5d13a17148125d9a264f949eb171fb3f120f9b04d85873cab054fa5"}, +] + +[package.dependencies] +annotated-doc = ">=0.0.2" +colorama = {version = "*", markers = "platform_system == \"Windows\""} +rich = ">=13.8.0" +shellingham = ">=1.3.0" + [[package]] name = "unidecode" version = "1.4.0" description = "ASCII transliterations of Unicode text" optional = false python-versions = ">=3.7" +groups = ["main"] files = [ {file = "Unidecode-1.4.0-py3-none-any.whl", hash = "sha256:c3c7606c27503ad8d501270406e345ddb480a7b5f38827eafe4fa82a137f0021"}, {file = "Unidecode-1.4.0.tar.gz", hash = "sha256:ce35985008338b676573023acc382d62c264f307c8f7963733405add37ea2b23"}, @@ -541,13 +682,14 @@ version = "2.5.0" description = "HTTP library with thread-safe connection pooling, file post, and more." optional = false python-versions = ">=3.9" +groups = ["main"] files = [ {file = "urllib3-2.5.0-py3-none-any.whl", hash = "sha256:e6b01673c0fa6a13e374b50871808eb3bf7046c4b125b216f6bf1cc604cff0dc"}, {file = "urllib3-2.5.0.tar.gz", hash = "sha256:3fc47733c7e419d4bc3f6b3dc2b4f890bb743906a30d56ba4a5bfa4bbff92760"}, ] [package.extras] -brotli = ["brotli (>=1.0.9)", "brotlicffi (>=0.8.0)"] +brotli = ["brotli (>=1.0.9) ; platform_python_implementation == \"CPython\"", "brotlicffi (>=0.8.0) ; platform_python_implementation != \"CPython\""] h2 = ["h2 (>=4,<5)"] socks = ["pysocks (>=1.5.6,!=1.5.7,<2.0)"] zstd = ["zstandard (>=0.18.0)"] @@ -558,12 +700,13 @@ version = "0.15.2" description = "Python data validation library" optional = false python-versions = ">=3.9" +groups = ["main"] files = [ {file = "voluptuous-0.15.2-py3-none-any.whl", hash = "sha256:016348bc7788a9af9520b1764ebd4de0df41fe2138ebe9e06fa036bf86a65566"}, {file = "voluptuous-0.15.2.tar.gz", hash = "sha256:6ffcab32c4d3230b4d2af3a577c87e1908a714a11f6f95570456b1849b0279aa"}, ] [metadata] -lock-version = "2.0" +lock-version = "2.1" python-versions = "^3.13" -content-hash = "c7ea8f7805f5b885c7ee0b0381c31b4d15812bed8adfba29d30b57ce6024ffc1" +content-hash = "6c38a52faf2fe38bb9a4800141fbc3a1ec02500cc2c4260974b9e54c9ab6f685" diff --git a/osfv_cli/pyproject.toml b/osfv_cli/pyproject.toml index 0d30c45..e9e7bde 100644 --- a/osfv_cli/pyproject.toml +++ b/osfv_cli/pyproject.toml @@ -24,3 +24,4 @@ requests = "^2.31.0" unidecode = "^1.3.6" importlib-resources = "^6.1.1" voluptuous = "^0.15.2" +typer = "^0.27.0" diff --git a/osfv_cli/src/osfv/cli/cli.py b/osfv_cli/src/osfv/cli/cli.py index b18124b..b8c49ec 100755 --- a/osfv_cli/src/osfv/cli/cli.py +++ b/osfv_cli/src/osfv/cli/cli.py @@ -1,91 +1,283 @@ #!/usr/bin/env python3 -import argparse import json +from collections.abc import Callable from copy import copy +from dataclasses import dataclass, field +from functools import partial, wraps from importlib import metadata +from pathlib import Path from time import sleep +from typing import Annotated, Literal, cast -import osfv.libs.utils as utils import pexpect import requests +import typer +from osfv.libs import utils from osfv.libs.models import Models from osfv.libs.rte import RTE from osfv.libs.snipeit_api import SnipeIT from osfv.libs.sonoff_api import SonoffDevice from osfv.libs.zabbix import Zabbix +from typer import Argument, Context, Option -def check_out_asset(snipeit_api, asset_id): - """ - Attempts to check out an asset.. - It checks if the asset is already checked out by the user. +class API: + """store API instances created e.g. in callbacks""" - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - asset_id (str): The unique identifier of the asset to be checked out. + def __init__(self) -> None: + self._sonoff_api: SonoffDevice | None = None + self._snipeit_api: SnipeIT | None = None + self._rte_api: RTE | None = None - Returns: - None - """ - success, data, already_checked_out = snipeit_api.check_out_asset(asset_id) + @property + def sonoff_api(self) -> SonoffDevice: + """Return SonoffDevice instance or raise error if None is set. - if already_checked_out: - print(f"Asset {asset_id} is already checked out by you") - return already_checked_out + Returns: + SonoffDevice: SonoffDevice instance + """ + assert self._sonoff_api is not None + return self._sonoff_api - if success: - print(f"Asset {asset_id} successfully checked out.") - else: - print(f"Error checking out asset {asset_id}") - print(f"Response data: {data}") - exit( - f"Exiting to avoid conflict. Check who is working on this device" - f" and contact them first." - ) + @property + def snipeit_api(self) -> SnipeIT: + """Return SnipeIT instance or raise error if None is set. - return already_checked_out + Returns: + SnipeIT: SnipeIT instance + """ + assert self._snipeit_api is not None + return self._snipeit_api + @property + def rte_api(self) -> RTE: + """Return RTE instance or raise error if None is set. -def check_in_asset(snipeit_api, asset_id): - """ - Checks in an asset to the system by its asset ID. + Returns: + RTE: RTE instance + """ + assert self._rte_api is not None + return self._rte_api - This method attempts to check in the specified asset identified by `asset_id` by making an HTTP POST request. - If the check-in is successful, it returns a success flag and the JSON response from the API. - If the check-in fails, it returns a failure flag and the error message from the API. + def get_or_create_snipeit(self) -> SnipeIT: + """returns snipeit_api instance, creates it if necessary - Parameters: - asset_id (str): The unique identifier of the asset to be checked in. + Returns: + SnipeIT: SnipeIT API instance + """ + if self._snipeit_api is None: + self._snipeit_api = SnipeIT() + return self._snipeit_api - Returns: - tuple: - bool: Indicates if the check-in operation was successful (True) or not (False). - dict: The JSON response from the API, either containing success information or error details. - """ - success, data = snipeit_api.check_in_asset(asset_id) - if success: - print(f"Asset {asset_id} successfully checked in.") - return True - else: - print(f"Error checking in asset {asset_id}") - print(f"Response data: {data}") - return False +@dataclass +class Hooks: + setup: Callable[[], tuple[bool, int | None]] + cleanup: Callable[[bool, int | None], None] + _already_ran: bool = field(default=False, init=False) + + +def with_setup(func): + """Use this decorator to call setup and cleanup check_out/check_in function + Requires the first argument of the decorated function to be of typer.Context + type. -def list_used_assets(snipeit_api, args): + Calls setup if ctx.obj is instance of Hooks, and setups cleanup to be called + during context tear down. Makes sure to only do it once even if mutltiple + decorated functions are called. + + Use this decorator so you don't have to setup/cleanup stuff manually in + each command or run heavy setup function before all arguments are validated. + + ctx.obj can be set in callback subcommand (check rte_options function for + examples) """ - Retrieves and displays all used assets. - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - args (object): Arguments that may contain additional parameters (not used in this function). + @wraps(func) + def wrapper(ctx: Context, *args, **kwargs): + if isinstance(ctx.obj, Hooks) and not ctx.obj._already_ran: + ctx.obj._already_ran = True + checked_out, asset_id = ctx.obj.setup() + ctx.call_on_close(partial(ctx.obj.cleanup, checked_out, asset_id)) + return func(ctx, *args, **kwargs) - Returns: - None + return wrapper + + +def check_in_cleanup(checked_out: bool, asset_id: int | None): + if checked_out and asset_id is not None: + _check_in_asset(apis.get_or_create_snipeit(), asset_id) + + +apis = API() + +## Intermediate subcommands + + +def add_typer(help: str, name: str, target_typer: typer.Typer): + new_typer = typer.Typer(help=help) + target_typer.add_typer(new_typer, name=name) + return new_typer + + +app = typer.Typer( + no_args_is_help=True, + help="Open Source Firmware Validation CLI", + context_settings={"help_option_names": ["-h", "--help"]}, +) + +## root subcommands +snipeit_t = add_typer("Snipe-IT commands", "snipeit", app) +rte_t = add_typer("RTE commands", "rte", app) +sonoff_t = add_typer("Sonoff commands", "sonoff", app) +# list_models and flash_image_check are final subcommands +## rte subcommands +rte_rel = add_typer("Control RTE relay", "rel", rte_t) +rte_gpio = add_typer("Control RTE GPIO", "gpio", rte_t) +rte_pwr = add_typer("Control DUT power via RTE", "pwr", rte_t) +rte_spi = add_typer("Control SPI lines of RTE", "spi", rte_t) +rte_flash = add_typer("DUT flash operations", "flash", rte_t) +# rte serial is final subcommand so it is defined with other commands +## rte pwr subcommands +rte_pwr_psu = add_typer( + "Generic control interface of the power supply", "psu", rte_pwr +) + + +def main(): + app() + + +def print_version(version: bool): + """Print version and exit""" + if version: + print(metadata.version("osfv")) + raise typer.Exit() + + +## root commands +@app.callback() +def root_options( + _: Annotated[ + bool, + typer.Option( + "--version", + "-v", + help="show program's version number and exit", + is_eager=True, + callback=print_version, + ), + ] = False, +): + pass + + +@app.command("list_models") +def list_models(): + """List of supported models""" + models = Models() + models.list_models() + + +def list_known_regions(list_regions: bool): + """List known region names and exit + + Args: + list_regions (bool): Whether to list regions """ - all_assets = snipeit_api.get_all_assets() + if list_regions: + print("Known flash regions:") + for reg_name in utils.get_list_of_known_image_regions(): + print(reg_name) + raise typer.Exit() + + +@app.command("flash_image_check") +def flash_image_check( + rom: Annotated[ + Path, + Option( + "--rom", + help="Path to read firmware file", + metavar="ROM", + exists=True, + dir_okay=False, + ), + ], + dry_mecheck: Annotated[ + bool, + Option( + "--dry-mecheck", + "-x", + help="Failed flash region checks won't change exit status", + ), + ] = False, + verbosity: Annotated[ + bool, + Option( + "--verbosity", + "-V", + help="Increase osfv.libs.flash_image verbosity", + ), + ] = False, + _: Annotated[ + bool, + Option( + "--list", + "-l", + help="list known region names and exit", + is_eager=True, + callback=list_known_regions, + ), + ] = False, + regions_to_check: Annotated[ + list[str] | None, + Option( + "--check", + "-c", + help="check named flash region", + metavar="REGIONS_TO_CHECK", + show_default="me", + ), + ] = None, + regions_to_dump: Annotated[ + list[str] | None, + Option( + "--dump", + "-d", + help="dump named flash region", + metavar="REGIONS_TO_DUMP", + ), + ] = None, +): + """Checks for existence & sane content of ME region in flash image""" + if regions_to_dump: + utils.dump_flash_image_regions(rom, verbosity, regions_to_dump) + if regions_to_check is None: + regions_to_check = ["me"] + if ( + utils.check_flash_image_regions( + rom, dry_mecheck, verbosity, regions_to_check + ) + == False + ): + print( + "Do not flash full image, unless you are skipping empty regions, and know what you are doing!" + ) + raise typer.Exit(1) + + +## snipeit commands +@snipeit_t.command("list_used") +def list_used_assets( + dump_json: Annotated[ + bool, Option("--json", "-j", help="Dump assets as JSON") + ] = False, +): + """List all already used assets""" + all_assets = apis.get_or_create_snipeit().get_all_assets() used_assets = [ asset for asset in all_assets if asset["assigned_to"] is not None ] @@ -94,52 +286,32 @@ def list_used_assets(snipeit_api, args): print("No used assets found.") return - if args.json: + if dump_json: print(json.dumps(used_assets)) else: for asset in used_assets: print_asset_details(asset) -def get_my_assets(snipeit_api): - """ - Gets a list of assets assigned to the current user - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - - Returns: - List of assets assigned to the current user - """ - all_assets = snipeit_api.get_all_assets() - used_assets = [ - asset for asset in all_assets if asset["assigned_to"] is not None - ] - return [ - asset - for asset in used_assets - if asset["assigned_to"]["id"] is snipeit_api.cfg_user_id - ] - - -def list_my_assets(snipeit_api, args): +@snipeit_t.command("list_my", help="List all my used assets") +def list_my_assets( + dump_json: Annotated[ + bool, Option("--json", "-j", help="Dump assets as JSON") + ] = False, +) -> bool: """ - Lists all assets assigned to the current user. - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - args (object): Arguments that may contain additional parameters (not used in this function). + List all my used assets Returns: Boolean: False if no assets were assigned to the user, True otherwise """ - my_assets = get_my_assets(snipeit_api) + my_assets = get_my_assets(apis.get_or_create_snipeit()) if not my_assets: print("No used assets found.") return False - if args.json: + if dump_json: print(json.dumps(my_assets)) else: for asset in my_assets: @@ -147,61 +319,14 @@ def list_my_assets(snipeit_api, args): return True -def check_in_my(snipeit_api, args): - """ - Lists all assets assigned to the current user, checks in all of them - except those which are in a category listed in `categories_to_ignore`. - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - args (object): Arguments that may contain additional parameters - (not used in this function). - - Returns: - None - """ - categories_to_ignore = ["Employee Laptop"] - my_assets = get_my_assets(snipeit_api) - - my_assets = [ - asset - for asset in my_assets - if not set(asset["category"].values()) & set(categories_to_ignore) - ] - if not list_my_assets(snipeit_api, args): - return - - if not args.yes: - print( - f"Are you sure you want to check in {len(my_assets)} assets? [y/N]" - ) - if input() != "y": - print(f"Checking in {len(my_assets)} assets aborted.") - return - - failed = [] - for asset in my_assets: - if not check_in_asset(snipeit_api, asset["id"]): - failed = failed.append(asset) - - if failed: - print(f"Failed to check-in {len(failed)} assets:") - else: - print(f"{len(my_assets)} assets checked in successfully.") - - -def list_unused_assets(snipeit_api, args): - """ - Retrieves all assets. - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - args (object): Arguments that may contain additional parameters (not used in this function). - - Returns: - None - """ - all_assets = snipeit_api.get_all_assets() +@snipeit_t.command("list_unused") +def list_unused_assets( + dump_json: Annotated[ + bool, Option("--json", "-j", help="Dump assets as JSON") + ] = False, +): + """List all unused assets""" + all_assets = apis.get_or_create_snipeit().get_all_assets() unused_assets = [ asset for asset in all_assets if asset["assigned_to"] is None ] @@ -210,49 +335,37 @@ def list_unused_assets(snipeit_api, args): print("No unused assets found.") return - if args.json: + if dump_json: print(json.dumps(unused_assets)) else: for asset in unused_assets: print_asset_details(asset) -def list_all_assets(snipeit_api, args): - """ - Retrieves all assets using. - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - args (object): Arguments that may contain additional parameters (not used in this function). - - Returns: - None - """ - all_assets = snipeit_api.get_all_assets() +@snipeit_t.command("list_all") +def list_all_assets( + dump_json: Annotated[ + bool, Option("--json", "-j", help="Dump assets as JSON") + ] = False, +): + """List all assets""" + all_assets = apis.get_or_create_snipeit().get_all_assets() if not all_assets: print("No assets found.") return - if args.json: + if dump_json: print(json.dumps(all_assets)) else: for asset in all_assets: print_asset_details(asset) -def list_for_zabbix(snipeit_api, args): - """ - Print asset details as JSON with specific custom fields. - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - args (object): Arguments that may contain additional parameters (not used in this function). - - Returns: - None - """ - all_assets = snipeit_api.get_all_assets() +@snipeit_t.command("list_for_zabbix") +def list_for_zabbix(): + """List assets in a format suitable for Zabbix integration""" + all_assets = apis.get_or_create_snipeit().get_all_assets() if all_assets: for asset in all_assets: @@ -261,261 +374,592 @@ def list_for_zabbix(snipeit_api, args): print("No assets found.") -def print_asset_details(asset): +@snipeit_t.command( + "update_zabbix", help="Syncs Zabbix assets with SnipeIT ones" +) +def update_zabbix_assets(): """ - Prints details of a given asset, including its basic information, assigned user (if any), - and custom fields. - - Args: - asset (dict): Dictionary containing asset details. - - Returns: - None + Updates Zabbix with the latest asset data from Snipe-IT, ensuring the IP addresses + are synchronized between Snipe-IT and Zabbix. """ + zabbix = Zabbix() + all_assets = apis.get_or_create_snipeit().get_all_assets() - print( - f'Asset Tag: {asset["asset_tag"]}, Asset ID: {asset["id"]},' - f'Name: {asset["name"]}, Serial: {asset["serial"]}' - ) + current_zabbix_assets = zabbix.get_all_hosts() + # snipeit assets but converted to zabbix-form assets + snipeit_assets = {} - if asset["assigned_to"]: - print(f'Assigned to: {asset["assigned_to"]["name"]}') + update_available = False - custom_fields = asset.get("custom_fields", {}) - if custom_fields: - for field_name, field_data in custom_fields.items(): - field_value = field_data.get("value") - print(f"{field_name}: {field_value}") + if all_assets: + for asset in all_assets: + snipeit_assets.update( + get_zabbix_compatible_assets_from_asset(asset) + ) - print() + snipeit_assets_keys = list(snipeit_assets.keys()) + snipeit_configuration_error = False -def get_zabbix_compatible_assets_from_asset(asset): - """ - Extracts an asset to zabbix assets. + forbidden_symbols = [ + "/", + "\\", + "{", + "}", + ";", + ":", + "~", + "`", + '"', + "'", + "[", + "]", + "|", + "<", + ">", + "$", + "#", + "@", + "%", + "^", + "&", + "*", + "(", + ")", + "+", + "=", + ] - Args: - asset (dict): Dictionary containing asset details. + for i in range(snipeit_assets.__len__()): + # check for duplicates + for j in range(i + 1, snipeit_assets.__len__()): + if ( + snipeit_assets[snipeit_assets_keys[i]] + == snipeit_assets[snipeit_assets_keys[j]] + ): + print( + f"{snipeit_assets_keys[i]} has the same IP as " + f"{snipeit_assets_keys[j]}!" + ) + snipeit_configuration_error = True + if snipeit_assets_keys[i] == snipeit_assets_keys[j]: + print( + f"There are at least 2 assets with name " + f"{snipeit_assets_keys[i]} present!" + ) + snipeit_configuration_error = True - Returns: - The dictionary with asset tags as keys and asset data as value. - """ - result = {} - custom_fields = asset.get("custom_fields", {}) - if custom_fields: - for field_name, field_data in custom_fields.items(): - if field_name in ["RTE IP", "Sonoff IP", "PiKVM IP"]: - field_value = field_data.get("value") - if field_value: - key = f'{asset["asset_tag"]}_{field_name}'.replace( - " ", "_" - ) - result[key] = field_value - return result + # check for forbidden symbols in asset names + if any( + symbol in snipeit_assets_keys[i] for symbol in forbidden_symbols + ): + print( + f"{snipeit_assets_keys[i]} contains forbidden symbols! They " + f"are going to be changed to '_'." + ) + new_key = copy(snipeit_assets_keys[i]) + for s in forbidden_symbols: + new_key = new_key.replace(s, "_") + snipeit_assets[new_key] = snipeit_assets.pop( + snipeit_assets_keys[i] + ) -def print_asset_details_for_zabbix(asset): - """ - Print asset details formatted as an input for Zabbix import script. + if snipeit_configuration_error: + print( + "\nSnipeIT configuration errors have been detected! " + "Fix them and then continue." + ) + return - Args: - asset (dict): Dictionary containing asset details. + keys_not_present_in_zabbix = set(snipeit_assets.keys()) - set( + current_zabbix_assets.keys() + ) - Returns: - None. - """ - assets = get_zabbix_compatible_assets_from_asset(asset) - for key in assets.keys(): - print(f"{key}: {assets[key]}") + keys_not_present_in_snipeit = set(current_zabbix_assets.keys()) - set( + snipeit_assets.keys() + ) + if ( + keys_not_present_in_zabbix.__len__() > 0 + or keys_not_present_in_snipeit.__len__() > 0 + ): + update_available = True -def relay_toggle(rte, args): - """ - Toggle the state of a relay controlled by the rte object. + common_keys = set(snipeit_assets.keys()) & set( + current_zabbix_assets.keys() + ) - Args: - rte: An object responsible for controlling the relay. - args (object): Arguments that may contain additional parameters (not used in this function). + if keys_not_present_in_zabbix.__len__() > 0: + print("Assets not present in Zabbix (these will be added):") + print("\n".join(keys_not_present_in_zabbix)) - Returns: - None. - """ - state_str = rte.relay_get() - if state_str == RTE.PSU_STATE_OFF: - new_state_str = RTE.PSU_STATE_ON - else: - new_state_str = RTE.PSU_STATE_OFF - rte.relay_set(new_state_str) - state = rte.relay_get() - print(f"Relay state toggled. New state: {state}") + if keys_not_present_in_snipeit.__len__() > 0: + print( + "\nAssets present in Zabbix but not in SnipeIT (these will be removed):" + ) + print("\n".join(keys_not_present_in_snipeit)) + print() + keys_for_ip_change = [] + for key in common_keys: + if snipeit_assets[key] != current_zabbix_assets[key]: + print( + f"{key} has wrong IP! (Zabbix one will be updated from " + f"{current_zabbix_assets[key]} to {snipeit_assets[key]})" + ) + keys_for_ip_change.append(key) -def relay_set(rte, args): - """ - Sets the relay state based on the provided args.state value. + if keys_for_ip_change.__len__() > 0: + update_available = True - Args: - rte: An object responsible for controlling the relay. - args (object): Arguments that may contain additional parameters (not used in this function). + if not update_available: + print("Zabbix is already synced with SnipeIT") + return - Returns: - None. - """ - rte.relay_set(args.state) - state = rte.relay_get() - print(f"Relay state set to {state}") + if not ask_to_proceed("Do you want to apply above changes? (y/n): "): + print("Changes were not applied") + return + # removing zabbix hosts + for key in keys_not_present_in_snipeit: + print(f"Removing {key}({current_zabbix_assets[key]})...") + result = zabbix.remove_host_by_name(key) + if "error" in result: + print("Failed to remove the host!") -def relay_get(rte, args): - """ - Get the state of the relay and print it. + # updating zabbix ips + for key in keys_for_ip_change: + print(f"Updating {key} IP to {snipeit_assets[key]}...") + result = zabbix.update_host_ip(key, snipeit_assets[key]) + if "error" in result: + print("Failed to change host's IP!") - Args: - rte (object): The object representing the relay control interface. - args (object): Arguments that may contain additional parameters (not used in this function). + # adding zabbix hosts + for key in keys_not_present_in_zabbix: + print(f"Adding {key}({snipeit_assets[key]})...") + try: + result = zabbix.add_host(key, snipeit_assets[key]) + except ValueError: + print("Failed to add the host!") - Returns: - None - """ - state = rte.relay_get() - print(f"Relay state: {state}") +@snipeit_t.command("check_out") +def check_out_asset( + asset_id: Annotated[ + int | None, Option("--asset_id", help="Asset ID", metavar="ASSET_ID") + ] = None, + rte_ip: Annotated[ + str | None, Option("--rte_ip", help="RTE IP address", metavar="RTE_IP") + ] = None, +): + """Check out an asset by providing the Asset ID or RTE IP -def power_on(rte, args): + It checks if the asset is already checked out by the user. """ - Power on the DUT if the power supply is enabled. - - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - - Returns: - None + snipeit_api = apis.get_or_create_snipeit() + asset_id = validate_and_return_asset_id(snipeit_api, asset_id, rte_ip) + _check_out_asset(snipeit_api, asset_id) + + +@snipeit_t.command("check_in") +def check_in_asset( + asset_id: Annotated[ + int | None, Option("--asset_id", help="Asset ID", metavar="ASSET_ID") + ] = None, + rte_ip: Annotated[ + str | None, Option("--rte_ip", help="RTE IP address", metavar="RTE_IP") + ] = None, +): + """Check in an asset by providing the Asset ID or RTE IP""" + snipeit_api = apis.get_or_create_snipeit() + asset_id = validate_and_return_asset_id(snipeit_api, asset_id, rte_ip) + if not _check_in_asset(snipeit_api, asset_id): + raise typer.Exit(1) + + +@snipeit_t.command("check_in_my") +def check_in_my( + dump_json: Annotated[ + bool, Option("--json", "-j", help="Dump assets as JSON") + ] = False, + yes: Annotated[ + bool, Option("--yes", "-y", help="Skips the confirmation") + ] = False, +): """ - state = rte.psu_get() - if state != rte.PSU_STATE_ON: - print(f"Power supply state: {state} !") - print( - "If you wanted to power on the DUT, you need to enable power suppl" - 'y first ("pwr psu on"), pushing the power button is not enough!' - ) - print(f"Powering on...") - rte.power_on(args.time) - - -def power_off(rte, args): + Lists all assets assigned to the current user, checks in all of them + except those which are in a category listed in `categories_to_ignore`. """ - Power off the DUT. - - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). + categories_to_ignore = ["Employee Laptop"] + snipeit_api = apis.get_or_create_snipeit() + my_assets = get_my_assets(snipeit_api) - Returns: - None - """ - print(f"Powering off...") - rte.power_off(args.time) + my_assets = [ + asset + for asset in my_assets + if not set(asset["category"].values()) & set(categories_to_ignore) + ] + if not list_my_assets(dump_json): + raise typer.Exit() + if not yes: + choice = typer.confirm( + f"Are you sure you want to check in {len(my_assets)} assets? [y/N]" + ) -def power_on_ex(rte, args): - power_on(rte, args) - for attempt in range(20): - if check_pwr_led(rte, args) == "high": - print("Power on successful.") - return True - sleep(0.25) - print("Power on failed.") - return False + if not choice: + print(f"Checking in {len(my_assets)} assets aborted.") + raise typer.Exit() + failed = [] + for asset in my_assets: + if not _check_in_asset(snipeit_api, asset["id"]): + failed.append(asset) -def power_off_ex(rte, args): - power_off(rte, args) - for attempt in range(20): - if check_pwr_led(rte, args) == "low": - print("Power off successful.") - return True - sleep(0.25) - print("Power off failed.") - return False + if failed: + print(f"Failed to check-in {len(failed)} assets: {failed}") + else: + print(f"{len(my_assets)} assets checked in successfully.") -def reset(rte, args): - """ - Reset the DUT by pressing the reset button. +@snipeit_t.command("user_add") +def user_add( + first_name: Annotated[ + str, + Option("--first-name", help="User First Name", metavar="FIRST_NAME"), + ], + last_name: Annotated[ + str, Option("--last-name", help="User Last Name", metavar="LAST_NAME") + ], + company_name: Annotated[ + str, + Option("--company-name", help="Company Name", metavar="COMPANY_NAME"), + ] = "3mdeb", +): + """Add a new user by providing user First Name, Last Name and Company Name""" + apis.get_or_create_snipeit().user_add(first_name, last_name, company_name) + + +@snipeit_t.command("user_del") +def user_del( + first_name: Annotated[ + str, + Option("--first-name", help="User First Name", metavar="FIRST_NAME"), + ], + last_name: Annotated[ + str, Option("--last-name", help="User Last Name", metavar="LAST_NAME") + ], +): + """Delete new user by providing user First Name and Last Name""" + apis.get_or_create_snipeit().user_del(first_name, last_name) + + +## rte commands +def setup_rte_subcommand( + rte_ip: str, model: str | None, skip_snipeit: bool +) -> tuple[bool, int | None]: + """Validate arguments, setup needed resources + + Sets up snipeit, sonoff, and rte in `apis`, checks out asset if required Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - - Returns: - None - """ - print(f"Pressing reset button...") - rte.reset(args.time) + rte_ip (str): RTE IP Address + model (str | None): Model name, used if skipping snipeit + skip_snipeit (bool): Whether to skip snipeit requests - -def psu_on(rte, args): - """ - Enable the power supply to the DUT. - - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). + Raises: + typer.Exit: When setup/checkout fails Returns: - None - """ - print(f"Enabling power supply...") - rte.psu_on() + tuple[bool, str]: (asset was checked_out?, asset_id) + """ + snipeit_api: SnipeIT | None = None + asset_id: int | None = None + dut_model_name: str | None = None + if not skip_snipeit: + snipeit_api = apis.get_or_create_snipeit() + asset_id = snipeit_api.get_asset_id_by_rte_ip(rte_ip) + if asset_id is None: + print(f"No asset found with RTE IP: {rte_ip}") + raise typer.Exit(1) + if model: + print("DUT model retrieved from cmdline, skipping Snipe-IT query") + dut_model_name = model + else: + if not skip_snipeit: + status, dut_model_name = apis.snipeit_api.get_asset_model_name( + asset_id + ) + if status: + print(f"DUT model retrieved from snipeit: {dut_model_name}") + else: + print( + "failed to retrieve model name from snipe-it. check " + "again arguments, or try providing model manually." + ) + raise typer.Exit(1) + else: + print("`--model MODEL` is required when skipping snipeit.") + raise typer.Exit(1) + # TODO: Add sonoff ip argument + apis._sonoff_api, _ = utils.init_sonoff(None, rte_ip, snipeit_api) + apis._rte_api = RTE(rte_ip, dut_model_name, apis._sonoff_api) + + if not skip_snipeit: + assert isinstance(asset_id, int) + print( + "Using rte command is invasive action, checking first if the " + "device is not used..." + ) + already_checked_out = _check_out_asset( + apis.snipeit_api, cast(int, asset_id) + ) + return not already_checked_out, asset_id + return False, None + + +@rte_t.callback() +def rte_options( + ctx: Context, + rte_ip: Annotated[ + str, Option("--rte_ip", help="RTE IP address", metavar="RTE_IP") + ], + model: Annotated[ + str | None, + Option( + "--model", + help="DUT model. If not given, will attempt to query from Snipe-IT.", + metavar="MODEL", + ), + ] = None, + skip_snipeit: Annotated[ + bool, + Option( + "--skip-snipeit", + help="Skips Snipe-IT related actions like checkout and check-in. Useful for OSFV homelab.", + ), + ] = False, +): + ctx.obj = Hooks( + setup=partial(setup_rte_subcommand, rte_ip, model, skip_snipeit), + cleanup=check_in_cleanup, + ) -def psu_off(rte, args): +@rte_t.command("serial") +@with_setup +def open_dut_serial(ctx: Context): + """Open DUT serial via telnet + + Open a Telnet session to interact with the DUT serial interface. """ - Disable the power supply to the DUT. + host = apis.rte_api.rte_ip + port = 13541 - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). + print(f"Opening telnet session with: {host}:{port}") + print("Press Ctrl+] to exit") + # Connect to the Telnet server + tn = pexpect.spawn(f"telnet {host} {port}") - Returns: - None - """ - print(f"Disabling power supply...") - rte.psu_off() + # Enter the interactive shell + tn.interact() -def psu_get(rte, args): - """ - Retrieve and print the current power supply state. +## rte rel commands +@rte_rel.command("tgl") +@with_setup +def relay_toggle(ctx: Context): + """Toggle relay state""" + rte = apis.rte_api + state_str = rte.relay_get() + if state_str == RTE.PSU_STATE_OFF: + new_state_str = RTE.PSU_STATE_ON + else: + new_state_str = RTE.PSU_STATE_OFF + rte.relay_set(new_state_str) + state = rte.relay_get() + print(f"Relay state toggled. New state: {state}") - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - state = rte.psu_get() - print(f"Power supply state: {state}") +@rte_rel.command("get") +@with_setup +def relay_get(ctx: Context): + """Get relay state""" + state = apis.rte_api.relay_get() + print(f"Relay state: {state}") -def gpio_get(rte, args): - """ - Retrieve and print the state of a specified GPIO pin. +@rte_rel.command("set") +@with_setup +def relay_set( + ctx: Context, + state: Annotated[Literal["on", "off"], Argument(help="Relay state")], +): + """Set relay state""" + rte = apis.rte_api + rte.relay_set(state) + current_state = rte.relay_get() + print(f"Relay state set to {current_state}") + + +## rte gpio commands +@rte_gpio.command("get") +@with_setup +def gpio_get( + ctx: Context, gpio_no: Annotated[int, Argument(help="GPIO number")] +): + """Get GPIO state""" + state = apis.rte_api.gpio_get(gpio_no) + print(f"GPIO {gpio_no} state: {state}") + + +@rte_gpio.command("set") +@with_setup +def gpio_set( + ctx: Context, + gpio_no: Annotated[int, Argument(help="GPIO number")], + state: Annotated[ + Literal["high", "low", "high-z"], Argument(help="GPIO state") + ], +): + """Set GPIO state""" + rte = apis.rte_api + rte.gpio_set(gpio_no, state) + state = rte.gpio_get(gpio_no) + print(f"GPIO {gpio_no} state set to {state}") + + +@rte_gpio.command("list") +@with_setup +def gpio_list(ctx: Context): + """List GPIO states""" + response = json.dumps(apis.rte_api.gpio_list(), indent=4) + print("GPIO list") + print(response) - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - state = rte.gpio_get(args.gpio_no) - print(f"GPIO {args.gpio_no} state: {state}") +## rte pwr commands +@rte_pwr.command("on") +@with_setup +def power_on( + ctx: Context, + time: Annotated[ + int, + Option( + "--time", + help="Power button press time in seconds", + metavar="TIME", + min=1, + ), + ] = 1, +): + """Short power button press, to power on DUT""" + rte = apis.rte_api + state = rte.psu_get() + if state != rte.PSU_STATE_ON: + print(f"Power supply state: {state} !") + print( + "If you wanted to power on the DUT, you need to enable power suppl" + 'y first ("pwr psu on"), pushing the power button is not enough!' + ) + print("Powering on...") + rte.power_on(time) + + +@rte_pwr.command("on_ex") +@with_setup +def power_on_ex( + ctx: Context, + time: Annotated[ + int, + Option( + "--time", + help="Power button press time in seconds", + metavar="TIME", + min=1, + ), + ] = 1, +): + """Short power button press, to power on DUT, and verify if power LED did turn on""" + power_on(ctx, time) + for _ in range(20): + if check_pwr_led(ctx) == "high": + print("Power on successful.") + return True + sleep(0.25) + print("Power on failed.") + return False -def check_pwr_led(rte, args): +@rte_pwr.command("off") +@with_setup +def power_off( + ctx: Context, + time: Annotated[ + int, + Option( + "--time", + help="Power button press time in seconds", + metavar="TIME", + min=1, + ), + ] = 6, +): + """Long power button press, to power off DUT""" + print("Powering off...") + apis.rte_api.power_off(time) + + +@rte_pwr.command("off_ex") +@with_setup +def power_off_ex( + ctx: Context, + time: Annotated[ + int, + Option( + "--time", + help="Power button press time in seconds", + metavar="TIME", + min=1, + ), + ] = 6, +): + """Long power button press, to power off DUT, and verify if power LED did turn off""" + power_off(ctx, time) + for _ in range(20): + if check_pwr_led(ctx) == "low": + print("Power off successful.") + raise typer.Exit() + sleep(0.25) + print("Power off failed.") + raise typer.Exit(1) + + +@rte_pwr.command() +@with_setup +def reset( + ctx: Context, + time: Annotated[ + int, + Option( + "--time", + help="Power button press time in seconds", + metavar="TIME", + min=1, + ), + ] = 1, +): + """Reset button press, to reset DUT""" + print("Pressing reset button...") + apis.rte_api.reset(time) + + +@rte_pwr.command("pwr_led") +@with_setup +def check_pwr_led(ctx: Context): + """Check the state of the DUT power LED""" + rte = apis.rte_api state = rte.gpio_get(RTE.GPIO_PWR_LED) polarity = rte.dut_data.get("pwr_led", {}).get("polarity") if polarity and polarity == "active low": @@ -528,274 +972,436 @@ def check_pwr_led(rte, args): return state -def gpio_set(rte, args): - """ - Set the state of a specified GPIO pin and print the new state. - - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - - Returns: - None - """ - rte.gpio_set(args.gpio_no, args.state) - state = rte.gpio_get(args.gpio_no) - print(f"GPIO {args.gpio_no} state set to {state}") +@rte_pwr.command("reset_cmos") +@with_setup +def reset_cmos(ctx: Context): + """Reset the DUT CMOS""" + print("Clearing CMOS...") + apis.rte_api.reset_cmos() -def gpio_list(rte, args): - """ - Retrieve and print the list of available GPIO pins. +## rte pwr psu commands +@rte_pwr_psu.command("on") +@with_setup +def psu_on(ctx: Context): + """Turn the power supply on""" + print("Enabling power supply...") + apis.rte_api.psu_on() - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - response = json.dumps(rte.gpio_list(), indent=4) - print(f"GPIO list") - print(response) +@rte_pwr_psu.command("off") +@with_setup +def psu_off(ctx: Context): + """Turn the power supply off""" + print("Disabling power supply...") + apis.rte_api.psu_off() -def rte_status(rte, args): - """ - Set the state of a GPIO pin and print its new state. +@rte_pwr_psu.command("get") +@with_setup +def psu_get(ctx: Context): + """Display information on DUT's power state""" + state = apis.rte_api.psu_get() + print(f"Power supply state: {state}") - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - rte.gpio_set(args.gpio_no, args.state) - state = rte.gpio_get(args.gpio_no) - print(f"GPIO {args.gpio_no} state set to {state}") +## rte spi commands +@rte_spi.command("on") +@with_setup +def spi_on(ctx: Context): + """Enable SPI lines""" + print("Enabling SPI...") + apis.rte_api.spi_enable() + + +@rte_spi.command("off") +@with_setup +def spi_off(ctx: Context): + """Disable SPI lines""" + print("Disabling SPI...") + apis.rte_api.spi_disable() + + +## rte flash commands +@rte_flash.command("probe") +@with_setup +def flash_probe(ctx: Context): + """Flash probe with flashrom""" + print("Probing flash...") + apis.rte_api.flash_probe() + + +@rte_flash.command("read") +@with_setup +def flash_read( + ctx: Context, + rom: Annotated[ + Path, + Option( + "--rom", + help="Path to read firmware file", + metavar="ROM", + dir_okay=False, + writable=True, + ), + ] = Path("read.rom"), +): + """Read from DUT flash with flashrom""" + print("Reading from flash...") + apis.rte_api.flash_read(rom) + print(f"Read flash content saved to {rom}") + + +@rte_flash.command("write") +@with_setup +def flash_write( + ctx: Context, + rom: Annotated[ + Path, + Option( + "--rom", + help="Path to read firmware file", + metavar="ROM", + exists=True, + dir_okay=False, + ), + ] = Path("write.rom"), + bios: Annotated[ + bool, + Option( + "--bios", "-b", help='Adds "-i bios --ifd" to flashrom command' + ), + ] = False, + dry_mecheck: Annotated[ + bool, + Option( + "--dry-mecheck", + "-x", + help="Failed flash region checks won't change exit status", + ), + ] = False, + verbosity: Annotated[ + bool, + Option( + "--verbosity", + "-V", + help="Increase osfv.libs.flash_image verbosity", + ), + ] = False, +): + """Write to DUT flash with flashrom""" + if utils.check_flash_image_regions(rom, dry_mecheck, verbosity) == False: + print( + "FATAL: Image could not be loaded, or some image's regions are empty, despite being defined in the flash descriptor. " + "Flashing full image in this form on Intel platform will result in a bricked platform. " + "If you wish to continue anyway (e.g. when using AMD platform), pass the -x option to skip the check. " + "When using Intel platform, you probably also want to pass the -b option to flash BIOS region only, " + "leaving other regions in platform's flash (such as ME) intact." + ) + raise typer.Exit(1) + print(f"Writing {rom} to flash...") + rc = apis.rte_api.flash_write(rom, bios) + if rc == 0: + print("Flash written successfully") + else: + print(f"Flash write failed with code {rc}") -def open_dut_serial(rte, args): - """ - Open a Telnet session to interact with the DUT serial interface. +@rte_flash.command("erase") +@with_setup +def flash_erase(ctx): + """Erase DUT flash with flashrom""" + print("Erasing DUT flash...") + apis.rte_api.flash_erase() + print("Flash erased") - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - host = args.rte_ip - port = 13541 +## sonoff commands - print(f"Opening telnet session with: {host}:{port}") - print(f"Press Ctrl+] to exit") - # Connect to the Telnet server - tn = pexpect.spawn(f"telnet {host} {port}") - # Enter the interactive shell - tn.interact() +def sonoff_setup( + sonoff_ip: str | None, rte_ip: str | None +) -> tuple[bool, int]: + if not sonoff_ip: + if not rte_ip: + print("Either sonoff_ip or rte_ip is required") + raise typer.Exit(1) + sonoff_ip = apis.get_or_create_snipeit().get_sonoff_ip_by_rte_ip( + rte_ip + ) + if not sonoff_ip: + print(f"No Sonoff Device found with RTE IP: {rte_ip}") + raise typer.Exit(1) + + asset_id = apis.get_or_create_snipeit().get_asset_id_by_sonoff_ip( + sonoff_ip + ) + if asset_id is None: + print(f"No asset found with Sonoff IP: {sonoff_ip}") + raise typer.Exit(1) + print( + "Using rte command is invasive action, checking first if the " + "device is not used..." + ) + already_checked_out = _check_out_asset(apis.snipeit_api, asset_id) + apis._sonoff_api = SonoffDevice(sonoff_ip) + return not already_checked_out, asset_id + + +@sonoff_t.callback() +def sonoff_options( + ctx: Context, + sonoff_ip: Annotated[ + str | None, + Option("--sonoff_ip", help="Sonoff IP address", metavar="SONOFF_IP"), + ] = None, + rte_ip: Annotated[ + str | None, Option("--rte_ip", help="RTE IP address", metavar="RTE_IP") + ] = None, +): + ctx.obj = Hooks( + setup=partial(sonoff_setup, sonoff_ip, rte_ip), + cleanup=check_in_cleanup, + ) + + +@sonoff_t.command("on") +@with_setup +def sonoff_on(ctx: Context): + """Turn Sonoff ON""" + print("Turning on Sonoff power switch...") + try: + response = apis.sonoff_api.turn_on() + print(response) + except requests.exceptions.RequestException as e: + print(f"Failed to turn on Sonoff power switch. Error: {e}") -def spi_on(rte, args): - """ - Enable SPI on the Device Under Test (DUT). - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). +@sonoff_t.command("off") +@with_setup +def sonoff_off(ctx: Context): + """Turn Sonoff OFF""" + print("Turning off Sonoff power switch...") + try: + response = apis.sonoff_api.turn_off() + print(response) + except requests.exceptions.RequestException as e: + print(f"Failed to turn off Sonoff power switch. Error: {e}") - Returns: - None - """ - print(f"Enabling SPI...") - rte.spi_enable() +@sonoff_t.command("tgl") +@with_setup +def sonoff_tgl(ctx: Context): + """Toggle Sonoff state""" + print("Toggling Sonoff power switch state...") + try: + sonoff_api = apis.sonoff_api + current_state = sonoff_api.get_state() -def spi_off(rte, args): - """ - Disable SPI on the Device Under Test (DUT). + if current_state == "ON": + _response = sonoff_api.turn_off() + print("Sonoff power switch state toggled off.") + elif current_state == "OFF": + _response = sonoff_api.turn_on() + print("Sonoff power switch state toggled on.") + else: + print(f"Unexpected Sonoff power switch state: {current_state}") + except requests.exceptions.RequestException as e: + print(f"Failed to toggle Sonoff power switch state. Error: {e}") - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - print(f"Disabling SPI...") - rte.spi_disable() +@sonoff_t.command("get") +@with_setup +def sonoff_get(ctx: Context): + """Get Sonoff state""" + print("Getting Sonoff power switch state...") + try: + state = apis.sonoff_api.get_state() + print(f"Sonoff power switch state: {state}") + except requests.exceptions.RequestException as e: + print(f"Failed to get Sonoff power switch state. Error: {e}") -def flash_probe(rte, args): - """ - Probe the flash memory on the Device Under Test (DUT). +def validate_and_return_asset_id( + snipeit_api: SnipeIT, asset_id: int | None, rte_ip: str | None +) -> int: + """Check which argument is set and return asset_id based on it + + If both arguments are the same (None or str) then raise typer.Exit, else + return asset_id or try to turn rte_ip to assed_id and return it. + Raise typer.Exit if cannot turn rte_ip to asset_id. Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). + snipeit_api (SnipeIT): SnipeIT instance + asset_id (str | None): Asset ID + rte_id (str | None): RTE IP address Returns: - None + str: Found asset_id or None """ - print(f"Probing flash...") - rte.flash_probe() + if isinstance(asset_id, int) and isinstance(rte_ip, str): + print("Only asset_id or rte_ip is allowed, not both") + raise typer.Exit(1) + if asset_id is None and rte_ip is None: + print("Either asset_id or rte_ip is required") + raise typer.Exit(1) + if rte_ip is not None: + asset_id = snipeit_api.get_asset_id_by_rte_ip(rte_ip) + if asset_id is None: + print(f"No asset found with RTE IP: {rte_ip}") + raise typer.Exit(1) + assert asset_id is not None + return asset_id -def flash_read(rte, args): - """ - Read the flash content from the Device Under Test (DUT). + +def _check_out_asset(snipeit_api: SnipeIT, asset_id: int) -> bool: + """Check out an asset by providing the Asset ID + + It checks if the asset is already checked out by the user. Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). + snipeit_api (SnipeIT): SnipeIT instance + asset_id (str): Asset ID to check-out + + Raises: + typer.Exit: If check-out failed Returns: - None + bool: True if asset was already checked-out """ - print(f"Reading from flash...") - rte.flash_read(args.rom) - print(f"Read flash content saved to {args.rom}") + success, data, already_checked_out = snipeit_api.check_out_asset(asset_id) + + if already_checked_out: + print(f"Asset {asset_id} is already checked out by you") + return already_checked_out + + if success: + print(f"Asset {asset_id} successfully checked out.") + else: + print(f"Error checking out asset {asset_id}") + print(f"Response data: {data}") + print( + "Exiting to avoid conflict. Check who is working on this device" + " and contact them first." + ) + raise typer.Exit(1) + + return already_checked_out -def flash_write(rte, args): - """ - Write a specified ROM file to the flash memory using the rte object. +def _check_in_asset(snipeit_api: SnipeIT, asset_id: int) -> bool: + """Check in an asset by providing the Asset ID + This method attempts to check in the specified asset identified by `asset_id` by making an HTTP POST request. + If the check-in is successful, it returns True. + If the check-in fails, it returns False and prints the error message from the API. Args: - rte: An object responsible for handling the flash memory operations. - args (object): Arguments that may contain additional parameters: - args.rom (str): Flash image file path & name - args.dry_mecheck (bool): runs ME check in dry run mode; - check status is always positive and does not - affect flash write process - args.verbosity (bool): increases osfv.libs.flash_image verbosity - + snipeit_api (SnipeIT): SnipeIT instance + asset_id (str): Asset ID to check-in Returns: - None. + bool: Whether check-in succeeded """ - if ( - utils.check_flash_image_regions( - args.rom, args.dry_mecheck, args.verbosity - ) - == False - ): - exit( - "FATAL: Image could not be loaded, or some image's regions are empty, despite being defined in the flash descriptor. " - "Flashing full image in this form on Intel platform will result in a bricked platform. " - "If you wish to continue anyway (e.g. when using AMD platform), pass the -x option to skip the check. " - "When using Intel platform, you probably also want to pass the -b option to flash BIOS region only, " - "leaving other regions in platform's flash (such as ME) intact." - ) - print(f"Writing {args.rom} to flash...") - rc = rte.flash_write(args.rom, args.bios) - if rc == 0: - print(f"Flash written successfully") + success, data = snipeit_api.check_in_asset(asset_id) + + if success: + print(f"Asset {asset_id} successfully checked in.") + return True else: - print(f"Flash write failed with code {rc}") + print(f"Error checking in asset {asset_id}") + print(f"Response data: {data}") + return False -def flash_erase(rte, args): +def get_my_assets(snipeit_api: SnipeIT): """ - Erases the flash memory of the device under test (DUT) using the rte object. + Gets a list of assets assigned to the current user Args: - rte: The object used to interact with the device, which includes methods for flash operations. - args (object): Arguments that may contain additional parameters (not used in this function). + snipeit_api: The API client used to interact with the Snipe-IT API. Returns: - None. + List of assets assigned to the current user """ - print(f"Erasing DUT flash...") - rte.flash_erase() - print(f"Flash erased") + all_assets = snipeit_api.get_all_assets() + used_assets = [ + asset for asset in all_assets if asset["assigned_to"] is not None + ] + return [ + asset + for asset in used_assets + if asset["assigned_to"]["id"] is snipeit_api.cfg_user_id + ] -def sonoff_on(sonoff, args): +def print_asset_details(asset): """ - Attempts to turn on the Sonoff power switch and prints the response. + Prints details of a given asset, including its basic information, assigned user (if any), + and custom fields. Args: - sonoff: An object responsible for controlling the Sonoff power switch. - args (object): Arguments that may contain additional parameters (not used in this function). + asset (dict): Dictionary containing asset details. Returns: - None. + None """ - print("Turning on Sonoff power switch...") - try: - response = sonoff.turn_on() - print(response) - except requests.exceptions.RequestException as e: - print(f"Failed to turn on Sonoff power switch. Error: {e}") + print( + f"Asset Tag: {asset['asset_tag']}, Asset ID: {asset['id']}," + f"Name: {asset['name']}, Serial: {asset['serial']}" + ) -def sonoff_off(sonoff, args): - """ - Attempts to turn off the Sonoff power switch and prints the response. + if asset["assigned_to"]: + print(f"Assigned to: {asset['assigned_to']['name']}") - Args: - sonoff: An object responsible for controlling the Sonoff power switch. - args (object): Arguments that may contain additional parameters (not used in this function). + custom_fields = asset.get("custom_fields", {}) + if custom_fields: + for field_name, field_data in custom_fields.items(): + field_value = field_data.get("value") + print(f"{field_name}: {field_value}") - Returns: - None. - """ - print("Turning off Sonoff power switch...") - try: - response = sonoff.turn_off() - print(response) - except requests.exceptions.RequestException as e: - print(f"Failed to turn off Sonoff power switch. Error: {e}") + print() -def sonoff_get(sonoff, args): +def get_zabbix_compatible_assets_from_asset(asset): """ - Retrieves the current state of the Sonoff power switch and prints it. + Extracts an asset to zabbix assets. Args: - sonoff: An object responsible for controlling the Sonoff power switch. - args (object): Arguments that may contain additional parameters (not used in this function). + asset (dict): Dictionary containing asset details. Returns: - None. + The dictionary with asset tags as keys and asset data as value. """ - print("Getting Sonoff power switch state...") - try: - state = sonoff.get_state() - print(f"Sonoff power switch state: {state}") - except requests.exceptions.RequestException as e: - print(f"Failed to get Sonoff power switch state. Error: {e}") + result = {} + custom_fields = asset.get("custom_fields", {}) + if custom_fields: + for field_name, field_data in custom_fields.items(): + if field_name in ["RTE IP", "Sonoff IP", "PiKVM IP"]: + field_value = field_data.get("value") + if field_value: + key = f"{asset['asset_tag']}_{field_name}".replace( + " ", "_" + ) + result[key] = field_value + return result -def sonoff_tgl(sonoff, args): +def print_asset_details_for_zabbix(asset): """ - Toggles the current state of the Sonoff power switch. + Print asset details formatted as an input for Zabbix import script. Args: - sonoff: An object responsible for controlling the Sonoff power switch. - args (object): Arguments that may contain additional parameters (not used in this function). + asset (dict): Dictionary containing asset details. Returns: None. """ - print("Toggling Sonoff power switch state...") - try: - current_state = sonoff.get_state() - - if current_state == "ON": - response = sonoff.turn_off() - print("Sonoff power switch state toggled off.") - elif current_state == "OFF": - response = sonoff.turn_on() - print("Sonoff power switch state toggled on.") - else: - print(f"Unexpected Sonoff power switch state: {current_state}") - except requests.exceptions.RequestException as e: - print(f"Failed to toggle Sonoff power switch state. Error: {e}") + assets = get_zabbix_compatible_assets_from_asset(asset) + for key in assets: + print(f"{key}: {assets[key]}") def ask_to_proceed(message="Do you want to proceed (y/n): "): @@ -808,7 +1414,7 @@ def ask_to_proceed(message="Do you want to proceed (y/n): "): Returns: bool: True if the user enters 'y', False if the user enters 'n'. """ - print("") + print() while True: choice = input(message).lower() if choice in ["y", "n"]: @@ -817,811 +1423,5 @@ def ask_to_proceed(message="Do you want to proceed (y/n): "): print("Invalid input. Please enter 'y' or 'n'.") -def update_zabbix_assets(snipeit_api): - """ - Updates Zabbix with the latest asset data from Snipe-IT, ensuring the IP addresses - are synchronized between Snipe-IT and Zabbix. - - Args: - snipeit_api: The API client used to interact with the Snipe-IT API. - - Returns: - None. - """ - zabbix = Zabbix() - all_assets = snipeit_api.get_all_assets() - - current_zabbix_assets = zabbix.get_all_hosts() - # snipeit assets but converted to zabbix-form assets - snipeit_assets = {} - - update_available = False - - if all_assets: - for asset in all_assets: - snipeit_assets.update( - get_zabbix_compatible_assets_from_asset(asset) - ) - - snipeit_assets_keys = list(snipeit_assets.keys()) - - snipeit_configuration_error = False - - forbidden_symbols = [ - "/", - "\\", - "{", - "}", - ";", - ":", - "~", - "`", - '"', - "'", - "[", - "]", - "|", - "<", - ">", - "$", - "#", - "@", - "%", - "^", - "&", - "*", - "(", - ")", - "+", - "=", - ] - - for i in range(snipeit_assets.__len__()): - # check for duplicates - for j in range(i + 1, snipeit_assets.__len__()): - if ( - snipeit_assets[snipeit_assets_keys[i]] - == snipeit_assets[snipeit_assets_keys[j]] - ): - print( - f"{snipeit_assets_keys[i]} has the same IP as " - f"{snipeit_assets_keys[j]}!" - ) - snipeit_configuration_error = True - if snipeit_assets_keys[i] == snipeit_assets_keys[j]: - print( - f"There are at least 2 assets with name " - f"{snipeit_assets_keys[i]} present!" - ) - snipeit_configuration_error = True - - # check for forbidden symbols in asset names - if any( - symbol in snipeit_assets_keys[i] for symbol in forbidden_symbols - ): - print( - f"{snipeit_assets_keys[i]} contains forbidden symbols! They " - f"are going to be changed to '_'." - ) - new_key = copy(snipeit_assets_keys[i]) - for s in forbidden_symbols: - new_key = new_key.replace(s, "_") - - snipeit_assets[new_key] = snipeit_assets.pop( - snipeit_assets_keys[i] - ) - - if snipeit_configuration_error: - print( - "\nSnipeIT configuration errors have been detected! " - "Fix them and then continue." - ) - return - - keys_not_present_in_zabbix = set(snipeit_assets.keys()) - set( - current_zabbix_assets.keys() - ) - - keys_not_present_in_snipeit = set(current_zabbix_assets.keys()) - set( - snipeit_assets.keys() - ) - - if ( - keys_not_present_in_zabbix.__len__() > 0 - or keys_not_present_in_snipeit.__len__() > 0 - ): - update_available = True - - common_keys = set(snipeit_assets.keys()) & set( - current_zabbix_assets.keys() - ) - - if keys_not_present_in_zabbix.__len__() > 0: - print("Assets not present in Zabbix (these will be added):") - print("\n".join(keys_not_present_in_zabbix)) - - if keys_not_present_in_snipeit.__len__() > 0: - print( - "\nAssets present in Zabbix but not in SnipeIT " - "(these will be removed):" - ) - print("\n".join(keys_not_present_in_snipeit)) - - print("") - keys_for_ip_change = [] - for key in common_keys: - if snipeit_assets[key] != current_zabbix_assets[key]: - print( - f"{key} has wrong IP! (Zabbix one will be updated from " - f"{current_zabbix_assets[key]} to {snipeit_assets[key]})" - ) - keys_for_ip_change.append(key) - - if keys_for_ip_change.__len__() > 0: - update_available = True - - if not update_available: - print("Zabbix is already synced with SnipeIT") - return - - if not ask_to_proceed("Do you want to apply above changes? (y/n): "): - print("Changes were not applied") - return - - # removing zabbix hosts - for key in keys_not_present_in_snipeit: - print(f"Removing {key}({current_zabbix_assets[key]})...") - result = zabbix.remove_host_by_name(key) - if "error" in result: - print("Failed to remove the host!") - - # updating zabbix ips - for key in keys_for_ip_change: - print(f"Updating {key} IP to {snipeit_assets[key]}...") - result = zabbix.update_host_ip(key, snipeit_assets[key]) - if "error" in result: - print("Failed to change host's IP!") - - # adding zabbix hosts - for key in keys_not_present_in_zabbix: - print(f"Adding {key}({snipeit_assets[key]})...") - try: - result = zabbix.add_host(key, snipeit_assets[key]) - except ValueError: - print("Failed to add the host!") - - -def list_models(args): - models = Models() - models.list_models() - - -def flash_image_check(args): - """ - Checks for existence & sane content of ME region in flash image. - - Args: - args (object): Arguments that may contain additional parameters: - args.list (bool): List known regions and exit - args.rom (str): Flash image file path & name - args.dry_mecheck (bool): runs osfv.libs.flash_image in dry run mode; - exit status is always positive - args.verbosity (bool): increases osfv.libs.flash_image verbosity - args.regions_to_check [str]: list of regions to check - args.regions_to_dump [str]: list of regions to dump to separate files - - Returns: - None. - """ - - regions_to_check = ["me"] - - if args.list: - print("Known flash regions:") - for reg_name in utils.get_list_of_known_image_regions(): - print(reg_name) - exit() - if args.regions_to_check: - regions_to_check = args.regions_to_check - if args.regions_to_dump: - utils.dump_flash_image_regions( - args.rom, args.verbosity, args.regions_to_dump - ) - if ( - utils.check_flash_image_regions( - args.rom, args.dry_mecheck, args.verbosity, regions_to_check - ) - == False - ): - exit( - "Do not flash full image, unless you are skipping empty regions, and know what you are doing!" - ) - - -def reset_cmos(rte, args): - """ - Resets the CMOS of the Device Under Test (DUT). - Args: - rte (object): The object representing the relay control and power supply interface. - args (object): Arguments that may contain additional parameters (not used in this function). - Returns: - None - """ - print(f"Clearing CMOS...") - rte.reset_cmos() - - -# Main function -def main(): - parser = argparse.ArgumentParser( - description="Open Source Firmware Validation CLI" - ) - parser.add_argument( - "-v", "--version", action="version", version=metadata.version("osfv") - ) - - parser.add_argument( - "-j", - "--json", - action="store_true", - help="Output as JSON (if applicable)", - ) - - subparsers = parser.add_subparsers( - title="commands", dest="command", help="Command to execute" - ) - - snipeit_parser = subparsers.add_parser("snipeit", help="Snipe-IT commands") - - rte_parser = subparsers.add_parser("rte", help="RTE commands") - sonoff_parser = subparsers.add_parser("sonoff", help="Sonoff commands") - list_models_parser = subparsers.add_parser( - "list_models", help="List of supported models" - ) - - # Sonoff subcommands - sonoff_group = sonoff_parser.add_mutually_exclusive_group(required=True) - sonoff_group.add_argument( - "--sonoff_ip", type=str, help="Sonoff IP address" - ) - sonoff_group.add_argument("--rte_ip", type=str, help="RTE IP address") - sonoff_subparsers = sonoff_parser.add_subparsers( - title="subcommands", dest="sonoff_cmd", help="Sonoff subcommands" - ) - - sonoff_subparsers.add_parser("on", help="Turn Sonoff ON") - sonoff_subparsers.add_parser("off", help="Turn Sonoff OFF") - sonoff_subparsers.add_parser("tgl", help="Toggle Sonoff state") - sonoff_subparsers.add_parser("get", help="Get Sonoff state") - - # Snipe-IT subcommands - snipeit_subparsers = snipeit_parser.add_subparsers( - title="subcommands", dest="snipeit_cmd", help="Snipe-IT subcommands" - ) - - list_used_parser = snipeit_subparsers.add_parser( - "list_used", help="List all already used assets" - ) - - list_my_parser = snipeit_subparsers.add_parser( - "list_my", help="List all my used assets" - ) - - list_unused_parser = snipeit_subparsers.add_parser( - "list_unused", help="List all unused assets" - ) - - list_all_parser = snipeit_subparsers.add_parser( - "list_all", help="List all assets" - ) - - list_zabbix_parser = snipeit_subparsers.add_parser( - "list_for_zabbix", - help="List assets in a format suitable for Zabbix integration", - ) - - update_zabbix_assets_parser = snipeit_subparsers.add_parser( - "update_zabbix", - help="Syncs Zabbix assets with SnipeIT ones", - ) - - check_out_parser = snipeit_subparsers.add_parser( - "check_out", - help="Check out an asset by providing the Asset ID or RTE IP", - ) - check_out_group = check_out_parser.add_mutually_exclusive_group( - required=True - ) - check_out_group.add_argument("--asset_id", type=int, help="Asset ID") - check_out_group.add_argument("--rte_ip", type=str, help="RTE IP") - check_out_parser = snipeit_subparsers.add_parser( - "user_add", - help="Add a new user by providing user First Name and Last Name", - ) - check_out_parser.add_argument( - "--first-name", type=str, help="User First Name", required=True - ) - check_out_parser.add_argument( - "--last-name", type=str, help="User Last Name", required=True - ) - check_out_parser.add_argument( - "--company-name", type=str, default="3mdeb", help="Company Name" - ) - check_out_parser = snipeit_subparsers.add_parser( - "user_del", - help="Delete new user by providing user First Name and Last Name", - ) - check_out_parser.add_argument( - "--first-name", type=str, help="User First Name", required=True - ) - check_out_parser.add_argument( - "--last-name", type=str, help="User Last Name", required=True - ) - - check_in_parser = snipeit_subparsers.add_parser( - "check_in", - help="Check in an asset by providing the Asset ID or RTE IP", - ) - check_in_group = check_in_parser.add_mutually_exclusive_group( - required=True - ) - check_in_group.add_argument("--asset_id", type=int, help="Asset ID") - check_in_group.add_argument("--rte_ip", type=str, help="RTE IP address") - - check_in_my_parser = snipeit_subparsers.add_parser( - "check_in_my", help="Check in all my used assets, except work laptops" - ) - check_in_my_parser.add_argument( - "-y", "--yes", action="store_true", help="Skips the confirmation" - ) - - # RTE subcommands - rte_parser.add_argument( - "--rte_ip", type=str, help="RTE IP address", required=True - ) - rte_parser.add_argument( - "--model", - type=str, - help="DUT model. If not given, will attempt to query from Snipe-IT.", - required=False, - ) - rte_parser.add_argument( - "--skip-snipeit", - action="store_true", - help=( - f"Skips Snipe-IT related actions like checkout and check-in. " - f"Useful for OSFV homelab." - ), - ) - rte_subparsers = rte_parser.add_subparsers( - title="subcommands", dest="rte_cmd", help="RTE subcommands" - ) - rel_parser = rte_subparsers.add_parser("rel", help="Control RTE relay") - gpio_parser = rte_subparsers.add_parser("gpio", help="Control RTE GPIO") - pwr_parser = rte_subparsers.add_parser( - "pwr", help="Control DUT power via RTE" - ) - spi_parser = rte_subparsers.add_parser( - "spi", help="Control SPI lines of RTE" - ) - serial_parser = rte_subparsers.add_parser( - "serial", help="Open DUT serial via telnet" - ) - flash_parser = rte_subparsers.add_parser( - "flash", help="DUT flash operations" - ) - - # Power subcommands - pwr_subparsers = pwr_parser.add_subparsers( - title="subcommands", dest="pwr_cmd" - ) - power_on_parser = pwr_subparsers.add_parser( - "on", help="Short power button press, to power on DUT" - ) - power_on_parser.add_argument( - "--time", - type=int, - default=1, - help="Power button press time in seconds (default: 1)", - ) - power_on_ex_parser = pwr_subparsers.add_parser( - "on_ex", help="Short power button press, to power on DUT" - ) - power_on_ex_parser.add_argument( - "--time", - type=int, - default=1, - help="Power button press time in seconds (default: 1) AND verify if power LED did light up", - ) - - power_off_parser = pwr_subparsers.add_parser( - "off_ex", help="Long power button press, to power off DUT" - ) - power_off_parser.add_argument( - "--time", - type=int, - default=6, - help="Power button press time in seconds (default: 6)", - ) - power_off_ex_parser = pwr_subparsers.add_parser( - "off", - help="Long power button press, to power off DUT and verify if power LED did turn off", - ) - power_off_ex_parser.add_argument( - "--time", - type=int, - default=6, - help="Power button press time in seconds (default: 6)", - ) - - reset_parser = pwr_subparsers.add_parser( - "reset", help="Reset button press, to reset DUT" - ) - reset_parser.add_argument( - "--time", - type=int, - default=1, - help="Reset button press time in seconds (default: 1)", - ) - psu_parser = pwr_subparsers.add_parser( - "psu", help="Generic control interface of the power supply" - ) - psu_subparsers = psu_parser.add_subparsers( - title="Power supply commands", dest="psu_cmd" - ) - psu_subparsers.add_parser("on", help="Turn the power supply on") - psu_subparsers.add_parser("off", help="Turn the power supply off") - psu_subparsers.add_parser( - "get", help="Display information on DUT's power state" - ) - check_pwr_led_parser = pwr_subparsers.add_parser( - "pwr_led", help="Check the state of the DUT power LED" - ) - - cmos_reset_subparser = pwr_subparsers.add_parser( - "reset_cmos", help="Reset the DUT CMOS" - ) - - # GPIO subcommands - gpio_subparsers = gpio_parser.add_subparsers( - title="subcommands", dest="gpio_cmd" - ) - get_gpio_parser = gpio_subparsers.add_parser("get", help="Get GPIO state") - get_gpio_parser.add_argument("gpio_no", type=int, help="GPIO number") - set_gpio_parser = gpio_subparsers.add_parser("set", help="Set GPIO state") - set_gpio_parser.add_argument("gpio_no", type=int, help="GPIO number") - set_gpio_parser.add_argument( - "state", choices=["high", "low", "high-z"], help="GPIO state" - ) - set_gpio_parser = gpio_subparsers.add_parser( - "list", help="List GPIO states" - ) - - # Relay subcommands - rel_subparsers = rel_parser.add_subparsers( - title="subcommands", dest="rel_cmd" - ) - tgl_rel_parser = rel_subparsers.add_parser( - "tgl", help="Toggle relay state" - ) - get_rel_parser = rel_subparsers.add_parser("get", help="Get relay state") - set_rel_parser = rel_subparsers.add_parser("set", help="Set relay state") - set_rel_parser.add_argument( - "state", - choices=[ - "on", - "off", - ], - help="Relay state", - ) - - # RTE SPI subcommands - spi_subparsers = spi_parser.add_subparsers( - title="subcommands", dest="spi_cmd" - ) - spi_on_parser = spi_subparsers.add_parser("on", help="Enable SPI lines") - spi_on_parser.add_argument( - "--voltage", - type=str, - default="1.8V", - help="SPI voltage (default: 1.8V)", - ) - spi_off_parser = spi_subparsers.add_parser("off", help="Disable SPI lines") - - # RTE flash subcommands - flash_subparsers = flash_parser.add_subparsers( - title="subcommands", dest="flash_cmd" - ) - flash_probe_parser = flash_subparsers.add_parser( - "probe", help="Flash probe with flashrom" - ) - flash_read_parser = flash_subparsers.add_parser( - "read", help="Read from DUT flash with flashrom" - ) - flash_read_parser.add_argument( - "--rom", - type=str, - default="read.rom", - help="Path to read firmware file (default: read.rom)", - ) - flash_write_parser = flash_subparsers.add_parser( - "write", help="Write to DUT flash with flashrom" - ) - flash_write_parser.add_argument( - "--rom", - type=str, - default="write.rom", - help="Path to read firmware file (default: write.rom)", - ) - flash_write_parser.add_argument( - "-b", - "--bios", - action="store_true", - help='Adds "-i bios --ifd" to flashrom command', - ) - flash_write_parser.add_argument( - "-x", - "--dry-mecheck", - help="Failed flash region checks won't forbid flashing", - action="store_true", - ) - flash_write_parser.add_argument( - "-V", - "--verbosity", - help="Increase osfv.libs.flash_image verbosity", - action="store_true", - ) - flash_erase_parser = flash_subparsers.add_parser( - "erase", help="Erase DUT flash with flashrom" - ) - - flash_image_check_parser = subparsers.add_parser( - "flash_image_check", - help="Check flash image completeness: descriptor & ME region existence", - ) - flash_image_check_parser.add_argument( - "--rom", - type=str, - required=True, - help="Path to read firmware file (default: write.rom)", - ) - flash_image_check_parser.add_argument( - "-x", - "--dry-mecheck", - help="Failed flash region checks won't change exit status", - action="store_true", - ) - flash_image_check_parser.add_argument( - "-V", - "--verbosity", - help="Increase osfv.libs.flash_image verbosity", - action="store_true", - ) - flash_image_check_parser.add_argument( - "-l", - "--list", - help="list known region names and exit", - action="store_true", - dest="list", - ) - flash_image_check_parser.add_argument( - "-c", - "--check", - help="check named flash region", - action="append", - type=str, - dest="regions_to_check", - ) - flash_image_check_parser.add_argument( - "-d", - "--dump", - help="dump named flash region", - action="append", - type=str, - dest="regions_to_dump", - ) - - args = parser.parse_args() - - snipeit_api = SnipeIT() - - if args.command == "snipeit": - if args.snipeit_cmd == "list_used": - list_used_assets(snipeit_api, args) - elif args.snipeit_cmd == "list_my": - list_my_assets(snipeit_api, args) - elif args.snipeit_cmd == "list_unused": - list_unused_assets(snipeit_api, args) - elif args.snipeit_cmd == "list_all": - list_all_assets(snipeit_api, args) - elif args.snipeit_cmd == "list_for_zabbix": - list_for_zabbix(snipeit_api, args) - elif args.snipeit_cmd == "check_out": - if args.asset_id: - check_out_asset(snipeit_api, args.asset_id) - elif args.rte_ip: - asset_id = snipeit_api.get_asset_id_by_rte_ip(args.rte_ip) - if asset_id: - check_out_asset(snipeit_api, asset_id) - else: - print(f"No asset found with RTE IP: {args.rte_ip}") - elif args.snipeit_cmd == "check_in": - if args.asset_id: - check_in_asset(snipeit_api, args.asset_id) - elif args.rte_ip: - asset_id = snipeit_api.get_asset_id_by_rte_ip(args.rte_ip) - if asset_id: - check_in_asset(snipeit_api, asset_id) - else: - print(f"No asset found with RTE IP: {args.rte_ip}") - elif args.snipeit_cmd == "check_in_my": - check_in_my(snipeit_api, args) - elif args.snipeit_cmd == "user_add": - snipeit_api.user_add( - args.first_name, args.last_name, args.company_name - ) - elif args.snipeit_cmd == "user_del": - snipeit_api.user_del(args.first_name, args.last_name) - elif args.snipeit_cmd == "update_zabbix": - update_zabbix_assets(snipeit_api) - - elif args.command == "rte": - if not args.skip_snipeit: - asset_id = snipeit_api.get_asset_id_by_rte_ip(args.rte_ip) - if not asset_id: - print(f"No asset found with RTE IP: {args.rte_ip}") - - if args.model: - print(f"DUT model retrieved from cmdline, skipping Snipe-IT query") - dut_model_name = args.model - else: - if not args.skip_snipeit: - status, dut_model_name = snipeit_api.get_asset_model_name( - asset_id - ) - if status: - print( - f"DUT model retrieved from snipeit: {dut_model_name}" - ) - else: - exit( - f"failed to retrieve model name from snipe-it. check " - f"again arguments, or try providing model manually." - ) - else: - exit(f"model name not present. check again arguments.") - sonoff, sonoff_ip = utils.init_sonoff(None, args.rte_ip, snipeit_api) - rte = RTE(args.rte_ip, dut_model_name, sonoff) - - if not args.skip_snipeit: - print( - f"Using rte command is invasive action, checking first if the " - f"device is not used..." - ) - already_checked_out = check_out_asset(snipeit_api, asset_id) - - if args.rte_cmd == "rel": - # Handle RTE relay related commands - if args.rel_cmd == "tgl": - relay_toggle(rte, args) - elif args.rel_cmd == "get": - relay_get(rte, args) - elif args.rel_cmd == "set": - relay_set(rte, args) - elif args.rte_cmd == "gpio": - # Handle RTE GPIO related commands - if args.gpio_cmd == "get": - gpio_get(rte, args) - elif args.gpio_cmd == "set": - gpio_set(rte, args) - elif args.gpio_cmd == "list": - gpio_list(rte, args) - elif args.rte_cmd == "pwr": - # Handle RTE power related commands - if args.pwr_cmd == "on": - power_on(rte, args) - elif args.pwr_cmd == "off": - power_off(rte, args) - if args.pwr_cmd == "on_ex": - power_on_ex(rte, args) - elif args.pwr_cmd == "off_ex": - power_off_ex(rte, args) - elif args.pwr_cmd == "reset": - reset(rte, args) - elif args.pwr_cmd == "pwr_led": - check_pwr_led(rte, args) - elif args.pwr_cmd == "psu": - if args.psu_cmd == "on": - psu_on(rte, args) - elif args.psu_cmd == "off": - psu_off(rte, args) - elif args.psu_cmd == "get": - psu_get(rte, args) - elif args.pwr_cmd == "reset_cmos": - reset_cmos(rte, args) - elif args.rte_cmd == "serial": - open_dut_serial(rte, args) - elif args.rte_cmd == "spi": - if args.spi_cmd == "on": - spi_on(rte, args) - elif args.spi_cmd == "off": - spi_off(rte, args) - elif args.rte_cmd == "flash": - if args.flash_cmd == "probe": - flash_probe(rte, args) - elif args.flash_cmd == "read": - flash_read(rte, args) - elif args.flash_cmd == "write": - flash_write(rte, args) - elif args.flash_cmd == "erase": - flash_erase(rte, args) - - if not args.skip_snipeit: - if already_checked_out: - print( - f"Since the asset {asset_id} has been checkout manually " - f"by you prior running this script, it will NOT be checked " - f"in automatically. Please return the device when work is " - f"finished." - ) - else: - print( - f"Since the asset {asset_id} has been checkout " - f"automatically by this script, it is automatically " - f"checked in as well." - ) - check_in_asset(snipeit_api, asset_id) - elif args.command == "sonoff": - sonoff_ip = "" - - if args.sonoff_ip: - sonoff_ip = args.sonoff_ip - elif args.rte_ip: - sonoff_ip = snipeit_api.get_sonoff_ip_by_rte_ip(args.rte_ip) - if not sonoff_ip: - print(f"No Sonoff Device found with RTE IP: {args.rte_ip}") - - asset_id = snipeit_api.get_asset_id_by_sonoff_ip(sonoff_ip) - if not asset_id: - print(f"No asset found with RTE IP: {args.rte_ip}") - - print( - f"Using rte command is invasive action, checking first if the " - f"device is not used..." - ) - already_checked_out = check_out_asset(snipeit_api, asset_id) - - sonoff = SonoffDevice(sonoff_ip) - - if args.sonoff_cmd == "on": - sonoff_on(sonoff, args) - if args.sonoff_cmd == "off": - sonoff_off(sonoff, args) - if args.sonoff_cmd == "get": - sonoff_get(sonoff, args) - if args.sonoff_cmd == "tgl": - sonoff_tgl(sonoff, args) - - if already_checked_out: - print( - f"Since the asset {asset_id} has been checkout manually by " - f"you prior running this script, it will NOT be checked in " - f"automatically. Please return the device when work is " - f"finished." - ) - else: - print( - f"Since the asset {asset_id} has been checkout automatically " - f"by this script, it is automatically checked in as well." - ) - check_in_asset(snipeit_api, asset_id) - elif args.command == "flash_image_check": - flash_image_check(args) - elif args.command == "list_models": - list_models(args) - else: - parser.print_help() - - if __name__ == "__main__": main() diff --git a/robocop.toml b/robocop.toml index 81cfdac..6983c3a 100644 --- a/robocop.toml +++ b/robocop.toml @@ -1,42 +1,40 @@ -[tool.robocop] -configure = [ +[tool.robocop.lint] +ignore = [ # TODO: add missing documentation for keywords and re-enable - "0201:enabled:False", + "DOC01", # TODO: add missing documentation for test cases and re-enable - "0202:enabled:False", + "DOC02", # TODO: add missing documentation for test suitdes and re-enable - "0203:enabled:False", + "DOC03", # TODO: consider re-enabling - "0301:enabled:False", + "NAME01", # TODO: consider fixing and re-enabling - "0316:enabled:False", + "VAR08", # TODO: fix and re-enable - "0319:enabled:False", + "DEPR05", # TODO: fix and re-enable - "0327:enabled:False", + "DEPR06", # TODO: fix and re-enable - "0328:enabled:False", + "COM01", + "MISC06", + # TODO: fix unused variables and re-enable + "VAR02", + # TODO: consider fixing and re-enabling + "VAR03", + # TODO: fix and re-enable + "MISC13", +] +configure = [ # TODO: consider reducing to default 40, or less - "0501:max_len:50", + "LEN01.max_len=50", # TODO: consider reducing to default 10, or less - "0503:max_calls:50", + "LEN03.max_calls=50", # TODO: consider reducing to dfeault 20, or less - "0504:max_len:50", + "LEN04.max_len=50", # TODO: consider reducing to dfeault 10, or less - "0505:max_calls:50", + "LEN06.max_calls=50", # TODO: reduce to default 400 or so - "0506:max_lines:5000", + "LEN28.max_lines=5000", # TODO: reduce to 120/80 - "0508:line_length:180", - "0701:enabled:False", - # TODO: fix and re-enable - "0908:enabled:False", - # TODO: re-enable after fixes - "0913:enabled:False", - # TODO: fix unused variables and re-enable - "0920:enabled:False", - # TODO: consider fixing and re-enabling - "0922:enabled:False", - # TODO: fix and re-enable - "0924:enabled:False" + "LEN08.line_length=180", ]