Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 33 additions & 22 deletions hmdriver2/hdc.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,38 @@ def _build_hdc_prefix() -> str:
return "hdc"


# Each mission ends at `isKeepAlive`. Do not require the dump's closing `}` —
# that token belongs to `current mission lists:{ ... }`, so including it made
# findall swallow every mission as one match (issue #52).
_MISSION_BLOCK_RE = re.compile(r'Mission ID #[\s\S]*?isKeepAlive: false\s*')


def _is_foreground_mission(block: str) -> bool:
"""True if the mission is in FOREGROUND (ability or app state)."""
return 'state #FOREGROUND' in block or 'app state #FOREGROUND' in block


def _parse_current_app(output: str) -> Tuple[Optional[str], Optional[str]]:
"""
Parse `aa dump -l` output and return the foreground (package, ability).

When several missions match, prefer FOREGROUND over BACKGROUND (issue #42).
"""
if not output:
return (None, None)

foreground = []
for block in _MISSION_BLOCK_RE.findall(output):
if not _is_foreground_mission(block):
continue
bundle_name_match = re.search(r'bundle name \[(.*?)\]', block)
main_name_match = re.search(r'main name \[(.*?)\]', block)
if bundle_name_match and main_name_match:
foreground.append((bundle_name_match.group(1), main_name_match.group(1)))

return foreground[0] if foreground else (None, None)


def list_devices() -> List[str]:
devices = []
hdc_prefix = _build_hdc_prefix()
Expand Down Expand Up @@ -216,29 +248,8 @@ def current_app(self) -> Tuple[str, str]:
Tuple[str, str]: A tuple contain the package_name andpage_name of the foreground application.
If no foreground application is found, returns (None, None).
"""

def __extract_info(output: str):
results = []

mission_blocks = re.findall(r'Mission ID #[\s\S]*?isKeepAlive: false\s*}', output)
if not mission_blocks:
return results

for block in mission_blocks:
if 'state #FOREGROUND' in block:
bundle_name_match = re.search(r'bundle name \[(.*?)\]', block)
main_name_match = re.search(r'main name \[(.*?)\]', block)
if bundle_name_match and main_name_match:
package_name = bundle_name_match.group(1)
page_name = main_name_match.group(1)
results.append((package_name, page_name))

return results

data: CommandResult = self.shell("aa dump -l")
output = data.output
results = __extract_info(output)
return results[0] if results else (None, None)
return _parse_current_app(data.output)

def wakeup(self):
self.shell("power-shell wakeup")
Expand Down
75 changes: 75 additions & 0 deletions tests/test_current_app.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-

from hmdriver2.hdc import HdcWrapper, _MISSION_BLOCK_RE, _parse_current_app
from hmdriver2.proto import CommandResult


# Sample from https://github.com/codematrixer/hmdriver2/issues/52
# photos is BACKGROUND; yang is FOREGROUND. current_app must return yang.
ISSUE_52_DUMP = """User ID #100
current mission lists:{
Mission ID #275 mission name #[#com.huawei.hmos.photos:phone_photos:com.huawei.hmos.photos.MainAbility] lockedState #0 mission affinity #[]
AbilityRecord ID #6842
app name [com.huawei.hmos.photos]
main name [com.huawei.hmos.photos.MainAbility]
bundle name [com.huawei.hmos.photos]
ability type [PAGE]
state #BACKGROUND start time [864068282]
app state #BACKGROUND
ready #1 window attached #0 launcher #0
callee connections:
isKeepAlive: false
Mission ID #298 mission name #[#com.xiaotuo.yang:entry:EntryAbility] lockedState #0 mission affinity #[]
AbilityRecord ID #7051
app name [com.xiaotuo.yang]
main name [EntryAbility]
bundle name [com.xiaotuo.yang]
ability type [PAGE]
state #FOREGROUND start time [892956626]
app state #FOREGROUND
ready #1 window attached #0 launcher #0
callee connections:
isKeepAlive: false
}
"""

PHOTOS = ("com.huawei.hmos.photos", "com.huawei.hmos.photos.MainAbility")
YANG = ("com.xiaotuo.yang", "EntryAbility")


def _hdc_with_dump(output: str) -> HdcWrapper:
"""Build an HdcWrapper that never talks to a device."""
hdc = HdcWrapper.__new__(HdcWrapper)
hdc.shell = lambda cmd, error_raise=True: CommandResult(output, "", 0)
return hdc


def test_mission_regex_matches_one_block_each():
blocks = _MISSION_BLOCK_RE.findall(ISSUE_52_DUMP)
assert len(blocks) == 2
assert "com.huawei.hmos.photos" in blocks[0]
assert "com.xiaotuo.yang" in blocks[1]
assert "com.xiaotuo.yang" not in blocks[0]


def test_parse_issue_52_dump_prefers_foreground_yang():
assert _parse_current_app(ISSUE_52_DUMP) == YANG
assert _parse_current_app(ISSUE_52_DUMP) != PHOTOS


def test_current_app_issue_52_dump_returns_yang_not_photos():
hdc = _hdc_with_dump(ISSUE_52_DUMP)
assert hdc.current_app() == YANG
assert hdc.current_app() != PHOTOS


def test_parse_empty_dump():
assert _parse_current_app("") == (None, None)
assert _parse_current_app("User ID #100\n current mission lists:{\n }\n") == (None, None)


def test_parse_background_only():
dump = ISSUE_52_DUMP.replace("state #FOREGROUND", "state #BACKGROUND").replace(
"app state #FOREGROUND", "app state #BACKGROUND"
)
assert _parse_current_app(dump) == (None, None)