1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
# Copyright 2022 The ChromiumOS Authors.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Provides patch utilities for PATCHES.json file handling."""
import collections
import dataclasses
import io
from pathlib import Path
import re
import subprocess
import sys
from typing import Any, Dict, List, Optional
CHECKED_FILE_RE = re.compile(r'^checking file\s+(.*)$')
HUNK_FAILED_RE = re.compile(r'^Hunk #(\d+) FAILED at.*')
HUNK_HEADER_RE = re.compile(r'^@@\s+-(\d+),(\d+)\s+\+(\d+),(\d+)\s+@@')
HUNK_END_RE = re.compile(r'^--\s*$')
PATCH_SUBFILE_HEADER_RE = re.compile(r'^\+\+\+ [ab]/(.*)$')
@dataclasses.dataclass
class Hunk:
"""Represents a patch Hunk."""
hunk_id: int
"""Hunk ID for the current file."""
orig_start: int
orig_hunk_len: int
patch_start: int
patch_hunk_len: int
patch_hunk_lineno_begin: int
patch_hunk_lineno_end: Optional[int]
def parse_patch_stream(patch_stream: io.TextIOBase) -> Dict[str, List[Hunk]]:
"""Parse a patch file-like into Hunks.
Args:
patch_stream: A IO stream formatted like a git patch file.
Returns:
A dictionary mapping filenames to lists of Hunks present
in the patch stream.
"""
current_filepath = None
current_hunk_id = 0
current_hunk = None
out = collections.defaultdict(list)
for lineno, line in enumerate(patch_stream.readlines()):
subfile_header = PATCH_SUBFILE_HEADER_RE.match(line)
if subfile_header:
current_filepath = subfile_header.group(1)
if not current_filepath:
raise RuntimeError('Could not get file header in patch stream')
# Need to reset the hunk id, as it's per-file.
current_hunk_id = 0
continue
hunk_header = HUNK_HEADER_RE.match(line)
if hunk_header:
if not current_filepath:
raise RuntimeError('Parsed hunk before file header in patch stream')
if current_hunk:
# Already parsing a hunk
current_hunk.patch_hunk_lineno_end = lineno
current_hunk_id += 1
current_hunk = Hunk(hunk_id=current_hunk_id,
orig_start=int(hunk_header.group(1)),
orig_hunk_len=int(hunk_header.group(2)),
patch_start=int(hunk_header.group(3)),
patch_hunk_len=int(hunk_header.group(4)),
patch_hunk_lineno_begin=lineno + 1,
patch_hunk_lineno_end=None)
out[current_filepath].append(current_hunk)
continue
if current_hunk and HUNK_END_RE.match(line):
current_hunk.patch_hunk_lineno_end = lineno
return out
def parse_failed_patch_output(text: str) -> Dict[str, List[int]]:
current_file = None
failed_hunks = collections.defaultdict(list)
for eline in text.split('\n'):
checked_file_match = CHECKED_FILE_RE.match(eline)
if checked_file_match:
current_file = checked_file_match.group(1)
continue
failed_match = HUNK_FAILED_RE.match(eline)
if failed_match:
if not current_file:
raise ValueError('Input stream was not parsable')
hunk_id = int(failed_match.group(1))
failed_hunks[current_file].append(hunk_id)
return failed_hunks
@dataclasses.dataclass(frozen=True)
class PatchResult:
"""Result of a patch application."""
succeeded: bool
failed_hunks: Dict[str, List[Hunk]] = dataclasses.field(default_factory=dict)
def __bool__(self):
return self.succeeded
@dataclasses.dataclass
class PatchEntry:
"""Object mapping of an entry of PATCHES.json."""
workdir: Path
metadata: Dict[str, Any]
platforms: List[str]
rel_patch_path: str
version_range: Dict[str, int]
_parsed_hunks = None
def __post_init__(self):
if not self.workdir.is_dir():
raise ValueError(f'workdir {self.workdir} is not a directory')
@classmethod
def from_dict(cls, workdir: Path, data: Dict[str, Any]):
"""Instatiate from a dictionary.
Dictionary must have at least the following keys:
{
'metadata': {
'title': '<title>'
},
'platforms': ['<platform>'],
'rel_patch_path': '<relative patch path to workdir>',
'version_range': {
'from': <int>,
'until': <int>,
},
}
Returns:
A new PatchEntry.
"""
return cls(workdir, data['metadata'], data['platforms'],
data['rel_patch_path'], data['version_range'])
def to_dict(self) -> Dict[str, Any]:
return {
'metadata': self.metadata,
'platforms': self.platforms,
'rel_patch_path': self.rel_patch_path,
'version_range': self.version_range,
}
def parsed_hunks(self) -> Dict[str, List[Hunk]]:
# Minor caching here because IO is slow.
if not self._parsed_hunks:
with self.patch_path().open(encoding='utf-8') as f:
self._parsed_hunks = parse_patch_stream(f)
return self._parsed_hunks
def patch_path(self) -> Path:
return self.workdir / self.rel_patch_path
def can_patch_version(self, svn_version: int) -> bool:
"""Is this patch meant to apply to `svn_version`?"""
# Sometimes the key is there, but it's set to None.
from_v = self.version_range.get('from') or 0
until_v = self.version_range.get('until')
if until_v is None:
until_v = sys.maxsize
return from_v <= svn_version < until_v
def is_old(self, svn_version: int) -> bool:
"""Is this patch old compared to `svn_version`?"""
until_v = self.version_range.get('until')
# Sometimes the key is there, but it's set to None.
if until_v is None:
until_v = sys.maxsize
return svn_version >= until_v
def apply(self,
root_dir: Path,
extra_args: Optional[List[str]] = None) -> PatchResult:
"""Apply a patch to a given directory."""
if not extra_args:
extra_args = []
# Cmd to apply a patch in the src unpack path.
cmd = [
'patch', '-d',
root_dir.absolute(), '-f', '-p1', '--no-backup-if-mismatch', '-i',
self.patch_path().absolute()
] + extra_args
try:
subprocess.run(cmd, encoding='utf-8', check=True, stdout=subprocess.PIPE)
except subprocess.CalledProcessError as e:
parsed_hunks = self.parsed_hunks()
failed_hunks_id_dict = parse_failed_patch_output(e.stdout)
failed_hunks = {}
for path, failed_hunk_ids in failed_hunks_id_dict.items():
hunks_for_file = parsed_hunks[path]
failed_hunks[path] = [
hunk for hunk in hunks_for_file if hunk.hunk_id in failed_hunk_ids
]
return PatchResult(succeeded=False, failed_hunks=failed_hunks)
return PatchResult(succeeded=True)
def test_apply(self, root_dir: Path) -> PatchResult:
"""Dry run applying a patch to a given directory."""
return self.apply(root_dir, ['--dry-run'])
def title(self) -> str:
return self.metadata['title']
|
