14968876494

Committed 12 May 2025 09:36AM UTC coverage: 89.09%. Remained the same

Build # 14968876494

Build Type

Pull #1174

github

Committed by

web-flow

Commit Message

Merge 6d2bd9a8e into 648912718

Pull Request Pull Request #1174: Work around changes in Click 8.2.0

Run Details

2044 of 2422 branches covered (84.39%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 1 file covered. (100.0%)

50 existing lines in 6 files now uncovered.

5542 of 6093 relevant lines covered (90.96%)

9.93 hits per line

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

89.86

/kopf/cli.py

import asyncio
import dataclasses
import functools
import os
from collections.abc import Collection
from typing import Any, Callable, Optional

import click

from kopf._cogs.aiokits import aioadapters
from kopf._cogs.configs import configuration
from kopf._cogs.helpers import loaders
from kopf._cogs.structs import credentials, references
from kopf._core.actions import loggers
from kopf._core.engines import peering
from kopf._core.intents import registries
from kopf._core.reactor import running
from kopf._kits import loops


@dataclasses.dataclass()
class CLIControls:
    """ `KopfRunner` controls, which are impossible to pass via CLI. """
    ready_flag: Optional[aioadapters.Flag] = None
    stop_flag: Optional[aioadapters.Flag] = None
    vault: Optional[credentials.Vault] = None
    registry: Optional[registries.OperatorRegistry] = None
    settings: Optional[configuration.OperatorSettings] = None
    loop: Optional[asyncio.AbstractEventLoop] = None


# With Click>=8.2.0, that should be click.Choice[LogFormat], but it is good for now, too.
class LogFormatParamType(click.Choice):  # type: ignore

    def __init__(self) -> None:
        super().__init__(choices=[v.name.lower() for v in loggers.LogFormat])

    def convert(self, value: Any, param: Any, ctx: Any) -> loggers.LogFormat:
        if isinstance(value, loggers.LogFormat):
            return value
        else:
            name: str = super().convert(value, param, ctx)
            return loggers.LogFormat[name.upper()]


def logging_options(fn: Callable[..., Any]) -> Callable[..., Any]:
    """ A decorator to configure logging in all commands the same way."""
    @click.option('-v', '--verbose', is_flag=True)
    @click.option('-d', '--debug', is_flag=True)
    @click.option('-q', '--quiet', is_flag=True)
    @click.option('--log-format', type=LogFormatParamType(), default='full')
    @click.option('--log-refkey', type=str)
    @click.option('--log-prefix/--no-log-prefix', default=None)
    @functools.wraps(fn)  # to preserve other opts/args
    def wrapper(verbose: bool, quiet: bool, debug: bool,
                log_format: loggers.LogFormat = loggers.LogFormat.FULL,
                log_prefix: Optional[bool] = False,
                log_refkey: Optional[str] = None,
                *args: Any, **kwargs: Any) -> Any:
        loggers.configure(debug=debug, verbose=verbose, quiet=quiet,
                          log_format=log_format, log_refkey=log_refkey, log_prefix=log_prefix)
        return fn(*args, **kwargs)

    return wrapper


@click.group(name='kopf', context_settings=dict(
    auto_envvar_prefix='KOPF',
))
@click.version_option(prog_name='kopf')
@click.make_pass_decorator(CLIControls, ensure=True)
def main(__controls: CLIControls) -> None:
    pass


@main.command()
@logging_options
@click.option('-A', '--all-namespaces', 'clusterwide', is_flag=True)
@click.option('-n', '--namespace', 'namespaces', multiple=True)
@click.option('--standalone', is_flag=True, default=None)
@click.option('--dev', 'priority', type=int, is_flag=True, flag_value=666)
@click.option('-L', '--liveness', 'liveness_endpoint', type=str)
@click.option('-P', '--peering', 'peering_name', type=str, envvar='KOPF_RUN_PEERING')
@click.option('-p', '--priority', type=int)
@click.option('-m', '--module', 'modules', multiple=True)
@click.argument('paths', nargs=-1)
@click.make_pass_decorator(CLIControls, ensure=True)
def run(
        __controls: CLIControls,
        paths: list[str],
        modules: list[str],
        peering_name: Optional[str],
        priority: Optional[int],
        standalone: Optional[bool],
        namespaces: Collection[references.NamespacePattern],
        clusterwide: bool,
        liveness_endpoint: Optional[str],
) -> None:
    """ Start an operator process and handle all the requests. """
    if os.environ.get('KOPF_RUN_NAMESPACE'):  # legacy for single-namespace mode
        namespaces = tuple(namespaces) + (os.environ.get('KOPF_RUN_NAMESPACE', ''),)
    if namespaces and clusterwide:
        raise click.UsageError("Either --namespace or --all-namespaces can be used, not both.")
    if __controls.registry is not None:
        registries.set_default_registry(__controls.registry)
    loaders.preload(
        paths=paths,
        modules=modules,
    )
    with loops.proper_loop(__controls.loop):
        return running.run(
            standalone=standalone,
            namespaces=namespaces,
            clusterwide=clusterwide,
            priority=priority,
            peering_name=peering_name,
            liveness_endpoint=liveness_endpoint,
            registry=__controls.registry,
            settings=__controls.settings,
            stop_flag=__controls.stop_flag,
            ready_flag=__controls.ready_flag,
            vault=__controls.vault,
            loop=__controls.loop,
        )


@main.command()
@logging_options
@click.option('-n', '--namespace', 'namespaces', multiple=True)
@click.option('-A', '--all-namespaces', 'clusterwide', is_flag=True)
@click.option('-i', '--id', type=str, default=None)
@click.option('--dev', 'priority', flag_value=666)
@click.option('-P', '--peering', 'peering_name', required=True, envvar='KOPF_FREEZE_PEERING')
@click.option('-p', '--priority', type=int, default=100, required=True)
@click.option('-t', '--lifetime', type=int, required=True)
@click.option('-m', '--message', type=str)
@click.make_pass_decorator(CLIControls, ensure=True)
def freeze(
        __controls: CLIControls,
        id: Optional[str],
        message: Optional[str],
        lifetime: int,
        namespaces: Collection[references.NamespacePattern],
        clusterwide: bool,
        peering_name: str,
        priority: int,
) -> None:
    """ Pause the resource handling in the operator(s). """
    identity = peering.Identity(id) if id else peering.detect_own_id(manual=True)
    insights = references.Insights()
    settings = configuration.OperatorSettings()
    settings.peering.name = peering_name
    settings.peering.priority = priority
    with loops.proper_loop(__controls.loop):
        return running.run(
            clusterwide=clusterwide,
            namespaces=namespaces,
            insights=insights,
            identity=identity,
            settings=settings,
            _command=peering.touch_command(
                insights=insights,
                identity=identity,
                settings=settings,
                lifetime=lifetime))


@main.command()
@logging_options
@click.option('-n', '--namespace', 'namespaces', multiple=True)
@click.option('-A', '--all-namespaces', 'clusterwide', is_flag=True)
@click.option('-i', '--id', type=str, default=None)
@click.option('-P', '--peering', 'peering_name', required=True, envvar='KOPF_RESUME_PEERING')
@click.make_pass_decorator(CLIControls, ensure=True)
def resume(
        __controls: CLIControls,
        id: Optional[str],
        namespaces: Collection[references.NamespacePattern],
        clusterwide: bool,
        peering_name: str,
) -> None:
    """ Resume the resource handling in the operator(s). """
    identity = peering.Identity(id) if id else peering.detect_own_id(manual=True)
    insights = references.Insights()
    settings = configuration.OperatorSettings()
    settings.peering.name = peering_name
    with loops.proper_loop(__controls.loop):
        return running.run(
            clusterwide=clusterwide,
            namespaces=namespaces,
            insights=insights,
            identity=identity,
            settings=settings,
            _command=peering.touch_command(
                insights=insights,
                identity=identity,
                settings=settings,
                lifetime=0))

1	import asyncio	11✔
2	import dataclasses	11✔
3	import functools	11✔
4	import os	11✔
5	from collections.abc import Collection	11✔
6	from typing import Any, Callable, Optional	11✔
7
8	import click	11✔
9
10	from kopf._cogs.aiokits import aioadapters	11✔
11	from kopf._cogs.configs import configuration	11✔
12	from kopf._cogs.helpers import loaders	11✔
13	from kopf._cogs.structs import credentials, references	11✔
14	from kopf._core.actions import loggers	11✔
15	from kopf._core.engines import peering	11✔
16	from kopf._core.intents import registries	11✔
17	from kopf._core.reactor import running	11✔
18	from kopf._kits import loops	11✔
19
20
21	@dataclasses.dataclass()	11✔
22	class CLIControls:	11✔
23	""" `KopfRunner` controls, which are impossible to pass via CLI. """
24	ready_flag: Optional[aioadapters.Flag] = None	11✔
25	stop_flag: Optional[aioadapters.Flag] = None	11✔
26	vault: Optional[credentials.Vault] = None	11✔
27	registry: Optional[registries.OperatorRegistry] = None	11✔
28	settings: Optional[configuration.OperatorSettings] = None	11✔
29	loop: Optional[asyncio.AbstractEventLoop] = None	11✔
30
31
32	# With Click>=8.2.0, that should be click.Choice[LogFormat], but it is good for now, too.
33	class LogFormatParamType(click.Choice): # type: ignore	11✔
34
35	def __init__(self) -> None:	11✔
36	super().__init__(choices=[v.name.lower() for v in loggers.LogFormat])	11!
37
38	def convert(self, value: Any, param: Any, ctx: Any) -> loggers.LogFormat:	11✔
39	if isinstance(value, loggers.LogFormat):	11!
UNCOV 40	return value	×
41	else:
42	name: str = super().convert(value, param, ctx)	11✔
43	return loggers.LogFormat[name.upper()]	11✔
44
45
46	def logging_options(fn: Callable[..., Any]) -> Callable[..., Any]:	11✔
47	""" A decorator to configure logging in all commands the same way."""
48	@click.option('-v', '--verbose', is_flag=True)	11✔
49	@click.option('-d', '--debug', is_flag=True)	11✔
50	@click.option('-q', '--quiet', is_flag=True)	11✔
51	@click.option('--log-format', type=LogFormatParamType(), default='full')	11✔
52	@click.option('--log-refkey', type=str)	11✔
53	@click.option('--log-prefix/--no-log-prefix', default=None)	11✔
54	@functools.wraps(fn) # to preserve other opts/args	11✔
55	def wrapper(verbose: bool, quiet: bool, debug: bool,	11✔
56	log_format: loggers.LogFormat = loggers.LogFormat.FULL,
57	log_prefix: Optional[bool] = False,
58	log_refkey: Optional[str] = None,
59	args: Any, *kwargs: Any) -> Any:
60	loggers.configure(debug=debug, verbose=verbose, quiet=quiet,	11✔
61	log_format=log_format, log_refkey=log_refkey, log_prefix=log_prefix)
62	return fn(args, *kwargs)	11✔
63
64	return wrapper	11✔
65
66
67	@click.group(name='kopf', context_settings=dict(	11✔
68	auto_envvar_prefix='KOPF',
69	))
70	@click.version_option(prog_name='kopf')	11✔
71	@click.make_pass_decorator(CLIControls, ensure=True)	11✔
72	def main(__controls: CLIControls) -> None:	11✔
73	pass	11✔
74
75
76	@main.command()	11✔
77	@logging_options	11✔
78	@click.option('-A', '--all-namespaces', 'clusterwide', is_flag=True)	11✔
79	@click.option('-n', '--namespace', 'namespaces', multiple=True)	11✔
80	@click.option('--standalone', is_flag=True, default=None)	11✔
81	@click.option('--dev', 'priority', type=int, is_flag=True, flag_value=666)	11✔
82	@click.option('-L', '--liveness', 'liveness_endpoint', type=str)	11✔
83	@click.option('-P', '--peering', 'peering_name', type=str, envvar='KOPF_RUN_PEERING')	11✔
84	@click.option('-p', '--priority', type=int)	11✔
85	@click.option('-m', '--module', 'modules', multiple=True)	11✔
86	@click.argument('paths', nargs=-1)	11✔
87	@click.make_pass_decorator(CLIControls, ensure=True)	11✔
88	def run(	11✔
89	__controls: CLIControls,
90	paths: list[str],
91	modules: list[str],
92	peering_name: Optional[str],
93	priority: Optional[int],
94	standalone: Optional[bool],
95	namespaces: Collection[references.NamespacePattern],
96	clusterwide: bool,
97	liveness_endpoint: Optional[str],
98	) -> None:
99	""" Start an operator process and handle all the requests. """
100	if os.environ.get('KOPF_RUN_NAMESPACE'): # legacy for single-namespace mode	11✔
101	namespaces = tuple(namespaces) + (os.environ.get('KOPF_RUN_NAMESPACE', ''),)	11✔
102	if namespaces and clusterwide:	11!
UNCOV 103	raise click.UsageError("Either --namespace or --all-namespaces can be used, not both.")	×
104	if __controls.registry is not None:	11✔
105	registries.set_default_registry(__controls.registry)	11✔
106	loaders.preload(	11✔
107	paths=paths,
108	modules=modules,
109	)
110	with loops.proper_loop(__controls.loop):	11✔
111	return running.run(	11✔
112	standalone=standalone,
113	namespaces=namespaces,
114	clusterwide=clusterwide,
115	priority=priority,
116	peering_name=peering_name,
117	liveness_endpoint=liveness_endpoint,
118	registry=__controls.registry,
119	settings=__controls.settings,
120	stop_flag=__controls.stop_flag,
121	ready_flag=__controls.ready_flag,
122	vault=__controls.vault,
123	loop=__controls.loop,
124	)
125
126
127	@main.command()	11✔
128	@logging_options	11✔
129	@click.option('-n', '--namespace', 'namespaces', multiple=True)	11✔
130	@click.option('-A', '--all-namespaces', 'clusterwide', is_flag=True)	11✔
131	@click.option('-i', '--id', type=str, default=None)	11✔
132	@click.option('--dev', 'priority', flag_value=666)	11✔
133	@click.option('-P', '--peering', 'peering_name', required=True, envvar='KOPF_FREEZE_PEERING')	11✔
134	@click.option('-p', '--priority', type=int, default=100, required=True)	11✔
135	@click.option('-t', '--lifetime', type=int, required=True)	11✔
136	@click.option('-m', '--message', type=str)	11✔
137	@click.make_pass_decorator(CLIControls, ensure=True)	11✔
138	def freeze(	11✔
139	__controls: CLIControls,
140	id: Optional[str],
141	message: Optional[str],
142	lifetime: int,
143	namespaces: Collection[references.NamespacePattern],
144	clusterwide: bool,
145	peering_name: str,
146	priority: int,
147	) -> None:
148	""" Pause the resource handling in the operator(s). """
UNCOV 149	identity = peering.Identity(id) if id else peering.detect_own_id(manual=True)	×
UNCOV 150	insights = references.Insights()	×
151	settings = configuration.OperatorSettings()	×
152	settings.peering.name = peering_name	×
153	settings.peering.priority = priority	×
154	with loops.proper_loop(__controls.loop):	×
155	return running.run(	×
156	clusterwide=clusterwide,
157	namespaces=namespaces,
158	insights=insights,
159	identity=identity,
160	settings=settings,
161	_command=peering.touch_command(
162	insights=insights,
163	identity=identity,
164	settings=settings,
165	lifetime=lifetime))
166
167
168	@main.command()	11✔
169	@logging_options	11✔
170	@click.option('-n', '--namespace', 'namespaces', multiple=True)	11✔
171	@click.option('-A', '--all-namespaces', 'clusterwide', is_flag=True)	11✔
172	@click.option('-i', '--id', type=str, default=None)	11✔
173	@click.option('-P', '--peering', 'peering_name', required=True, envvar='KOPF_RESUME_PEERING')	11✔
174	@click.make_pass_decorator(CLIControls, ensure=True)	11✔
175	def resume(	11✔
176	__controls: CLIControls,
177	id: Optional[str],
178	namespaces: Collection[references.NamespacePattern],
179	clusterwide: bool,
180	peering_name: str,
181	) -> None:
182	""" Resume the resource handling in the operator(s). """
UNCOV 183	identity = peering.Identity(id) if id else peering.detect_own_id(manual=True)	×
UNCOV 184	insights = references.Insights()	×
185	settings = configuration.OperatorSettings()	×
186	settings.peering.name = peering_name	×
187	with loops.proper_loop(__controls.loop):	×
188	return running.run(	×
189	clusterwide=clusterwide,
190	namespaces=namespaces,
191	insights=insights,
192	identity=identity,
193	settings=settings,
194	_command=peering.touch_command(
195	insights=insights,
196	identity=identity,
197	settings=settings,
198	lifetime=0))

nolar / kopf / 14968876494

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