Coverage for putki/discover.py: 45.35%

1import pathlib 

2from typing import Union, no_type_check 

3 

4import yaml 

5from git.exc import GitCommandError, InvalidGitRepositoryError, NoSuchPathError 

6from git.repo import Repo 

7from putki import ENCODING 

8 

9 

10@no_type_check 

11def root(within: Union[str, pathlib.Path] = '.') -> str: 

12 """Detect the root folder for tasks.""" 

13 try: 

14 repo = Repo(within, search_parent_directories=True) 

15 try: 

16 repo_root = repo.git.rev_parse(show_toplevel=True) 

17 start_here = pathlib.Path(repo_root) 

18 for path in start_here.rglob('*'): 18 ↛ 21line 18 didn't jump to line 21, because the loop on line 18 didn't complete

19 if path.name == 'tasks' and path.is_dir(): 

20 return str(path) 

21 return '' 

22 except GitCommandError: 

23 return '' 

24 except (InvalidGitRepositoryError, NoSuchPathError): 

25 return '' 

26 

27 

28@no_type_check 

29def tasks(below: Union[str, pathlib.Path] = '.') -> dict[str, list[dict[str, str]]]: 

30 """Collect the tasks below by mapping the paths to lists of key value string maps.""" 

31 jobs: dict[str, list[dict[str, str]]] = {} 

32 start_here = pathlib.Path(below) 

33 for path in start_here.rglob('**/*'): 

34 if path.is_file() and path.stat().st_size and path.suffix.lower() in ('.yaml', '.yml'): 

35 with open(path, 'rt', encoding=ENCODING) as handle: 

36 data = yaml.safe_load(handle) 

37 if data and data.get('tasks', None): 

38 jobs[str(path.parent)] = data['tasks'] 

39 return jobs 

40 

41 

42@no_type_check 

43def combine(jobs: dict[str, list[dict[str, str]]]) -> list[dict[str, str]]: 

44 """Combine the tasks by mapping the local ids to path prefixed ids.""" 

45 tasks_seq: list[dict[str, str]] = [] 

46 common = min(sorted(jobs.keys())) 

47 for path, job_seq in jobs.items(): 

48 prefix = path.replace(common, '', 1) 

49 for task in job_seq: 

50 local_id = task['id'] 

51 task['id'] = f'{prefix}/{local_id}' 

52 tasks_seq.append(task) 

53 return tasks_seq 

54 

55 

56@no_type_check 

57def assemble_path(path_elements: dict[str, str]) -> str: 

58 """Assemble a connection string from /source/path_elements.""" 

59 address_template = path_elements['address_template'] 

60 user = path_elements.get('user') 

61 return ( 

62 address_template.replace('{{protocol}}', path_elements['protocol']) 

63 .replace('{{user}}', user if user is not None else '') 

64 .replace('{{host}}', path_elements['host']) 

65 .replace('{{port}}', path_elements['port']) 

66 .replace('{{service_root}}', path_elements['service_root']) 

67 ) 

68 

69 

70@no_type_check 

71def derive( 

72 tasks_seq: list[dict[str, Union[str, dict[str, str]]]] 

73) -> dict[str, dict[str, Union[str, int, dict[str, str]]]]: 

74 """Derive map with actionable names by mapping the path prefixed ids and assembling path elements.""" 

75 actions: dict[str, dict[str, Union[str, int, dict[str, str]]]] = {} 

76 for slot, task in enumerate(tasks_seq): 

77 folder_slug = task['id'].replace('/', '_') 

78 

79 action: dict[str, Union[str, int]] = {**task, 'rank': slot} 

80 source: dict[str, Union[str, dict[str, str]]] = task['source'] 

81 action['url'] = source['path'] if source.get('path') else assemble_path(source['path_elements']) 

82 actions[folder_slug] = action 

83 

84 return actions