Coverage for src/backoffice/check_compatibility.py: 0%

1"""check compatibility of tools against resources in the index.json""" 

2 

3import json 

4import shutil 

5import traceback 

6import warnings 

7from pathlib import Path 

8from typing import TYPE_CHECKING, Callable 

9 

10from loguru import logger 

11 

12from .utils_pure import get_tool_report_path 

13 

14try: 

15 from tqdm import tqdm 

16except ImportError: 

17 tqdm = list 

18 

19 

20if TYPE_CHECKING: 

21 from .compatibility import ToolCompatibilityReport 

22 from .compatibility_pure import ( 

23 ToolCompatibilityReportDict as ToolCompatibilityReportDict, 

24 ) 

25 

26 

27ItemId = str 

28ItemVersion = str 

29Url = str 

30Sha256 = str 

31 

32 

33def check_tool_compatibility( 

34 tool_name: str, 

35 tool_version: str, 

36 *, 

37 index_path: Path = Path("index.json"), 

38 check_tool_compatibility_impl: Callable[ 

39 [ItemId, ItemVersion, Url, Sha256], 

40 "ToolCompatibilityReport | ToolCompatibilityReportDict", 

41 ], 

42 applicable_types: set[str], 

43 id_startswith: str = "", 

44): 

45 """helper to implement tool compatibility checks 

46 

47 Args: 

48 tool_name: name of the tool (without version), e.g. "ilastik" 

49 tool_version: version of the tool, e.g. "1.4" 

50 index_path: Path to the `index.json` file. 

51 check_tool_compatibility_impl: 

52 Function accepting two positional arguments: 

53 URL to an rdf.yaml, SHA-256 of that rdf.yaml. 

54 And returning a compatibility report. 

55 applicable_types: Set of resource types 

56 **check_tool_compatibility_impl** is applicable to. 

57 """ 

58 with index_path.open() as f: 

59 items = json.load(f)["items"] 

60 

61 filtered_items = [ 

62 item 

63 for item in items 

64 if item["type"] in applicable_types and item["id"].startswith(id_startswith) 

65 ] 

66 print(f"found {len(filtered_items)} starting with '{id_startswith}'") 

67 

68 for item in tqdm(filtered_items): 

69 for version in item["versions"]: 

70 rdf_url = version["source"] 

71 sha256 = version["sha256"] 

72 

73 report_path = get_tool_report_path( 

74 item["id"], version["version"], tool_name, tool_version 

75 ) 

76 if report_path.exists(): 

77 logger.info("found existing report at {}", report_path) 

78 continue 

79 

80 try: 

81 report = check_tool_compatibility_impl( 

82 item["id"], version["version"], rdf_url, sha256 

83 ) 

84 except Exception as e: 

85 traceback.print_exc() 

86 warnings.warn(f"failed to check '{rdf_url}': {e}") 

87 else: 

88 if not isinstance(report, dict): 

89 report = report.model_dump(mode="json") 

90 

91 report_path.parent.mkdir(parents=True, exist_ok=True) 

92 with report_path.open("wt", encoding="utf-8") as f: 

93 json.dump(report, f, indent=4, sort_keys=True, ensure_ascii=False) 

94 

95 _total, _used, free = shutil.disk_usage(".") 

96 if free < 7_000_000_000: 

97 raise RuntimeError("less than 7GB disk space left, stopping now")