Coverage for app/venv/lib/python3.14/site-packages/weblate/vcs/base.py: 72%
320 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
5"""Version control system abstraction for Weblate needs."""
7from __future__ import annotations
9import hashlib
10import logging
11import os
12import os.path
13import subprocess
14from typing import TYPE_CHECKING, ClassVar, Self, TypedDict
16from dateutil import parser
17from django.core.cache import cache
18from django.utils.functional import cached_property
19from django.utils.translation import gettext_lazy
20from packaging.version import Version
22from weblate.trans.util import get_clean_env, path_separator
23from weblate.utils.data import data_path
24from weblate.utils.errors import add_breadcrumb
25from weblate.utils.lock import WeblateLock
26from weblate.vcs.ssh import SSH_WRAPPER
28if TYPE_CHECKING: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true
29 from collections.abc import Generator, Iterator
30 from datetime import datetime
32 import requests
33 from django_stubs_ext import StrOrPromise
35 from weblate.trans.models import Component
37LOGGER = logging.getLogger("weblate.vcs")
40class SubprocessArgs(TypedDict, total=False):
41 stdin: int
42 input: str
45class RepositoryError(Exception):
46 """Error while working with a repository."""
48 def __init__(self, retcode: int, message: str) -> None:
49 super().__init__(message)
50 self.retcode = retcode
52 def get_message(self):
53 if self.retcode != 0: 53 ↛ 55line 53 didn't jump to line 55 because the condition on line 53 was always true
54 return f"{self.args[0]} ({self.retcode})"
55 return self.args[0]
57 def __str__(self) -> str:
58 return self.get_message()
61class Repository:
62 """Basic repository object."""
64 _cmd: ClassVar[str] = "false"
65 _cmd_last_revision: ClassVar[list[str]]
66 _cmd_last_remote_revision: ClassVar[list[str]]
67 _cmd_status: ClassVar[list[str]] = ["status"]
68 _cmd_list_changed_files: ClassVar[list[str]]
70 name: ClassVar[StrOrPromise] = ""
71 identifier: ClassVar[str] = ""
72 req_version: ClassVar[str | None] = None
73 default_branch: ClassVar[str] = ""
74 needs_push_url: ClassVar[bool] = True
75 supports_push: ClassVar[bool] = True
76 pushes_to_different_location: ClassVar[bool] = False
77 push_label: ClassVar[StrOrPromise] = gettext_lazy(
78 "This will push changes to the upstream repository."
79 )
80 ref_to_remote: ClassVar[str]
81 ref_from_remote: ClassVar[str]
82 _version: ClassVar[str | None] = None
84 @classmethod
85 def get_identifier(cls) -> str:
86 return cls.identifier or cls.name.lower()
88 def __init__(
89 self,
90 path: str,
91 *,
92 branch: str | None = None,
93 component: Component | None = None,
94 local: bool = False,
95 skip_init: bool = False,
96 ) -> None:
97 self.path: str = path
98 if branch is None:
99 self.branch = self.default_branch
100 else:
101 self.branch = branch
102 self.component = component
103 self.last_output = ""
104 base_path = self.path.rstrip("/").rstrip("\\")
105 self.lock = WeblateLock(
106 lock_path=os.path.dirname(base_path),
107 scope="repo",
108 key=component.pk if component else os.path.basename(base_path),
109 slug=os.path.basename(base_path),
110 file_template="{slug}.lock",
111 timeout=120,
112 origin=component.full_slug if component else base_path,
113 )
114 self._config_updated = False
115 self.local = local
116 if not local:
117 # Create ssh wrapper for possible use
118 SSH_WRAPPER.create()
119 if not skip_init and not self.is_valid():
120 with self.lock:
121 self.create_blank_repository(self.path)
123 @classmethod
124 def get_remote_branch(cls, repo: str) -> str: # noqa: ARG003
125 return cls.default_branch
127 @classmethod
128 def add_breadcrumb(cls, message: str, **data) -> None:
129 add_breadcrumb(category="vcs", message=message, **data)
131 @classmethod
132 def add_response_breadcrumb(cls, response: requests.Response) -> None:
133 cls.add_breadcrumb(
134 "http.response",
135 status_code=response.status_code,
136 text=response.text,
137 headers=response.headers,
138 )
140 @classmethod
141 def log(cls, message: str, level: int = logging.DEBUG) -> None:
142 return LOGGER.log(level, "%s: %s", cls._cmd, message)
144 def ensure_config_updated(self) -> None:
145 """Ensure the configuration is periodically checked."""
146 if self._config_updated:
147 return
148 if self.component is None: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true
149 msg = "Component not set!"
150 raise TypeError(msg)
151 cache_key = f"sp-config-check-{self.component.pk}"
152 if cache.get(cache_key) is None:
153 self.check_config()
154 cache.set(cache_key, True, 86400)
155 self._config_updated = True
157 def check_config(self) -> None:
158 """Check VCS configuration."""
159 raise NotImplementedError
161 def is_valid(self) -> bool:
162 """Check whether this is a valid repository."""
163 raise NotImplementedError
165 @classmethod
166 def create_blank_repository(cls, path: str) -> None:
167 """Initialize the repository."""
168 raise NotImplementedError
170 def resolve_symlinks(self, path: str) -> str:
171 """Resolve any symlinks in the path."""
172 # Resolve symlinks first
173 real_path = path_separator(os.path.realpath(os.path.join(self.path, path)))
174 repository_path = path_separator(os.path.realpath(self.path))
176 if not real_path.startswith(repository_path): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 msg = "Too many symlinks or link outside tree"
178 raise ValueError(msg)
180 return real_path[len(repository_path) :].lstrip("/")
182 @staticmethod
183 def _getenv(environment: dict[str, str] | None = None) -> dict[str, str]:
184 """Generate environment for process execution."""
185 base: dict[str, str] = {
186 # Avoid prompts from Git
187 "GIT_TERMINAL_PROMPT": "0",
188 # Avoid Git traversing outside the data dir
189 "GIT_CEILING_DIRECTORIES": data_path("vcs").as_posix(),
190 # Use ssh wrapper
191 "GIT_SSH": SSH_WRAPPER.filename.as_posix(),
192 "SVN_SSH": SSH_WRAPPER.filename.as_posix(),
193 }
194 if environment: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 base.update(environment)
196 return get_clean_env(base, extra_path=SSH_WRAPPER.path.as_posix())
198 @classmethod
199 def _popen(
200 cls,
201 args: list[str],
202 *,
203 cwd: str | None = None,
204 merge_err: bool = True,
205 fullcmd: bool = False,
206 raw: bool = False,
207 local: bool = False,
208 stdin: str | None = None,
209 environment: dict[str, str] | None = None,
210 ):
211 """Execute the command using popen."""
212 if args is None: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 raise RepositoryError(0, "Not supported functionality")
214 if not fullcmd:
215 args = [cls._cmd, *list(args)]
216 text_cmd = " ".join(args)
217 try:
218 # These are mutually exclusive, gevent actually checks
219 # for their presence, not a avalue
220 kwargs: SubprocessArgs = {}
221 if stdin is None:
222 kwargs["stdin"] = subprocess.PIPE
223 else:
224 kwargs["input"] = stdin
226 process = subprocess.run(
227 args=args,
228 cwd=cwd,
229 env=environment or {} if local else cls._getenv(environment),
230 stdout=subprocess.PIPE,
231 stderr=subprocess.STDOUT if merge_err else subprocess.PIPE,
232 text=not raw,
233 check=False,
234 # Excessively long timeout to catch misbehaving processes
235 timeout=3600,
236 **kwargs,
237 )
238 except subprocess.TimeoutExpired as error:
239 stdout = (
240 error.stdout.decode()
241 if isinstance(error.stdout, bytes)
242 else error.stdout
243 )
244 stderr = (
245 error.stderr.decode()
246 if isinstance(error.stderr, bytes)
247 else error.stderr
248 )
249 raise RepositoryError(
250 0,
251 f"Subprocess didn't complete before {error.timeout} seconds\n{stdout}{stderr or ''}",
252 ) from error
253 cls.add_breadcrumb(
254 text_cmd,
255 retcode=process.returncode,
256 output=process.stdout,
257 stderr=process.stderr,
258 cwd=cwd,
259 )
260 if process.returncode:
261 raise RepositoryError(
262 process.returncode, process.stdout + (process.stderr or "")
263 )
264 return process.stdout
266 def execute(
267 self,
268 args: list[str],
269 *,
270 needs_lock: bool = True,
271 fullcmd: bool = False,
272 merge_err: bool = True,
273 stdin: str | None = None,
274 environment: dict[str, str] | None = None,
275 ):
276 """Execute command and caches its output."""
277 if needs_lock:
278 if not self.lock.is_locked: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 msg = "Repository operation without lock held!"
280 raise RuntimeError(msg)
281 if self.component:
282 self.ensure_config_updated()
283 is_status = args[0] == self._cmd_status[0]
284 try:
285 self.last_output = self._popen(
286 args,
287 cwd=self.path,
288 fullcmd=fullcmd,
289 local=self.local,
290 merge_err=merge_err,
291 stdin=stdin,
292 environment=environment,
293 )
294 except RepositoryError as error:
295 if not is_status and not self.local:
296 self.log_status(error)
297 raise
298 return self.last_output
300 def log_status(self, error: str | RepositoryError) -> None:
301 try:
302 self.log(f"failure {error}")
303 self.log(self.status())
304 except RepositoryError:
305 pass
307 def clean_revision_cache(self) -> None:
308 if "last_revision" in self.__dict__:
309 del self.__dict__["last_revision"]
310 if "last_remote_revision" in self.__dict__: 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 del self.__dict__["last_remote_revision"]
313 @cached_property
314 def last_revision(self):
315 """Return last local revision."""
316 return self.get_last_revision()
318 def get_last_revision(self):
319 return self.execute(self._cmd_last_revision, needs_lock=False, merge_err=False)
321 @cached_property
322 def last_remote_revision(self):
323 """Return last remote revision."""
324 return self.execute(
325 self._cmd_last_remote_revision, needs_lock=False, merge_err=False
326 )
328 @classmethod
329 def _clone(cls, source: str, target: str, branch: str) -> None:
330 """Clone repository."""
331 raise NotImplementedError
333 @classmethod
334 def clone(
335 cls, source: str, target: str, branch: str, component: Component | None = None
336 ) -> Self:
337 """Clone repository and return object for cloned repository."""
338 repo = cls(target, branch=branch, component=component, skip_init=True)
339 with repo.lock:
340 cls._clone(source, target, branch)
341 return repo
343 def update_remote(self) -> None:
344 """Update remote repository."""
345 raise NotImplementedError
347 def status(self) -> str:
348 """Return status of the repository."""
349 return self.execute(self._cmd_status, needs_lock=False)
351 def push(self, branch: str) -> None:
352 """Push given branch to remote repository."""
353 raise NotImplementedError
355 def unshallow(self) -> None:
356 """Unshallow working copy."""
357 return
359 def reset(self) -> None:
360 """Reset working copy to match remote branch."""
361 raise NotImplementedError
363 def merge(
364 self, abort: bool = False, message: str | None = None, no_ff: bool = False
365 ) -> None:
366 """Merge remote branch or reverts the merge."""
367 raise NotImplementedError
369 def rebase(self, abort: bool = False) -> None:
370 """Rebase working copy on top of remote branch."""
371 raise NotImplementedError
373 def needs_commit(self, filenames: list[str] | None = None) -> bool:
374 """Check whether repository needs commit."""
375 raise NotImplementedError
377 def count_missing(self):
378 """Count missing commits."""
379 return len(
380 self.log_revisions(self.ref_to_remote.format(self.get_remote_branch_name()))
381 )
383 def count_outgoing(self, branch: str | None = None):
384 """Count outgoing commits."""
385 return len(
386 self.log_revisions(
387 self.ref_from_remote.format(self.get_remote_branch_name(branch))
388 )
389 )
391 def needs_merge(self):
392 """
393 Check whether repository needs merge with upstream.
395 It is missing some revisions.
396 """
397 return self.count_missing() > 0
399 def needs_push(self):
400 """
401 Check whether repository needs push to upstream.
403 It has additional revisions.
404 """
405 return self.count_outgoing() > 0
407 def _get_revision_info(self, revision: str) -> dict[str, str]:
408 """Return dictionary with detailed revision information."""
409 raise NotImplementedError
411 def get_revision_info(self, revision: str) -> dict[str, str]:
412 """Return dictionary with detailed revision information."""
413 key = f"rev-info-{self.get_identifier()}-{revision}"
414 result = cache.get(key)
415 if not result:
416 result = self._get_revision_info(revision)
417 # Keep the cache for one day
418 cache.set(key, result, 86400)
420 # Parse timestamps into datetime objects
421 for name, value in result.items():
422 if "date" in name:
423 result[name] = parser.parse(value)
425 return result
427 @classmethod
428 def is_configured(cls) -> bool:
429 return True
431 @classmethod
432 def validate_configuration(cls) -> list[str]:
433 return []
435 @classmethod
436 def is_supported(cls):
437 """Check whether this VCS backend is supported."""
438 try:
439 version = cls.get_version()
440 except Exception:
441 return False
442 return cls.req_version is None or Version(version) >= Version(cls.req_version)
444 @classmethod
445 def get_version(cls):
446 """Get cached backend version."""
447 if cls._version is None:
448 try:
449 cls._version = cls._get_version()
450 except Exception as error:
451 cls._version = error
452 if isinstance(cls._version, Exception): 452 ↛ 453line 452 didn't jump to line 453 because the condition on line 452 was never true
453 raise cls._version
454 return cls._version
456 @classmethod
457 def _get_version(cls):
458 """Return VCS program version."""
459 return cls._popen(["--version"], merge_err=False)
461 def set_committer(self, name: str, mail: str) -> None:
462 """Configure committer name."""
463 raise NotImplementedError
465 def commit(
466 self,
467 message: str,
468 author: str | None = None,
469 timestamp: datetime | None = None,
470 files: list[str] | None = None,
471 ) -> bool:
472 """Create new revision."""
473 raise NotImplementedError
475 def remove(self, files: list[str], message: str, author: str | None = None) -> None:
476 """Remove files and creates new revision."""
477 raise NotImplementedError
479 @staticmethod
480 def update_hash(
481 objhash: hashlib._Hash, filename: str, extra: str | None = None
482 ) -> None:
483 if os.path.islink(filename): 483 ↛ 484line 483 didn't jump to line 484 because the condition on line 483 was never true
484 objtype = "symlink"
485 data = os.readlink(filename).encode()
486 else:
487 objtype = "blob"
488 with open(filename, "rb") as handle:
489 data = handle.read()
490 if extra: 490 ↛ 491line 490 didn't jump to line 491 because the condition on line 490 was never true
491 objhash.update(extra.encode())
492 objhash.update(f"{objtype} {len(data)}\0".encode("ascii"))
493 objhash.update(data)
495 def get_object_hash(self, path: str) -> str:
496 """
497 Return hash of object in the VCS.
499 For files in a way compatible with Git (equivalent to git ls-tree HEAD), for
500 dirs it behaves differently as we do not need to track some attributes (for
501 example permissions).
502 """
503 real_path = os.path.join(self.path, self.resolve_symlinks(path))
504 objhash = hashlib.sha1(usedforsecurity=False)
506 if os.path.isdir(real_path): 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true
507 files = []
508 for root, _unused, filenames in os.walk(real_path):
509 for filename in filenames:
510 full_name = os.path.join(root, filename)
511 files.append((full_name, os.path.relpath(full_name, self.path)))
512 for filename, name in sorted(files):
513 self.update_hash(objhash, filename, name)
514 else:
515 self.update_hash(objhash, real_path)
517 return objhash.hexdigest()
519 def configure_remote(
520 self, pull_url: str, push_url: str, branch: str, fast: bool = True
521 ) -> None:
522 """Configure remote repository."""
523 raise NotImplementedError
525 def configure_branch(self, branch: str) -> None:
526 """Configure repository branch."""
527 raise NotImplementedError
529 def describe(self) -> str:
530 """Verbosely describes current revision."""
531 raise NotImplementedError
533 def get_file(self, path: str, revision: str) -> str:
534 """Return content of file at given revision."""
535 raise NotImplementedError
537 @staticmethod
538 def get_examples_paths() -> Generator[str]:
539 """
540 List possible paths for shipped examples.
542 Used to locate merge drivers which are shipped there.
543 """
544 yield os.path.join(os.path.dirname(os.path.dirname(__file__)), "examples")
546 @classmethod
547 def find_merge_driver(cls, name: str) -> str | None:
548 for path in cls.get_examples_paths(): 548 ↛ 552line 548 didn't jump to line 552 because the loop on line 548 didn't complete
549 result = os.path.join(path, name)
550 if os.path.exists(result): 550 ↛ 548line 550 didn't jump to line 548 because the condition on line 550 was always true
551 return os.path.abspath(result)
552 return None
554 @classmethod
555 def get_merge_driver(cls, file_format: str) -> str | None:
556 merge_driver = None
557 if file_format == "po": 557 ↛ 559line 557 didn't jump to line 559 because the condition on line 557 was always true
558 merge_driver = cls.find_merge_driver("git-merge-gettext-po")
559 if merge_driver is None or not os.path.exists(merge_driver): 559 ↛ 560line 559 didn't jump to line 560 because the condition on line 559 was never true
560 return None
561 return merge_driver
563 def cleanup(self) -> None:
564 """Remove not tracked files from the repository."""
565 raise NotImplementedError
567 def log_revisions(self, refspec: str) -> list[str]:
568 """
569 Log revisions for given refspec.
571 This is not universal as refspec is different per vcs.
572 """
573 raise NotImplementedError
575 def list_changed_files(self, refspec: str) -> list:
576 """
577 List changed files for given refspec.
579 This is not universal as refspec is different per vcs.
580 """
581 lines = self.execute(
582 [*self._cmd_list_changed_files, refspec], needs_lock=False, merge_err=False
583 ).splitlines()
584 return list(self.parse_changed_files(lines))
586 def parse_changed_files(self, lines: list[str]) -> Iterator[str]:
587 """Parse output with changed files."""
588 raise NotImplementedError
590 def get_changed_files(self, compare_to: str | None = None):
591 """Get files missing upstream or changes between revisions."""
592 if compare_to is None:
593 compare_to = self.get_remote_branch_name()
595 return self.list_changed_files(self.ref_to_remote.format(compare_to))
597 def get_remote_branch_name(self, branch: str | None = None) -> str:
598 return f"origin/{self.branch if branch is None else branch}"
600 def list_remote_branches(self):
601 return []
603 def compact(self) -> None:
604 return
606 def show(self, revision: str) -> str:
607 raise NotImplementedError