16466672479

Committed 23 Jul 2025 09:20AM UTC coverage: 91.417%. First build

Build # 16466672479

Build Type

Pull #138

github

Pull Request Pull Request #138: Hotfix SPADE cli error + lint

Run Details

1 of 2 new or added lines in 2 files covered. (50.0%)

1097 of 1200 relevant lines covered (91.42%)

3.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/spade/cli.py

import sys
import click
import spade
import socket

from pyjabber.server import Server
from pyjabber.server_parameters import Parameters
from rich.console import Console
from rich.panel import Panel
from loguru import logger


def check_port_in_use(port, host="0.0.0.0"):
    """Checks if a port is in use"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        result = s.connect_ex((host, port))
        return result == 0


def create_cli():
    """Factory function to create the CLI"""

    @click.group()
    def cli():
        """SPADE command line tool"""
        pass

    @cli.command()
    @click.option("--host", default="0.0.0.0", help="Server host")
    @click.option("--client_port", default=5222, help="Client port")
    @click.option("--server_port", default=5269, help="Server port")
    @click.option("--debug", is_flag=True, default=False, help="Enables debug mode")
    @click.option("--timeout", default=600, help="Connection timeout")
    @click.option(
        "--db",
        type=str,
        default="server.db",
        show_default=True,
        help="Path for database file",
    )
    @click.option(
        "--purge",
        is_flag=True,
        default=False,
        help="Restore database file to default state (empty)",
    )
    @click.option(
        "--memory", is_flag=True, default=False, help="Use an in-memory database"
    )
    def run(host, client_port, server_port, debug, timeout, db, purge, memory):
        """Launch an XMPP server"""
        if check_port_in_use(client_port, host):
            click.echo(f"Error: The port {client_port} is already in use.")
            sys.exit(1)
        if check_port_in_use(server_port, host):
            click.echo(f"Error: The port {server_port} is already in use.")
            sys.exit(1)

        logger.remove()
        if debug:
            level = "DEBUG"
        else:
            level = "INFO"

        logger.add(
            sys.stderr,
            enqueue=True,
            format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> - <level>{level}: {message}</level>",
            level=level,
        )

        server = Server(
            Parameters(
                host=host,
                client_port=client_port,
                server_port=server_port,
                connection_timeout=timeout,
                database_path=db,
                database_purge=purge,
                database_in_memory=memory,
            )
        )

        print_spade_info()
        container = spade.container.Container()
        loop = container.loop
        loop.run_until_complete(server.start())

    def print_spade_info():
        """Prints SPADE information"""

        console = Console()

        project_name = "[bold green]SPADE[/bold green] - [yellow]Smart Python Agent Development Environment[/yellow]"
        author = (
            "[green]Development Lead:[/green]\n"
            "  - [cyan]Javi Palanca[/cyan]\n"
            "[green]Funded by:[/green]\n"
            "  - [cyan]Valencian Research Institute for Artificial Intelligence (VRAIN)[/cyan]\n"
            "[green]URL:[/green]\n"
            "  - [cyan] https://spadeagents.eu[/cyan]\n"
            "[green]Documentation:[/green]\n"
            "  - [cyan] http://spade-mas.readthedocs.io/[/cyan]"
        )

        footer = f"[cyan]Version:[/cyan] {spade.__version__}   [magenta]License:[/magenta] MIT"

        console.print(
            Panel.fit(
                f"{project_name}\n\n{author}",
                title="SPADE",
                subtitle=footer,
                border_style="blue",
            )
        )

    @cli.command()
    def version():
        """Displays the SPADE version"""
        click.echo(spade.__version__)

    return cli


# Create a single instance of the CLI
cli = create_cli()

if __name__ == "__main__":
    sys.exit(cli())  # pragma: no cover

1	import sys	×
2	import click	×
3	import spade	×
4	import socket	×
5
6	from pyjabber.server import Server	×
NEW 7	from pyjabber.server_parameters import Parameters	×
8	from rich.console import Console	×
9	from rich.panel import Panel	×
10	from loguru import logger	×
11
12
13	def check_port_in_use(port, host="0.0.0.0"):	×
14	"""Checks if a port is in use"""
15	with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:	×
16	result = s.connect_ex((host, port))	×
17	return result == 0	×
18
19
20	def create_cli():	×
21	"""Factory function to create the CLI"""
22
23	@click.group()	×
24	def cli():	×
25	"""SPADE command line tool"""
26	pass	×
27
28	@cli.command()	×
29	@click.option("--host", default="0.0.0.0", help="Server host")	×
30	@click.option("--client_port", default=5222, help="Client port")	×
31	@click.option("--server_port", default=5269, help="Server port")	×
32	@click.option("--debug", is_flag=True, default=False, help="Enables debug mode")	×
33	@click.option("--timeout", default=600, help="Connection timeout")	×
34	@click.option(	×
35	"--db",
36	type=str,
37	default="server.db",
38	show_default=True,
39	help="Path for database file",
40	)
41	@click.option(	×
42	"--purge",
43	is_flag=True,
44	default=False,
45	help="Restore database file to default state (empty)",
46	)
47	@click.option(	×
48	"--memory", is_flag=True, default=False, help="Use an in-memory database"
49	)
50	def run(host, client_port, server_port, debug, timeout, db, purge, memory):	×
51	"""Launch an XMPP server"""
52	if check_port_in_use(client_port, host):	×
53	click.echo(f"Error: The port {client_port} is already in use.")	×
54	sys.exit(1)	×
55	if check_port_in_use(server_port, host):	×
56	click.echo(f"Error: The port {server_port} is already in use.")	×
57	sys.exit(1)	×
58
59	logger.remove()	×
60	if debug:	×
61	level = "DEBUG"	×
62	else:
63	level = "INFO"	×
64
65	logger.add(	×
66	sys.stderr,
67	enqueue=True,
68	format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> - <level>{level}: {message}</level>",
69	level=level,
70	)
71
72	server = Server(	×
73	Parameters(
74	host=host,
75	client_port=client_port,
76	server_port=server_port,
77	connection_timeout=timeout,
78	database_path=db,
79	database_purge=purge,
80	database_in_memory=memory,
81	)
82	)
83		×
84	print_spade_info()	×
85	container = spade.container.Container()	×
86	loop = container.loop	×
87	loop.run_until_complete(server.start())
88		×
89	def print_spade_info():
90	"""Prints SPADE information"""
91		×
92	console = Console()
93		×
94	project_name = "[bold green]SPADE[/bold green] - [yellow]Smart Python Agent Development Environment[/yellow]"	×
95	author = (
96	"[green]Development Lead:[/green]\n"
97	" - [cyan]Javi Palanca[/cyan]\n"
98	"[green]Funded by:[/green]\n"
99	" - [cyan]Valencian Research Institute for Artificial Intelligence (VRAIN)[/cyan]\n"
100	"[green]URL:[/green]\n"
101	" - [cyan] https://spadeagents.eu[/cyan]\n"
102	"[green]Documentation:[/green]\n"
103	" - [cyan] http://spade-mas.readthedocs.io/[/cyan]"
104	)
105		×
106	footer = f"[cyan]Version:[/cyan] {spade.__version__} [magenta]License:[/magenta] MIT"
107		×
108	console.print(
109	Panel.fit(
110	f"{project_name}\n\n{author}",
111	title="SPADE",
112	subtitle=footer,
113	border_style="blue",
114	)
115	)
116		×
117	@cli.command()	×
118	def version():
119	"""Displays the SPADE version"""	×
120	click.echo(spade.__version__)
121		×
122	return cli
123
124
125	# Create a single instance of the CLI	×
126	cli = create_cli()
127		×
128	if __name__ == "__main__":
129	sys.exit(cli()) # pragma: no cover

javipalanca / spade / 16466672479

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous