mmintel/instagram-bot
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat: complete modular plugin refactor with 100% E2E coverage for interactions

Browse Source
This commit is contained in:
Marc Mintel
2026-04-25 20:58:07 +02:00
parent 77e8251aa7
commit 144d6401b5
232 changed files with 66259 additions and 5410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
tests/chaos/__init__.py
View File

@@ -1,6 +1,7 @@
"""
Shared fixtures and utilities for chaos engineering tests.
"""
import pytest
@@ -23,26 +24,19 @@ def generate_corrupted_xml(corruption_type: str) -> str:
'enabled="true" focusable="true" focused="false" scrollable="false" '
'long-clickable="false" password="false" selected="false" '
'bounds="[50,500][150,600]" />'
'</node>'
'</hierarchy>'
"</node>"
"</hierarchy>"
)
generators = {
"EMPTY_STRING": lambda: "",
"NONE_VALUE": lambda: None,
"TRUNCATED_MID_TAG": lambda: base_valid[:len(base_valid) // 2],
"UNICODE_INJECTION": lambda: base_valid.replace(
'text="Like"',
'text="L̵̡̧̢̛̛̛̘̗̣̥̱̲̲̝̪̣̗̝̠̫̲̤̱̪̞̻̙̜̺̩̰̫̝̥̩̭̩̫̦̠̦̣̣̬̤̤̠̗̣̲̬̟̣̰̝̥̤̜̻̫̙̥̘̻̝̯̗̼̣̮̲̻̝̹̩̗̥̖̝̝̪̣̜̜̱̣̱̻̮̬̮̬̗̖̟̩̭̜̀̀̈̀̀̀̑́̀̀̆̈́̐̑̈̈́̈́̉̿̈̉̆̂̃̉̆̉̑̉̈̊̏̀̒̌̽̈́̃̓̏̏͋̾̈́́̄̊̈́̽̅̒̓̈̈́̆̈̐̓̋̏̃͑̋̊̅̿̌̇̎̀̀̀̕̕̕͘̕̕̕̕̕͘͜͝͝i̷ke"'
),
"TRUNCATED_MID_TAG": lambda: base_valid[: len(base_valid) // 2],
"UNICODE_INJECTION": lambda: base_valid.replace('text="Like"', 'text="L̵̡̧̢̛̛̛̘̗̣̥̱̲̲̝̪̣̗̝̠̫̲̤̱̪̞̻̙̜̺̩̰̫̝̥̩̭̩̫̦̠̦̣̣̬̤̤̠̗̣̲̬̟̣̰̝̥̤̜̻̫̙̥̘̻̝̯̗̼̣̮̲̻̝̹̩̗̥̖̝̝̪̣̜̜̱̣̱̻̮̬̮̬̗̖̟̩̭̜̀̀̈̀̀̀̑́̀̀̆̈́̐̑̈̈́̈́̉̿̈̉̆̂̃̉̆̉̑̉̈̊̏̀̒̌̽̈́̃̓̏̏͋̾̈́́̄̊̈́̽̅̒̓̈̈́̆̈̐̓̋̏̃͑̋̊̅̿̌̇̎̀̀̀̕̕̕͘̕̕̕̕̕͘͜͝͝i̷ke"'),
"MASSIVE_DOM_10K_NODES": lambda: _generate_massive_dom(10000),
"ZERO_SIZE_BOUNDS": lambda: base_valid.replace(
'bounds="[50,500][150,600]"',
'bounds="[500,500][500,500]"'
),
"ZERO_SIZE_BOUNDS": lambda: base_valid.replace('bounds="[50,500][150,600]"', 'bounds="[500,500][500,500]"'),
"NEGATIVE_COORDINATES": lambda: base_valid.replace(
'bounds="[50,500][150,600]"',
'bounds="[-100,-200][50,100]"'
'bounds="[50,500][150,600]"', 'bounds="[-100,-200][50,100]"'
),
"MISSING_CLOSING_TAGS": lambda: (
'<hierarchy rotation="0">'
@@ -52,17 +46,11 @@ def generate_corrupted_xml(corruption_type: str) -> str:
),
"RECURSIVE_NESTING_500_DEEP": lambda: _generate_deep_nesting(500),
"NULL_BYTES": lambda: base_valid.replace("Like", "Li\x00ke\x00"),
"MALFORMED_BOUNDS": lambda: base_valid.replace(
'bounds="[50,500][150,600]"',
'bounds="NOT_A_BOUND"'
),
"MALFORMED_BOUNDS": lambda: base_valid.replace('bounds="[50,500][150,600]"', 'bounds="NOT_A_BOUND"'),
"ONLY_WHITESPACE": lambda: " \n\t\n ",
"HTML_NOT_XML": lambda: "<html><body><div>Not XML at all</div></body></html>",
"BINARY_GARBAGE": lambda: bytes(range(256)).decode("latin-1"),
"EXTREMELY_LONG_TEXT": lambda: base_valid.replace(
'text="Like"',
f'text="{"A" * 100000}"'
),
"EXTREMELY_LONG_TEXT": lambda: base_valid.replace('text="Like"', f'text="{"A" * 100000}"'),
}
generator = generators.get(corruption_type)
@@ -82,7 +70,7 @@ def _generate_massive_dom(count: int) -> str:
f'package="com.instagram.android" '
f'clickable="true" bounds="[0,{i}][100,{i+50}]" />'
)
parts.append('</hierarchy>')
parts.append("</hierarchy>")
return "".join(parts)
@@ -91,11 +79,11 @@ def _generate_deep_nesting(depth: int) -> str:
xml = '<hierarchy rotation="0">'
for i in range(depth):
xml += f'<node index="{i}" text="level_{i}" class="android.widget.FrameLayout" '
xml += f'package="com.instagram.android" bounds="[0,0][1080,2400]">'
xml += 'package="com.instagram.android" bounds="[0,0][1080,2400]">'
# Close all tags
for _ in range(depth):
xml += '</node>'
xml += '</hierarchy>'
xml += "</node>"
xml += "</hierarchy>"
return xml
@@ -120,6 +108,6 @@ VALID_FEED_XML = (
'<node index="4" text="" resource-id="com.instagram.android:id/search_tab" '
'class="android.widget.ImageView" package="com.instagram.android" '
'content-desc="Search and explore" clickable="true" bounds="[216,2300][432,2400]" />'
'</node>'
'</hierarchy>'
"</node>"
"</hierarchy>"
)

102
tests/chaos/test_chaos_network.py
View File

@@ -6,24 +6,33 @@ Verifies that the bot degrades gracefully when external services
Tesla's FSD doesn't crash if the map server is unreachable — neither should we.
"""
import pytest
from unittest.mock import MagicMock, patch, PropertyMock
from tests.chaos import VALID_FEED_XML
from unittest.mock import MagicMock, patch
import pytest
from tests.chaos import VALID_FEED_XML
# ──────────────────────────────────────────────────
# Qdrant Failure Tests
# ──────────────────────────────────────────────────
@pytest.mark.chaos
class TestQdrantFailure:
"""Bot must survive total Qdrant outage."""
def test_telepathic_works_without_qdrant(self):
"""TelepathicEngine must still resolve nodes via keyword fast-path when Qdrant is down."""
with patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None), \
patch("GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)):
with (
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
patch(
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
new_callable=lambda: property(lambda self: False),
),
):
from GramAddict.core.telepathic_engine import TelepathicEngine
TelepathicEngine._instance = None
engine = TelepathicEngine.__new__(TelepathicEngine)
engine.ui_memory = MagicMock()
@@ -42,37 +51,54 @@ class TestQdrantFailure:
def test_sae_recall_returns_none_without_qdrant(self):
"""SAE episodic memory must return None (not crash) when Qdrant is down."""
with patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None), \
patch("GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)):
with (
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
patch(
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
new_callable=lambda: property(lambda self: False),
),
):
from GramAddict.core.situational_awareness import SituationEpisodeDB
db = SituationEpisodeDB()
db._db = MagicMock()
db._db.is_connected = False
result = db.recall("test_situation_signature")
assert result is None
def test_sae_learn_silently_fails_without_qdrant(self):
"""SAE learning must silently skip (not crash) when Qdrant is down."""
with patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None), \
patch("GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)):
from GramAddict.core.situational_awareness import SituationEpisodeDB, EscapeAction
with (
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
patch(
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
new_callable=lambda: property(lambda self: False),
),
):
from GramAddict.core.situational_awareness import EscapeAction, SituationEpisodeDB
db = SituationEpisodeDB()
db._db = MagicMock()
db._db.is_connected = False
action = EscapeAction("back", reason="test")
# Must not raise
db.learn("test_signature", action, True)
def test_qdrant_timeout_doesnt_hang_extraction(self):
"""If Qdrant queries time out, node extraction must still complete."""
import time
with patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None), \
patch("GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)):
with (
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
patch(
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
new_callable=lambda: property(lambda self: False),
),
):
from GramAddict.core.telepathic_engine import TelepathicEngine
TelepathicEngine._instance = None
engine = TelepathicEngine.__new__(TelepathicEngine)
engine.ui_memory = MagicMock()
@@ -83,11 +109,11 @@ class TestQdrantFailure:
engine.positive_memory.recall = MagicMock(side_effect=TimeoutError("Qdrant timeout"))
engine._edge_model = None
engine._edge_tokenizer = None
start = time.time()
nodes = engine._extract_semantic_nodes(VALID_FEED_XML)
elapsed = time.time() - start
assert elapsed < 5.0
assert isinstance(nodes, list)
TelepathicEngine._instance = None
@@ -97,6 +123,7 @@ class TestQdrantFailure:
# LLM (Ollama/OpenRouter) Failure Tests
# ──────────────────────────────────────────────────
@pytest.mark.chaos
class TestLLMFailure:
"""Bot must survive LLM outages."""
@@ -104,48 +131,48 @@ class TestLLMFailure:
def test_sae_perceive_defaults_to_normal_on_llm_failure(self):
"""If LLM classification fails, SAE must default to NORMAL (safe fallback)."""
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
SituationalAwarenessEngine.reset()
device = MagicMock()
device.app_id = "com.instagram.android"
device.deviceV2 = MagicMock()
device.deviceV2.info = {"screenOn": True}
sae = SituationalAwarenessEngine(device)
sae.episodes = MagicMock()
sae.episodes.recall = MagicMock(return_value=None)
with patch("GramAddict.core.qdrant_memory.ScreenMemoryDB") as MockScreenDB:
mock_screen_db = MagicMock()
mock_screen_db.get_screen_type = MagicMock(return_value=None)
MockScreenDB.return_value = mock_screen_db
with patch("GramAddict.core.llm_provider.query_telepathic_llm", side_effect=ConnectionError("Ollama down")):
result = sae.perceive(VALID_FEED_XML)
# Must default to NORMAL, not crash
assert result == SituationType.NORMAL
SituationalAwarenessEngine.reset()
def test_sae_escape_planning_defaults_to_back_on_llm_failure(self):
"""If LLM escape planning fails, SAE must default to BACK press."""
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
SituationalAwarenessEngine.reset()
device = MagicMock()
device.app_id = "com.instagram.android"
device.deviceV2 = MagicMock()
device.deviceV2.info = {"screenOn": True}
sae = SituationalAwarenessEngine(device)
with patch("GramAddict.core.llm_provider.query_llm", side_effect=ConnectionError("LLM down")):
action = sae._plan_escape_via_llm(
VALID_FEED_XML, "compressed_sig", SituationType.OBSTACLE_MODAL
)
action = sae._plan_escape_via_llm(VALID_FEED_XML, "compressed_sig", SituationType.OBSTACLE_MODAL)
assert action.action_type == "back"
assert "failed" in action.reason.lower() or "default" in action.reason.lower()
SituationalAwarenessEngine.reset()
@@ -153,6 +180,7 @@ class TestLLMFailure:
# Active Inference Resilience
# ──────────────────────────────────────────────────
@pytest.mark.chaos
class TestActiveInferenceChaos:
"""Active Inference engine must survive edge cases."""
@@ -160,35 +188,39 @@ class TestActiveInferenceChaos:
def test_evaluate_with_empty_history(self):
"""Evaluating without any predictions must return True (no-op)."""
from GramAddict.core.active_inference import ActiveInferenceEngine
ai = ActiveInferenceEngine("test_user")
assert ai.evaluate_prediction("<hierarchy/>") is True
def test_extreme_free_energy_doesnt_overflow(self):
"""Repeated errors must not cause float overflow."""
from GramAddict.core.active_inference import ActiveInferenceEngine
ai = ActiveInferenceEngine("test_user")
for _ in range(1000):
ai.predict_state(["nonexistent_element"])
ai.evaluate_prediction("<hierarchy><node text='wrong'/></hierarchy>")
assert ai.free_energy < float('inf')
assert ai.free_energy < float("inf")
assert ai.free_energy >= 0
def test_surprise_with_identical_prediction_is_zero(self):
"""Perfect prediction (predicted == observed) must produce near-zero surprise."""
from GramAddict.core.active_inference import ActiveInferenceEngine
ai = ActiveInferenceEngine("test_user")
ai.free_energy = 0.0
result = ai.calculate_surprise(1.0, 1.0)
assert result < 0.1 # Near-zero free energy
def test_sleep_modifier_bounds(self):
"""Sleep modifier must always be between 1.0 and 5.0."""
from GramAddict.core.active_inference import ActiveInferenceEngine
ai = ActiveInferenceEngine("test_user")
for policy in ["STABLE", "CAUTIOUS", "DORMANT"]:
ai.policy = policy
mod = ai.get_sleep_modifier()

91
tests/chaos/test_chaos_xml_corruption.py
View File

@@ -8,22 +8,30 @@ or empty lists) without raising unhandled exceptions.
These tests are the "crash barrier" of autonomous navigation — ensuring that
no matter what Android dumps to us, the bot survives and recovers.
"""
import pytest
import time
from unittest.mock import MagicMock, patch
from tests.chaos import generate_corrupted_xml
import pytest
from tests.chaos import generate_corrupted_xml
# ──────────────────────────────────────────────────
# Telepathic Engine Chaos Tests
# ──────────────────────────────────────────────────
@pytest.fixture
def telepathic_engine():
"""Creates a real TelepathicEngine instance with mocked Qdrant."""
with patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None), \
patch("GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)):
with (
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
patch(
"GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)
),
):
from GramAddict.core.telepathic_engine import TelepathicEngine
TelepathicEngine._instance = None
engine = TelepathicEngine.__new__(TelepathicEngine)
engine.ui_memory = MagicMock()
@@ -65,30 +73,35 @@ class TestTelepathicEngineChaos:
def test_extract_semantic_nodes_survives(self, telepathic_engine, corruption_type):
"""Engine's XML parser must return empty list on any corruption."""
xml = generate_corrupted_xml(corruption_type)
# Must NOT raise. May return empty list.
if xml is None:
# None input — directly test defense
result = telepathic_engine._extract_semantic_nodes("")
else:
result = telepathic_engine._extract_semantic_nodes(xml)
assert isinstance(result, list)
@pytest.mark.parametrize("corruption_type", [
"EMPTY_STRING", "NONE_VALUE", "TRUNCATED_MID_TAG",
"MISSING_CLOSING_TAGS", "ONLY_WHITESPACE", "HTML_NOT_XML",
"BINARY_GARBAGE",
])
@pytest.mark.parametrize(
"corruption_type",
[
"EMPTY_STRING",
"NONE_VALUE",
"TRUNCATED_MID_TAG",
"MISSING_CLOSING_TAGS",
"ONLY_WHITESPACE",
"HTML_NOT_XML",
"BINARY_GARBAGE",
],
)
def test_find_best_node_survives_garbage(self, telepathic_engine, corruption_type):
"""find_best_node must return None on garbage XML, never crash."""
xml = generate_corrupted_xml(corruption_type)
if xml is None:
xml = ""
result = telepathic_engine._find_best_node_inner(
xml, "tap like button", min_confidence=0.82
)
result = telepathic_engine._find_best_node_inner(xml, "tap like button", min_confidence=0.82)
# Must be None or a dict, never an exception
assert result is None or isinstance(result, dict)
@@ -109,7 +122,7 @@ class TestTelepathicEngineChaos:
start = time.time()
nodes = telepathic_engine._extract_semantic_nodes(xml)
elapsed = time.time() - start
assert elapsed < 5.0, f"Parsing 10K nodes took {elapsed:.2f}s (limit: 5s)"
assert isinstance(nodes, list)
@@ -117,7 +130,7 @@ class TestTelepathicEngineChaos:
"""500 levels of nesting must not cause stack overflow."""
xml = generate_corrupted_xml("RECURSIVE_NESTING_500_DEEP")
# This would crash Python's default recursion limit (1000) if
# we used recursive parsing. ElementTree uses iterative parsing,
# we used recursive parsing. ElementTree uses iterative parsing,
# so it should survive.
nodes = telepathic_engine._extract_semantic_nodes(xml)
assert isinstance(nodes, list)
@@ -136,24 +149,26 @@ class TestTelepathicEngineChaos:
# SAE (Situational Awareness Engine) Chaos Tests
# ──────────────────────────────────────────────────
@pytest.fixture
def sae_engine():
"""Creates a SAE instance with mocked device."""
from GramAddict.core.situational_awareness import SituationalAwarenessEngine
SituationalAwarenessEngine.reset()
device = MagicMock()
device.app_id = "com.instagram.android"
device.deviceV2 = MagicMock()
device.deviceV2.info = {"screenOn": True}
engine = SituationalAwarenessEngine(device)
# Mock the episode DB to avoid Qdrant dependency
engine.episodes = MagicMock()
engine.episodes.recall = MagicMock(return_value=None)
engine.episodes.learn = MagicMock()
yield engine
SituationalAwarenessEngine.reset()
@@ -162,16 +177,23 @@ def sae_engine():
class TestSAEChaos:
"""SAE perception must be bulletproof against XML corruption."""
@pytest.mark.parametrize("corruption_type", [
"EMPTY_STRING", "TRUNCATED_MID_TAG", "MISSING_CLOSING_TAGS",
"ONLY_WHITESPACE", "HTML_NOT_XML", "BINARY_GARBAGE",
])
@pytest.mark.parametrize(
"corruption_type",
[
"EMPTY_STRING",
"TRUNCATED_MID_TAG",
"MISSING_CLOSING_TAGS",
"ONLY_WHITESPACE",
"HTML_NOT_XML",
"BINARY_GARBAGE",
],
)
def test_compress_xml_survives_garbage(self, sae_engine, corruption_type):
"""XML compression must never crash, even on garbage."""
xml = generate_corrupted_xml(corruption_type)
if xml is None:
xml = ""
result = sae_engine._compress_xml(xml)
assert isinstance(result, str)
assert len(result) > 0 # Should always return something
@@ -181,16 +203,23 @@ class TestSAEChaos:
assert sae_engine._compress_xml("") == "EMPTY_SCREEN"
assert sae_engine._compress_xml(None) == "EMPTY_SCREEN"
@pytest.mark.parametrize("corruption_type", [
"EMPTY_STRING", "TRUNCATED_MID_TAG", "BINARY_GARBAGE", "ONLY_WHITESPACE",
])
@pytest.mark.parametrize(
"corruption_type",
[
"EMPTY_STRING",
"TRUNCATED_MID_TAG",
"BINARY_GARBAGE",
"ONLY_WHITESPACE",
],
)
def test_perceive_survives_garbage(self, sae_engine, corruption_type):
"""perceive() must return a valid SituationType on any input."""
from GramAddict.core.situational_awareness import SituationType
xml = generate_corrupted_xml(corruption_type)
if xml is None:
xml = ""
result = sae_engine.perceive(xml)
assert isinstance(result, SituationType)
@@ -208,6 +237,6 @@ class TestSAEChaos:
start = time.time()
result = sae_engine._compress_xml(xml)
elapsed = time.time() - start
assert len(result) <= 3000, f"Compressed output is {len(result)} chars (limit: 3000)"
assert elapsed < 5.0, f"Compression took {elapsed:.2f}s"