Lines Matching refs:self
42 def __post_init__(self):
43 section_is_created = bool(self.current_section)
45 self.current_section and self.current_section.has_started
47 self.log_hints = LAVALogHints(self)
53 def phase(self) -> LogSectionType:
55 self.current_section.type
56 if self.current_section
60 def __enter__(self):
61 return self
63 def __exit__(self, exc_type, exc_val, exc_tb):
65 self.clear_current_section()
66 last_lines = self.flush()
70 def watchdog(self):
71 if not self.current_section:
74 timeout_duration = self.timeout_durations.get(
75 self.current_section.type, self.fallback_timeout
78 if self.current_section.delta_time() > timeout_duration:
80 f"Gitlab Section {self.current_section} has timed out",
84 def clear_current_section(self):
85 if self.current_section and not self.current_section.has_finished:
86 self._buffer.append(self.current_section.end())
87 self.current_section = None
89 def update_section(self, new_section: GitlabSection):
92 if self.current_section and self.current_section.id == new_section.id:
94 self.clear_current_section()
95 self.current_section = new_section
96 self._buffer.append(new_section.start())
98 def manage_gl_sections(self, line):
105 self.update_section(new_section)
107 def detect_kernel_dump_line(self, line: dict[str, Union[str, list]]) -> bool:
123 def feed(self, new_lines: list[dict[str, str]]) -> bool:
128 self.watchdog()
134 if self.detect_kernel_dump_line(line):
140 self.manage_gl_sections(line)
142 self._buffer.append(parsed_line)
144 self.log_hints.detect_failure(new_lines)
148 def flush(self) -> list[str]:
149 buffer = self._buffer
150 self._buffer = []