test: e2e regression suite for SAE structural fast-paths

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 = '<node package="com.android.systemui" />'
+        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 = '''
+        <node package="com.instagram.android">
+            <node resource-id="android:id/dialog" />
+            <node text="Action Blocked" />
+            <node text="Try again later to protect our community." />
+        </node>
+        '''
+        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 = '''
+        <node package="com.google.android.permissioncontroller">
+            <node text="Allow Instagram to record audio?" />
+        </node>
+        '''
+        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 = '''
+        <node package="com.instagram.android">
+            <node resource-id="com.instagram.android:id/quick_capture_button" />
+        </node>
+        '''
+        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 = '''
+        <node package="com.instagram.android">
+            <node resource-id="com.instagram.android:id/gallery_cancel_button" />
+        </node>
+        '''
+        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 = '''
+        <?xml version="1.0" encoding="UTF-8"?>
+        <hierarchy rotation="0">
+            <node index="0" text="" resource-id="" class="android.widget.FrameLayout" package="com.instagram.android" content-desc="" checkable="false" checked="false" clickable="false" enabled="true" focusable="false" focused="false" scrollable="false" long-clickable="false" password="false" selected="false" bounds="[0,0][1080,2400]">
+                <node text="Following" resource-id="com.instagram.android:id/follow_button" clickable="true" bounds="[10,10][100,50]" />
+                <node content-desc="Profile Picture" clickable="true" />
+            </node>
+        </hierarchy>
+        '''
+        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 = '<<node package="com.instagram.android" text="Hello" />'
+        compressed = sae._compress_xml(broken_xml)
+        assert "PACKAGES: com.instagram.android" in compressed
+        assert "TEXTS: Hello" in compressed