Files
modulus/modulus/cli.py
T
khodowany 0c9ba1bff4
publish / publish (push) Successful in 10m36s
Cleanup
2026-09-24 19:40:03 -05:00

1087 lines
34 KiB
Python

import argparse
import hashlib
import os
import pathlib
import shlex
import shutil
import subprocess
import sys
import tarfile
import urllib.parse
import modulus.config
import modulus.export
import modulus.clangd
import modulus.formatting
build_root = pathlib.Path("build")
common_dir = build_root / "build_common"
tools_dir = build_root / "build_tools"
class build_target:
def __init__(self, name, profile, build_dir, build_type, can_run, cmake_definitions):
self.name = name
self.profile = profile
self.build_dir = build_dir
self.build_type = build_type
self.can_run = can_run
self.cmake_definitions = cmake_definitions
def main(command):
repo_root = find_repo_root(pathlib.Path.cwd())
config = modulus.config.load_repo_config(repo_root)
context = build_context(repo_root, config)
if command == "format":
return modulus.formatting.launch(context, sys.argv[1:])
if command == "clangd":
return modulus.clangd.launch(context, sys.argv[1:])
if command in {"configure", "lsp"}:
modulus.export.sync_editor_files(repo_root, config)
if command == "configure":
return configure(context, sys.argv[1:])
if command == "build":
return build(context, sys.argv[1:])
if command == "test":
return test(context, sys.argv[1:])
if command == "package":
return package(context, sys.argv[1:])
if command == "run":
return product_run(context, sys.argv[1:])
if command == "lsp":
return lsp(context, sys.argv[1:])
raise SystemExit(f"unknown modulus command: {command}")
def find_repo_root(start):
current = start.resolve()
for candidate in (current, *current.parents):
if (candidate / "modulus.json").is_file():
return candidate
raise SystemExit("could not find modulus.json from current directory")
class build_context:
def __init__(self, repo_root, config):
self.repo_root = repo_root
self.config = config
self.package = config["package"]
self.artifact = self.package["artifact"]
self.tooling = config["tooling"]
self.run_settings = self.tooling["run"]
self.lsp_settings = self.tooling.get("lsp", {"enabled": False, "compile_databases": []})
self.build_root = repo_root / build_root
self.common_dir = repo_root / common_dir
self.tools_rel = tools_dir if (repo_root / tools_dir).exists() else pathlib.Path("tools")
self.tools_dir = repo_root / self.tools_rel
self.conan_cache_dir = self.common_dir / "conan_cache"
self.default_target_name = str(self.tooling["default_target"])
self.targets = self._targets()
self.default_target = self.targets[self.default_target_name]
docker = self.tooling["docker"]
self.docker_image = str(docker["image"])
self.dockerfile = repo_root / docker["dockerfile"] if docker.get("dockerfile") else self.tools_dir / "docker" / "Dockerfile"
self.docker_inputs = [self.tools_dir / "docker" / "format-requirements.txt", *(repo_root / path for path in docker.get("inputs", []))]
self.docker_workdir = str(repo_root)
self.docker_marker = pathlib.Path(str(docker["marker"]))
self.upload_env = str(self.tooling["upload_env"])
self.package_upload_env = str(self.artifact["upload_env"])
def _targets(self):
result = {}
targets = self.tooling["targets"]
for name, target in targets.items():
validate_build_directory(self.repo_root, self.repo_root, self.repo_root / str(target["build_dir"]))
result[name] = build_target(
name=name,
profile=self.repo_root / str(target["profile"]),
build_dir=self.repo_root / str(target["build_dir"]),
build_type=str(target.get("build_type", "Release")),
can_run=bool(target.get("can_run", True)),
cmake_definitions=dict(target.get("cmake_definitions", {})),
)
return result
def target(self, name):
try:
return self.targets[name]
except KeyError:
valid = ", ".join(sorted(self.targets))
raise SystemExit(f"unknown target '{name}'. Valid targets: {valid}") from None
def selected_targets(self, names, all_targets):
if all_targets:
return list(self.targets.values())
if not names:
return [self.default_target]
return [self.target(name) for name in names]
def in_docker(context):
return direct_mode() or context.docker_marker.exists()
def env_flag(name):
return os.environ.get(name, "").lower() in {"1", "true", "yes", "on"}
def direct_mode():
return env_flag("MODULUS_DIRECT")
def in_container():
return pathlib.Path("/.dockerenv").exists()
def current_container_name():
return os.environ.get("HOSTNAME", "").strip()
def docker_platform_args():
platform = os.environ.get("MODULUS_DOCKER_PLATFORM", "").strip()
if not platform:
return []
return [
"--platform",
platform,
]
def conan_remote(context):
return str(context.tooling["remote"]["name"])
def conan_remote_url(context):
return context.tooling["remote"]["url"]
def conan_user():
return os.environ.get("CONAN_USER")
def conan_token():
return os.environ.get("CONAN_TOKEN")
def conan_credentials():
return conan_user(), conan_token()
def require_conan_credentials(operation):
user, token = conan_credentials()
if bool(user) != bool(token):
print("CONAN_USER and CONAN_TOKEN must either both be set or both be unset", file=sys.stderr)
return None
if not user or not token:
print(f"CONAN_USER and CONAN_TOKEN are required to {operation}", file=sys.stderr)
return None
return user, token
def subprocess_environment(context, command):
env = os.environ.copy()
env["CONAN_HOME"] = str(context.conan_cache_dir)
if str(command[0]) == "conan":
for name in list(env):
if name in {"CONAN_USER", "CONAN_TOKEN"} or name.startswith(("CONAN_LOGIN_USERNAME", "CONAN_PASSWORD")):
del env[name]
env["CONAN_NON_INTERACTIVE"] = "1"
return env
def run(context, command, cwd=None, env_overrides=None, env_unset=None, display_command=None):
visible = display_command if display_command is not None else command
print("+", " ".join(shlex.quote(str(part)) for part in visible))
env = subprocess_environment(context, command)
for name in env_unset or ():
env.pop(name, None)
if env_overrides:
env.update(env_overrides)
return subprocess.run([str(part) for part in command], cwd=cwd or context.repo_root, check=False, env=env).returncode
def run_capture(context, command, cwd=None, env_overrides=None, display_command=None):
visible = display_command if display_command is not None else command
print("+", " ".join(shlex.quote(str(part)) for part in visible))
env = subprocess_environment(context, command)
if env_overrides:
env.update(env_overrides)
completed = subprocess.run(
[str(part) for part in command],
cwd=cwd or context.repo_root,
check=False,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
if completed.returncode != 0:
if completed.stdout:
print(completed.stdout, end="")
if completed.stderr:
print(completed.stderr, end="", file=sys.stderr)
return completed
def configure_conan_remote(context, authenticate=False):
remote = conan_remote(context)
credentials = require_conan_credentials("upload Conan packages") if authenticate else None
if authenticate and credentials is None:
return 1
status = run(
context,
[
"conan",
"remote",
"add",
"-f",
remote,
conan_remote_url(context),
],
)
if status != 0:
return status
if not authenticate:
return run(context, ["conan", "remote", "logout", remote])
user, token = credentials
return run(
context,
[
"conan",
"remote",
"login",
remote,
user,
"-p",
token,
],
display_command=[
"conan",
"remote",
"login",
remote,
user,
"-p",
"<hidden>",
],
)
def docker_passthrough_env(context, publishing=False):
names = (
"CI",
"CONAN_HOME",
"MODULUS_DIRECT",
"MODULUS_ALLOW_PACKAGE_REPLACE",
context.upload_env,
context.package_upload_env,
)
if publishing:
names += ("CONAN_TOKEN", "CONAN_USER", "PACKAGE_REGISTRY_TOKEN", "PACKAGE_REGISTRY_USER")
return names
def docker_socket_args(context, command):
if command not in context.tooling["docker"].get("socket_commands", []):
return []
socket = pathlib.Path("/var/run/docker.sock")
if not socket.is_socket():
return []
return ["--volume", f"{socket}:{socket}", "--group-add", str(socket.stat().st_gid)]
def docker_platform_arch():
platform = os.environ.get("MODULUS_DOCKER_PLATFORM", "").strip()
if not platform:
return None
parts = platform.split("/")
if len(parts) < 2:
return None
return {
"amd64": "amd64",
"arm64": "arm64",
"arm64v8": "arm64",
}.get(parts[1])
def docker_image_available(context):
return modulus.clangd.current_image(context, docker_platform_arch()) is not None
def build_image(context):
return run(
context,
[
"docker",
"build",
*docker_platform_args(),
"--file",
context.dockerfile,
"--tag",
context.docker_image,
"--label",
f"{modulus.clangd.image_label}={modulus.clangd.recipe_digest(context)}",
"--build-arg",
f"MODULUS_TOOLS={context.tools_rel}",
".",
],
)
def ensure_build_image(context):
if docker_image_available(context):
return 0
return build_image(context)
def docker_run(context, command, docker_args=None, env_overrides=None):
status = ensure_build_image(context)
if status != 0:
return status
resolved_env_overrides = dict(env_overrides or {})
operation = pathlib.Path(command[0]).stem
publishing = operation == "package" and "--no-remote" not in command and (should_upload(context) or should_upload_artifacts(context))
docker_command = [
"docker",
"run",
"--rm",
"--interactive",
*docker_platform_args(),
"--user",
f"{os.getuid()}:{os.getgid()}",
"--env",
"HOME=/tmp",
*docker_socket_args(context, operation),
]
if docker_args:
docker_command.extend(docker_args)
for name in docker_passthrough_env(context, publishing):
if name in os.environ:
docker_command.extend(["--env", name])
if resolved_env_overrides:
for name in resolved_env_overrides:
docker_command.extend(["--env", name])
if in_container():
container_name = current_container_name()
if not container_name:
raise SystemExit("HOSTNAME is required to share the CI job workspace with the build container")
docker_command.extend([
"--volumes-from",
container_name,
])
else:
docker_command.extend([
"--volume",
f"{context.repo_root}:{context.docker_workdir}",
])
docker_command.extend([
"--workdir",
context.docker_workdir,
context.docker_image,
*command,
])
return run(context, docker_command, env_overrides=resolved_env_overrides)
def docker_publish_args(context):
args = []
for port in context.run_settings.get("ports", []):
args.extend(["--publish", port])
return args
def clear_cmake_cache(target):
cache = target.build_dir / "CMakeCache.txt"
cmake_files = target.build_dir / "CMakeFiles"
if cache.exists():
cache.unlink()
if cmake_files.exists():
shutil.rmtree(cmake_files)
def clear_conan_deps(target):
conan_deps = target.build_dir / "conan"
if conan_deps.exists():
shutil.rmtree(conan_deps)
def cmake_cache_values(path):
result = {}
try:
lines = path.read_text(encoding="utf-8").splitlines()
except FileNotFoundError:
return result
for line in lines:
if not line or line.startswith(("//", "#")) or "=" not in line:
continue
key, value = line.split("=", 1)
name = key.split(":", 1)[0]
result[name] = value
return result
def cmake_bool_value(value):
text = str(value).upper()
if text.endswith("-NOTFOUND"):
return False
if text in {"1", "ON", "TRUE", "YES", "Y"}:
return True
if text in {"0", "OFF", "FALSE", "NO", "N", "IGNORE", "NOTFOUND", ""}:
return False
return None
def cmake_cache_value_matches(configured, expected):
if isinstance(expected, bool):
return cmake_bool_value(configured) is expected
return configured == cmake_value(expected)
def target_cmake_definitions_match(target):
definitions = target.cmake_definitions
if not definitions:
return True
values = cmake_cache_values(target.build_dir / "CMakeCache.txt")
for name, expected in definitions.items():
configured = values.get(name)
if configured is None or not cmake_cache_value_matches(configured, expected):
return False
return True
def target_is_configured(target):
return (
(target.build_dir / "CMakeCache.txt").is_file()
and (target.build_dir / "build.ninja").is_file()
and target_cmake_definitions_match(target)
)
def package_name(context):
return context.package["name"]
def dependency_build_args(context):
return ["--build=missing"]
def configure_one(context, target, update, no_remote=False):
clear_conan_deps(target)
command = [
"conan",
"install",
".",
f"--output-folder={target.build_dir / 'conan'}",
f"--profile:host={target.profile}",
f"--profile:build={context.default_target.profile}",
*dependency_build_args(context),
"-o",
"&:with_tests=True" if target.can_run else "&:with_tests=False",
]
if update:
command.append("--update")
if no_remote:
command.append("--no-remote")
status = run(context, command)
if status != 0:
return status
clear_cmake_cache(target)
return run(
context,
[
"cmake",
"-S",
".",
"-B",
target.build_dir,
"-G",
"Ninja",
f"-DCMAKE_TOOLCHAIN_FILE={target.build_dir / 'conan' / 'conan_toolchain.cmake'}",
f"-DCMAKE_BUILD_TYPE={target.build_type}",
"-DCMAKE_EXPORT_COMPILE_COMMANDS=ON",
f"-DBUILD_TESTING={'ON' if target.can_run else 'OFF'}",
f"-DCMAKE_PREFIX_PATH={target.build_dir / 'conan'}",
*cmake_definition_args(target.cmake_definitions),
],
)
def configure(context, argv):
if not in_docker(context):
status = docker_run(
context,
[
f"./{context.tools_rel}/configure.py",
*argv,
],
)
return status or modulus.clangd.prepare(context)
parser = argparse.ArgumentParser(description=f"Configure {context.package['name']} for Conan/CMake targets.")
parser.add_argument(
"--target",
action="append",
default=[],
choices=sorted(context.targets),
help="Target to configure. May be passed more than once. Defaults to the default target.",
)
parser.add_argument("--all", action="store_true", help="Configure all supported targets.")
parser.add_argument("--update", action="store_true", help="Check Conan remotes for newer dependency revisions.")
parser.add_argument("--no-remote", action="store_true", help="Do not query Conan remotes. Requires all dependencies to be available locally or buildable from local recipes.")
args = parser.parse_args(argv)
context.conan_cache_dir.mkdir(parents=True, exist_ok=True)
if not args.no_remote:
status = configure_conan_remote(context)
if status != 0:
return status
update = args.update
for target in context.selected_targets(args.target, args.all):
target.build_dir.mkdir(parents=True, exist_ok=True)
status = configure_one(context, target, update, no_remote=args.no_remote)
if status != 0:
return status
return 0
def build(context, argv):
if not in_docker(context):
return docker_run(
context,
[
f"./{context.tools_rel}/build.py",
*argv,
],
)
parser = argparse.ArgumentParser(description=f"Build {context.package['name']} for configured targets.")
parser.add_argument(
"--target",
action="append",
default=[],
choices=sorted(context.targets),
help="Target to build. May be passed more than once. Defaults to the default target.",
)
parser.add_argument(
"--cmake-target",
action="append",
default=[],
help="CMake target to build. May be passed more than once.",
)
parser.add_argument("--all", action="store_true", help="Build all supported targets.")
args = parser.parse_args(argv)
selected_targets = context.selected_targets(args.target, args.all)
if any(not target_is_configured(target) for target in selected_targets):
context.conan_cache_dir.mkdir(parents=True, exist_ok=True)
status = configure_conan_remote(context)
if status != 0:
return status
for target in selected_targets:
if not target_is_configured(target):
target.build_dir.mkdir(parents=True, exist_ok=True)
status = configure_one(context, target, update=False)
if status != 0:
return status
cmake_targets = list(args.cmake_target)
if not cmake_targets:
cmake_targets = [None]
for cmake_target in cmake_targets:
command = [
"cmake",
"--build",
target.build_dir,
]
if cmake_target is not None:
command.extend(["--target", cmake_target])
command.extend(["--parallel", "2"])
status = run(context, command)
if status != 0:
return status
return 0
def test(context, argv):
if not in_docker(context):
return docker_run(
context,
[
f"./{context.tools_rel}/test.py",
*argv,
],
)
parser = argparse.ArgumentParser(description=f"Run {context.package['name']} tests for configured targets.")
parser.add_argument(
"--target",
action="append",
default=[],
choices=sorted(context.targets),
help="Target to test. May be passed more than once. Defaults to the default target.",
)
parser.add_argument("--all", action="store_true", help="Test all supported targets.")
args = parser.parse_args(argv)
test_name = context.config["tests"]["name"]
for target in context.selected_targets(args.target, args.all):
if not target.can_run:
print(f"Skipping tests for {target.name}: target binaries cannot run in this container.")
continue
status = run(
context,
[
"ctest",
"--test-dir",
target.build_dir,
"-R",
f"^{test_name}$",
"--output-on-failure",
],
)
if status != 0:
return status
return 0
def lsp_compile_databases(context):
if not context.lsp_settings["enabled"]:
return []
return list(context.lsp_settings["compile_databases"])
def lsp_database_map(context):
return {
str(database["name"]): database
for database in lsp_compile_databases(context)
}
def clear_lsp_cmake_cache(build_dir):
cache = build_dir / "CMakeCache.txt"
cmake_files = build_dir / "CMakeFiles"
if cache.exists():
cache.unlink()
if cmake_files.exists():
shutil.rmtree(cmake_files)
def validate_build_directory(repo_root, source_dir, build_dir):
resolved = pathlib.Path(build_dir).resolve()
if resolved in {pathlib.Path(repo_root).resolve(), pathlib.Path(source_dir).resolve()}:
raise modulus.config.config_error(f"build directory must differ from the project and source roots: {build_dir}")
def configure_lsp_database(context, database):
build_dir = context.repo_root / str(database["build_dir"])
source_dir = context.repo_root / str(database.get("source_dir", "."))
validate_build_directory(context.repo_root, source_dir, build_dir)
build_dir.mkdir(parents=True, exist_ok=True)
clear_lsp_cmake_cache(build_dir)
return run(
context,
[
"cmake",
"-S",
source_dir,
"-B",
build_dir,
"-G",
str(database.get("generator", "Ninja")),
"-DCMAKE_EXPORT_COMPILE_COMMANDS=ON",
*cmake_definition_args(database.get("cmake_definitions", {})),
],
env_overrides=dict(database.get("env", {})),
env_unset=list(database.get("unset_env", [])),
)
def lsp(context, argv):
if not in_docker(context):
status = docker_run(
context,
[
f"./{context.tools_rel}/lsp.py",
*argv,
],
)
return status or modulus.clangd.prepare(context)
database_map = lsp_database_map(context)
parser = argparse.ArgumentParser(description=f"Refresh {context.package['name']} compile databases for clangd.")
parser.add_argument(
"--target",
action="append",
default=[],
choices=sorted(context.targets),
help="Configured build target compile database to refresh. May be passed more than once. Defaults to the default target.",
)
parser.add_argument(
"--database",
action="append",
default=[],
choices=sorted(database_map),
help="Extra LSP compile database to refresh. May be passed more than once. Defaults to all configured LSP databases.",
)
parser.add_argument("--no-build-targets", action="store_true", help="Skip Modulus build-target compile databases.")
parser.add_argument("--update", action="store_true", help="Check Conan remotes for newer dependency revisions.")
parser.add_argument("--no-remote", action="store_true", help="Do not query Conan remotes for Modulus build-target compile databases.")
args = parser.parse_args(argv)
if not args.no_build_targets:
context.conan_cache_dir.mkdir(parents=True, exist_ok=True)
if not args.no_remote:
status = configure_conan_remote(context)
if status != 0:
return status
for target in context.selected_targets(args.target, all_targets=False):
target.build_dir.mkdir(parents=True, exist_ok=True)
status = configure_one(context, target, args.update, no_remote=args.no_remote)
if status != 0:
return status
database_names = args.database or [str(database["name"]) for database in lsp_compile_databases(context)]
for name in database_names:
status = configure_lsp_database(context, database_map[name])
if status != 0:
return status
return 0
def product_run(context, argv):
if not context.run_settings["enabled"]:
print(f"{package_name(context)} does not enable the Modulus run command", file=sys.stderr)
return 1
if not in_docker(context):
return docker_run(
context,
[
f"./{context.tools_rel}/run.py",
*argv,
],
docker_args=docker_publish_args(context),
)
parser = argparse.ArgumentParser(description=f"Run {context.package['name']} from a configured build target.")
parser.add_argument(
"--target",
default=str(context.run_settings["target"]),
choices=sorted(context.targets),
help="Configured target to run. Defaults to tooling.run.target.",
)
parser.add_argument(
"args",
nargs=argparse.REMAINDER,
help="Arguments passed to the product executable. A leading -- separator is optional.",
)
args = parser.parse_args(argv)
target = context.target(args.target)
if not target.can_run:
print(f"Target {target.name} is not runnable in this environment", file=sys.stderr)
return 1
executable = pathlib.Path(str(context.run_settings["executable"]))
if not executable.is_absolute():
executable = target.build_dir / executable
if not executable.is_file():
print(f"Executable was not found: {executable}. Run python3 modulus.py build --target {target.name} first.", file=sys.stderr)
return 1
command_args = list(context.run_settings["args"])
extra_args = list(args.args)
if extra_args and extra_args[0] == "--":
extra_args = extra_args[1:]
command_args.extend(extra_args)
return run(
context,
[
executable,
*command_args,
],
env_overrides=dict(context.run_settings["env"]),
)
def create_package(context, target, update, no_remote=False):
command = [
"conan",
"create",
".",
f"--profile:host={target.profile}",
f"--profile:build={context.default_target.profile}",
f"--build={package_name(context)}/*",
*dependency_build_args(context),
"-o",
"&:with_tests=False",
]
if update:
command.append("--update")
if no_remote:
command.append("--no-remote")
return run(context, command)
def upload_all_packages(context, remote, force=False):
command = [
"conan",
"upload",
f"{package_name(context)}/latest:*",
"-r",
remote,
"--confirm",
]
if force:
command.append("--force")
return run(context, command)
def allow_package_replace():
return env_flag("MODULUS_ALLOW_PACKAGE_REPLACE")
def upload_packages(context, replace=False):
if replace and not allow_package_replace():
print("--replace requires MODULUS_ALLOW_PACKAGE_REPLACE=1", file=sys.stderr)
return 1
remote = conan_remote(context)
try:
status = configure_conan_remote(context, authenticate=True)
if status == 0:
status = upload_all_packages(context, remote, force=replace)
finally:
logout_status = run(context, ["conan", "remote", "logout", remote])
status = status or logout_status
if status != 0 and not replace:
print(f"upload failed for {package_name(context)}/latest. Use --replace with MODULUS_ALLOW_PACKAGE_REPLACE=1 to intentionally replace an existing version.", file=sys.stderr)
return status
def should_upload(context):
return env_flag(context.upload_env)
def artifact_packaging_enabled(context):
return bool(context.artifact["enabled"])
def should_upload_artifacts(context):
return env_flag(context.package_upload_env)
def package_registry_credentials():
user = os.environ.get("PACKAGE_REGISTRY_USER")
token = os.environ.get("PACKAGE_REGISTRY_TOKEN")
if bool(user) != bool(token):
print("PACKAGE_REGISTRY_USER and PACKAGE_REGISTRY_TOKEN must either both be set or both be unset", file=sys.stderr)
return None
if not user or not token:
print("PACKAGE_REGISTRY_USER and PACKAGE_REGISTRY_TOKEN are required to upload package artifacts", file=sys.stderr)
return None
return user, token
def option_value(value):
if isinstance(value, bool):
return "True" if value else "False"
return str(value)
def cmake_value(value):
if isinstance(value, bool):
return "ON" if value else "OFF"
return str(value)
def cmake_definition_args(definitions):
result = []
for name, value in definitions.items():
result.append(f"-D{name}={cmake_value(value)}")
return result
def package_artifact_conan_option_args(context):
result = []
for name, value in context.artifact["conan_options"].items():
result.extend(["-o", f"&:{name}={option_value(value)}"])
return result
def package_artifact_cmake_definition_args(context):
return cmake_definition_args(context.artifact["cmake_definitions"])
def package_artifact_targets(context, names, all_targets):
if all_targets or names:
return context.selected_targets(names, all_targets)
configured = context.artifact["targets"]
if configured:
return [context.target(name) for name in configured]
return [context.default_target]
def git_version(context):
completed = run_capture(
context,
[
"git",
"rev-parse",
"--short=12",
"HEAD",
],
)
if completed.returncode != 0:
return "dev"
version = completed.stdout.strip()
return version if version else "dev"
def package_artifact_build_root(context, target):
return context.common_dir / "artifacts" / target.name
def package_artifact_dist_dir(context):
return context.build_root / "packages"
def package_artifact_install_prefix(context, stage_root):
install_prefix = pathlib.Path(str(context.artifact["install_prefix"]))
if str(install_prefix) == ".":
return stage_root
return stage_root / install_prefix
def create_archive(source_dir, archive):
with tarfile.open(archive, "w:gz") as output:
for path in sorted(source_dir.rglob("*")):
output.add(path, arcname=path.relative_to(source_dir), recursive=False)
def write_checksum(archive):
digest = hashlib.sha256()
with archive.open("rb") as input_file:
for chunk in iter(lambda: input_file.read(1024 * 1024), b""):
digest.update(chunk)
checksum = archive.with_name(archive.name + ".sha256")
checksum.write_text(f"{digest.hexdigest()} {archive.name}\n", encoding="utf-8")
return checksum
def create_artifact_package(context, target, version, update, no_remote=False):
root = package_artifact_build_root(context, target)
build_dir = root / "build"
stage_root = root / "stage"
conan_dir = build_dir / "conan"
if root.exists():
shutil.rmtree(root)
build_dir.mkdir(parents=True)
stage_root.mkdir(parents=True)
install_prefix = package_artifact_install_prefix(context, stage_root)
install_prefix.mkdir(parents=True, exist_ok=True)
conan_command = [
"conan",
"install",
".",
f"--output-folder={conan_dir}",
f"--profile:host={target.profile}",
f"--profile:build={context.default_target.profile}",
*dependency_build_args(context),
"-o",
"&:with_tests=False",
*package_artifact_conan_option_args(context),
]
if update:
conan_command.append("--update")
if no_remote:
conan_command.append("--no-remote")
status = run(context, conan_command)
if status != 0:
return status, []
status = run(
context,
[
"cmake",
"-S",
".",
"-B",
build_dir,
"-G",
"Ninja",
f"-DCMAKE_TOOLCHAIN_FILE={conan_dir / 'conan_toolchain.cmake'}",
f"-DCMAKE_BUILD_TYPE={target.build_type}",
"-DBUILD_TESTING=OFF",
f"-DCMAKE_PREFIX_PATH={conan_dir}",
f"-DCMAKE_INSTALL_PREFIX={install_prefix}",
*package_artifact_cmake_definition_args(context),
],
)
if status != 0:
return status, []
status = run(
context,
[
"cmake",
"--build",
build_dir,
"--parallel",
"2",
],
)
if status != 0:
return status, []
status = run(
context,
[
"cmake",
"--install",
build_dir,
],
)
if status != 0:
return status, []
dist_dir = package_artifact_dist_dir(context)
dist_dir.mkdir(parents=True, exist_ok=True)
archive = dist_dir / f"{package_name(context)}-{version}-{target.name}.tar.gz"
if archive.exists():
archive.unlink()
create_archive(stage_root, archive)
checksum = write_checksum(archive)
print(f"Created {archive}")
print(f"Created {checksum}")
return 0, [archive, checksum]
def generic_package_upload_url(context, version, path):
base = str(context.artifact["registry_url"]).rstrip("/")
package = urllib.parse.quote(package_name(context), safe="")
version = urllib.parse.quote(version, safe="")
filename = urllib.parse.quote(path.name, safe="")
return f"{base}/{package}/{version}/{filename}"
def upload_generic_artifact(context, version, path):
credentials = package_registry_credentials()
if credentials is None:
return 1
user, token = credentials
url = generic_package_upload_url(context, version, path)
return run(
context,
[
"curl",
"--fail",
"--silent",
"--show-error",
"--user",
f"{user}:{token}",
"--upload-file",
path,
url,
],
display_command=[
"curl",
"--fail",
"--silent",
"--show-error",
"--user",
f"{user}:<hidden>",
"--upload-file",
path,
url,
],
)
def package_artifacts(context, argv):
if not in_docker(context):
return docker_run(
context,
[
f"./{context.tools_rel}/package.py",
*argv,
],
)
parser = argparse.ArgumentParser(description=f"Create {context.package['name']} application tarballs.")
parser.add_argument(
"--target",
action="append",
default=[],
choices=sorted(context.targets),
help="Target to package. May be passed more than once. Defaults to package.artifact.targets.",
)
parser.add_argument("--all", action="store_true", help="Package all supported targets.")
parser.add_argument("--update", action="store_true", help="Check Conan remotes for newer dependency revisions.")
parser.add_argument("--no-remote", action="store_true", help="Do not query Conan remotes or upload package artifacts.")
parser.add_argument("--version", default=None, help="Artifact version. Defaults to the current git commit.")
args = parser.parse_args(argv)
context.conan_cache_dir.mkdir(parents=True, exist_ok=True)
if should_upload_artifacts(context) and not args.no_remote and package_registry_credentials() is None:
return 1
if not args.no_remote:
status = configure_conan_remote(context)
if status != 0:
return status
version = args.version or git_version(context)
artifacts = []
for target in package_artifact_targets(context, args.target, args.all):
status, created = create_artifact_package(context, target, version, args.update, no_remote=args.no_remote)
if status != 0:
return status
artifacts.extend(created)
if should_upload_artifacts(context) and not args.no_remote:
for artifact in artifacts:
status = upload_generic_artifact(context, version, artifact)
if status != 0:
return status
return 0
def package(context, argv):
if artifact_packaging_enabled(context):
return package_artifacts(context, argv)
if not in_docker(context):
return docker_run(
context,
[
f"./{context.tools_rel}/package.py",
*argv,
],
)
parser = argparse.ArgumentParser(description=f"Create {context.package['name']} Conan packages.")
parser.add_argument(
"--target",
action="append",
default=[],
choices=sorted(context.targets),
help="Target to package. May be passed more than once. Defaults to the default target.",
)
parser.add_argument("--all", action="store_true", help="Package all supported targets.")
parser.add_argument("--update", action="store_true", help="Check Conan remotes for newer dependency revisions.")
parser.add_argument("--no-remote", action="store_true", help="Do not query Conan remotes. Requires all dependencies to be available locally or buildable from local recipes.")
parser.add_argument("--replace", action="store_true", help="Delete the remote package version before upload. Requires MODULUS_ALLOW_PACKAGE_REPLACE=1.")
args = parser.parse_args(argv)
context.conan_cache_dir.mkdir(parents=True, exist_ok=True)
if not args.no_remote:
status = configure_conan_remote(context)
if status != 0:
return status
update = args.update
for target in context.selected_targets(args.target, args.all):
status = create_package(context, target, update, no_remote=args.no_remote)
if status != 0:
return status
if should_upload(context) and not args.no_remote:
return upload_packages(context, replace=args.replace)
return 0