Commit 4f3ab5b6 authored by Marius Kriegerowski's avatar Marius Kriegerowski
Browse files

Added basic rcli implementation

parent 8db3887b
Pipeline #20678 passed with stage
in 2 minutes and 42 seconds
image: python:3-buster
before_script:
- python3 -V
- pip3 install pytest pytest-cov
- pip3 install git+https://git.gfz-potsdam.de/dynamicexposure/rabotnik/rabotnik.git
- pip3 install .[tests]
- pip3 install .
formatting:
script:
- pip3 install black flake8 pylint
- make check
test:
script:
- pytest tests
coverage:
script:
- pytest --cov=rcli/ tests/
- coverage xml
artifacts:
reports:
cobertura: coverage.xml
SOURCES=rcli tests
LENGTH=96
check: $(SOURCES)
flake8 --max-line-length=$(LENGTH) $^
black --check --line-length $(LENGTH) $^
pylint -E $^
format: $(SOURCES)
black --line-length $(LENGTH) $^
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
import re
from argparse import ArgumentParser
def format_argument(argument):
"""Format arguments the same way as the `argparse.ArgumentParser` does."""
# strip possible preceding dashes
argument = re.sub(r"^-+", "", argument)
# replace intermediate dashes with underscore
return re.sub(r"-", "_", argument)
class BuildingIDArgument:
def __init__(self):
self.argument = "--building-id"
self.rabotnik_message = "building"
def add_argument_to_parser(self, parser: ArgumentParser):
"""Attach this instance to a `parser`."""
parser.add_argument(
self.argument, type=str, help="Single `building_id` as integer", required=True
)
def process_arguments(self, arguments: dict):
arg = format_argument(self.argument)
return {arg: arguments[arg]}
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
import argparse
import sys
import asyncio
import logging
from rabotnik.bus import MessageBus, PASSWORD, USERNAME, URL
from .building_id_argument import BuildingIDArgument
logger = logging.getLogger(__name__)
class RCLI:
"""The Rabotnik Command-Line Interface (RCLI).
Communicate with nodes running within the `rabotnik` ecosystem via the `message_bus`.
Args:
parser: Creates and parses command line arguments.
Attributes:
argument: A `BuildingIDArgument` that is added the `parser`.
"""
def __init__(self, parser: argparse.ArgumentParser):
self.parser = parser
self.argument = BuildingIDArgument()
self.message_bus = None
def setup_parser(self):
self.argument.add_argument_to_parser(self.parser)
def set_message_bus(self, message_bus: MessageBus):
self.message_bus = message_bus
async def execute(self, args):
await self.ensure_message_bus_connected()
try:
payload = self.argument.process_arguments(args)
except KeyError as ex:
logger.debug(f"Argument {ex} not provided")
return
message = self.argument.rabotnik_message
logger.info(f"Dispatching {payload} on {message}")
await self.message_bus.send(message=message, payload=payload)
async def ensure_message_bus_connected(self):
if not self.message_bus:
raise AttributeError("message_bus not set")
logger.info("Waiting for message bus to connect")
await self.message_bus.connected.wait()
def main(args: list = None, message_bus: MessageBus = None):
"""RCLI entry point.
Args:
args: List of arguments. If not provided, read arguments from commandline
message_bus: Dispatches messages. If not provided, create a new
`rabotnik.bus.MessageBus` instance
"""
logging.basicConfig(level=logging.INFO)
args = args or sys.argv[1:]
parser = argparse.ArgumentParser()
rcli = RCLI(parser=parser)
rcli.setup_parser()
group = parser.add_argument_group("message bus")
group.add_argument(
"--hostname", type=str, default=URL, help=f"set message bus hostname. Default: {URL}"
)
group.add_argument(
"--username",
type=str,
default=USERNAME,
help=f"set message bus username. Default: {USERNAME}",
)
group.add_argument(
"--password",
type=str,
default=PASSWORD,
help=f"set message bus password. Default: {PASSWORD}",
)
parser.add_argument(
"-v", "--verbose", action="store_true", help="verbose mode", dest="verbose"
)
args = parser.parse_args(args)
if args.verbose:
logging.getLogger("rabotnik").setLevel(logging.DEBUG)
logger.setLevel(logging.DEBUG)
if message_bus is None:
message_bus = MessageBus(
url=args.hostname, username=args.username, password=args.password
)
rcli.set_message_bus(message_bus)
# Parse `args` to dictionary.
args_dict = vars(args)
loop = asyncio.get_event_loop()
loop.run_until_complete(rcli.execute(args_dict))
#!/usr/bin/python3
from setuptools import setup, find_packages
tests_require = [
"pytest",
"pytest-asyncio",
"pytest-mock",
"pytest-cov",
]
linters_require = ["black>=20.8b1", "pylint", "flake8"]
setup(
name="rcli",
version="0.1",
description="Rabotnik command-line interface",
author="Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ",
license="AGPLv3+",
install_requires=[
"rabotnik@https://git.gfz-potsdam.de/dynamicexposure/rabotnik/rabotnik/-/archive/master/rabotnik-master.zip"
],
tests_require=tests_require,
extras_require={
"tests": tests_require,
"linters": linters_require,
},
entry_points={"console_scripts": ["rcli = rcli.rcli:main"]},
packages=find_packages(),
python_requires=">=3.6",
)
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
from rcli.building_id_argument import format_argument
def test_format_argument():
assert format_argument("--x") == "x"
assert format_argument("--x-x") == "x_x"
assert format_argument("x-x") == "x_x"
assert format_argument("x_x") == "x_x"
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
import asyncio
import pytest
import argparse
from rcli.rcli import RCLI
class MockMessageBus:
def __init__(self):
self.calls = []
self.connected = asyncio.Event()
self.connected.set()
async def send(self, *args, **kwargs):
self.calls.append((args, kwargs))
@pytest.mark.asyncio
async def test_rcli():
message_bus = MockMessageBus()
parser = argparse.ArgumentParser()
rcli = RCLI(parser=parser)
rcli.set_message_bus(message_bus)
rcli.setup_parser()
help_formatted = parser.format_help()
assert help_formatted.startswith("usage:")
arguments = {"building_id": "1"}
await rcli.execute(arguments)
assert message_bus.calls == [((), {"message": "building", "payload": {"building_id": "1"}})]
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment