Фиксация наработок

obsdsnmp/__init__.py

@@ -0,0 +1,84 @@
+# coding: utf-8
+"""Command line interface."""
+
+import asyncio
+import logging
+from typing import Optional, Tuple
+
+import click
+from pysnmp.hlapi.asyncio import SnmpEngine
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+
+try:
+    import uvloop  # noqa: WPS433
+except ImportError:
+    logging.info('For performance boost install uvloop')
+else:
+    uvloop.install()
+
+
+DEFAULT_INTERVAL = 10.0
+
+
+async def worker(
+    interval: float,
+    host: str,
+    db_engine: Optional[AsyncEngine],
+):
+    """Connect to host, gather metrics and store to database.
+
+    Args:
+        interval: seconds to wait between iterations.
+        host: hostname or IP of the host.
+        db_engine: SQLAlchemy Engine object.
+
+    """
+    snmp_engine = SnmpEngine()
+    while True:
+        print(host)
+        await asyncio.sleep(interval)
+
+
+async def scheduler(interval: float, dsn: str, hosts: Tuple[str]):
+    """Schdule async worker for each host.
+
+    Args:
+        interval: seconds to wait between iterations.
+        dsn: SQLAlchemy connection string.
+        hosts: hosts to monitor.
+
+    """
+    db_engine = None
+    if dsn:
+        db_engine = create_async_engine(dsn, echo='debug', future=True)
+        async with db_engine.begin() as conn:
+            # await conn.run_sync(meta.drop_all)
+            # await conn.run_sync(meta.create_all)
+            pass
+    await asyncio.gather(
+        *(worker(interval, host, db_engine) for host in hosts),
+    )
+
+
+@click.command()
+@click.option(
+    '-i',
+    '--interval',
+    type=click.FloatRange(0, min_open=True),
+    default=DEFAULT_INTERVAL,
+    help='Collect metrics every interval seconds.',
+    show_default=True,
+)
+@click.option(
+    '-d',
+    '--dsn',
+    help='DSN connection string to store metrics.',
+    show_default=True,
+)
+@click.argument('hosts', nargs=-1, required=True)
+def command(*args, **kwargs):
+    """Collect OpenBSD SNMP metrics from HOST.
+
+    Multiple hosts can be specified with spaces.
+    """  # noqa: DAR101
+    asyncio.run(scheduler(*args, **kwargs))

obsdsnmp/__main__.py

@@ -0,0 +1,6 @@
+# coding: utf-8
+"""Code to be run if invoked with python -m."""
+
+from obsdsnmp import command
+
+command(auto_envvar_prefix='OBSDSNMP')

obsdsnmp/models.py

@@ -0,0 +1,25 @@
+# coding: utf-8
+"""SQLAlchemy models for monitored hosts."""
+
+import sqlalchemy as sa
+
+metadata = sa.MetaData()
+
+MAX_DOMAIN_NAME = 255
+host = sa.Table(
+    'host',
+    metadata,
+    sa.Column('id', sa.Integer, primary_key=True),
+    sa.Column(
+        'hostname',
+        sa.String(MAX_DOMAIN_NAME),
+        unique=True,
+        nullable=False,
+    ),
+    sa.Column('sysname', sa.Text, nullable=False),
+    sa.Column('description', sa.Text, nullable=False),
+    sa.Column('contact', sa.Text, nullable=False),
+    sa.Column('location', sa.Text, nullable=False),
+    sa.Column('uptime', sa.Time, nullable=False),
+    sa.Column('modified', sa.DateTime, onupdate=sa.func.utc_timestamp()),
+)

tests/snapshots/test_cli/test_empty_run/empty_run

@@ -0,0 +1,4 @@
+Usage: command [OPTIONS] HOSTS...
+Try 'command --help' for help.
+
+Error: Missing argument 'HOSTS...'.

tests/snapshots/test_cli/test_help/help

@@ -0,0 +1,11 @@
+Usage: command [OPTIONS] HOSTS...
+
+  Collect OpenBSD SNMP metrics from HOST.
+
+  Multiple hosts can be specified with spaces.
+
+Options:
+  -i, --interval FLOAT RANGE  Collect metrics every interval seconds.  [default:
+                              10.0;x>0]
+  -d, --dsn TEXT              DSN connection string to store metrics.
+  --help                      Show this message and exit.

tests/test_cli.py

@@ -0,0 +1,88 @@
+# coding: utf-8
+"""Test CLI of the application."""
+
+from runpy import run_module
+
+from click.testing import CliRunner
+from pytest import MonkeyPatch, CaptureFixture
+from pytest_snapshot.plugin import Snapshot
+
+import obsdsnmp
+
+
+def printer(*args, **kwargs):
+    """Monkeypatch function, that prints args kwargs to stdout.
+
+    Args:
+        args: positional arguments.
+        kwargs: key-value arguments.
+    """
+    print(args, kwargs)  # noqa: WPS421
+
+
+async def async_printer(*args, **kwargs):
+    """Monkeypatch function, that prints args kwargs to stdout.
+
+    Args:
+        args: positional arguments.
+        kwargs: key-value arguments.
+    """
+    print(args, kwargs)  # noqa: WPS421
+
+
+def test_empty_run(snapshot: Snapshot):
+    """Test invoking without arguments.
+
+    The app shall give error information.
+
+    Args:
+        snapshot: instance of pytest_snapshot fixture.
+
+    """
+    runner = CliRunner()
+    invoke_result = runner.invoke(obsdsnmp.command, [])
+    assert invoke_result.exit_code == 2
+    snapshot.assert_match(invoke_result.output, 'empty_run')
+
+
+def test_help(snapshot: Snapshot):
+    """Test invoking help.
+
+    The app shall give help.
+
+    Args:
+        snapshot: instance of pytest_snapshot fixture.
+
+    """
+    runner = CliRunner()
+    invoke_result = runner.invoke(obsdsnmp.command, ['--help'])
+    assert invoke_result.exit_code == 0
+    snapshot.assert_match(invoke_result.output, 'help')
+
+
+def test_worker(monkeypatch: MonkeyPatch):
+    """Test setting interval.
+
+    Args:
+        monkeypatch: pytest monkeypatch fixture.
+    """
+    monkeypatch.setattr(obsdsnmp, 'worker', async_printer)
+    runner = CliRunner()
+    invoke_result = runner.invoke(obsdsnmp.command, ['-i', '1', 'test'])
+    assert invoke_result.exit_code == 0
+    assert invoke_result.output == '{0} {1}\n'.format((1.0, 'test', None), {})
+
+
+def test_package(monkeypatch: MonkeyPatch, capsys: CaptureFixture):
+    """Test __main__ invokation.
+
+    Args:
+        monkeypatch: pytest monkeypatch fixture.
+        capsys: capture stdout fixture.
+    """
+    monkeypatch.setattr(obsdsnmp, 'command', printer)
+    run_module('obsdsnmp')
+    captured = capsys.readouterr()
+    assert captured.out == '{0} {1}\n'.format((), {
+        'auto_envvar_prefix': 'OBSDSNMP',
+    })