Jan Lindemann
1385b1a4ba
ExecApp is a ready-made base class for applications that operate through an ExecContext: it adds the --interactive, --verbose and --target options, exposes interactive, verbose and exec_context properties, and closes the exec context when the async context manager exits. The code for that has lived in jw.pkg.App code before, which now inherits from ExecApp. A fix along the way: __aexit__() closes the exec context and then chains to super().__aexit__(), so App.close() runs when the async context manager exits. Before the change, exiting the async context left the app unclosed; close() ran only on the run() path. Add a unit test that builds an ExecApp with a root command and asserts that close() runs on context exit and that the exec options are registered. The exec options are now registered before App's own options, which moves them up in the rendered --help output. Update the golden file of the help integration test to match. Signed-off-by: Jan Lindemann <jan@janware.com> Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v0.84.2
544 lines
19 KiB
Python
544 lines
19 KiB
Python
#
|
|
# This source code file is a merge of various build tools and a horrible mess.
|
|
#
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
|
|
from enum import Enum, auto
|
|
from functools import cache
|
|
from typing import TYPE_CHECKING, override
|
|
|
|
from .lib.ExecApp import ExecApp as Base
|
|
from .lib.Distro import Distro
|
|
from .lib.log import DEBUG, ERR, log
|
|
from .lib.ProjectConf import ProjectConf
|
|
|
|
if TYPE_CHECKING:
|
|
import argparse
|
|
|
|
from argparse import ArgumentParser
|
|
from typing import TypeAlias
|
|
|
|
from .lib.PackageFilter import PackageFilter
|
|
|
|
# Meaning of pkg.requires.xxx variables
|
|
# build: needs to be built and installed before this can be built
|
|
# devel: needs to be installed before this-devel can be installed,
|
|
# i.e. before _other_ packages can be built against this
|
|
# run: needs to be installed before this-run can be installed,
|
|
# i.e. before this and other packages can run with this
|
|
|
|
# --------------------------------------------------------------------- Helpers
|
|
|
|
class Scope(Enum):
|
|
Self = auto()
|
|
One = auto()
|
|
Subtree = auto()
|
|
|
|
Graph: TypeAlias = dict[str, set[str]]
|
|
|
|
# ----------------------------------------------------------------- class App
|
|
|
|
class App(Base):
|
|
|
|
def __format_topdir(self, path: None | str, fmt: str) -> str | None:
|
|
if path is None:
|
|
return None
|
|
match fmt:
|
|
case 'unaltered':
|
|
return path
|
|
case 'relative':
|
|
return os.path.relpath(path)
|
|
case 'absolute':
|
|
return os.path.abspath(path)
|
|
case _:
|
|
m = re.search(r'^make:(\S+)$', fmt)
|
|
if m is None:
|
|
raise Exception(
|
|
f'Can\'t interpret "{fmt}" as valid topdir reference, '
|
|
'expecting "absolute", "relative", "unaltered", '
|
|
'or "make:<variable-name>"'
|
|
)
|
|
return '$(' + m.group(1) + ')'
|
|
|
|
@property
|
|
def __topdir(self) -> str:
|
|
if self.___topdir is None:
|
|
raise Exception('Tried to access undefined top directory')
|
|
return self.___topdir
|
|
|
|
@property
|
|
def __pretty_topdir(self) -> str:
|
|
if self.___pretty_topdir is None:
|
|
raise Exception('Tried to access undefined pretty top directory')
|
|
return self.___pretty_topdir
|
|
|
|
def __proj_dir(self, name: str, pretty: bool) -> str | None:
|
|
if name == self.__top_name:
|
|
if pretty:
|
|
return self.__pretty_topdir
|
|
return self.__topdir
|
|
for d in [self.__projs_root, self.___opt_root]:
|
|
if d is None:
|
|
continue
|
|
ret = d + '/' + name
|
|
if os.path.exists(ret):
|
|
return ret
|
|
if os.path.exists(f'/usr/share/doc/packages/{name}/VERSION'):
|
|
# The package exists but does not have a dedicated project directory
|
|
return None
|
|
raise Exception('No project path found for module "{}"'.format(name))
|
|
|
|
def __find_dir(
|
|
self,
|
|
name: str,
|
|
search_subdirs: list[str] | None = None,
|
|
search_absdirs: list[str] | None = None,
|
|
pretty: bool = True,
|
|
) -> str | None:
|
|
if search_subdirs is None:
|
|
search_subdirs = []
|
|
if search_absdirs is None:
|
|
search_absdirs = []
|
|
|
|
def __format_relpath(path: str) -> str:
|
|
if path.startswith('./'):
|
|
return path[2:]
|
|
if path.endswith('/.'):
|
|
return path[:-2]
|
|
return path
|
|
|
|
def __relpath(target: str, base: str) -> str:
|
|
return __format_relpath(os.path.relpath(target, base))
|
|
|
|
def __format_pd(name: str, pd: str, pretty: bool) -> str | None:
|
|
if not pretty:
|
|
return pd
|
|
if self.__topdir_fmt == 'absolute':
|
|
return str(os.path.abspath(pd))
|
|
if self.__topdir_fmt == 'unaltered':
|
|
return pd
|
|
if self.__topdir_fmt == 'relative':
|
|
return __relpath(pd, self.__topdir)
|
|
if self.__topdir_fmt.startswith('make:'):
|
|
relpath = __relpath(pd, self.__topdir)
|
|
var = self.__topdir_fmt.split(':')[1]
|
|
return __format_relpath(f'$({var})/{relpath}')
|
|
if name == self.__top_name:
|
|
return self.__pretty_topdir
|
|
raise NotImplementedError(
|
|
f'Tried to pretty-format directory {pd}, not implemented'
|
|
)
|
|
|
|
pd = self.__proj_dir(name, False)
|
|
if pd is None:
|
|
return None
|
|
if not search_subdirs and not search_absdirs:
|
|
return __format_pd(name, pd, pretty)
|
|
for sd in search_subdirs:
|
|
path = pd + '/' + sd
|
|
if os.path.isdir(path):
|
|
ret = __format_pd(name, pd, pretty)
|
|
assert ret is not None
|
|
if sd and sd[0] != '/':
|
|
if ret == '.':
|
|
ret = ''
|
|
else:
|
|
ret += '/'
|
|
ret += sd
|
|
return ret
|
|
for ret in search_absdirs:
|
|
if os.path.isdir(ret):
|
|
return ret
|
|
return None
|
|
|
|
def __read_project_conf(self, project_dir: str) -> ProjectConf:
|
|
return ProjectConf.read(project_dir + '/make/project.conf')
|
|
|
|
@cache
|
|
def __get_project_conf(self, project: str) -> ProjectConf | None:
|
|
pd = self.__proj_dir(project, False)
|
|
if pd is None:
|
|
raise Exception(f'Failed to find directory of project {project}')
|
|
try:
|
|
return self.__read_project_conf(pd)
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
def __get_project_refs(
|
|
self,
|
|
buf: list[str],
|
|
visited: set[str],
|
|
spec: str,
|
|
section: str,
|
|
key: str,
|
|
add_self: bool,
|
|
scope: Scope,
|
|
names_only: bool,
|
|
) -> None:
|
|
name = self.strip_module_from_spec(spec)
|
|
if names_only:
|
|
spec = name
|
|
if spec in buf:
|
|
return
|
|
if spec in visited:
|
|
if add_self:
|
|
buf.append(spec)
|
|
return
|
|
visited.add(spec)
|
|
vals = self.get_value(name, section, key)
|
|
log(
|
|
DEBUG,
|
|
(
|
|
f'name={name}, section={section}, key={key}, deps={vals}, '
|
|
f'scope={scope.name}, visited={visited}'
|
|
),
|
|
)
|
|
vals_list = vals.split(',') if vals else []
|
|
match scope:
|
|
case Scope.Self:
|
|
buf += vals_list
|
|
case Scope.One | Scope.Subtree:
|
|
subscope = scope.Self if scope == Scope.One else scope
|
|
for val in vals_list:
|
|
val = val.strip()
|
|
if not (len(val)):
|
|
continue
|
|
self.__get_project_refs(
|
|
buf,
|
|
visited,
|
|
val,
|
|
section,
|
|
key,
|
|
add_self = True,
|
|
scope = subscope,
|
|
names_only = names_only,
|
|
)
|
|
if add_self:
|
|
buf.append(spec)
|
|
|
|
def __read_dep_graph(
|
|
self,
|
|
projects: list[str],
|
|
sections: str | list[str],
|
|
graph: Graph,
|
|
) -> None:
|
|
if isinstance(sections, str):
|
|
sections = [sections]
|
|
for project in projects:
|
|
if project in graph:
|
|
continue
|
|
deps = self.get_project_refs(
|
|
[project],
|
|
['pkg.requires.jw'],
|
|
sections,
|
|
scope = Scope.One,
|
|
add_self = False,
|
|
names_only = True,
|
|
)
|
|
graph[project] = set(deps)
|
|
for dep in deps:
|
|
self.__read_dep_graph([dep], sections, graph)
|
|
|
|
def __flip_dep_graph(self, graph: Graph) -> Graph:
|
|
ret: Graph = {}
|
|
for project, deps in graph.items():
|
|
for d in deps:
|
|
if d not in ret:
|
|
ret[d] = set()
|
|
ret[d].add(project)
|
|
return ret
|
|
|
|
def __find_circular_deps_recursive(
|
|
self,
|
|
project: str,
|
|
graph: Graph,
|
|
unvisited: list[str],
|
|
stack: list[str],
|
|
) -> list[str] | None:
|
|
if project in stack:
|
|
log(DEBUG, 'found circular dependency at project', project)
|
|
idx = stack.index(project)
|
|
return stack[idx:] + [project]
|
|
if project not in unvisited:
|
|
return None
|
|
stack.append(project)
|
|
if project in graph:
|
|
for dep in graph[project]:
|
|
cycle = self.__find_circular_deps_recursive(
|
|
dep, graph, unvisited, stack
|
|
)
|
|
if cycle is not None:
|
|
return cycle
|
|
unvisited.remove(project)
|
|
stack.pop()
|
|
return None
|
|
|
|
def __find_circular_deps(self, projects: list[str],
|
|
flavours: list[str]) -> list[str]:
|
|
graph: Graph = {}
|
|
self.__read_dep_graph(projects, flavours, graph)
|
|
unvisited = list(graph.keys())
|
|
flipped = self.__flip_dep_graph(graph)
|
|
while unvisited:
|
|
project = unvisited[0]
|
|
log(DEBUG, 'Checking circular dependency of', project)
|
|
cycle = self.__find_circular_deps_recursive(project, flipped, unvisited, [])
|
|
if cycle is not None:
|
|
# An edge a -> b in the flipped graph means that b
|
|
# depends on a, so reverse to report the cycle in the
|
|
# original direction
|
|
cycle = list(reversed(cycle))
|
|
log(DEBUG, f'Found circular dependency: {" -> ".join(cycle)}')
|
|
return cycle
|
|
return []
|
|
|
|
def __init__(self, distro: Distro | None = None) -> None:
|
|
|
|
super().__init__(
|
|
description = 'jw-pkg swiss army knife', modules = ['jw.pkg.cmds']
|
|
)
|
|
|
|
# -- Members without default values
|
|
self.__top_name: str | None = None
|
|
self.__distro = distro
|
|
self.___topdir: str | None = None
|
|
self.___pretty_topdir: str | None = None
|
|
|
|
# -- Members with default values
|
|
self.__topdir_fmt = 'absolute'
|
|
self.__projs_root: str | None = None
|
|
self.___opt_root = '/opt'
|
|
self.__pretty_projs_root = None
|
|
|
|
async def __init_async(self) -> None:
|
|
if self.__distro is None:
|
|
pkg_filter_str = self.args.pkg_filter
|
|
if pkg_filter_str is None:
|
|
pkg_filter_str = os.getenv('JW_DEFAULT_PKG_FILTER')
|
|
pkg_filter: PackageFilter | None = None
|
|
if pkg_filter_str is not None:
|
|
from .lib.PackageFilter import PackageFilterString
|
|
|
|
pkg_filter = PackageFilterString(pkg_filter_str)
|
|
self.__distro = await Distro.instantiate(
|
|
ec = self.exec_context,
|
|
id = self.args.distro_id,
|
|
default_pkg_filter = pkg_filter,
|
|
)
|
|
|
|
@override
|
|
def _add_arguments(self, parser: ArgumentParser) -> None:
|
|
super()._add_arguments(parser)
|
|
parser.add_argument('-t', '--topdir', default = None, help = 'Project Path')
|
|
parser.add_argument(
|
|
'--topdir-format',
|
|
default = 'absolute',
|
|
help = (
|
|
'Output references to topdir as one of "make:<var-name>", '
|
|
'"unaltered", "relative", "absolute". Absolute topdir by default'
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
'-p',
|
|
'--prefix',
|
|
default = None,
|
|
help = 'Parent directory of project source directories',
|
|
)
|
|
parser.add_argument(
|
|
'--distro-id',
|
|
default = None,
|
|
help = 'Distribution ID (default is taken from /etc/os-release)',
|
|
)
|
|
parser.add_argument(
|
|
'--pkg-filter',
|
|
help = 'Default filter for all distribution package-related operations',
|
|
)
|
|
|
|
@override
|
|
async def _run(self, args: argparse.Namespace) -> None:
|
|
self.___topdir = args.topdir
|
|
self.___pretty_topdir = self.__format_topdir(self.___topdir, args.topdir_format)
|
|
self.__topdir_fmt = args.topdir_format
|
|
if self.___topdir is not None:
|
|
try:
|
|
conf = self.__read_project_conf(self.__topdir)
|
|
self.__top_name = conf.get_str_or_none('build', 'name')
|
|
except FileNotFoundError:
|
|
pass
|
|
if not self.__top_name:
|
|
self.__top_name = re.sub(
|
|
'-[0-9.-]*$',
|
|
'',
|
|
os.path.basename(os.path.realpath(self.___topdir))
|
|
)
|
|
if args.prefix is not None:
|
|
self.__projs_root = args.prefix
|
|
self.__pretty_projs_root = args.prefix
|
|
await self.__init_async()
|
|
await super()._run(args)
|
|
|
|
@property
|
|
def top_name(self) -> str | None:
|
|
return self.__top_name
|
|
|
|
@property
|
|
def projs_root(self) -> str:
|
|
if self.__projs_root is None:
|
|
raise Exception('Tried to get unknown projects root directory')
|
|
return self.__projs_root
|
|
|
|
@property
|
|
def distro(self) -> Distro:
|
|
if self.__distro is None:
|
|
raise Exception('No distro object')
|
|
return self.__distro
|
|
|
|
def find_dir(
|
|
self,
|
|
name: str,
|
|
search_subdirs: list[str] | None = None,
|
|
search_absdirs: list[str] | None = None,
|
|
pretty: bool = True,
|
|
throw: bool = False,
|
|
) -> str | None:
|
|
ret = self.__find_dir(name, search_subdirs, search_absdirs, pretty)
|
|
if ret is not None:
|
|
return ret
|
|
if not throw:
|
|
return None
|
|
msg = f'Failed to find directory for "{name}":'
|
|
log(ERR, msg)
|
|
for search_name, search in [
|
|
('subdirs', search_subdirs),
|
|
('absdirs', search_absdirs),
|
|
]:
|
|
if search:
|
|
log(ERR, f'Searched {search_name}:')
|
|
for d in search:
|
|
log(ERR, f' - {d}')
|
|
raise FileNotFoundError(msg)
|
|
|
|
# TODO: add support for customizing this in project.conf
|
|
def htdocs_dir(self, project: str) -> str | None:
|
|
return self.find_dir(
|
|
project,
|
|
['/src/html/htdocs', '/tools/html/htdocs', '/htdocs'],
|
|
['/srv/www/proj/' + project],
|
|
)
|
|
|
|
# TODO: add support for customizing this in project.conf
|
|
def tmpl_dir(self, name: str) -> str | None:
|
|
return self.find_dir(name, ['/tmpl'], ['/opt/' + name + '/share/tmpl'])
|
|
|
|
def strip_module_from_spec(self, mod: str) -> str:
|
|
return re.sub(r'-dev$|-devel$|-run$', '', re.split('([=><]+)', mod)[0].strip())
|
|
|
|
@cache
|
|
def get_value(self, project: str, section: str, key: str) -> str | None:
|
|
ret: str | None
|
|
if section == 'version':
|
|
proj_dir = self.__proj_dir(project, pretty = False)
|
|
if proj_dir is None:
|
|
raise Exception(f"Can't get project directory for {project}")
|
|
proj_version_dirs = [proj_dir]
|
|
if proj_dir != self.___topdir:
|
|
proj_version_dirs.append('/usr/share/doc/packages/' + project)
|
|
for d in proj_version_dirs:
|
|
version_path = d + '/VERSION'
|
|
try:
|
|
with open(version_path) as fd:
|
|
ret = fd.read().replace('\n', '').replace('-dev', '')
|
|
return ret
|
|
except EnvironmentError:
|
|
log(DEBUG, f'Ignoring unreadable file "{version_path}"')
|
|
continue
|
|
raise Exception(f'No version file found for project "{project}"')
|
|
proj_conf = self.__get_project_conf(project)
|
|
if proj_conf is None:
|
|
return None
|
|
ret = proj_conf.get_str_or_none(section, key)
|
|
log(
|
|
DEBUG,
|
|
'Lookup %s -> %s / [%s%s] -> "%s"' %
|
|
(self.__top_name, project, section, '.' + key if key else '', ret),
|
|
)
|
|
return ret
|
|
|
|
@cache
|
|
def get_version(self, project: str) -> str:
|
|
ret = self.get_value(project, 'version', '')
|
|
if ret is None:
|
|
raise Exception(f"Can't get version of project {project}")
|
|
return ret
|
|
|
|
def get_values(self, projects: list[str], sections: list[str],
|
|
keys: list[str]) -> list[str]:
|
|
"""
|
|
Collect a list of values from a list of given projects, sections and
|
|
keys, maintaining order
|
|
"""
|
|
ret: list[str] = []
|
|
for p in projects:
|
|
for section in sections:
|
|
for key in keys:
|
|
vals = self.get_value(p, section, key)
|
|
if vals:
|
|
for val in vals.split(','):
|
|
stripped = val.strip()
|
|
if stripped:
|
|
ret.append(stripped)
|
|
return list(dict.fromkeys(ret)) # Remove duplicates, keep ordering
|
|
|
|
def get_project_refs(
|
|
self,
|
|
projects: list[str],
|
|
sections: list[str],
|
|
keys: str | list[str],
|
|
scope: Scope = Scope.One,
|
|
add_self: bool = False,
|
|
names_only: bool = False,
|
|
) -> list[str]:
|
|
if isinstance(keys, str):
|
|
keys = [keys]
|
|
ret: list[str] = []
|
|
for section in sections:
|
|
for key in keys:
|
|
visited: set[str] = set()
|
|
for name in projects:
|
|
rr: list[str] = []
|
|
self.__get_project_refs(
|
|
rr, visited, name, section, key, add_self, scope, names_only
|
|
)
|
|
# TODO: this looks like a performance hogger
|
|
for m in rr:
|
|
if m not in ret:
|
|
ret.append(m)
|
|
return ret
|
|
|
|
def get_libname(self, spec: str) -> str | None:
|
|
project_name = self.strip_module_from_spec(spec)
|
|
ret = self.get_value(project_name, 'build', 'libname')
|
|
if ret == 'none':
|
|
return None
|
|
if ret is None:
|
|
return project_name
|
|
return ret
|
|
|
|
def is_excluded_from_build(self, project: str) -> str | None:
|
|
log(DEBUG, 'checking if project ' + project + ' is excluded from build')
|
|
exclude = self.get_value(project, 'build', 'exclude')
|
|
if exclude is None:
|
|
return None
|
|
exclude_arr = re.split(r'[, ]+', exclude)
|
|
cascade = self.distro.os_cascade + ['all']
|
|
intersection = [x for x in cascade if x in set(exclude_arr)]
|
|
if intersection:
|
|
return ', '.join(intersection)
|
|
return None
|
|
|
|
def find_circular_deps(self, projects: list[str], flavours: list[str]) -> list[str]:
|
|
return self.__find_circular_deps(projects, flavours)
|