Atomically append one coverage record to filename. Each record is a single line '{bolt} {si} {start} {end} {src_file} {src_line}\\n', written via a single os.write() call so parallel invocations don't interleave partial lines.
(filename: str, bolt: int, section_idx: int, start: int, end: int,
src_file: str, src_line: int)
| 232 | |
| 233 | |
| 234 | def write_coverage(filename: str, bolt: int, section_idx: int, start: int, end: int, |
| 235 | src_file: str, src_line: int) -> None: |
| 236 | """Atomically append one coverage record to filename. |
| 237 | |
| 238 | Each record is a single line '{bolt} {si} {start} {end} {src_file} {src_line}\\n', |
| 239 | written via a single os.write() call so parallel invocations don't |
| 240 | interleave partial lines. |
| 241 | """ |
| 242 | record = "{} {} {} {} {} {}\n".format(bolt, section_idx, start, end, src_file, src_line).encode() |
| 243 | fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o666) |
| 244 | try: |
| 245 | os.write(fd, record) |
| 246 | finally: |
| 247 | os.close(fd) |
| 248 | |
| 249 | |
| 250 | def main(args: Namespace) -> None: |