ffdd3d13-9fd2-42d6-8817-f9a076d0bad6

Committed 21 Jan 2026 02:47AM UTC coverage: 79.551% (-8.3%) from 87.869%

Build # ffdd3d13-9fd2-42d6-8817-f9a076d0bad6

Build Type

Pull #239

circleci

Committed by

ThrawnCA

Commit Message

[QOLSVC-12515] add support for specifying facet sorts via query parameters

- Use '_{facet}_sort' as a parallel to '_{facet}_limit', with comma separation for the ordering, eg
'_tags_sort=count,desc' to mimic the default popularity-based sorting.

Pull Request Pull Request #239: QOLSVC-12515 alphabetical facet sort

Run Details

11 of 35 new or added lines in 2 files covered. (31.43%)

13656 existing lines in 309 files now uncovered.

42766 of 53759 relevant lines covered (79.55%)

1.63 hits per line

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

95.73

/ckan/cli/cli.py

# encoding: utf-8
from __future__ import annotations

import logging
from collections import defaultdict
from typing import Optional
from pkg_resources import iter_entry_points

import click
import sys

import ckan.plugins as p
import ckan.cli as ckan_cli
from ckan.config.middleware import make_app
from ckan.exceptions import CkanConfigurationException
from . import (
    asset,
    config,
    clean,
    dataset,
    db, search_index, server,
    generate,
    jobs,
    notify,
    plugin_info,
    profile,
    sass,
    sysadmin,
    translation,
    user,
    views,
    config_tool,
    error_shout,
    shell
)

META_ATTR = u'_ckan_meta'
CMD_TYPE_PLUGIN = u'plugin'
CMD_TYPE_ENTRY = u'entry_point'

log = logging.getLogger(__name__)

_no_config_commands = [
    [u'config-tool'],
    [u'generate', u'config'],
    [u'generate', u'extension'],
]


class CtxObject(object):

    def __init__(self, conf: Optional[str] = None):
        # Don't import `load_config` by itself, rather call it using
        # module so that it can be patched during tests
        raw_config = ckan_cli.load_config(conf)
        self.app = make_app(raw_config)

        # Attach the actual CKAN config object to the context
        from ckan.common import config
        self.config = config


class ExtendableGroup(click.Group):
    _section_titles = {
        CMD_TYPE_PLUGIN: u'Plugins',
        CMD_TYPE_ENTRY: u'Entry points',
    }

    def format_commands(
            self, ctx: click.Context, formatter: click.HelpFormatter):
        """Print help message.

        Includes information about commands that were registered by extensions.
        """
        # click won't parse config file from envvar if no other options
        # provided, except for `--help`. In this case it has to be done
        # manually.
        if not ctx.obj:
            _add_ctx_object(ctx)
            _add_external_commands(ctx)

        commands = []
        ext_commands = defaultdict(lambda: defaultdict(list))

        for subcommand in self.list_commands(ctx):
            cmd = self.get_command(ctx, subcommand)
            if cmd is None:
                continue
            if cmd.hidden:
                continue
            help = cmd.short_help or u''

            meta = getattr(cmd, META_ATTR, None)
            if meta:
                ext_commands[meta[u'type']][meta[u'name']].append(
                    (subcommand, help))
            else:
                commands.append((subcommand, help))

        if commands:
            with formatter.section(u'Commands'):
                formatter.write_dl(commands)

        for section, group in ext_commands.items():
            with formatter.section(self._section_titles.get(section, section)):
                for rows in group.values():
                    formatter.write_dl(rows)

    def parse_args(self, ctx: click.Context, args: list[str]):
        """Preprocess options and arguments.

        As long as at least one option is provided, click won't fallback to
        printing help message. That means that `ckan -c config.ini` will be
        executed as command, instead of just printing help message(as `ckan -c
        config.ini --help`).
        In order to fix it, we have to check whether there is at least one
        argument. If no, let's print help message manually

        """
        result = super().parse_args(ctx, args)
        if not ctx.protected_args and not ctx.args:
            click.echo(ctx.get_help(), color=ctx.color)
            ctx.exit()
        return result


def _init_ckan_config(ctx: click.Context, param: str, value: str):
    if any(sys.argv[1:len(cmd) + 1] == cmd for cmd in _no_config_commands):
        return
    _add_ctx_object(ctx, value)
    _add_external_commands(ctx)


def _add_ctx_object(ctx: click.Context, path: Optional[str] = None):
    """Initialize CKAN App using config file available under provided path.

    """
    try:
        ctx.obj = CtxObject(path)
    except CkanConfigurationException as e:
        error_shout(e)
        ctx.abort()

    ctx.meta["flask_app"] = ctx.obj.app._wsgi_app

    # Remove all commands that were registered by extensions before
    # adding new ones. Such situation is possible only during tests,
    # because we are using singleton as main entry point, so it
    # preserves its state even between tests
    commands = getattr(ctx.command, "commands")
    for key, cmd in list(commands.items()):
        if hasattr(cmd, META_ATTR):
            commands.pop(key)


def _add_external_commands(ctx: click.Context):
    add = getattr(ctx.command, "add_command")
    for cmd in _get_commands_from_entry_point():
        add(cmd)

    plugins = p.PluginImplementations(p.IClick)
    for cmd in _get_commands_from_plugins(plugins):
        add(cmd)


def _command_with_ckan_meta(cmd: click.Command, name: str, type_: str):
    """Mark command as one retrieved from CKAN extension.

    This information is used when CLI help text is generated.
    """
    setattr(cmd, META_ATTR, {u'name': name, u'type': type_})
    return cmd


def _get_commands_from_plugins(plugins: p.PluginImplementations[p.IClick]):
    """Register commands that are available when plugin enabled.

    """
    for plugin in plugins:
        for cmd in plugin.get_commands():
            yield _command_with_ckan_meta(cmd, plugin.name, CMD_TYPE_PLUGIN)


def _get_commands_from_entry_point(entry_point: str = 'ckan.click_command'):
    """Register commands that are available even if plugin is not enabled.

    """
    registered_entries = {}
    for entry in iter_entry_points(entry_point):
        if entry.name in registered_entries:
            error_shout((
                u'Attempt to override entry_point `{name}`.\n'
                u'First encounter:\n\t{first!r}\n'
                u'Second encounter:\n\t{second!r}\n'
                u'Either uninstall one of mentioned extensions or update'
                u' corresponding `setup.py` and re-install the extension.'
            ).format(
                name=entry.name,
                first=registered_entries[entry.name].dist,
                second=entry.dist))
            raise click.Abort()
        registered_entries[entry.name] = entry

        yield _command_with_ckan_meta(entry.load(), entry.name, CMD_TYPE_ENTRY)


@click.group(cls=ExtendableGroup)
@click.option(
    u'-c', u'--config', metavar=u'CONFIG',
    is_eager=True, callback=_init_ckan_config, expose_value=False,
    help=u'Config file to use (default: ckan.ini)')
@click.help_option(u'-h', u'--help')
def ckan():
    pass


ckan.add_command(asset.asset)
ckan.add_command(config.config)
ckan.add_command(config_tool.config_tool)
ckan.add_command(dataset.dataset)
ckan.add_command(db.db)
ckan.add_command(generate.generate)
ckan.add_command(jobs.jobs)
ckan.add_command(notify.notify)
ckan.add_command(plugin_info.plugin_info)
ckan.add_command(profile.profile)
ckan.add_command(sass.sass)
ckan.add_command(search_index.search_index)
ckan.add_command(server.run)
ckan.add_command(sysadmin.sysadmin)
ckan.add_command(translation.translation)
ckan.add_command(user.user)
ckan.add_command(views.views)
ckan.add_command(shell.shell)
ckan.add_command(clean.clean)

1	# encoding: utf-8
2	from __future__ import annotations	3✔
3
4	import logging	3✔
5	from collections import defaultdict	3✔
6	from typing import Optional	3✔
7	from pkg_resources import iter_entry_points	3✔
8
9	import click	3✔
10	import sys	3✔
11
12	import ckan.plugins as p	3✔
13	import ckan.cli as ckan_cli	3✔
14	from ckan.config.middleware import make_app	3✔
15	from ckan.exceptions import CkanConfigurationException	3✔
16	from . import (	3✔
17	asset,
18	config,
19	clean,
20	dataset,
21	db, search_index, server,
22	generate,
23	jobs,
24	notify,
25	plugin_info,
26	profile,
27	sass,
28	sysadmin,
29	translation,
30	user,
31	views,
32	config_tool,
33	error_shout,
34	shell
35	)
36
37	META_ATTR = u'_ckan_meta'	3✔
38	CMD_TYPE_PLUGIN = u'plugin'	3✔
39	CMD_TYPE_ENTRY = u'entry_point'	3✔
40
41	log = logging.getLogger(__name__)	3✔
42
43	_no_config_commands = [	3✔
44	[u'config-tool'],
45	[u'generate', u'config'],
46	[u'generate', u'extension'],
47	]
48
49
50	class CtxObject(object):	3✔
51
52	def __init__(self, conf: Optional[str] = None):	3✔
53	# Don't import `load_config` by itself, rather call it using
54	# module so that it can be patched during tests
55	raw_config = ckan_cli.load_config(conf)	3✔
56	self.app = make_app(raw_config)	3✔
57
58	# Attach the actual CKAN config object to the context
59	from ckan.common import config	3✔
60	self.config = config	3✔
61
62
63	class ExtendableGroup(click.Group):	3✔
64	_section_titles = {	3✔
65	CMD_TYPE_PLUGIN: u'Plugins',
66	CMD_TYPE_ENTRY: u'Entry points',
67	}
68
69	def format_commands(	3✔
70	self, ctx: click.Context, formatter: click.HelpFormatter):
71	"""Print help message.
72
73	Includes information about commands that were registered by extensions.
74	"""
75	# click won't parse config file from envvar if no other options
76	# provided, except for `--help`. In this case it has to be done
77	# manually.
78	if not ctx.obj:	2✔
UNCOV 79	_add_ctx_object(ctx)	1✔
UNCOV 80	_add_external_commands(ctx)	1✔
81
82	commands = []	2✔
83	ext_commands = defaultdict(lambda: defaultdict(list))	2✔
84
85	for subcommand in self.list_commands(ctx):	2✔
86	cmd = self.get_command(ctx, subcommand)	2✔
87	if cmd is None:	2✔
88	continue	×
89	if cmd.hidden:	2✔
90	continue	×
91	help = cmd.short_help or u''	2✔
92
93	meta = getattr(cmd, META_ATTR, None)	2✔
94	if meta:	2✔
95	ext_commands[meta[u'type']][meta[u'name']].append(	2✔
96	(subcommand, help))
97	else:
98	commands.append((subcommand, help))	2✔
99
100	if commands:	2✔
101	with formatter.section(u'Commands'):	2✔
102	formatter.write_dl(commands)	2✔
103
104	for section, group in ext_commands.items():	2✔
105	with formatter.section(self._section_titles.get(section, section)):	2✔
106	for rows in group.values():	2✔
107	formatter.write_dl(rows)	2✔
108
109	def parse_args(self, ctx: click.Context, args: list[str]):	3✔
110	"""Preprocess options and arguments.
111
112	As long as at least one option is provided, click won't fallback to
113	printing help message. That means that `ckan -c config.ini` will be
114	executed as command, instead of just printing help message(as `ckan -c
115	config.ini --help`).
116	In order to fix it, we have to check whether there is at least one
117	argument. If no, let's print help message manually
118
119	"""
120	result = super().parse_args(ctx, args)	3✔
121	if not ctx.protected_args and not ctx.args:	3✔
122	click.echo(ctx.get_help(), color=ctx.color)	1✔
123	ctx.exit()	1✔
124	return result	3✔
125
126
127	def _init_ckan_config(ctx: click.Context, param: str, value: str):	3✔
128	if any(sys.argv[1:len(cmd) + 1] == cmd for cmd in _no_config_commands):	3✔
129	return	×
130	_add_ctx_object(ctx, value)	3✔
131	_add_external_commands(ctx)	3✔
132
133
134	def _add_ctx_object(ctx: click.Context, path: Optional[str] = None):	3✔
135	"""Initialize CKAN App using config file available under provided path.
136
137	"""
138	try:	3✔
139	ctx.obj = CtxObject(path)	3✔
UNCOV 140	except CkanConfigurationException as e:	1✔
UNCOV 141	error_shout(e)	1✔
UNCOV 142	ctx.abort()	1✔
143
144	ctx.meta["flask_app"] = ctx.obj.app._wsgi_app	3✔
145
146	# Remove all commands that were registered by extensions before
147	# adding new ones. Such situation is possible only during tests,
148	# because we are using singleton as main entry point, so it
149	# preserves its state even between tests
150	commands = getattr(ctx.command, "commands")	3✔
151	for key, cmd in list(commands.items()):	3✔
152	if hasattr(cmd, META_ATTR):	3✔
153	commands.pop(key)	3✔
154
155
156	def _add_external_commands(ctx: click.Context):	3✔
157	add = getattr(ctx.command, "add_command")	3✔
158	for cmd in _get_commands_from_entry_point():	3✔
159	add(cmd)	3✔
160
161	plugins = p.PluginImplementations(p.IClick)	3✔
162	for cmd in _get_commands_from_plugins(plugins):	3✔
163	add(cmd)	3✔
164
165
166	def _command_with_ckan_meta(cmd: click.Command, name: str, type_: str):	3✔
167	"""Mark command as one retrieved from CKAN extension.
168
169	This information is used when CLI help text is generated.
170	"""
171	setattr(cmd, META_ATTR, {u'name': name, u'type': type_})	3✔
172	return cmd	3✔
173
174
175	def _get_commands_from_plugins(plugins: p.PluginImplementations[p.IClick]):	3✔
176	"""Register commands that are available when plugin enabled.
177
178	"""
179	for plugin in plugins:	3✔
180	for cmd in plugin.get_commands():	3✔
181	yield _command_with_ckan_meta(cmd, plugin.name, CMD_TYPE_PLUGIN)	3✔
182
183
184	def _get_commands_from_entry_point(entry_point: str = 'ckan.click_command'):	3✔
185	"""Register commands that are available even if plugin is not enabled.
186
187	"""
188	registered_entries = {}	3✔
189	for entry in iter_entry_points(entry_point):	3✔
190	if entry.name in registered_entries:	3✔
191	error_shout((	×
192	u'Attempt to override entry_point `{name}`.\n'
193	u'First encounter:\n\t{first!r}\n'
194	u'Second encounter:\n\t{second!r}\n'
195	u'Either uninstall one of mentioned extensions or update'
196	u' corresponding `setup.py` and re-install the extension.'
197	).format(
198	name=entry.name,
199	first=registered_entries[entry.name].dist,
200	second=entry.dist))
201	raise click.Abort()	×
202	registered_entries[entry.name] = entry	3✔
203
204	yield _command_with_ckan_meta(entry.load(), entry.name, CMD_TYPE_ENTRY)	3✔
205
206
207	@click.group(cls=ExtendableGroup)	3✔
208	@click.option(	3✔
209	u'-c', u'--config', metavar=u'CONFIG',
210	is_eager=True, callback=_init_ckan_config, expose_value=False,
211	help=u'Config file to use (default: ckan.ini)')
212	@click.help_option(u'-h', u'--help')	3✔
213	def ckan():	3✔
214	pass	3✔
215
216
217	ckan.add_command(asset.asset)	3✔
218	ckan.add_command(config.config)	3✔
219	ckan.add_command(config_tool.config_tool)	3✔
220	ckan.add_command(dataset.dataset)	3✔
221	ckan.add_command(db.db)	3✔
222	ckan.add_command(generate.generate)	3✔
223	ckan.add_command(jobs.jobs)	3✔
224	ckan.add_command(notify.notify)	3✔
225	ckan.add_command(plugin_info.plugin_info)	3✔
226	ckan.add_command(profile.profile)	3✔
227	ckan.add_command(sass.sass)	3✔
228	ckan.add_command(search_index.search_index)	3✔
229	ckan.add_command(server.run)	3✔
230	ckan.add_command(sysadmin.sysadmin)	3✔
231	ckan.add_command(translation.translation)	3✔
232	ckan.add_command(user.user)	3✔
233	ckan.add_command(views.views)	3✔
234	ckan.add_command(shell.shell)	3✔
235	ckan.add_command(clean.clean)	3✔

qld-gov-au / ckan / ffdd3d13-9fd2-42d6-8817-f9a076d0bad6

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