mirror
/
amqtt
kopia lustrzana https://github.com/Yakifo/amqtt
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

Merge pull request #168 from ajmirsky/replace_docopts 

replace `docopt` with `typer`
pull/184/head^2
Andrew Mirsky 2025-05-29 13:16:49 -04:00 zatwierdzone przez GitHub
rodzic 28edc4938e 00cebb6a23
commit ab117d61d8
Nie znaleziono w bazie danych klucza dla tego podpisu
ID klucza GPG: B5690EEEBB952194
9 zmienionych plików z 622 dodań i 186 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

1
amqtt/mqtt/packet.py
Wyświetl plik

@ -5,6 +5,7 @@ try:
from datetime import UTC, datetime
except ImportError:
from datetime import datetime, timezone
UTC = timezone.utc
from struct import unpack

3
amqtt/plugins/sys/broker.py
Wyświetl plik

@ -6,15 +6,18 @@ try:
from collections.abc import Buffer
except ImportError:
from typing import Protocol, runtime_checkable
@runtime_checkable
class Buffer(Protocol): # type: ignore[no-redef]
def __buffer__(self, flags: int = ...) -> memoryview:
"""Mimic the behavior of `collections.abc.Buffer` for python 3.10-3.12."""
try:
from datetime import UTC, datetime
except ImportError:
from datetime import datetime, timezone
UTC = timezone.utc

59
amqtt/scripts/broker_script.py
Wyświetl plik

@ -16,10 +16,12 @@ import asyncio
import logging
from pathlib import Path
from docopt import docopt
import typer
from yaml.parser import ParserError
import amqtt
from amqtt import __version__ as amqtt_version
from amqtt.broker import Broker
from amqtt.errors import BrokerError
from amqtt.utils import read_yaml_config
default_config = {
@ -41,28 +43,63 @@ default_config = {
logger = logging.getLogger(__name__)
app = typer.Typer(rich_markup_mode=None)
def main() -> None:
"""Run the MQTT broker."""
arguments = docopt(__doc__, version=amqtt.__version__)
app()
def _version(v:bool) -> None:
if v:
typer.echo(f"{amqtt_version}")
raise typer.Exit(code=0)
@app.command()
def broker_main(
config_file: str | None = typer.Option(None, "-c", help="Broker configuration file (YAML format)"),
debug: bool = typer.Option(False, "-d", help="Enable debug messages"),
version: bool = typer.Option( # noqa : ARG001
False,
"--version",
callback=_version,
is_eager=True,
help="Show version and exit",
),
) -> None:
"""Run the MQTT broker."""
formatter = "[%(asctime)s] :: %(levelname)s - %(message)s"
level = logging.DEBUG if arguments["-d"] else logging.INFO
level = logging.DEBUG if debug else logging.INFO
logging.basicConfig(level=level, format=formatter)
try:
if config_file:
config = read_yaml_config(config_file)
else:
config = read_yaml_config(Path(__file__).parent / "default_broker.yaml")
logger.debug("Using default configuration")
except FileNotFoundError as exc:
typer.echo(f"❌ Config file error: {exc}", err=True)
raise typer.Exit(code=1) from exc
config = None
if arguments["-c"]:
config = read_yaml_config(arguments["-c"])
else:
config = read_yaml_config(Path(__file__).parent / "default_broker.yaml")
logger.debug("Using default configuration")
loop = asyncio.get_event_loop()
broker = Broker(config)
try:
broker = Broker(config)
except (BrokerError, ParserError) as exc:
typer.echo(f"❌ Broker failed to start: {exc}", err=True)
raise typer.Exit(code=1) from exc
try:
loop.run_until_complete(broker.start())
loop.run_forever()
except KeyboardInterrupt:
loop.run_until_complete(broker.shutdown())
except Exception as exc:
typer.echo("❌ Connection failed", err=True)
raise typer.Exit(code=1) from exc
finally:
loop.close()

256
amqtt/scripts/pub_script.py
Wyświetl plik

@ -1,38 +1,7 @@
"""amqtt_pub - MQTT 3.1.1 publisher.
Usage:
amqtt_pub --version
amqtt_pub (-h | --help)
amqtt_pub --url BROKER_URL -t TOPIC (-f FILE | -l | -m MESSAGE | -n | -s) [-c CONFIG_FILE] [-i CLIENT_ID] [-q | --qos QOS] [-d] [-k KEEP_ALIVE] [--clean-session] [--ca-file CAFILE] [--ca-path CAPATH] [--ca-data CADATA] [ --will-topic WILL_TOPIC [--will-message WILL_MESSAGE] [--will-qos WILL_QOS] [--will-retain] ] [--extra-headers HEADER] [-r]
Options:
-h --help Show this screen.
--version Show version.
--url BROKER_URL Broker connection URL (must conform to MQTT URI scheme (see https://github.com/mqtt/mqtt.github.io/wiki/URI-Scheme>)
-c CONFIG_FILE Broker configuration file (YAML format)
-i CLIENT_ID Id to use as client ID.
-q | --qos QOS Quality of service to use for the message, from 0, 1, and 2. Defaults to 0.
-r Set retain flag on connect
-t TOPIC Message topic
-m MESSAGE Message data to send
-f FILE Read file by line and publish message for each line
-s Read from stdin and publish message for each line
-k KEEP_ALIVE Keep alive timeout in seconds
--clean-session Clean session on connect (defaults to False)
--ca-file CAFILE CA file
--ca-path CAPATH CA Path
--ca-data CADATA CA data
--will-topic WILL_TOPIC
--will-message WILL_MESSAGE
--will-qos WILL_QOS
--will-retain
--extra-headers EXTRA_HEADERS JSON object with key-value pairs of additional headers for websocket connections
-d Enable debug messages
""" # noqa: E501
import asyncio
from collections.abc import Generator
import contextlib
from dataclasses import dataclass
import json
import logging
import os
@ -41,11 +10,11 @@ import socket
import sys
from typing import Any
from docopt import docopt
import typer
import amqtt
from amqtt import __version__ as amqtt_version
from amqtt.client import MQTTClient
from amqtt.errors import ConnectError
from amqtt.errors import ClientError, ConnectError
from amqtt.utils import read_yaml_config
logger = logging.getLogger(__name__)
@ -57,64 +26,79 @@ def _gen_client_id() -> str:
return f"amqtt_pub/{pid}-{hostname}"
def _get_qos(arguments: dict[str, Any]) -> int | None:
def _get_extra_headers(extra_headers_json: str | None = None) -> dict[str, Any]:
try:
return int(arguments["--qos"][0])
except (ValueError, IndexError):
return None
def _get_extra_headers(arguments: dict[str, Any]) -> dict[str, Any]:
try:
extra_headers: dict[str, Any] = json.loads(arguments["--extra-headers"])
extra_headers: dict[str, Any] = json.loads(extra_headers_json or "{}")
except (json.JSONDecodeError, TypeError):
return {}
return extra_headers
def _get_message(arguments: dict[str, Any]) -> Generator[bytes | bytearray]:
if arguments["-n"]:
yield b""
if arguments["-m"]:
yield arguments["-m"].encode(encoding="utf-8")
if arguments["-f"]:
try:
with Path(arguments["-f"]).open(encoding="utf-8") as f:
for line in f:
@dataclass
class MessageInput:
message_str: str | None = None
file: str | None = None
stdin: bool | None = False
lines: bool | None = False
no_message: bool | None = False
def get_message(self) -> Generator[bytes | bytearray]:
if self.no_message:
yield b""
if self.message_str:
yield self.message_str.encode(encoding="utf-8")
if self.file:
try:
with Path(self.file).open(encoding="utf-8") as f:
for line in f:
yield line.encode(encoding="utf-8")
except Exception:
logger.exception(f"Failed to read file '{self.file}'")
if self.lines:
for line in sys.stdin:
if line:
yield line.encode(encoding="utf-8")
except Exception:
logger.exception(f"Failed to read file '{arguments['-f']}'")
if arguments["-l"]:
for line in sys.stdin:
if line:
yield line.encode(encoding="utf-8")
if arguments["-s"]:
message = bytearray()
for line in sys.stdin:
message.extend(line.encode(encoding="utf-8"))
yield message
if self.stdin:
messages = bytearray()
for line in sys.stdin:
messages.extend(line.encode(encoding="utf-8"))
yield messages
async def do_pub(client: MQTTClient, arguments: dict[str, Any]) -> None:
"""Perform the publish."""
@dataclass
class CAInfo:
ca_file: str | None = None
ca_path: str | None = None
ca_data: str | None = None
async def do_pub(
client: MQTTClient,
url: str,
topic: str,
message_input: MessageInput,
ca_info: CAInfo,
clean_session: bool = False,
retain: bool = False,
extra_headers_json: str | None = None,
qos: int | None = None,
) -> None:
"""Publish the message."""
running_tasks = []
try:
logger.info(f"{client.client_id} Connecting to broker")
await client.connect(
uri=arguments["--url"],
cleansession=arguments["--clean-session"],
cafile=arguments["--ca-file"],
capath=arguments["--ca-path"],
cadata=arguments["--ca-data"],
additional_headers=_get_extra_headers(arguments),
uri=url,
cleansession=clean_session,
cafile=ca_info.ca_file,
capath=ca_info.ca_path,
cadata=ca_info.ca_data,
additional_headers=_get_extra_headers(extra_headers_json),
)
qos = _get_qos(arguments)
topic = arguments["-t"]
retain = arguments["-r"]
for message in _get_message(arguments):
for message in message_input.get_message():
logger.info(f"{client.client_id} Publishing to '{topic}'")
task = asyncio.ensure_future(client.publish(topic, message, qos, retain))
running_tasks.append(task)
@ -128,22 +112,79 @@ async def do_pub(client: MQTTClient, arguments: dict[str, Any]) -> None:
await client.disconnect()
logger.info(f"{client.client_id} Disconnected from broker")
except ConnectError as ce:
logger.fatal(f"Connection to '{arguments['--url']}' failed: {ce!r}")
except asyncio.CancelledError:
logger.fatal(f"Connection to '{url}' failed: {ce!r}")
raise ConnectError from ce
except asyncio.CancelledError as ce:
logger.fatal("Publish canceled due to previous error")
raise asyncio.CancelledError from ce
app = typer.Typer(rich_markup_mode=None)
def main() -> None:
"""Entry point for the amqtt publisher."""
app()
def _version(v: bool) -> None:
if v:
typer.echo(f"{amqtt_version}")
raise typer.Exit(code=0)
@app.command()
def publisher_main( # pylint: disable=R0914,R0917 # noqa : PLR0913
url: str = typer.Option(..., "--url", help="Broker connection URL (must conform to MQTT URI scheme: mqtt://<username:password>@HOST:port)"),
config_file: str | None = typer.Option(None, "-c", "--config-file", help="Broker configuration file (YAML format)"),
client_id: str | None = typer.Option(None, "-i", "--client-id", help="Client ID to use for the connection"),
qos: int = typer.Option(0, "--qos", "-q", help="Quality of service (0, 1, or 2)"),
retain: bool = typer.Option(False, "-r", help="Set retain flag on connect"),
topic: str = typer.Option(..., "-t", help="Message topic"),
message: str | None = typer.Option(None, "-m", help="Message data to send"),
file: str | None = typer.Option(None, "-f", help="Read file by line and publish each line as a message"),
stdin: bool = typer.Option(False, "-s", help="Read from stdin and publish message for first line"),
lines: bool = typer.Option(False, "-l", help="Read from stdin and publish message for each line"),
no_message: bool = typer.Option(False, "-n", help="Publish an empty message"),
keep_alive: int | None = typer.Option(None, "-k", help="Keep alive timeout in seconds"),
clean_session: bool = typer.Option(False, "--clean-session", help="Clean session on connect (defaults to False)"),
ca_file: str | None = typer.Option(None, "--ca-file", help="Define the path to a file containing PEM encoded CA certificates that are trusted. Used to enable SSL communication."),
ca_path: str | None = typer.Option(None, "--ca-path", help="Define the path to a directory containing PEM encoded CA certificates that are trusted. Used to enable SSL communication."),
ca_data: str | None = typer.Option(None, "--ca-data", help="Set the PEM encoded CA certificates that are trusted. Used to enable SSL communication."),
will_topic: str | None = typer.Option(None, "--will-topic", help="The topic on which to send a Will, in the event that the client disconnects unexpectedly."),
will_message: str | None = typer.Option(None, "--will-message", help="Specify a message that will be stored by the broker and sent out if this client disconnects unexpectedly. [required if `--will-topic` is specified]."),
will_qos: int | None = typer.Option(0, "--will-qos", help="The QoS to use for the Will. [default: 0, only valid if `--will-topic` is specified]."),
will_retain: bool = typer.Option(False, "--will-retain", help="If the client disconnects unexpectedly the message sent out will be treated as a retained message. [optional, only valid if `--will-topic` is specified]."),
extra_headers_json: str | None = typer.Option(
None, "--extra-headers", help="Specify a JSON object string with key-value pairs representing additional headers that are transmitted on the initial connection (websocket connections only)."
),
debug: bool = typer.Option(False, "-d", help="Enable debug messages"),
version: bool | None = typer.Option( # noqa : ARG001
None,
"--version",
callback=_version,
is_eager=True,
help="Show version and exit",
),
) -> None:
"""Run the MQTT publisher."""
arguments = docopt(__doc__, version=amqtt.__version__)
provided = [bool(message), bool(file), stdin, lines, no_message]
if sum(provided) != 1:
typer.echo("❌ You must provide exactly one of --config, --file, or --stdin.", err=True)
raise typer.Exit(code=1)
if bool(will_message) != bool(will_topic):
typer.echo("❌ must specify both 'will_message' and 'will_topic' ")
raise typer.Exit(code=1)
if will_retain and not (will_message and will_topic):
typer.echo("❌ 'will-retain' only valid if 'will_message' and 'will_topic' are specified.", err=True)
raise typer.Exit(code=1)
formatter = "[%(asctime)s] :: %(levelname)s - %(message)s"
level = logging.DEBUG if arguments["-d"] else logging.INFO
level = logging.DEBUG if debug else logging.INFO
logging.basicConfig(level=level, format=formatter)
config = None
if arguments["-c"]:
config = read_yaml_config(arguments["-c"])
if config_file:
config = read_yaml_config(config_file)
else:
default_config_path = Path(__file__).parent / "default_client.yaml"
logger.debug(f"Using default configuration from {default_config_path}")
@ -151,7 +192,6 @@ def main() -> None:
loop = asyncio.get_event_loop()
client_id = arguments.get("-i", None)
if not client_id:
client_id = _gen_client_id()
@ -159,22 +199,52 @@ def main() -> None:
logger.debug("Failed to correctly initialize config")
return
if arguments["-k"]:
config["keep_alive"] = int(arguments["-k"])
if keep_alive:
config["keep_alive"] = int(keep_alive)
if arguments["--will-topic"] and arguments["--will-message"] and arguments["--will-qos"]:
if will_topic and will_message and will_qos is not None and will_retain:
config["will"] = {
"topic": arguments["--will-topic"],
"message": arguments["--will-message"].encode("utf-8"),
"qos": int(arguments["--will-qos"]),
"retain": arguments["--will-retain"],
"topic": will_topic,
"message": will_message.encode(),
"qos": will_qos,
"retain": will_retain,
}
client = MQTTClient(client_id=client_id, config=config)
message_input = MessageInput(
message_str=message,
file=file,
stdin=stdin,
no_message=no_message,
lines=lines,
)
ca_info = CAInfo(
ca_file=ca_file,
ca_path=ca_path,
ca_data=ca_data,
)
with contextlib.suppress(KeyboardInterrupt):
loop.run_until_complete(do_pub(client, arguments))
try:
loop.run_until_complete(
do_pub(
client=client,
message_input=message_input,
url=url,
topic=topic,
retain=retain,
clean_session=clean_session,
ca_info=ca_info,
extra_headers_json=extra_headers_json,
qos=qos,
)
)
except (ClientError, ConnectError) as exc:
typer.echo("❌ Connection failed", err=True)
raise typer.Exit(code=1) from exc
loop.close()
if __name__ == "__main__":
main()
typer.run(main)

187
amqtt/scripts/sub_script.py
Wyświetl plik

@ -1,34 +1,6 @@
"""amqtt_sub - MQTT 3.1.1 publisher.
Usage:
amqtt_sub --version
amqtt_sub (-h | --help)
amqtt_sub --url BROKER_URL -t TOPIC... [-n COUNT] [-c CONFIG_FILE] [-i CLIENT_ID] [-q | --qos QOS] [-d] [-k KEEP_ALIVE] [--clean-session] [--ca-file CAFILE] [--ca-path CAPATH] [--ca-data CADATA] [ --will-topic WILL_TOPIC [--will-message WILL_MESSAGE] [--will-qos WILL_QOS] [--will-retain] ] [--extra-headers HEADER]
Options:
-h --help Show this screen.
--version Show version.
--url BROKER_URL Broker connection URL (must conform to MQTT URI scheme)
-c CONFIG_FILE Broker configuration file (YAML format)
-i CLIENT_ID Id to use as client ID.
-n COUNT Number of messages to read before ending.
-q | --qos QOS Quality of service desired to receive messages, from 0, 1 and 2. Defaults to 0.
-t TOPIC... Topic filter to subscribe
-k KEEP_ALIVE Keep alive timeout in seconds
--clean-session Clean session on connect (defaults to False)
--ca-file CAFILE CA file
--ca-path CAPATH CA Path
--ca-data CADATA CA data
--will-topic WILL_TOPIC
--will-message WILL_MESSAGE
--will-qos WILL_QOS
--will-retain
--extra-headers EXTRA_HEADERS JSON object with key-value pairs of additional headers for websocket connections
-d Enable debug messages
""" # noqa: E501
import asyncio
import contextlib
from dataclasses import dataclass
import json
import logging
import os
@ -37,11 +9,11 @@ import socket
import sys
from typing import Any
from docopt import docopt
import typer
import amqtt
from amqtt import __version__ as amqtt_version
from amqtt.client import MQTTClient
from amqtt.errors import ConnectError, MQTTError
from amqtt.errors import ClientError, ConnectError, MQTTError
from amqtt.mqtt.constants import QOS_0
from amqtt.utils import read_yaml_config
@ -54,40 +26,46 @@ def _gen_client_id() -> str:
return f"amqtt_sub/{pid}-{hostname}"
def _get_qos(arguments: dict[str, Any]) -> int:
def _get_extra_headers(extra_headers_json: str | None = None) -> dict[str, Any]:
try:
return int(arguments["--qos"][0])
except (ValueError, IndexError):
return QOS_0
def _get_extra_headers(arguments: dict[str, Any]) -> dict[str, Any]:
try:
extra_headers: dict[str, Any] = json.loads(arguments["--extra-headers"])
extra_headers: dict[str, Any] = json.loads(extra_headers_json or "{}")
except (json.JSONDecodeError, TypeError):
return {}
return extra_headers
async def do_sub(client: MQTTClient, arguments: dict[str, Any]) -> None:
@dataclass
class CAInfo:
ca_file: str | None = None
ca_path: str | None = None
ca_data: str | None = None
async def do_sub(client: MQTTClient,
url: str,
topics: list[str],
ca_info: CAInfo,
max_count: int | None = None,
clean_session: bool = False,
extra_headers_json: str | None = None,
qos: int | None = None,
) -> None:
"""Perform the subscription."""
try:
logger.info(f"{client.client_id} Connecting to broker")
await client.connect(
uri=arguments["--url"],
cleansession=arguments["--clean-session"],
cafile=arguments["--ca-file"],
capath=arguments["--ca-path"],
cadata=arguments["--ca-data"],
additional_headers=_get_extra_headers(arguments),
uri=url,
cleansession=clean_session,
cafile=ca_info.ca_file,
capath=ca_info.ca_path,
cadata=ca_info.ca_data,
additional_headers=_get_extra_headers(extra_headers_json),
)
qos = _get_qos(arguments)
filters = [(topic, qos) for topic in arguments["-t"]]
filters = [(topic, qos) for topic in topics]
await client.subscribe(filters)
max_count = int(arguments["-n"]) if arguments["-n"] else None
count = 0
while True:
if max_count and count >= max_count:
@ -106,23 +84,78 @@ async def do_sub(client: MQTTClient, arguments: dict[str, Any]) -> None:
except KeyboardInterrupt:
await client.disconnect()
logger.info(f"{client.client_id} Disconnected from broker")
except ConnectError as ce:
logger.fatal(f"Connection to '{arguments['--url']}' failed: {ce!r}")
except asyncio.CancelledError:
except ConnectError as exc:
logger.fatal(f"Connection to '{url}' failed: {exc!r}")
raise ConnectError from exc
except asyncio.CancelledError as exc:
logger.fatal("Publish canceled due to previous error")
raise asyncio.CancelledError from exc
app = typer.Typer(rich_markup_mode=None)
def main() -> None:
"""Run the MQTT subscriber."""
arguments = docopt(__doc__, version=amqtt.__version__)
"""Entry point for the amqtt subscriber."""
app()
def _version(v:bool) -> None:
if v:
typer.echo(f"{amqtt_version}")
raise typer.Exit(code=0)
@app.command()
def subscribe_main( # pylint: disable=R0914,R0917 # noqa : PLR0913
url: str = typer.Option(..., help="Broker connection URL (must conform to MQTT URI scheme)", show_default=False),
config_file: str | None = typer.Option(None, "-c", help="Broker configuration file (YAML format)"),
client_id: str | None = typer.Option(None, "-i", help="Id to use as client ID. [default: process id and the hostname of the client.]"),
max_count: int | None = typer.Option(None, "-n", help="Number of messages to read before ending (optional) [default: read indefinitely]"),
qos: int = typer.Option(0, "--qos", "-q", help="Quality of service (0, 1, or 2)"),
topics: list[str] = typer.Option(..., "-t", help="Topic filter to subscribe"), # noqa: B008
keep_alive: int | None = typer.Option(None, "-k", help="Keep alive timeout in seconds"),
clean_session: bool = typer.Option(False, help="Clean session on connect (defaults to False)"),
ca_file: str | None = typer.Option(None, "--ca-file", help="Define the path to a file containing PEM encoded CA certificates that are trusted. Used to enable SSL communication."),
ca_path: str | None = typer.Option(None, "--ca-path", help="Define the path to a directory containing PEM encoded CA certificates that are trusted. Used to enable SSL communication."),
ca_data: str | None = typer.Option(None, "--ca-data", help="Set the PEM encoded CA certificates that are trusted. Used to enable SSL communication."),
will_topic: str | None = typer.Option(None, "--will-topic", help="The topic on which to send a Will, in the event that the client disconnects unexpectedly."),
will_message: str | None = typer.Option(None, "--will-message", help="Specify a message that will be stored by the broker and sent out if this client disconnects unexpectedly. [required if `--will-topic` is specified]."),
will_qos: int | None = typer.Option(None, "--will-qos", help="The QoS to use for the Will. [default: 0, only valid if `--will-topic` is specified]."),
will_retain: bool = typer.Option(False, "--will-retain", help="If the client disconnects unexpectedly the message sent out will be treated as a retained message. [optional, only valid if `--will-topic` is specified]."),
extra_headers_json: str | None = typer.Option(None, "--extra-headers", help="Specify a JSON object string with key-value pairs representing additional headers that are transmitted on the initial connection (websocket connections only)."),
debug: bool = typer.Option(False, "-d", help="Enable debug messages"),
version: bool = typer.Option( # noqa : ARG001
False,
"--version",
callback=_version,
is_eager=True,
help="Show version and exit",
),
) -> None:
"""Run the MQTT subscriber.
Examples:
\n
Subscribe with QoS 0 to all messages published under $SYS/:
`amqtt_sub --url mqtt://localhost -t '$SYS/#' -q 0`
Subscribe to 10 messages with QoS 2 from /#:
`amqtt_sub --url mqtt://localhost -t # -q 2 -n 10`
Subscribe with QoS 0 to all messages published under $SYS/ over mqtt encapsulated in a websocket connection and additional headers:
`amqtt_sub --url wss://localhost -t '$SYS/#' -q 0 --extra-headers '{"Authorization": "Bearer <token>"}'`
"""
formatter = "[%(asctime)s] :: %(levelname)s - %(message)s"
level = logging.DEBUG if arguments["-d"] else logging.INFO
level = logging.DEBUG if debug else logging.INFO
logging.basicConfig(level=level, format=formatter)
config = None
if arguments["-c"]:
config = read_yaml_config(arguments["-c"])
if config_file:
config = read_yaml_config(config_file)
else:
default_config_path = Path(__file__).parent / "default_client.yaml"
logger.debug(f"Using default configuration from {default_config_path}")
@ -130,7 +163,6 @@ def main() -> None:
loop = asyncio.get_event_loop()
client_id = arguments.get("-i", None)
if not client_id:
client_id = _gen_client_id()
@ -138,20 +170,37 @@ def main() -> None:
logger.debug("Failed to correctly initialize config")
return
if arguments["-k"]:
config["keep_alive"] = int(arguments["-k"])
if keep_alive:
config["keep_alive"] = keep_alive
if arguments["--will-topic"] and arguments["--will-message"] and arguments["--will-qos"]:
if will_topic and will_message and will_qos:
config["will"] = {
"topic": arguments["--will-topic"],
"message": arguments["--will-message"].encode("utf-8"),
"qos": int(arguments["--will-qos"]),
"retain": arguments["--will-retain"],
"topic": will_topic,
"message": will_message.encode("utf-8"),
"qos": int(will_qos),
"retain": will_retain,
}
client = MQTTClient(client_id=client_id, config=config)
ca_info = CAInfo(
ca_file=ca_file,
ca_path=ca_path,
ca_data=ca_data,
)
with contextlib.suppress(KeyboardInterrupt):
loop.run_until_complete(do_sub(client, arguments))
try:
loop.run_until_complete(do_sub(client,
url=url,
topics=topics,
ca_info=ca_info,
extra_headers_json=extra_headers_json,
qos=qos or QOS_0,
max_count=max_count,
clean_session=clean_session,
))
except (ClientError, ConnectError) as exc:
typer.echo("❌ Connection failed", err=True)
raise typer.Exit(code=1) from exc
loop.close()

1
amqtt/version.py
Wyświetl plik

@ -2,6 +2,7 @@ try:
from datetime import UTC, datetime
except ImportError:
from datetime import datetime, timezone
UTC = timezone.utc
import logging

9
pyproject.toml
Wyświetl plik

@ -1,5 +1,6 @@
[build-system]
requires = ["hatchling", "hatch-vcs"]
requires = ["hatchling", "hatch-vcs", "uv-dynamic-versioning"]
build-backend = "hatchling.build"
[project]
@ -26,8 +27,8 @@ dependencies = [
"transitions==0.9.2", # https://pypi.org/project/transitions
"websockets==15.0.1", # https://pypi.org/project/websockets
"passlib==1.7.4", # https://pypi.org/project/passlib
"docopt==0.6.2", # https://pypi.org/project/docopt
"PyYAML==6.0.2", # https://pypi.org/project/PyYAML
"typer==0.15.4"
]
[dependency-groups]
@ -117,11 +118,13 @@ ignore = [
"TD003", # TODO Missing issue link for this TODO.
"ANN401", # Dynamically typed expressions (typing.Any) are disallowed
"ARG002", # Unused method argument
"PERF203" # try-except penalty within loops (3.10 only)
"PERF203",# try-except penalty within loops (3.10 only),
"COM812" # rule causes conflicts when used with the formatter
]
[tool.ruff.lint.per-file-ignores]
"tests/**" = ["ALL"]
"amqtt/scripts/*_script.py" = ["FBT003"]
[tool.ruff.lint.flake8-pytest-style]
fixture-parentheses = false

206
tests/test_cli.py
Wyświetl plik

@ -1,18 +1,210 @@
import asyncio
import json
import os
import subprocess
import tempfile
from pathlib import Path
import pytest
import yaml
from amqtt.broker import Broker
from amqtt.client import MQTTClient
from amqtt.mqtt.constants import QOS_1
def test_smometest():
@pytest.fixture
def broker_config():
return {
"listeners": {
"default": {
"type": "tcp",
"bind": "127.0.0.1:1884", # Use non-standard port for testing
},
},
"sys_interval": 0,
"auth": {
"allow-anonymous": True,
"plugins": ["auth_anonymous"],
},
"topic-check": {"enabled": False},
}
@pytest.fixture
def config_file(broker_config, tmp_path):
config_path = tmp_path / "config.yaml"
with config_path.open("w") as f:
yaml.dump(broker_config, f)
return str(config_path)
@pytest.fixture
async def broker(config_file):
proc = subprocess.Popen(
["amqtt", "-c", config_file],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
# Give broker time to start
await asyncio.sleep(1)
yield proc
proc.terminate()
proc.wait()
def test_cli_help_messages():
"""Test that help messages are displayed correctly."""
amqtt_path = "amqtt"
output = subprocess.check_output([amqtt_path, "--help"])
assert b"Usage" in output
assert b"aMQTT" in output
assert "Usage: amqtt" in output.decode("utf-8")
amqtt_sub_path = "amqtt_sub"
output = subprocess.check_output([amqtt_sub_path, "--help"])
assert b"Usage" in output
assert b"amqtt_sub" in output
assert "Usage: amqtt_sub" in output.decode("utf-8")
amqtt_pub_path = "amqtt_pub"
output = subprocess.check_output([amqtt_pub_path, "--help"])
assert b"Usage" in output
assert b"amqtt_pub" in output
assert "Usage: amqtt_pub" in output.decode("utf-8")
def test_broker_version():
"""Test broker version command."""
output = subprocess.check_output(["amqtt", "--version"])
assert output.strip()
@pytest.mark.asyncio
async def test_broker_start_stop(config_file):
"""Test broker start and stop with config file."""
proc = subprocess.Popen(
["amqtt", "-c", config_file],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
# Give broker time to start
await asyncio.sleep(1)
# Verify broker is running by connecting a client
client = MQTTClient()
await client.connect("mqtt://127.0.0.1:1884")
await client.disconnect()
# Stop broker
proc.terminate()
proc.wait()
@pytest.mark.asyncio
async def test_publish_subscribe(broker):
"""Test pub/sub CLI tools with running broker."""
# Create a temporary file with test message
with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp:
tmp.write("test message\n")
tmp.write("another message\n")
tmp_path = tmp.name
# Start subscriber in background
sub_proc = subprocess.Popen(
[
"amqtt_sub",
"--url", "mqtt://127.0.0.1:1884",
"-t", "test/topic",
"-n", "2", # Exit after 2 messages
"--qos", "1",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
# Give subscriber time to connect
await asyncio.sleep(0.5)
#
# # Publish messages from file
pub_proc = subprocess.run(
[
"amqtt_pub",
"--url", "mqtt://127.0.0.1:1884",
"-t", "test/topic",
"-f", tmp_path,
"--qos", "1",
],
capture_output=True,
)
assert pub_proc.returncode == 0
#
# # Wait for subscriber to receive messages
stdout, stderr = sub_proc.communicate(timeout=5)
# Clean up temp file
os.unlink(tmp_path)
# Verify messages were received
print(stdout.decode("utf-8"))
assert "test message" in str(stdout)
assert "another message" in str(stdout)
assert sub_proc.returncode == 0
@pytest.mark.asyncio
async def test_pub_sub_options(broker):
"""Test various pub/sub options."""
# Test publishing with retain flag
pub_proc = subprocess.run(
[
"amqtt_pub",
"--url", "mqtt://127.0.0.1:1884",
"-t", "topic/test",
"-m", "standard message",
"--will-topic", "topic/retain",
"--will-message", "last will message",
"--will-retain",
],
capture_output=True,
)
assert pub_proc.returncode == 0, "publisher error code"
# Verify retained message is received by new subscriber
sub_proc = subprocess.run(
[
"amqtt_sub",
"--url", "mqtt://127.0.0.1:1884",
"-t", "topic/retain",
"-n", "1",
],
capture_output=True,
)
assert sub_proc.returncode == 0, "subscriber error code"
assert "last will message" in str(sub_proc.stdout)
@pytest.mark.asyncio
async def test_pub_sub_errors():
"""Test error handling in pub/sub tools."""
# Test connection to non-existent broker
pub_proc = subprocess.run(
[
"amqtt_pub",
"--url", "mqtt://127.0.0.1:1885", # Wrong port
"-t", "test/topic",
"-m", "test",
],
capture_output=True,
)
assert pub_proc.returncode != 0, f"publisher error code {pub_proc.returncode} != 0"
assert "Connection failed" in str(pub_proc.stderr)
# Test invalid URL format
sub_proc = subprocess.run(
[
"amqtt_sub",
"--url", "invalid://url",
"-t", "test/topic",
],
capture_output=True,
)
assert sub_proc.returncode != 0, f"subscriber error code {sub_proc.returncode} != 0"

86
uv.lock
Wyświetl plik

@ -12,10 +12,10 @@ name = "amqtt"
version = "0.11.0rc1"
source = { editable = "." }
dependencies = [
{ name = "docopt" },
{ name = "passlib" },
{ name = "pyyaml" },
{ name = "transitions" },
{ name = "typer" },
{ name = "websockets" },
]
@ -46,10 +46,10 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "coveralls", marker = "extra == 'ci'", specifier = "==4.0.1" },
{ name = "docopt", specifier = "==0.6.2" },
{ name = "passlib", specifier = "==1.7.4" },
{ name = "pyyaml", specifier = "==6.0.2" },
{ name = "transitions", specifier = "==0.9.2" },
{ name = "typer", specifier = "==0.15.4" },
{ name = "websockets", specifier = "==15.0.1" },
]
provides-extras = ["ci"]
@ -173,6 +173,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/20/94/c5790835a017658cbfabd07f3bfb549140c3ac458cfc196323996b10095a/charset_normalizer-3.4.2-py3-none-any.whl", hash = "sha256:7f56930ab0abd1c45cd15be65cc741c28b1c9a34876ce8c17a2fa107810c0af0", size = 52626, upload-time = "2025-05-02T08:34:40.053Z" },
]
[[package]]
name = "click"
version = "8.1.8"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b9/2e/0090cbf739cee7d23781ad4b89a9894a41538e4fcf4c31dcdd705b78eb8b/click-8.1.8.tar.gz", hash = "sha256:ed53c9d8990d83c2a27deae68e4ee337473f6330c040a31d4225c9574d16096a", size = 226593, upload-time = "2024-12-21T18:38:44.339Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7e/d4/7ebdbd03970677812aac39c869717059dbb71a4cfc033ca6e5221787892c/click-8.1.8-py3-none-any.whl", hash = "sha256:63c132bbbed01578a06712a2d1f497bb62d9c1c0d329b7903a866228027263b2", size = 98188, upload-time = "2024-12-21T18:38:41.666Z" },
]
[[package]]
name = "colorama"
version = "0.4.6"
@ -290,7 +302,7 @@ name = "exceptiongroup"
version = "1.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
{ name = "typing-extensions", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/0b/9f/a65090624ecf468cdca03533906e7c69ed7588582240cfe7cc9e770b50eb/exceptiongroup-1.3.0.tar.gz", hash = "sha256:b241f5885f560bc56a59ee63ca4c6a8bfa46ae4ad651af316d4e81817bb9fd88", size = 29749, upload-time = "2025-05-10T17:42:51.123Z" }
wheels = [
@ -356,6 +368,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/c1/11/114d0a5f4dabbdcedc1125dee0888514c3c3b16d3e9facad87ed96fad97c/isort-6.0.1-py3-none-any.whl", hash = "sha256:2dc5d7f65c9678d94c88dfc29161a320eec67328bc97aad576874cb4be1e9615", size = 94186, upload-time = "2025-02-26T21:13:14.911Z" },
]
[[package]]
name = "markdown-it-py"
version = "3.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "mdurl" },
]
sdist = { url = "https://files.pythonhosted.org/packages/38/71/3b932df36c1a044d397a1f92d1cf91ee0a503d91e470cbd670aa66b07ed0/markdown-it-py-3.0.0.tar.gz", hash = "sha256:e3f60a94fa066dc52ec76661e37c851cb232d92f9886b15cb560aaada2df8feb", size = 74596, upload-time = "2023-06-03T06:41:14.443Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/42/d7/1ec15b46af6af88f19b8e5ffea08fa375d433c998b8a7639e76935c14f1f/markdown_it_py-3.0.0-py3-none-any.whl", hash = "sha256:355216845c60bd96232cd8d8c40e8f9765cc86f46880e43a8fd22dc1a1a8cab1", size = 87528, upload-time = "2023-06-03T06:41:11.019Z" },
]
[[package]]
name = "mccabe"
version = "0.7.0"
@ -365,6 +389,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/27/1a/1f68f9ba0c207934b35b86a8ca3aad8395a3d6dd7921c0686e23853ff5a9/mccabe-0.7.0-py2.py3-none-any.whl", hash = "sha256:6c2d30ab6be0e4a46919781807b4f0d834ebdd6c6e3dca0bda5a15f863427b6e", size = 7350, upload-time = "2022-01-24T01:14:49.62Z" },
]
[[package]]
name = "mdurl"
version = "0.1.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" },
]
[[package]]
name = "mypy"
version = "1.15.0"
@ -488,6 +521,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/50/1b/6921afe68c74868b4c9fa424dad3be35b095e16687989ebbb50ce4fceb7c/psutil-7.0.0-cp37-abi3-win_amd64.whl", hash = "sha256:4cf3d4eb1aa9b348dec30105c55cd9b7d4629285735a102beb4441e38db90553", size = 244885, upload-time = "2025-02-13T21:54:37.486Z" },
]
[[package]]
name = "pygments"
version = "2.19.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/7c/2d/c3338d48ea6cc0feb8446d8e6937e1408088a72a39937982cc6111d17f84/pygments-2.19.1.tar.gz", hash = "sha256:61c16d2a8576dc0649d9f39e089b5f02bcd27fba10d8fb4dcc28173f7a45151f", size = 4968581, upload-time = "2025-01-06T17:26:30.443Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/8a/0b/9fcc47d19c48b59121088dd6da2488a49d5f72dacf8262e2790a1d2c7d15/pygments-2.19.1-py3-none-any.whl", hash = "sha256:9ea1544ad55cecf4b8242fab6dd35a93bbce657034b0611ee383099054ab6d8c", size = 1225293, upload-time = "2025-01-06T17:26:25.553Z" },
]
[[package]]
name = "pylint"
version = "3.3.7"
@ -632,6 +674,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/f9/9b/335f9764261e915ed497fcdeb11df5dfd6f7bf257d4a6a2a686d80da4d54/requests-2.32.3-py3-none-any.whl", hash = "sha256:70761cfe03c773ceb22aa2f671b4757976145175cdfca038c02654d061d6dcc6", size = 64928, upload-time = "2024-05-29T15:37:47.027Z" },
]
[[package]]
name = "rich"
version = "14.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py" },
{ name = "pygments" },
{ name = "typing-extensions", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a1/53/830aa4c3066a8ab0ae9a9955976fb770fe9c6102117c8ec4ab3ea62d89e8/rich-14.0.0.tar.gz", hash = "sha256:82f1bc23a6a21ebca4ae0c45af9bdbc492ed20231dcb63f297d6d1021a9d5725", size = 224078, upload-time = "2025-03-30T14:15:14.23Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0d/9b/63f4c7ebc259242c89b3acafdb37b41d1185c07ff0011164674e9076b491/rich-14.0.0-py3-none-any.whl", hash = "sha256:1c9491e1951aac09caffd42f448ee3d04e58923ffe14993f6e83068dc395d7e0", size = 243229, upload-time = "2025-03-30T14:15:12.283Z" },
]
[[package]]
name = "ruff"
version = "0.11.10"
@ -666,6 +722,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/a1/18/0e835c3a557dc5faffc8f91092f62fc337c1dab1066715842e7a4b318ec4/setuptools-80.7.1-py3-none-any.whl", hash = "sha256:ca5cc1069b85dc23070a6628e6bcecb3292acac802399c7f8edc0100619f9009", size = 1200776, upload-time = "2025-05-15T02:40:58.887Z" },
]
[[package]]
name = "shellingham"
version = "1.5.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/58/15/8b3609fd3830ef7b27b655beb4b4e9c62313a4e8da8c676e142cc210d58e/shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de", size = 10310, upload-time = "2023-10-24T04:13:40.426Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" },
]
[[package]]
name = "six"
version = "1.17.0"
@ -744,6 +809,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/ec/47/852f96b115425618382472ea06860069da5bb078bdec3e4449f185a40e07/transitions-0.9.2-py2.py3-none-any.whl", hash = "sha256:f7b40c9b4a93869f36c4d1c33809aeb18cdeeb065fd1adba018ee39c3db216f3", size = 111773, upload-time = "2024-08-06T13:32:46.703Z" },
]
[[package]]
name = "typer"
version = "0.15.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "click" },
{ name = "rich" },
{ name = "shellingham" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/6c/89/c527e6c848739be8ceb5c44eb8208c52ea3515c6cf6406aa61932887bf58/typer-0.15.4.tar.gz", hash = "sha256:89507b104f9b6a0730354f27c39fae5b63ccd0c95b1ce1f1a6ba0cfd329997c3", size = 101559, upload-time = "2025-05-14T16:34:57.704Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c9/62/d4ba7afe2096d5659ec3db8b15d8665bdcb92a3c6ff0b95e99895b335a9c/typer-0.15.4-py3-none-any.whl", hash = "sha256:eb0651654dcdea706780c466cf06d8f174405a659ffff8f163cfbfee98c0e173", size = 45258, upload-time = "2025-05-14T16:34:55.583Z" },
]
[[package]]
name = "types-mock"
version = "5.2.0.20250516"