(contents: str)
| 56 | |
| 57 | |
| 58 | def _migrate_composed(contents: str) -> str: |
| 59 | tree = yaml_compose(contents) |
| 60 | rewrites: list[tuple[ScalarNode, Callable[[ScalarNode], str]]] = [] |
| 61 | |
| 62 | # sha -> rev |
| 63 | sha_to_rev_replace = functools.partial(_preserve_style, s='rev') |
| 64 | sha_to_rev_matcher = ( |
| 65 | MappingValue('repos'), |
| 66 | SequenceItem(), |
| 67 | MappingKey('sha'), |
| 68 | ) |
| 69 | for node in match(tree, sha_to_rev_matcher): |
| 70 | rewrites.append((node, sha_to_rev_replace)) |
| 71 | |
| 72 | # python_venv -> python |
| 73 | language_matcher = ( |
| 74 | MappingValue('repos'), |
| 75 | SequenceItem(), |
| 76 | MappingValue('hooks'), |
| 77 | SequenceItem(), |
| 78 | MappingValue('language'), |
| 79 | ) |
| 80 | python_venv_replace = functools.partial(_preserve_style, s='python') |
| 81 | for node in match(tree, language_matcher): |
| 82 | if node.value == 'python_venv': |
| 83 | rewrites.append((node, python_venv_replace)) |
| 84 | |
| 85 | # stages rewrites |
| 86 | default_stages_matcher = (MappingValue('default_stages'), SequenceItem()) |
| 87 | default_stages_match = match(tree, default_stages_matcher) |
| 88 | hook_stages_matcher = ( |
| 89 | MappingValue('repos'), |
| 90 | SequenceItem(), |
| 91 | MappingValue('hooks'), |
| 92 | SequenceItem(), |
| 93 | MappingValue('stages'), |
| 94 | SequenceItem(), |
| 95 | ) |
| 96 | hook_stages_match = match(tree, hook_stages_matcher) |
| 97 | for node in itertools.chain(default_stages_match, hook_stages_match): |
| 98 | if node.value in {'commit', 'push', 'merge-commit'}: |
| 99 | rewrites.append((node, _fix_stage)) |
| 100 | |
| 101 | rewrites.sort(reverse=True, key=lambda nf: nf[0].start_mark.index) |
| 102 | |
| 103 | src_parts = [] |
| 104 | end: int | None = None |
| 105 | for node, func in rewrites: |
| 106 | src_parts.append(contents[node.end_mark.index:end]) |
| 107 | src_parts.append(func(node)) |
| 108 | end = node.start_mark.index |
| 109 | src_parts.append(contents[:end]) |
| 110 | src_parts.reverse() |
| 111 | return ''.join(src_parts) |
| 112 | |
| 113 | |
| 114 | def migrate_config(config_file: str, quiet: bool = False) -> int: |
no test coverage detected