diff --git a/tests/e2e/test_system_sae.py b/tests/e2e/test_system_sae.py new file mode 100644 index 0000000..b520c34 --- /dev/null +++ b/tests/e2e/test_system_sae.py @@ -0,0 +1,136 @@ +""" +E2E: Situational Awareness Engine (SAE) Integrity +================================================= +Tests the deterministic perception and planning logic of the SAE. +Ensures that the bot correctly identifies obstacles via structural fast-paths +without relying on brittle mock LLM responses. +""" + +import pytest + +from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType + + +class DummyDeviceInfo: + def __init__(self, screen_on=True): + self.info = {"screenOn": screen_on} + + +class DummyDeviceV2: + def __init__(self, screen_on=True): + self.info = {"screenOn": screen_on} + + +class DummyDevice: + def __init__(self, screen_on=True): + self.app_id = "com.instagram.android" + self.deviceV2 = DummyDeviceV2(screen_on) + + +@pytest.fixture +def sae(): + SituationalAwarenessEngine.reset() + device = DummyDevice() + engine = SituationalAwarenessEngine.get_instance(device) + # Clear the global ScreenMemoryDB so tests don't pollute each other + from GramAddict.core.qdrant_memory import ScreenMemoryDB + db = ScreenMemoryDB() + if db.is_connected: + try: + db.client.delete( + collection_name=db.collection_name, + points_selector={"filter": {}} # Delete all points + ) + except Exception: + pass + return engine + + +class TestSAEPerception: + """Tests the perception fast-paths in SAE.""" + + def test_perceive_empty_xml_is_foreign_app(self, sae): + """Empty or invalid XML should be considered a foreign app / unknown state.""" + assert sae.perceive("") == SituationType.OBSTACLE_FOREIGN_APP + assert sae.perceive(None) == SituationType.OBSTACLE_FOREIGN_APP + + def test_perceive_locked_screen(self, sae): + """If the hardware reports the screen is off, it must be LOCKED_SCREEN.""" + sae.device.deviceV2.info["screenOn"] = False + xml_dump = '' + assert sae.perceive(xml_dump) == SituationType.OBSTACLE_LOCKED_SCREEN + + def test_perceive_action_blocked(self, sae): + """Critical account safety check: 'Action Blocked' must return DANGER_ACTION_BLOCKED.""" + xml_dump = ''' + + + + + + ''' + assert sae.perceive(xml_dump) == SituationType.DANGER_ACTION_BLOCKED + + def test_perceive_system_dialog(self, sae): + """System permission dialogs must be identified structurally.""" + xml_dump = ''' + + + + ''' + assert sae.perceive(xml_dump) == SituationType.OBSTACLE_SYSTEM + + def test_perceive_creation_flow_modal(self, sae): + """ + Creation flows (camera, story gallery) are inside Instagram but block navigation. + They must be caught by structural markers. + """ + xml_dump = ''' + + + + ''' + assert sae.perceive(xml_dump) == SituationType.OBSTACLE_MODAL + + def test_perceive_story_gallery_cancel_modal(self, sae): + """Story gallery cancel button blocks navigation.""" + xml_dump = ''' + + + + ''' + assert sae.perceive(xml_dump) == SituationType.OBSTACLE_MODAL + + +class TestSAECompression: + """Tests the XML compression logic to ensure signatures are compact but retain identity.""" + + def test_compress_strips_noise_keeps_important_data(self, sae): + xml_dump = ''' + + + + + + + + ''' + compressed = sae._compress_xml(xml_dump) + + # Must extract packages + assert "PACKAGES: com.instagram.android" in compressed + # Must extract resource-ids without the package prefix + assert "id=follow_button" in compressed + # Must extract text + assert "text='Following'" in compressed + # Must extract content-desc + assert "desc='Profile Picture'" in compressed + # Must preserve CLICKABLE + assert "CLICKABLE" in compressed + + def test_compress_handles_broken_xml(self, sae): + """If XML is malformed, it should use regex fallback.""" + broken_xml = '<' + compressed = sae._compress_xml(broken_xml) + assert "PACKAGES: com.instagram.android" in compressed + assert "TEXTS: Hello" in compressed