Coverage for src / competitive_verifier / oj / languages / rust.py: 72%
217 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-03 23:40 +0900
1import abc
2import enum
3import functools
4import itertools
5import json
6import pathlib
7import shutil
8from collections import defaultdict
9from collections.abc import Sequence
10from enum import Enum
11from logging import getLogger
12from typing import Any, Literal
14from pydantic import BaseModel, Field
16from competitive_verifier.exec import command_stdout
17from competitive_verifier.models import ShellCommand
18from competitive_verifier.util import read_text_normalized
20from .base import Language, LanguageEnvironment, OjVerifyLanguageConfig
22# ruff: noqa: PLR2004
24logger = getLogger(__name__)
26_metadata_by_manifest_path: dict[pathlib.Path, dict[str, Any]] = {}
27_cargo_checked_workspaces: set[pathlib.Path] = set()
28_related_source_files_by_workspace: dict[
29 pathlib.Path, dict[pathlib.Path, frozenset[pathlib.Path]]
30] = {}
33class OjVerifyRustListDependenciesBackend(BaseModel):
34 kind: Literal["none", "cargo-udeps"]
35 toolchain: str | None = None
38class OjVerifyRustConfig(OjVerifyLanguageConfig):
39 list_dependencies_backend: OjVerifyRustListDependenciesBackend | None = None
42class _ListDependenciesBackend(abc.ABC, BaseModel):
43 @abc.abstractmethod
44 def list_dependencies(
45 self, path: pathlib.Path, *, basedir: pathlib.Path
46 ) -> list[pathlib.Path]: ...
49class _NoBackend(_ListDependenciesBackend):
50 def list_dependencies(
51 self, path: pathlib.Path, *, basedir: pathlib.Path
52 ) -> list[pathlib.Path]:
53 return _list_dependencies_by_crate(
54 path, basedir=basedir, cargo_udeps_toolchain=None
55 )
58class _CargoUdeps(_ListDependenciesBackend):
59 toolchain: str = "nightly"
61 def list_dependencies(
62 self, path: pathlib.Path, *, basedir: pathlib.Path
63 ) -> list[pathlib.Path]:
64 return _list_dependencies_by_crate(
65 path, basedir=basedir, cargo_udeps_toolchain=self.toolchain
66 )
69@functools.cache
70def _list_dependencies_by_crate(
71 path: pathlib.Path, *, basedir: pathlib.Path, cargo_udeps_toolchain: str | None
72) -> list[pathlib.Path]:
73 """The `list_dependencies` implementation for `_NoBackend` and `CargoUdeps`.
75 Args:
76 path (pathlib.Path): A main source file path of a target
77 basedir (pathlib.Path): A parameter from `Language.list_dependencies`
78 cargo_udeps_toolchain (str | None): If not `None`, use `cargo-udeps` with the specified toolchain to detect unused dependencies
79 Returns:
80 list[pathlib.Path]: A list of dependent `.rs` file paths
81 Raises:
82 RuntimeError: If any cargo command fails
83 """
84 basedir = basedir.resolve()
85 path = basedir / path
87 # We regard that a generated file does not depend on any files.
88 for parent in path.parents:
89 if (parent.parent / "Cargo.toml").exists() and parent.parts[-1] == "target": 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 logger.warning("This is a generated file!: %s", path)
91 return [path]
93 metadata = _cargo_metadata(cwd=path.parent)
95 # First, collects source files in the same crate.
96 common_result = set(
97 _source_files_in_same_targets(path, _related_source_files(basedir, metadata))
98 )
100 main_package_and_target = _find_target(metadata, path)
101 if not main_package_and_target: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 return sorted(common_result)
103 main_package, main_target = main_package_and_target
105 packages_by_id = {p["id"]: p for p in metadata["packages"]}
107 class DependencyNamespace(Enum):
108 NORMAL_DEVELOPMENT = enum.auto()
109 BUILD = enum.auto()
111 @classmethod
112 def from_dep_kind(cls, kind: str):
113 if kind == "build":
114 return cls.BUILD
115 return cls.NORMAL_DEVELOPMENT
117 # Collect the `(|dev-|build-)dependencies` into a <is a `build-dependency`> → (<"extern crate name"> → <package>) dictionary.
118 dependencies: defaultdict[DependencyNamespace, dict[str, dict[str, Any]]] = (
119 defaultdict(dict)
120 )
121 for dep in next(
122 n["deps"] for n in metadata["resolve"]["nodes"] if n["id"] == main_package["id"]
123 ):
124 if _need_dev_deps(main_target) or any( 124 ↛ 130line 124 didn't jump to line 130 because the condition on line 124 was always true
125 k["kind"] is None for k in dep["dep_kinds"]
126 ):
127 dependencies[DependencyNamespace.NORMAL_DEVELOPMENT][dep["name"]] = (
128 packages_by_id[dep["pkg"]]
129 )
130 if any(k["kind"] == "build" for k in dep["dep_kinds"]): 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 dependencies[DependencyNamespace.BUILD][dep["name"]] = packages_by_id[
132 dep["pkg"]
133 ]
135 # If `cargo_udeps_toolchain` is present, collects packages that are "unused" by `target`.
136 unused_packages: defaultdict[DependencyNamespace, set[Any]] = defaultdict(set)
137 if cargo_udeps_toolchain is not None: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 explicit_names_in_toml = {
139 (DependencyNamespace.from_dep_kind(d["kind"]), d["rename"])
140 for d in main_package["dependencies"]
141 if d["rename"]
142 }
143 if not shutil.which("cargo-udeps"):
144 raise RuntimeError("`cargo-udeps` not in $PATH")
145 args: list[str] = [
146 "rustup",
147 "run",
148 cargo_udeps_toolchain,
149 "cargo",
150 "udeps",
151 "--output",
152 "json",
153 "--manifest-path",
154 main_package["manifest_path"],
155 *_target_option(main_target),
156 ]
157 unused_deps = json.loads(
158 command_stdout(args, cwd=metadata["workspace_root"], check=False)
159 )["unused_deps"].values()
160 unused_dep = next(
161 (
162 u
163 for u in unused_deps
164 if u["manifest_path"] == main_package["manifest_path"]
165 ),
166 None,
167 )
168 if unused_dep:
169 names_in_toml: list[tuple[DependencyNamespace, Any]] = [
170 (DependencyNamespace.NORMAL_DEVELOPMENT, name_in_toml)
171 for name_in_toml in [*unused_dep["normal"], *unused_dep["development"]]
172 ]
173 names_in_toml.extend(
174 (DependencyNamespace.BUILD, name_in_toml)
175 for name_in_toml in unused_dep["build"]
176 )
177 for dependency_namespace, name_in_toml in names_in_toml:
178 if (dependency_namespace, name_in_toml) in explicit_names_in_toml:
179 # If the `name_in_toml` is explicitly renamed one, it equals to the `extern_crate_name`.
180 unused_package: Any = dependencies[dependency_namespace][
181 name_in_toml
182 ]["id"]
183 else:
184 # Otherwise, it equals to the `package.name`.
185 unused_package = next(
186 p["id"]
187 for p in dependencies[dependency_namespace].values()
188 if p["name"] == name_in_toml
189 )
190 unused_packages[dependency_namespace].add(unused_package)
192 # Finally, adds source files related to the depended crates except:
193 #
194 # - those detected by cargo-udeps
195 # - those come from Crates.io or Git repositories (e.g. `proconio`, other people's libraries including `ac-library-rs`)
197 # `main_package` should always be included.
198 # Note that cargo-udeps does not detect it if it is unused.
199 # https://github.com/est31/cargo-udeps/pull/35
200 depended_packages = [main_package]
201 for dependency_namespace, values in dependencies.items():
202 depended_packages.extend(
203 depended_package
204 for depended_package in values.values()
205 if (
206 depended_package["id"] not in unused_packages[dependency_namespace]
207 and not depended_package["source"]
208 )
209 )
211 ret = common_result
213 for depended_package in depended_packages:
214 depended_targets = [
215 t
216 for t in depended_package["targets"]
217 if t != main_target and (_is_build(t) or _is_lib_or_proc_macro(t))
218 ]
219 assert len(depended_targets) <= 2
220 for depended_target in depended_targets:
221 related_source_files = _related_source_files(
222 basedir,
223 _cargo_metadata_by_manifest_path(
224 pathlib.Path(depended_package["manifest_path"])
225 ),
226 )
227 ret |= _source_files_in_same_targets(
228 pathlib.Path(depended_target["src_path"]).resolve(strict=True),
229 related_source_files,
230 )
231 return sorted(ret)
234def _related_source_files(
235 basedir: pathlib.Path, metadata: dict[str, Any]
236) -> dict[pathlib.Path, frozenset[pathlib.Path]]:
237 """Collects all of the `.rs` files recognized by a workspace.
239 Args:
240 basedir (pathlib.Path): A parameter from `Language.list_dependencies`
241 metadata (dict[str, Any]): "metadata" for a Cargo.toml file in the workspace
242 Returns:
243 dict[pathlib.Path, frozenset[pathlib.Path]]: A (main source file) → (other related files) map
244 Raises:
245 RuntimeError: If any cargo command fails
246 """
247 if pathlib.Path(metadata["workspace_root"]) in _related_source_files_by_workspace:
248 return _related_source_files_by_workspace[
249 pathlib.Path(metadata["workspace_root"])
250 ]
252 # Runs `cargo check` to generate `$target_directory/debug/deps/*.d`.
253 if pathlib.Path(metadata["workspace_root"]) not in _cargo_checked_workspaces: 253 ↛ 267line 253 didn't jump to line 267 because the condition on line 253 was always true
254 command_stdout(
255 [
256 "cargo",
257 "check",
258 "--manifest-path",
259 str(pathlib.Path(metadata["workspace_root"], "Cargo.toml")),
260 "--workspace",
261 "--all-targets",
262 ],
263 cwd=metadata["workspace_root"],
264 )
265 _cargo_checked_workspaces.add(pathlib.Path(metadata["workspace_root"]))
267 ret: dict[pathlib.Path, frozenset[pathlib.Path]] = {}
269 targets_in_workspace = itertools.chain.from_iterable(
270 p["targets"]
271 for p in metadata["packages"]
272 if p["id"] in metadata["workspace_members"]
273 )
274 for target in targets_in_workspace:
275 # Finds the **latest** "dep-info" file that contains a line in the following format, and parses the line.
276 #
277 # ```
278 # <relative/absolute path to the `.d` file itself>: <relative/absolute path to the root source file> <relative/aboslute paths to the other related files>...
279 # ```
280 #
281 # - https://github.com/rust-lang/cargo/blob/rust-1.49.0/src/cargo/core/compiler/fingerprint.rs#L1979-L1997
282 # - https://github.com/rust-lang/cargo/blob/rust-1.49.0/src/cargo/core/compiler/fingerprint.rs#L1824-L1830
283 if _is_build(target): 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 dep_info_paths = pathlib.Path(
285 metadata["target_directory"], "debug", "build"
286 ).rglob(f"{_crate_name(target)}-*.d")
287 elif _is_example(target): 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 dep_info_paths = pathlib.Path(
289 metadata["target_directory"], "debug", "examples"
290 ).glob(f"{_crate_name(target)}-*.d")
291 else:
292 dep_info_paths = pathlib.Path(
293 metadata["target_directory"], "debug", "deps"
294 ).glob(f"{_crate_name(target)}-*.d")
295 for dep_info_path in sorted( 295 ↛ 326line 295 didn't jump to line 326 because the loop on line 295 didn't complete
296 dep_info_paths, key=lambda p: p.stat().st_mtime_ns, reverse=True
297 ):
298 dep_info = read_text_normalized(dep_info_path)
299 for line in dep_info.splitlines(): 299 ↛ 323line 299 didn't jump to line 323 because the loop on line 299 didn't complete
300 ss = line.split(": ")
301 if (
302 len(ss) == 2
303 and pathlib.Path(metadata["workspace_root"], ss[0]) == dep_info_path
304 ):
305 paths: list[pathlib.Path] = []
306 it = iter(ss[1].split())
307 for s in it:
308 ss = s
309 while ss.endswith("\\"): 309 ↛ 310line 309 didn't jump to line 310 because the condition on line 309 was never true
310 ss = ss.rstrip("\\") + " " + next(it)
311 path = pathlib.Path(metadata["workspace_root"], s).resolve(
312 strict=True
313 )
314 # Ignores paths that don't start with the `basedir`. (e.g. `/dev/null`, `/usr/local/share/foo/bar`)
315 if path.is_relative_to(basedir): 315 ↛ 307line 315 didn't jump to line 307 because the condition on line 315 was always true
316 paths.append(path)
317 if paths[:1] == [ 317 ↛ 299line 317 didn't jump to line 299 because the condition on line 317 was always true
318 pathlib.Path(target["src_path"]).resolve(strict=True)
319 ]:
320 ret[paths[0]] = frozenset(paths[1:])
321 break
322 else:
323 continue
324 break
325 else:
326 logger.error("no `.d` file for `%s`", target["name"])
328 _related_source_files_by_workspace[pathlib.Path(metadata["workspace_root"])] = ret
329 return ret
332def _source_files_in_same_targets(
333 path: pathlib.Path,
334 related_source_files: dict[pathlib.Path, frozenset[pathlib.Path]],
335) -> frozenset[pathlib.Path]:
336 """Returns `.rs` file paths relating to `path`.
338 Args:
339 path (pathlib.Path): A main source file path of a target
340 related_source_files (dict[pathlib.Path, frozenset[pathlib.Path]]): A (main source file) → (other related files) map
341 Returns:
342 frozenset[pathlib.Path]: A set of `.rs` file paths relating to `path`
343 Raises:
344 RuntimeError: If `path` is not found in `related_source_files` and is not related to any other files.
345 """
346 # If `p` is `src_path` of a target, it does not belong to any other target unless it's weirdly symlinked,
347 if path in related_source_files: 347 ↛ 351line 347 didn't jump to line 351 because the condition on line 347 was always true
348 return frozenset({path, *related_source_files[path]})
350 # Otherwise, it may be used by multiple targets with `#[path = ".."] mod foo;` or something.
351 return frozenset(
352 itertools.chain.from_iterable(
353 {k, *v} for (k, v) in related_source_files.items() if path in v
354 )
355 ) or frozenset({path})
358class RustLanguageEnvironment(LanguageEnvironment):
359 @property
360 def name(self) -> str:
361 return "Rust"
363 def get_compile_command(
364 self, path: pathlib.Path, *, basedir: pathlib.Path, tempdir: pathlib.Path
365 ) -> ShellCommand:
366 abs_path = basedir.resolve() / path
367 metadata = _cargo_metadata(cwd=abs_path.parent)
368 target = _ensure_target(metadata, abs_path)
369 return ShellCommand(
370 command=["cargo", "build", "--release", *_target_option(target)],
371 cwd=(basedir / path).parent,
372 )
374 def get_execute_command(
375 self, path: pathlib.Path, *, basedir: pathlib.Path, tempdir: pathlib.Path
376 ) -> str:
377 abs_basedir = basedir.resolve()
378 abs_path = abs_basedir / path
379 metadata = _cargo_metadata(cwd=abs_path.parent)
380 target = _ensure_target(metadata, abs_path)
381 target_directory = pathlib.Path(metadata["target_directory"])
382 if target_directory.is_relative_to(abs_basedir): 382 ↛ 384line 382 didn't jump to line 384 because the condition on line 382 was always true
383 target_directory = basedir / target_directory.relative_to(abs_basedir)
384 return str(
385 pathlib.Path(
386 target_directory,
387 "release",
388 *([] if _is_bin(target) else ["examples"]),
389 target["name"],
390 )
391 )
394class RustLanguage(Language):
395 config: OjVerifyRustConfig = Field(default_factory=OjVerifyRustConfig)
397 @functools.cached_property
398 def _list_dependencies_backend(self) -> _ListDependenciesBackend:
399 list_dependencies_backend = self.config.list_dependencies_backend
400 if ( 400 ↛ 405line 400 didn't jump to line 405 because the condition on line 400 was always true
401 list_dependencies_backend is None
402 or list_dependencies_backend.kind == "none"
403 ):
404 return _NoBackend()
405 if list_dependencies_backend.kind == "cargo-udeps":
406 if toolchain := list_dependencies_backend.toolchain:
407 return _CargoUdeps(toolchain=toolchain)
408 return _CargoUdeps()
409 raise RuntimeError(
410 "expected 'none' or 'cargo-udeps' for `languages.rust.list_dependencies_backend.kind`"
411 )
413 def list_dependencies(
414 self, path: pathlib.Path, *, basedir: pathlib.Path
415 ) -> list[pathlib.Path]:
416 return self._list_dependencies_backend.list_dependencies(path, basedir=basedir)
418 def list_environments(
419 self, path: pathlib.Path, *, basedir: pathlib.Path
420 ) -> Sequence[RustLanguageEnvironment]:
421 return [RustLanguageEnvironment()]
424def _cargo_metadata(cwd: pathlib.Path) -> dict[str, Any]:
425 """Returns "metadata" for a Cargo.toml file in `cwd` or its parent directories.
427 Args:
428 cwd (pathlib.Path): The current working directory
429 Returns:
430 dict[str, Any]: Output of `cargo metadata` command
431 Raises:
432 ValueError: If `cwd` is not absolute or contains `..`
433 RuntimeError: If no `Cargo.toml` is found
434 """
435 if not cwd.is_absolute() or ".." in cwd.parts: 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true
436 raise ValueError(
437 f"the `cwd` parameter must be absolute and must not contain `..`: {cwd}"
438 )
440 # https://docs.rs/cargo/0.49.0/src/cargo/util/important_paths.rs.html#6-20
441 for directory in [cwd, *cwd.parents]: 441 ↛ 445line 441 didn't jump to line 445 because the loop on line 441 didn't complete
442 manifest_path = directory / "Cargo.toml"
443 if manifest_path.exists():
444 return _cargo_metadata_by_manifest_path(manifest_path)
445 raise RuntimeError(
446 f"could not find `Cargo.toml` in `{cwd}` or any parent directory"
447 )
450def _cargo_metadata_by_manifest_path(manifest_path: pathlib.Path) -> dict[str, Any]:
451 """Returns "metadata" for a certain `Cargo.toml`.
453 Args:
454 manifest_path (pathlib.Path): Path to a `Cargo.toml`
455 Returns:
456 dict[str, Any]: Output of `cargo metadata` command
457 Raises:
458 RuntimeError: If the `cargo metadata` command fails
459 """
460 if manifest_path in _metadata_by_manifest_path:
461 return _metadata_by_manifest_path[manifest_path]
463 metadata = _run_cargo_metadata(manifest_path)
464 root_manifest_path = pathlib.Path(metadata["workspace_root"], "Cargo.toml")
465 if root_manifest_path != manifest_path: 465 ↛ 468line 465 didn't jump to line 468 because the condition on line 465 was always true
466 metadata = _run_cargo_metadata(root_manifest_path)
468 for key in [
469 root_manifest_path,
470 *(
471 pathlib.Path(p["manifest_path"])
472 for p in metadata["packages"]
473 if p["id"] in metadata["workspace_members"]
474 ),
475 ]:
476 _metadata_by_manifest_path[key] = metadata
478 return metadata
481def _run_cargo_metadata(manifest_path: pathlib.Path) -> dict[str, Any]:
482 """Runs `cargo metadata` for a certain `Cargo.toml`.
484 This function is considered to be executed just once for every Cargo.toml in the repository.
485 For detailed information about `cargo metadata`, see:
487 - <https://doc.rust-lang.org/cargo/commands/cargo-metadata.html#output-format>
488 - <https://docs.rs/cargo_metadata>
490 Args:
491 manifest_path (pathlib.Path): Path to a `Cargo.toml`
492 Returns:
493 dict[str, Any]: Output of `cargo metadata` command
494 Raises:
495 RuntimeError: If the `cargo metadata` command fails
496 """
497 return json.loads(
498 command_stdout(
499 [
500 "cargo",
501 "metadata",
502 "--format-version",
503 "1",
504 "--manifest-path",
505 str(manifest_path),
506 ],
507 cwd=manifest_path.parent,
508 )
509 )
512def _find_target(
513 metadata: dict[str, Any],
514 src_path: pathlib.Path,
515) -> tuple[dict[str, Any], dict[str, Any]] | None:
516 for package in metadata["packages"]: 516 ↛ 522line 516 didn't jump to line 522 because the loop on line 516 didn't complete
517 for target in package["targets"]:
518 # A `src_path` may contain `..`
519 # The path may not actually exist by being excluded from the package.
520 if pathlib.Path(target["src_path"]).resolve() == src_path:
521 return package, target
522 return None
525def _ensure_target(metadata: dict[str, Any], src_path: pathlib.Path) -> dict[str, Any]:
526 package_and_target = _find_target(metadata, src_path)
527 if not package_and_target: 527 ↛ 528line 527 didn't jump to line 528 because the condition on line 527 was never true
528 raise RuntimeError(f"{src_path} is not a main source file of any target")
529 _, target = package_and_target
530 return target
533def _crate_name(target: dict[str, Any]) -> bool:
534 return target["name"].replace("-", "_")
537def _is_build(target: dict[str, Any]) -> bool:
538 return target["kind"] == ["custom-build"]
541def _is_lib_or_proc_macro(target: dict[str, Any]) -> bool:
542 return target["kind"] in [["lib"], ["proc-macro"]]
545def _is_bin(target: dict[str, Any]) -> bool:
546 return target["kind"] == ["bin"]
549def _is_example(target: dict[str, Any]) -> bool:
550 return target["kind"] == ["example"]
553def _need_dev_deps(target: dict[str, Any]) -> bool:
554 # Comes from https://docs.rs/cargo/0.49.0/cargo/ops/enum.CompileFilter.html#method.need_dev_deps
555 return not (_is_lib_or_proc_macro(target) or _is_bin(target))
558def _target_option(target: dict[str, Any]) -> list[str]:
559 if target["kind"] == ["bin"]: 559 ↛ 561line 559 didn't jump to line 561 because the condition on line 559 was always true
560 return ["--bin", target["name"]]
561 if target["kind"] == ["example"]:
562 return ["--example", target["name"]]
563 if target["kind"] == ["test"]:
564 return ["--test", target["name"]]
565 if target["kind"] == ["bench"]:
566 return ["--bench", target["name"]]
567 return ["--lib"]