diff --git a/docs/mobly.controllers.android_device_lib.rst b/docs/mobly.controllers.android_device_lib.rst index 8f1d3790..9f74e60b 100644 --- a/docs/mobly.controllers.android_device_lib.rst +++ b/docs/mobly.controllers.android_device_lib.rst @@ -67,6 +67,14 @@ mobly.controllers.android\_device\_lib.jsonrpc\_shell\_base module :undoc-members: :show-inheritance: +mobly.controllers.android\_device\_lib.logcat\_processor module +--------------------------------------------------------------- + +.. automodule:: mobly.controllers.android_device_lib.logcat_processor + :members: + :undoc-members: + :show-inheritance: + mobly.controllers.android\_device\_lib.service\_manager module -------------------------------------------------------------- diff --git a/mobly/controllers/android_device_lib/logcat_processor.py b/mobly/controllers/android_device_lib/logcat_processor.py new file mode 100644 index 00000000..ab785f0e --- /dev/null +++ b/mobly/controllers/android_device_lib/logcat_processor.py @@ -0,0 +1,612 @@ +# Copyright 2026 Google Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Logcat line parsing, timestamp comparison, and file reader utilities.""" + +import collections +from collections.abc import Iterable +import dataclasses +import os +import queue +import re +import threading +import time +from typing import Any, ClassVar, Iterator, Optional, Pattern, Sequence, Set, Union + + +_LEVEL_NORM_MAP = { + 'V': 'V', + 'VERBOSE': 'V', + 'D': 'D', + 'DEBUG': 'D', + 'I': 'I', + 'INFO': 'I', + 'W': 'W', + 'WARN': 'W', + 'WARNING': 'W', + 'E': 'E', + 'ERROR': 'E', + 'F': 'F', + 'FATAL': 'F', + 'A': 'F', + 'ASSERT': 'F', + 'S': 'S', + 'SILENT': 'S', +} + + +@dataclasses.dataclass(frozen=True) +class LogcatPosition: + """A position marker representing a specific point in the logcat stream. + + Attributes: + timestamp: Optional string timestamp corresponding to this position. + creation_time: Host epoch time when this position was marked. + """ + + timestamp: Optional[str] = None + creation_time: float = dataclasses.field(default_factory=time.time) + _byte_offset: int = 0 + + @classmethod + def from_file( + cls, file_path: str, timestamp: Optional[str] = None + ) -> 'LogcatPosition': + """Captures a LogcatPosition snapshot of a logcat file at the current moment.""" + try: + file_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0 + except OSError: + file_size = 0 + return cls( + timestamp=timestamp, + creation_time=time.time(), + _byte_offset=file_size, + ) + + @staticmethod + def _parse_timestamp(t: str) -> tuple[int, int, int, int, int, int, int]: + """Parses a logline timestamp into (year, month, day, hour, minute, second, microsecond).""" + if not t: + raise ValueError('Empty timestamp string') + + date_part, time_part = re.split(r'[\sT]+', t.strip(), maxsplit=1) + date_elements = [int(x) for x in re.split(r'[-/]', date_part)] + if len(date_elements) == 3: + year, month, day = date_elements + elif len(date_elements) == 2: + year = 0 + month, day = date_elements + else: + raise ValueError(f'Invalid date elements in timestamp: {t}') + + time_parts = time_part.split(':') + hour = int(time_parts[0]) + minute = int(time_parts[1]) if len(time_parts) > 1 else 0 + second, microsecond = 0, 0 + if len(time_parts) > 2: + s_ms = time_parts[2].split('.', 1) + second = int(s_ms[0]) + if len(s_ms) > 1: + microsecond = int(s_ms[1].ljust(6, '0')[:6]) + + return (year, month, day, hour, minute, second, microsecond) + + @classmethod + def _compare_timestamps(cls, t1: Optional[str], t2: Optional[str]) -> int: + """Compares two logline timestamps chronologically.""" + if not t1 and not t2: + return 0 + if not t1: + return -1 + if not t2: + return 1 + try: + p1 = cls._parse_timestamp(t1) + p2 = cls._parse_timestamp(t2) + if p1[0] == 0 or p2[0] == 0: + p1 = (0,) + p1[1:] + p2 = (0,) + p2[1:] + return (p1 > p2) - (p1 < p2) + except (ValueError, IndexError): + str_t1, str_t2 = str(t1 or ''), str(t2 or '') + return (str_t1 > str_t2) - (str_t1 < str_t2) + + def __lt__(self, other: Any) -> bool: + if not isinstance(other, LogcatPosition): + return NotImplemented + if self._byte_offset != other._byte_offset: + return self._byte_offset < other._byte_offset + return self._compare_timestamps(self.timestamp, other.timestamp) < 0 + + def __le__(self, other: Any) -> bool: + if not isinstance(other, LogcatPosition): + return NotImplemented + if self._byte_offset != other._byte_offset: + return self._byte_offset <= other._byte_offset + return self._compare_timestamps(self.timestamp, other.timestamp) <= 0 + + def __gt__(self, other: Any) -> bool: + if not isinstance(other, LogcatPosition): + return NotImplemented + if self._byte_offset != other._byte_offset: + return self._byte_offset > other._byte_offset + return self._compare_timestamps(self.timestamp, other.timestamp) > 0 + + def __ge__(self, other: Any) -> bool: + if not isinstance(other, LogcatPosition): + return NotImplemented + if self._byte_offset != other._byte_offset: + return self._byte_offset >= other._byte_offset + return self._compare_timestamps(self.timestamp, other.timestamp) >= 0 + + +@dataclasses.dataclass(frozen=True) +class LogLine: + """Represents a single parsed Android logcat line in threadtime format. + + Attributes: + position: LogcatPosition, position marker and timestamp of this log line. + pid: int, process ID. + tid: int, thread ID. + level: str, single-letter severity level ('V', 'D', 'I', 'W', 'E', 'F', 'S'). + tag: str, log tag. + message: str, log message payload. + raw: str, original raw log line string without line endings. + """ + + position: LogcatPosition + pid: int + tid: int + level: str + tag: str + message: str + raw: str + + _PATTERN: ClassVar[Pattern[str]] = re.compile( + r'^(?P(?:\d{4}[-/])?\d{2}[-/]\d{2}\s+\d{2}:\d{2}:\d{2}(?:\.\d+)?)' + r'\s+(?P\d+)' + r'\s+(?P\d+)' + r'\s+(?P[VDIWEFSA])' + r'\s+(?P.*?)' + r'\s*:\s?' + r'(?P.*)$' + ) + + @property + def timestamp(self) -> str: + """Returns the string timestamp of this log line.""" + return self.position.timestamp or '' + + @classmethod + def from_string(cls, line: str, byte_offset: int = 0) -> Optional['LogLine']: + """Parses a raw logcat line in threadtime format into a LogLine object.""" + if not line or not isinstance(line, str): + return None + + clean_line = line.rstrip('\r\n') + match = cls._PATTERN.match(clean_line) + if not match: + return None + + try: + pos = LogcatPosition( + timestamp=match.group('timestamp'), + _byte_offset=byte_offset, + ) + return cls( + position=pos, + pid=int(match.group('pid')), + tid=int(match.group('tid')), + level=match.group('level'), + tag=match.group('tag'), + message=match.group('message'), + raw=clean_line, + ) + except (ValueError, TypeError, IndexError): + return None + + def matches( + self, + pattern: Optional[Union[str, Pattern[str]]] = None, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + ) -> bool: + """Checks if this log line matches the given pattern, tag, and/or level.""" + if pattern is not None: + regex = re.compile(pattern) if isinstance(pattern, str) else pattern + if not (regex.search(self.message) or regex.search(self.raw)): + return False + + if tag is not None: + if isinstance(tag, str): + if self.tag != tag: + return False + elif hasattr(tag, 'search'): + if not tag.search(self.tag): + return False + elif isinstance(tag, Iterable) and self.tag not in tag: + return False + + if level is not None: + levels = {level} if isinstance(level, str) else set(level) + norm_levels = { + _LEVEL_NORM_MAP.get(str(l).upper(), str(l).upper()) for l in levels + } + self_norm = _LEVEL_NORM_MAP.get(self.level.upper(), self.level.upper()) + if self.level not in levels and self_norm not in norm_levels: + return False + + return True + + @property + def is_error(self) -> bool: + """Returns True if this line represents an error or fatal severity.""" + return self.level.upper() in ('E', 'F', 'A') + + def __lt__(self, other: Any) -> bool: + if not isinstance(other, LogLine): + return NotImplemented + return self.position < other.position + + def __le__(self, other: Any) -> bool: + if not isinstance(other, LogLine): + return NotImplemented + return self.position <= other.position + + def __gt__(self, other: Any) -> bool: + if not isinstance(other, LogLine): + return NotImplemented + return self.position > other.position + + def __ge__(self, other: Any) -> bool: + if not isinstance(other, LogLine): + return NotImplemented + return self.position >= other.position + + +class LogcatListenerContext: + """Context manager for listening to real-time logcat events.""" + + def __init__( + self, + processor: 'LogcatProcessor', + pattern: Optional[Union[str, Pattern[str]]] = None, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + position: Optional[Union[LogcatPosition, LogLine]] = None, + max_events: int = 1000, + timeout_error_cls: type[Exception] = TimeoutError, + ): + self._processor = processor + self._pattern = pattern + self._tag = tag + self._level = level + self._position = ( + position.position if isinstance(position, LogLine) else position + ) + self._max_events = max_events + self._timeout_error_cls = timeout_error_cls + self._events: collections.deque[LogLine] = collections.deque( + maxlen=max_events + ) + self._queue: queue.Queue[LogLine] = queue.Queue(maxsize=max_events) + self._lock = threading.Lock() + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + + @property + def events(self) -> list[LogLine]: + """Returns a snapshot list of captured events.""" + with self._lock: + return list(self._events) + + def has_events(self) -> bool: + """Returns True if any events have been captured.""" + with self._lock: + return bool(self._events) + + def get_next_event(self, timeout: Optional[float] = None) -> LogLine: + """Gets the next event from the queue, blocking up to timeout seconds.""" + try: + return self._queue.get(block=True, timeout=timeout) + except queue.Empty: + raise self._timeout_error_cls( + f'Timed out after {timeout}s waiting for next logcat event ' + f'(pattern={self._pattern!r}, tag={self._tag!r},' + f' level={self._level!r})' + ) + + def _dispatch(self, line: LogLine) -> None: + if line.matches(pattern=self._pattern, tag=self._tag, level=self._level): + with self._lock: + self._events.append(line) + try: + self._queue.put_nowait(line) + except queue.Full: + pass + + def _listen_loop(self) -> None: + current_offset = ( + self._position._byte_offset + if self._position + else LogcatPosition.from_file(self._processor.file_path)._byte_offset + ) + while not self._stop_event.is_set(): + for offset, line in self._processor._iter_lines(offset=current_offset): + current_offset = offset + self._dispatch(line) + if self._stop_event.is_set(): + break + time.sleep(0.05) + + def __enter__(self) -> 'LogcatListenerContext': + self._stop_event.clear() + self._thread = threading.Thread(target=self._listen_loop, daemon=True) + self._thread.start() + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + self._stop_event.set() + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=2.0) + self._thread = None + + +class LogcatProcessor: + """Thread-safe processor for querying and streaming logcat files on the host.""" + + def __init__( + self, + file_path: str, + timeout_error_cls: type[Exception] = TimeoutError, + ): + self._file_path = file_path + self._timeout_error_cls = timeout_error_cls + + @property + def file_path(self) -> str: + return self._file_path + + def _iter_lines(self, offset: int = 0) -> Iterator[tuple[int, LogLine]]: + """Yields (line_offset, LogLine) pairs from the file from the given offset.""" + if not os.path.exists(self._file_path): + return + try: + with open( + self._file_path, 'r', encoding='utf-8', errors='replace', newline='' + ) as f: + if offset > 0: + f.seek(offset) + while True: + line_offset = f.tell() + line = f.readline() + if not line: + break + current_offset = f.tell() + parsed = LogLine.from_string(line, byte_offset=line_offset) + if parsed is not None: + yield current_offset, parsed + except OSError: + return + + def get_lines( + self, + pattern: Optional[Union[str, Pattern[str]]] = None, + *, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + since: Optional[Union[LogcatPosition, LogLine]] = None, + max_lines: Optional[int] = None, + ) -> list[LogLine]: + """Gets log lines from the file satisfying filter criteria.""" + if ( + pattern is None + and tag is None + and level is None + and since is None + and max_lines is None + ): + raise ValueError( + 'At least one filter criteria (pattern, tag, level, since, or' + ' max_lines) must be specified. To inspect the latest logs, use' + ' tail() instead.' + ) + + pos = since.position if isinstance(since, LogLine) else since + offset = pos._byte_offset if pos else 0 + begin_time = pos.timestamp if pos and offset == 0 else None + + results: list[LogLine] = [] + for _, parsed in self._iter_lines(offset=offset): + if ( + begin_time + and LogcatPosition._compare_timestamps(parsed.timestamp, begin_time) + < 0 + ): + continue + if parsed.matches(pattern=pattern, tag=tag, level=level): + results.append(parsed) + if max_lines is not None and len(results) >= max_lines: + break + return results + + def tail( + self, + num_lines: int = 100, + pattern: Optional[Union[str, Pattern[str]]] = None, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + ) -> list[LogLine]: + """Tails the last num_lines matching log lines by reading backwards from EOF.""" + if num_lines <= 0 or not os.path.exists(self._file_path): + return [] + + buf: collections.deque[LogLine] = collections.deque() + block_size = 64 * 1024 # 64KB chunks + + try: + with open(self._file_path, 'rb') as f: + f.seek(0, os.SEEK_END) + file_size = f.tell() + if file_size == 0: + return [] + + remaining = file_size + remainder = b'' + lines_to_process: list[tuple[int, bytes]] = [] + + while remaining > 0 and len(buf) < num_lines: + read_size = min(block_size, remaining) + remaining -= read_size + f.seek(remaining) + chunk = f.read(read_size) + data = chunk + remainder + + # Split lines from chunk + split = data.split(b'\n') + if remaining > 0: + remainder = split[0] + lines_chunk = split[1:] + else: + remainder = b'' + lines_chunk = split + + # Calculate offsets and parse lines in reverse order within this block + current_offset = remaining + len(remainder) + block_lines: list[tuple[int, LogLine]] = [] + for line_bytes in lines_chunk: + line_offset = current_offset + current_offset += len(line_bytes) + 1 # count \n byte + line_str = line_bytes.decode('utf-8', errors='replace') + parsed = LogLine.from_string(line_str, byte_offset=line_offset) + if parsed is not None: + block_lines.append((line_offset, parsed)) + + for _, parsed in reversed(block_lines): + if parsed.matches(pattern=pattern, tag=tag, level=level): + buf.appendleft(parsed) + if len(buf) >= num_lines: + break + except OSError: + return [] + + return list(buf) + + def listen( + self, + pattern: Optional[Union[str, Pattern[str]]] = None, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + position: Optional[Union[LogcatPosition, LogLine]] = None, + ) -> LogcatListenerContext: + """Listens for real-time logcat events in a context manager.""" + return LogcatListenerContext( + processor=self, + pattern=pattern, + tag=tag, + level=level, + position=position, + timeout_error_cls=self._timeout_error_cls, + ) + + def wait_for( + self, + patterns: Sequence[Union[str, Pattern[str]]], + timeout_sec: float = 60.0, + in_order: bool = True, + since: Optional[Union[LogcatPosition, LogLine]] = None, + ) -> list[LogLine]: + """Waits until a sequence of patterns appears in logcat.""" + if not patterns: + return [] + + deadline = time.perf_counter() + timeout_sec + + if in_order: + matched_lines: list[LogLine] = [] + current_since = since + for pat in patterns: + remaining = deadline - time.perf_counter() + if remaining <= 0: + raise self._timeout_error_cls( + f'Timed out after {timeout_sec}s waiting for in-order pattern:' + f' {pat!r}' + ) + matched, next_offset = self._wait_for_single( + pattern=pat, + timeout_sec=remaining, + since=current_since, + ) + matched_lines.append(matched) + current_since = LogcatPosition(_byte_offset=next_offset) + return matched_lines + + unmatched = list(enumerate(patterns)) + matched_dict: dict[int, LogLine] = {} + pos = since.position if isinstance(since, LogLine) else since + offset = pos._byte_offset if pos else 0 + begin_time = pos.timestamp if pos and offset == 0 else None + scan_offset = offset + + while time.perf_counter() < deadline: + for current_offset, parsed in self._iter_lines(offset=scan_offset): + scan_offset = current_offset + if ( + begin_time + and LogcatPosition._compare_timestamps(parsed.timestamp, begin_time) + < 0 + ): + continue + for idx, pat in list(unmatched): + if parsed.matches(pattern=pat): + matched_dict[idx] = parsed + unmatched.remove((idx, pat)) + if not unmatched: + return [matched_dict[i] for i in range(len(patterns))] + time.sleep(0.1) + + remaining_patterns = [pat for _, pat in unmatched] + raise self._timeout_error_cls( + f'Timed out after {timeout_sec}s waiting for patterns:' + f' {remaining_patterns!r}' + ) + + def _wait_for_single( + self, + pattern: Union[str, Pattern[str]], + timeout_sec: float = 60.0, + since: Optional[Union[LogcatPosition, LogLine]] = None, + ) -> tuple[LogLine, int]: + deadline = time.perf_counter() + timeout_sec + pos = since.position if isinstance(since, LogLine) else since + offset = pos._byte_offset if pos else 0 + begin_time = pos.timestamp if pos and offset == 0 else None + scan_offset = offset + + while time.perf_counter() < deadline: + for current_offset, parsed in self._iter_lines(offset=scan_offset): + scan_offset = current_offset + if ( + begin_time + and LogcatPosition._compare_timestamps(parsed.timestamp, begin_time) + < 0 + ): + continue + if parsed.matches(pattern=pattern): + return parsed, current_offset + time.sleep(0.1) + + raise self._timeout_error_cls( + f'Timed out after {timeout_sec}s waiting for logcat pattern:' + f' {pattern!r}' + ) diff --git a/mobly/controllers/android_device_lib/services/logcat.py b/mobly/controllers/android_device_lib/services/logcat.py index a478b0a9..9203c166 100644 --- a/mobly/controllers/android_device_lib/services/logcat.py +++ b/mobly/controllers/android_device_lib/services/logcat.py @@ -11,14 +11,16 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. + import logging import os import time +from typing import Any, Callable, Optional, Pattern, Sequence, Set, Union -from mobly import logger as mobly_logger from mobly import utils from mobly.controllers.android_device_lib import adb from mobly.controllers.android_device_lib import errors +from mobly.controllers.android_device_lib import logcat_processor from mobly.controllers.android_device_lib.services import base_service CREATE_LOGCAT_FILE_TIMEOUT_SEC = 5 @@ -30,15 +32,19 @@ class Error(errors.ServiceError): SERVICE_TYPE = 'Logcat' +class LogcatTimeoutError(Error): + """Raised when a logcat wait operation times out.""" + + class Config: """Config object for logcat service. Attributes: clear_log: bool, clears the logcat before collection if True. logcat_params: string, extra params to be added to logcat command. - output_file_path: string, the path on the host to write the log file - to, including the actual filename. The service will automatically - generate one if not specified. + output_file_path: string, the path on the host to write the log file to, + including the actual filename. The service will automatically generate one + if not specified. """ def __init__(self, logcat_params=None, clear_log=True, output_file_path=None): @@ -51,8 +57,8 @@ class Logcat(base_service.BaseService): """Android logcat service for Mobly's AndroidDevice controller. Attributes: - adb_logcat_file_path: string, path to the file that the service writes - adb logcat to by default. + adb_logcat_file_path: string, path to the file that the service writes adb + logcat to by default. """ OUTPUT_FILE_TYPE = 'logcat' @@ -63,11 +69,22 @@ def __init__(self, android_device, configs=None): self._adb_logcat_process = None self._adb_logcat_file_obj = None self.adb_logcat_file_path = None + self._processor = None self._last_connection_time = None # Logcat service uses a single config obj, using singular internal # name: `_config`. self._config = configs if configs else Config() + def _get_processor(self) -> logcat_processor.LogcatProcessor: + if self._processor is None: + if not self.adb_logcat_file_path: + raise Error(self._ad, 'Logcat service has not been started.') + self._processor = logcat_processor.LogcatProcessor( + self.adb_logcat_file_path, + timeout_error_cls=lambda msg: LogcatTimeoutError(self._ad, msg), + ) + return self._processor + def _enable_logpersist(self): """Attempts to enable logpersist daemon to persist logs.""" # Logpersist is only allowed on rootable devices because of excessive @@ -76,8 +93,8 @@ def _enable_logpersist(self): return logpersist_warning = ( - '%s encountered an error enabling persistent' - ' logs, logs may not get saved.' + '%s encountered an error enabling persistent logs, logs may not get' + ' saved.' ) # Android L and older versions do not have logpersist installed, # so check that the logpersist scripts exists before trying to use @@ -95,13 +112,212 @@ def _enable_logpersist(self): except adb.AdbError: logging.warning(logpersist_warning, self) - def _is_timestamp_in_range(self, target, begin_time, end_time): - low = mobly_logger.logline_timestamp_comparator(begin_time, target) <= 0 - high = mobly_logger.logline_timestamp_comparator(end_time, target) >= 0 - return low and high + def _get_device_time(self) -> Optional[str]: + """Retrieves current timestamp from device via ADB date command.""" + try: + response = self._ad.adb.shell(['date', r'+%Y-%m-%d\ %H:%M:%S.%3N']) + if response: + return response.decode('utf-8').strip() + except (adb.AdbError, UnicodeDecodeError): + self._ad.log.debug('Failed to get device timestamp.') + return None + + def now(self) -> logcat_processor.LogcatPosition: + """Captures the current logcat position and device timestamp. + + Creates a position marker representing the logcat stream right now to bound + subsequent queries or wait operations. + + Examples:: + + start = ad.services.logcat.now() + ad.droid.wifiEnable() + lines = ad.services.logcat.get_lines('STATE_CONNECTED', since=start) + + Returns: + A :class:`~mobly.controllers.android_device_lib.logcat_processor.LogcatPosition` + instance representing the current log state. + """ + return logcat_processor.LogcatPosition.from_file( + self.adb_logcat_file_path or '', + timestamp=self._get_device_time(), + ) + + def get_lines( + self, + pattern: Optional[Union[str, Pattern[str]]] = None, + *, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + since: Optional[ + Union[logcat_processor.LogcatPosition, logcat_processor.LogLine] + ] = None, + max_lines: Optional[int] = None, + ) -> list[logcat_processor.LogLine]: + """Gets log lines from the logcat file matching the given filters. + + Filters are evaluated conjunctively. At least one filter criteria + (``pattern``, ``tag``, ``level``, ``since``, or ``max_lines``) must be + specified. To retrieve the latest un-filtered logs, use :meth:`tail` + instead. + + Examples:: + + # Find error logs for a specific tag + lines = ad.services.logcat.get_lines(tag='ActivityManager', level='E') + + # Find pattern matches since a marked position + start = ad.services.logcat.now() + ad.droid.wifiEnable() + lines = ad.services.logcat.get_lines('WiFi connected', since=start) + + Args: + pattern: Regular expression pattern matched against message and raw line. + tag: Tag string, compiled regex pattern, or collection of tags to match. + level: Severity level string ('V', 'D', 'I', 'W', 'E', 'F') or collection. + since: Optional + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogcatPosition` + or + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogLine` + bounding the search start. + max_lines: Maximum number of matching log lines to return. + + Returns: + A list of matching + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogLine` + objects in file order. + + Raises: + ValueError: If no filter criteria are provided. + """ + return self._get_processor().get_lines( + pattern=pattern, + tag=tag, + level=level, + since=since, + max_lines=max_lines, + ) + + def tail( + self, + num_lines: int = 100, + pattern: Optional[Union[str, Pattern[str]]] = None, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + ) -> list[logcat_processor.LogLine]: + """Tails the last matching log lines from the logcat file. + + Examples:: + + # Get the last 50 log lines + recent_lines = ad.services.logcat.tail(num_lines=50) + + # Get the last 5 fatal errors + fatal_lines = ad.services.logcat.tail(num_lines=5, level='F') + + Args: + num_lines: Number of matching lines to return from the end of the file. + pattern: Optional regex pattern filter. + tag: Optional tag filter. + level: Optional severity level filter. + + Returns: + A list of the last ``num_lines`` matching + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogLine` + objects. + """ + return self._get_processor().tail( + num_lines=num_lines, + pattern=pattern, + tag=tag, + level=level, + ) + + def listen( + self, + pattern: Optional[Union[str, Pattern[str]]] = None, + tag: Optional[Union[str, Pattern[str], Sequence[str], Set[str]]] = None, + level: Optional[Union[str, Sequence[str], Set[str]]] = None, + ) -> logcat_processor.LogcatListenerContext: + """Listens for real-time logcat events within a scoped context manager. + + Automatically captures current logcat position on entry and cleans up + background workers upon exiting the context. + + Examples:: + + with ad.services.logcat.listen(tag='WifiService') as listener: + ad.droid.wifiEnable() + event = listener.get_next_event(timeout=10.0) + assert 'STATE_CONNECTED' in event.message + + Args: + pattern: Optional regex pattern to filter incoming events. + tag: Optional tag filter. + level: Optional severity level filter. + + Returns: + A + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogcatListenerContext` + object managing the event queue and background stream. + """ + cursor = self.now() + return self._get_processor().listen( + pattern=pattern, + tag=tag, + level=level, + position=cursor, + ) + + def wait_for( + self, + patterns: Sequence[Union[str, Pattern[str]]], + timeout_sec: float = 60.0, + in_order: bool = True, + since: Optional[ + Union[logcat_processor.LogcatPosition, logcat_processor.LogLine] + ] = None, + ) -> list[logcat_processor.LogLine]: + """Waits until pattern(s) appear in logcat within the given timeout. + + Examples:: + + # Wait for a single pattern + lines = ad.services.logcat.wait_for(['Bluetooth connected'], timeout_sec=10.0) + + # Wait for multiple patterns in sequential order + steps = ['DHCP DISCOVER', 'DHCP OFFER', 'DHCP ACK'] + lines = ad.services.logcat.wait_for(steps, in_order=True, timeout_sec=15.0) + + Args: + patterns: Sequence of string patterns or compiled regular expressions. + timeout_sec: Maximum wall-clock time in seconds to wait before timing out. + in_order: Whether patterns must occur in the specified sequential order + (True) or any order (False). + since: Optional + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogcatPosition` + or + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogLine` + bounding the search start. + + Returns: + A list of matching + :class:`~mobly.controllers.android_device_lib.logcat_processor.LogLine` + objects corresponding to each pattern in ``patterns``. + + Raises: + LogcatTimeoutError: If matching pattern(s) are not found within + ``timeout_sec``. + """ + return self._get_processor().wait_for( + patterns=patterns, + timeout_sec=timeout_sec, + in_order=in_order, + since=since, + ) def create_output_excerpts(self, test_info): - """Convenient method for creating excerpts of adb logcat. + """Creates excerpts of adb logcat copied from current stream. This copies logcat lines from self.adb_logcat_file_path to an excerpt file, starting from the location where the previous excerpt ended. @@ -126,7 +342,7 @@ def create_output_excerpts(self, test_info): 'w', encoding='utf-8', errors='replace', - # When newline is '', line endings are written without conversion. + # When newline is '', line endings are read without conversion. newline='', ) as out: # Devices may accidentally go offline during test, @@ -186,9 +402,11 @@ def update_config(self, new_config): new_config, ) self._config = new_config + self._processor = None def _open_logcat_file(self): - """Create a file object that points to the beginning of the logcat file. + """Creates a file object that points to the beginning of the logcat file. + Wait for the logcat file to be created by the subprocess if it doesn't exist. """ @@ -233,7 +451,7 @@ def start(self): self._open_logcat_file() def _start(self): - """The actual logic of starting logcat.""" + """Starts the actual subprocess logic of starting logcat.""" self._enable_logpersist() if self._config.output_file_path: self._close_logcat_file() @@ -244,6 +462,10 @@ def _start(self): ) logcat_file_path = os.path.join(self._ad.log_path, f_name) self.adb_logcat_file_path = logcat_file_path + self._processor = logcat_processor.LogcatProcessor( + self.adb_logcat_file_path, + timeout_error_cls=lambda msg: LogcatTimeoutError(self._ad, msg), + ) utils.create_dir(os.path.dirname(self.adb_logcat_file_path)) # In debugging mode of IntelijIDEA, "patch_args" remove # double quotes in args if starting and ending with it. @@ -280,7 +502,7 @@ def _stop(self): self._last_connection_time = None def pause(self): - """Pauses logcat. + """Pauses logcat collection. Note: the service is unable to collect the logs when paused, if more logs are generated on the device than the device's log buffer can hold, diff --git a/tests/mobly/controllers/android_device_lib/services/logcat_test.py b/tests/mobly/controllers/android_device_lib/services/logcat_test.py index 493ef311..91167878 100755 --- a/tests/mobly/controllers/android_device_lib/services/logcat_test.py +++ b/tests/mobly/controllers/android_device_lib/services/logcat_test.py @@ -14,8 +14,11 @@ import logging import os +import re import shutil import tempfile +import threading +import time import unittest from unittest import mock @@ -23,6 +26,7 @@ from mobly import runtime_test_info from mobly.controllers import android_device from mobly.controllers.android_device_lib import adb +from mobly.controllers.android_device_lib import logcat_processor from mobly.controllers.android_device_lib.services import logcat from tests.lib import mock_android_device @@ -814,5 +818,150 @@ def test_clear_adb_log(self, MockFastboot, MockAdbProxy): logcat_service.clear_adb_log() +SAMPLE_REALISTIC_LOGCAT = ( + '--------- beginning of system\n' + '08-09 22:00:00.100 1000 1010 I SystemServer: Entered SystemServer main\n' + '08-09 22:00:01.200 1000 1020 I ActivityManager: Starting activity' + ' com.example.app/.MainActivity\n' + '2026-08-09 22:00:02.300 1000 1030 D WifiService: Enabling Wi-Fi' + ' interface wlan0\n' + '2026-08-09 22:00:02.500 1000 1030 I DhcpClient: DHCP DISCOVER sent on' + ' wlan0\n' + '--------- beginning of main\n' + '2026-08-09 22:00:02.700 2050 2050 I ExampleApp: App initialized' + ' successfully\n' + '2026-08-09 22:00:03.100 1000 1030 I DhcpClient: DHCP OFFER received from' + ' 192.168.1.1\n' + '08-09 22:00:03.400 1000 1040 W BtGatt: Connection retry count 1 for' + ' device AA:BB:CC:DD:EE:FF\n' + '2026-08-09 22:00:03.600 1000 1030 I DhcpClient: DHCP ACK received,' + ' assigned IP 192.168.1.50\n' + '08-09 22:00:04.000 1000 1020 I WifiService: Network STATE_CONNECTED on' + ' wlan0\n' + '08-09 22:00:05.150 2050 2060 E ExampleApp: Failed to connect to' + ' server\n' + '\tat com.example.app.NetworkClient.connect(NetworkClient.java:42)\n' + '\tat com.example.app.MainActivity.onStart(MainActivity.java:108)\n' + '08-09 22:00:05.800 1000 1040 F BtGatt: Fatal hardware controller error' +) + + +class LogcatServiceUserBehaviorTest(unittest.TestCase): + """User-facing behavior tests for Logcat service.""" + + def setUp(self): + self.tmp_dir = tempfile.mkdtemp() + self.log_file = os.path.join(self.tmp_dir, 'logcat.txt') + with open(self.log_file, 'w', encoding='utf-8') as f: + f.write(SAMPLE_REALISTIC_LOGCAT) + + self.mock_serial = '12345' + self.ad = mock.MagicMock(name='AndroidDevice', serial=self.mock_serial) + self.ad.log = logging.getLogger('mock_ad') + self.ad.adb = mock.MagicMock() + self.ad.adb.shell.return_value = b'2026-08-09 22:00:00.000' + + self.logcat_service = logcat.Logcat(self.ad) + self.logcat_service.adb_logcat_file_path = self.log_file + + def tearDown(self): + self.logcat_service.stop() + shutil.rmtree(self.tmp_dir) + + def _append_log(self, text: str): + with open(self.log_file, 'a', encoding='utf-8', newline='') as f: + f.write(text) + + def test_query_logs_by_tag_level_and_pattern(self): + # Search for error logs + error_logs = self.logcat_service.get_lines(level=['E', 'F']) + self.assertEqual(len(error_logs), 2) + self.assertEqual([r.tag for r in error_logs], ['ExampleApp', 'BtGatt']) + self.assertTrue(error_logs[0].is_error) + + # Filter logs by tag exact match + wifi_logs = self.logcat_service.get_lines(tag='WifiService') + self.assertEqual(len(wifi_logs), 2) + self.assertEqual( + [r.message for r in wifi_logs], + ['Enabling Wi-Fi interface wlan0', 'Network STATE_CONNECTED on wlan0'], + ) + + # Search by pattern + dhcp_offer = self.logcat_service.get_lines(pattern=r'DHCP OFFER.*192\.168') + self.assertEqual(len(dhcp_offer), 1) + self.assertEqual(dhcp_offer[0].tag, 'DhcpClient') + + def test_tail_recent_logs(self): + recent_logs = self.logcat_service.tail(num_lines=3) + self.assertEqual(len(recent_logs), 3) + self.assertEqual(recent_logs[-1].tag, 'BtGatt') + self.assertEqual(recent_logs[-1].level, 'F') + + def test_now_and_bounded_query(self): + # Take position marker before triggering an action + start = self.logcat_service.now() + + # Append new logs simulating device activity after start + self._append_log( + '08-09 22:00:06.000 1000 1030 I WifiService: Disconnected from' + ' wlan0\n' + ) + + lines_since = self.logcat_service.get_lines( + pattern='Disconnected', since=start + ) + self.assertEqual(len(lines_since), 1) + self.assertEqual(lines_since[0].tag, 'WifiService') + self.assertTrue(lines_since[0].position > start) + + # Test passing a LogLine directly to since + self._append_log( + '08-09 22:00:07.000 1000 1030 I WifiService: Reconnected to wlan0\n' + ) + reconnected_lines = self.logcat_service.get_lines( + pattern='Reconnected', since=lines_since[0] + ) + self.assertEqual(len(reconnected_lines), 1) + self.assertTrue(reconnected_lines[0] > lines_since[0]) + + def test_wait_for_sequential_protocol_handshake(self): + handshake_steps = [ + 'DHCP DISCOVER', + 'DHCP OFFER', + 'DHCP ACK', + 'STATE_CONNECTED', + ] + matched_lines = self.logcat_service.wait_for( + handshake_steps, in_order=True, timeout_sec=2.0 + ) + self.assertEqual(len(matched_lines), 4) + self.assertEqual( + [r.tag for r in matched_lines], + ['DhcpClient', 'DhcpClient', 'DhcpClient', 'WifiService'], + ) + self.assertTrue(matched_lines[0] < matched_lines[1] < matched_lines[2]) + + def test_wait_for_timeout_when_event_does_not_occur(self): + with self.assertRaises(logcat.LogcatTimeoutError): + self.logcat_service.wait_for( + ['Nonexistent System Event'], timeout_sec=0.1 + ) + + def test_listen_realtime_event_stream(self): + with self.logcat_service.listen(tag='WifiService') as listener: + self.assertFalse(listener.has_events()) + + # Simulate background service writing new logcat entry + self._append_log( + '08-09 22:00:07.000 1000 1030 I WifiService: Reconnected to wlan0\n' + ) + + event = listener.get_next_event(timeout=2.0) + self.assertEqual(event.tag, 'WifiService') + self.assertEqual(event.message, 'Reconnected to wlan0') + self.assertTrue(listener.has_events()) + + if __name__ == '__main__': unittest.main()