Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / _build / build_utils.py: 0%

27 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-11-15 05:03 +0800

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Small standard-library filesystem and timing primitives shared by build modules.""" 

16 

17from contextlib import contextmanager 

18import hashlib 

19import json 

20from pathlib import Path 

21import re 

22import time 

23from collections.abc import Iterator 

24 

25 

26def version_label(version: str) -> str: 

27 """Encode a framework label for filenames without imposing version syntax.""" 

28 label = version.lower().replace("+", ".").replace("!", ".") 

29 if len(label) <= 80 and re.fullmatch(r"[a-z0-9]+(?:[._-][a-z0-9]+)*", label): 

30 return re.sub(r"[._-]+", ".", label) 

31 readable = re.sub(r"[^a-z0-9]+", ".", label).strip(".")[:64].rstrip(".") or "unknown" 

32 digest = hashlib.sha256(version.encode()).hexdigest()[:12] 

33 return f"{readable}.{digest}" 

34 

35 

36def atomic_json(path: Path, value: object) -> None: 

37 """Replace complete JSON and remove private partial writes on failure or cancellation.""" 

38 path.parent.mkdir(parents=True, exist_ok=True) 

39 temporary = path.with_name(path.name + ".tmp") 

40 try: 

41 temporary.write_text(json.dumps(value, indent=2) + "\n") 

42 temporary.replace(path) 

43 finally: 

44 temporary.unlink(missing_ok=True) 

45 

46 

47@contextmanager 

48def measured(events: list, stage: str) -> Iterator[None]: 

49 """Record one controller phase, including a failed or interrupted phase.""" 

50 started = time.monotonic() 

51 try: 

52 yield 

53 finally: 

54 events.append({"stage": stage, "elapsed_seconds": time.monotonic() - started})