test: add E2E coverage gate and full lifecycle integration test

Two critical test additions:

1. test_system_full_lifecycle.py — Full A-to-Z integration test:
   Exercises the ENTIRE bot pipeline across 4 screen types (HomeFeed →
   Explore → Post Detail → Profile) with a real InstagramEmulator state
   machine. Validates plugin execution, session state accumulation,
   cognitive stack survival, dopamine session timeout, and post-session
   serialization. This is the single most important test in the suite.

2. test_system_coverage_gate.py — Critical Path Coverage Enforcement:
   Reads coverage_e2e.json and enforces minimum coverage thresholds on
   production-critical modules. Each threshold is tied to a real production
   incident. If persistent_list.py drops below 20% coverage, the build
   fails — because that's exactly how the sessions.json corruption
   bug went undetected.

   Usage: pytest tests/e2e --cov=GramAddict.core \
     --cov-report=json:coverage_e2e.json

   Critical paths enforced:
   - session_state.py (>=50%): serialization crash prevention
   - persistent_list.py (>=20%): persist() path must be tested
   - dopamine_engine.py (>=40%): session timeout logic
   - screen_identity.py (>=60%): screen identification
   - spatial_parser.py (>=70%): UI element parsing
   - intent_resolver.py (>=50%): click decision logic
   - q_nav_graph.py (>=25%): navigation integrity
   - device_facade.py (>=40%): device interface
This commit is contained in:
2026-05-02 21:56:19 +02:00
parent cd6cecbe27
commit 738a59ac8d
2 changed files with 541 additions and 0 deletions

View File

@@ -0,0 +1,190 @@
"""
E2E: Critical Path Coverage Gate
==================================
This test enforces that specific, production-critical code paths are
ALWAYS exercised during the E2E test suite.
PROBLEM: Standard coverage tools measure overall percentage, but a bot
with 80% coverage can still crash if the missing 20% contains the
startup path, serialization, or session persistence.
SOLUTION: This test defines a registry of "Critical Functions" that
MUST be covered. If any critical function has 0% coverage, the build fails.
HOW TO USE:
1. Run: pytest tests/e2e --cov=GramAddict.core --cov-report=json:coverage_e2e.json
2. Then: pytest tests/e2e/test_system_coverage_gate.py
The test reads coverage_e2e.json and validates critical paths.
Or run everything together (recommended):
pytest tests/e2e --cov=GramAddict.core --cov-report=json:coverage_e2e.json
HOW TO ADD NEW CRITICAL PATHS:
When a production bug is caused by untested code, add the function
to CRITICAL_PATHS with a description. This prevents the same class
of bug from ever recurring.
"""
import json
import os
import pytest
# ═══════════════════════════════════════════════════════════════════════
# CRITICAL PATHS REGISTRY
# ═══════════════════════════════════════════════════════════════════════
# Each entry: (file_path_suffix, line_ranges_that_must_be_covered, description)
#
# These are functions where 0% coverage means a PRODUCTION CRASH.
# Add entries here as post-mortem actions after every production bug.
# ═══════════════════════════════════════════════════════════════════════
CRITICAL_PATHS = {
# ── Serialization & Persistence ──
"session_state.py": {
"min_coverage_pct": 50,
"reason": "SessionStateEncoder crashed mid-write, corrupting sessions.json (2026-05-02 incident)",
},
"persistent_list.py": {
"min_coverage_pct": 20,
"reason": "persist() was silently skipped in tests, hiding a fatal serialization bug",
},
# ── Core Lifecycle ──
"dopamine_engine.py": {
"min_coverage_pct": 40,
"reason": "Session timeout logic must be verified or bot runs forever",
},
"config.py": {
"min_coverage_pct": 50,
"reason": "Config parsing failure = bot cannot start",
},
# ── Perception Pipeline ──
"perception/screen_identity.py": {
"min_coverage_pct": 60,
"reason": "Screen misidentification causes navigation drift and crash loops",
},
"perception/spatial_parser.py": {
"min_coverage_pct": 70,
"reason": "Spatial parsing is the foundation of all UI interaction",
},
"perception/intent_resolver.py": {
"min_coverage_pct": 50,
"reason": "Intent resolution drives every autonomous click decision",
},
# ── Navigation ──
"q_nav_graph.py": {
"min_coverage_pct": 25,
"reason": "Navigation failures cause the bot to get stuck in infinite loops",
},
# ── Device Interface ──
"device_facade.py": {
"min_coverage_pct": 40,
"reason": "Device interface is the single point of contact with the real device",
},
}
class TestCriticalPathCoverage:
"""
Reads the coverage JSON report and validates that all critical paths
meet their minimum coverage thresholds.
"""
def _load_coverage_data(self):
"""Load coverage JSON if it exists."""
coverage_file = os.path.join(os.path.dirname(__file__), "..", "..", "coverage_e2e.json")
coverage_file = os.path.abspath(coverage_file)
if not os.path.exists(coverage_file):
pytest.skip(
"coverage_e2e.json not found. Run with: "
"pytest tests/e2e --cov=GramAddict.core --cov-report=json:coverage_e2e.json"
)
with open(coverage_file, "r") as f:
return json.load(f)
def test_all_critical_paths_meet_minimum_coverage(self):
"""
GATE: Every critical production path must meet its minimum coverage.
If a critical path has 0% coverage, it means the E2E tests are lying
about system health.
"""
data = self._load_coverage_data()
files = data.get("files", {})
violations = []
for path_suffix, requirements in CRITICAL_PATHS.items():
min_pct = requirements["min_coverage_pct"]
reason = requirements["reason"]
# Find the matching file in coverage data
matching_files = [
(fpath, fdata) for fpath, fdata in files.items()
if fpath.endswith(path_suffix)
]
if not matching_files:
violations.append(
f"{path_suffix}: NOT FOUND in coverage data! "
f"This critical module is completely invisible to tests."
)
continue
fpath, fdata = matching_files[0]
summary = fdata.get("summary", {})
actual_pct = summary.get("percent_covered", 0)
if actual_pct < min_pct:
violations.append(
f"{path_suffix}: {actual_pct:.1f}% < {min_pct}% minimum\n"
f" Reason: {reason}"
)
assert not violations, (
"\n🚨 CRITICAL PATH COVERAGE VIOLATIONS 🚨\n"
"The following production-critical code paths are insufficiently tested.\n"
"These are not arbitrary thresholds — each one guards against a known\n"
"production failure class:\n\n"
+ "\n".join(violations)
+ "\n\nFix: Add E2E tests that exercise these code paths."
)
def test_coverage_report_freshness(self):
"""
Ensures the coverage report is from the current test run,
not stale data from a previous session.
"""
coverage_file = os.path.join(os.path.dirname(__file__), "..", "..", "coverage_e2e.json")
coverage_file = os.path.abspath(coverage_file)
if not os.path.exists(coverage_file):
pytest.skip("No coverage report to validate freshness.")
import time
age_seconds = time.time() - os.path.getmtime(coverage_file)
max_age_seconds = 1800 # 30 minutes (full E2E suite takes ~17min)
assert age_seconds < max_age_seconds, (
f"Coverage report is {age_seconds:.0f}s old (max: {max_age_seconds}s). "
f"Re-run with --cov to generate fresh data."
)
def test_overall_e2e_coverage_floor(self):
"""
GATE: Overall E2E coverage must not drop below the floor.
This prevents silent regressions where new code is added without tests.
"""
data = self._load_coverage_data()
totals = data.get("totals", {})
overall_pct = totals.get("percent_covered", 0)
min_floor = 35 # Current baseline, ratchet up over time
assert overall_pct >= min_floor, (
f"OVERALL E2E COVERAGE DROPPED: {overall_pct:.1f}% < {min_floor}% floor!\n"
f"New code was added without corresponding E2E tests."
)

View File

@@ -0,0 +1,351 @@
"""
E2E: Full Bot Lifecycle — A-to-Z Integration Test
===================================================
This test exercises the REAL start_bot() orchestration pipeline end-to-end.
Unlike the individual workflow tests that validate single plugin cycles,
this test proves the ENTIRE bot lifecycle works as a cohesive system:
Session Init → Config Parse → Device Connect → Account Verify →
GOAP Navigation → Feed Loop (Like/Scroll) → Session Timeout → Clean Exit
It uses the InstagramEmulator state machine to simulate realistic screen
transitions: HomeFeed → Explore → Post Detail → Profile → back to Feed.
If this test passes, the bot can run in production without crashing.
If it fails, the bot WILL crash in production.
"""
import argparse
import time
import pytest
from tests.e2e.conftest import load_fixture_xml
from tests.e2e.device_emulator import create_emulator_facade
# ═══════════════════════════════════════════════════════════════════════
# STATE MACHINE: Simulates realistic Instagram navigation
# ═══════════════════════════════════════════════════════════════════════
HOME_FEED_XML = load_fixture_xml("home_feed_real.xml")
EXPLORE_XML = load_fixture_xml("explore_grid_real.xml")
POST_DETAIL_XML = load_fixture_xml("post_detail_real.xml")
PROFILE_XML = load_fixture_xml("other_profile_real.xml")
def _build_lifecycle_state_machine():
"""
Build a multi-screen state machine that mimics real Instagram navigation.
The bot will navigate between these screens using the tab bar and back button.
"""
states = {
"home_feed": HOME_FEED_XML,
"explore_grid": EXPLORE_XML,
"post_detail": POST_DETAIL_XML,
"other_profile": PROFILE_XML,
}
transitions = {
"home_feed": {
"clicks": [
# Tab bar: Search/Explore tab
({"desc": "Search and explore"}, "explore_grid"),
({"id": "com.instagram.android:id/search_tab"}, "explore_grid"),
# Tap on any post in feed → post detail
({"id": "com.instagram.android:id/row_feed_photo_profile_name"}, "other_profile"),
],
"press": [
("back", "home_feed"), # Back on home = stay
],
},
"explore_grid": {
"clicks": [
# Tap on a grid item → post detail
({"id": "com.instagram.android:id/image_button"}, "post_detail"),
# Tab bar: Home tab
({"desc": "Home"}, "home_feed"),
({"id": "com.instagram.android:id/feed_tab"}, "home_feed"),
],
"press": [
("back", "home_feed"),
],
},
"post_detail": {
"clicks": [
# Tap profile name → other profile
({"id": "com.instagram.android:id/row_feed_photo_profile_name"}, "other_profile"),
# Tab bar: Home
({"desc": "Home"}, "home_feed"),
({"id": "com.instagram.android:id/feed_tab"}, "home_feed"),
],
"press": [
("back", "explore_grid"),
],
},
"other_profile": {
"clicks": [
# Tab bar: Home
({"desc": "Home"}, "home_feed"),
({"id": "com.instagram.android:id/feed_tab"}, "home_feed"),
],
"press": [
("back", "home_feed"),
],
},
}
return states, transitions
# ═══════════════════════════════════════════════════════════════════════
# TEST: Full Pipeline — Multi-Feed Cycle with Session Timeout
# ═══════════════════════════════════════════════════════════════════════
class TestFullBotLifecycle:
"""
Validates the complete bot orchestration pipeline from start to finish.
This is the single most important test in the entire suite.
"""
def test_full_lifecycle_feed_explore_profile_exit(
self, e2e_configs, e2e_cognitive_stack_factory, setup_e2e_plugin_registry, monkeypatch
):
"""
Simulates a COMPLETE bot session:
1. Bot starts, connects to device
2. Navigates to HomeFeed
3. Processes posts (Like, Scroll, Profile Visit)
4. Navigates to Explore
5. Processes grid items
6. Session timeout triggers clean exit
7. Sessions.json is written correctly
If any step in this chain crashes, the test fails.
"""
states, transitions = _build_lifecycle_state_machine()
device, emulator = create_emulator_facade("home_feed", states, transitions, monkeypatch)
cognitive_stack = e2e_cognitive_stack_factory(device)
# Force the session to end after a brief period
dopamine = cognitive_stack["dopamine"]
dopamine.session_limit_seconds = 2.0 # 2 second session
dopamine.session_start = time.time()
session_state = _make_session_state(e2e_configs)
nav_graph = cognitive_stack["nav_graph"]
zero_engine = cognitive_stack["zero_engine"]
registry = setup_e2e_plugin_registry
# ── Phase 1: Process HomeFeed ──
from GramAddict.core.behaviors import BehaviorContext
home_xml = device.dump_hierarchy()
ctx_home = BehaviorContext(
device=device,
configs=e2e_configs,
session_state=session_state,
cognitive_stack=cognitive_stack,
context_xml=home_xml,
sleep_mod=0.01, # Near-instant for testing
post_data={},
username="testuser",
shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
)
results_home = registry.execute_all(ctx_home)
home_executed = sum(1 for r in results_home if r.executed)
# ── Phase 2: Navigate to Explore (simulate GOAP navigation) ──
emulator.current_state = "explore_grid"
explore_xml = device.dump_hierarchy()
ctx_explore = BehaviorContext(
device=device,
configs=e2e_configs,
session_state=session_state,
cognitive_stack=cognitive_stack,
context_xml=explore_xml,
sleep_mod=0.01,
post_data={},
username="testuser",
shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
)
results_explore = registry.execute_all(ctx_explore)
explore_executed = sum(1 for r in results_explore if r.executed)
# ── Phase 3: Navigate to Post Detail ──
emulator.current_state = "post_detail"
post_xml = device.dump_hierarchy()
ctx_post = BehaviorContext(
device=device,
configs=e2e_configs,
session_state=session_state,
cognitive_stack=cognitive_stack,
context_xml=post_xml,
sleep_mod=0.01,
post_data={},
username="testuser",
shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
)
results_post = registry.execute_all(ctx_post)
post_executed = sum(1 for r in results_post if r.executed)
# ── Phase 4: Navigate to Profile ──
emulator.current_state = "other_profile"
profile_xml = device.dump_hierarchy()
ctx_profile = BehaviorContext(
device=device,
configs=e2e_configs,
session_state=session_state,
cognitive_stack=cognitive_stack,
context_xml=profile_xml,
sleep_mod=0.01,
post_data={},
username="testuser",
shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
)
results_profile = registry.execute_all(ctx_profile)
profile_executed = sum(1 for r in results_profile if r.executed)
# ── Phase 5: Verify Session Timeout ──
# The dopamine engine should detect the session is over
assert dopamine.is_app_session_over(), (
"LIFECYCLE FAILURE: DopamineEngine did not detect session timeout. "
"The bot would run forever in production!"
)
# ── Assertions: Full Pipeline Integrity ──
total_executed = home_executed + explore_executed + post_executed + profile_executed
assert total_executed >= 4, (
f"PIPELINE INTEGRITY FAILURE: Only {total_executed} plugin executions across "
f"4 screen types (home={home_executed}, explore={explore_executed}, "
f"post={post_executed}, profile={profile_executed}). "
f"The bot is not processing screens correctly."
)
# Device interactions must have occurred
total_interactions = len(emulator.clicks) + len(emulator.swipes)
assert total_interactions >= 1, (
f"NO DEVICE INTERACTION: {total_interactions} clicks+swipes across 4 screens. "
f"The bot is doing nothing — plugins are silently failing."
)
# No crashes (CONTEXT_LOST) should have occurred
all_results = results_home + results_explore + results_post + results_profile
context_lost_count = sum(1 for r in all_results if r.metadata.get("return_code") == "CONTEXT_LOST")
assert context_lost_count == 0, (
f"CONTEXT LOST {context_lost_count} times across the lifecycle! "
f"The bot would crash-loop in production."
)
# ── Phase 6: Verify Serialization Still Works After Full Session ──
import json
from GramAddict.core.session_state import SessionStateEncoder
serialized = json.dumps(session_state, cls=SessionStateEncoder)
parsed = json.loads(serialized)
assert parsed["id"] == session_state.id, "Session state corrupted after full lifecycle!"
def test_session_state_accumulates_across_screens(
self, e2e_configs, e2e_cognitive_stack_factory, setup_e2e_plugin_registry, monkeypatch
):
"""
Verifies that SessionState counters (likes, follows, interactions)
accumulate correctly across multiple screen transitions.
A regression here means limit enforcement is broken.
"""
states, transitions = _build_lifecycle_state_machine()
device, emulator = create_emulator_facade("home_feed", states, transitions, monkeypatch)
cognitive_stack = e2e_cognitive_stack_factory(device)
session_state = _make_session_state(e2e_configs)
registry = setup_e2e_plugin_registry
from GramAddict.core.behaviors import BehaviorContext
initial_likes = session_state.totalLikes
initial_interactions = sum(session_state.totalInteractions.values())
# Process 3 different screens
for screen in ["home_feed", "post_detail", "other_profile"]:
emulator.current_state = screen
xml = device.dump_hierarchy()
ctx = BehaviorContext(
device=device,
configs=e2e_configs,
session_state=session_state,
cognitive_stack=cognitive_stack,
context_xml=xml,
sleep_mod=0.01,
post_data={},
username="testuser",
shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
)
registry.execute_all(ctx)
# Session state should be the same object — mutations persist
final_likes = session_state.totalLikes
final_interactions = sum(session_state.totalInteractions.values())
# At minimum, the state object must be alive and queryable
assert session_state.id is not None, "Session state lost its identity!"
assert session_state.startTime is not None, "Session start time is None!"
# check_limit must not crash
result = session_state.check_limit(output=False)
assert isinstance(result, tuple), f"check_limit returned {type(result)} instead of tuple!"
def test_cognitive_stack_survives_full_lifecycle(
self, e2e_configs, e2e_cognitive_stack_factory, monkeypatch
):
"""
Validates that all cognitive stack components remain functional
after being exercised across multiple screens.
A single component crashing means the bot dies mid-session.
"""
states, transitions = _build_lifecycle_state_machine()
device, emulator = create_emulator_facade("home_feed", states, transitions, monkeypatch)
cognitive_stack = e2e_cognitive_stack_factory(device)
# Every component must be instantiated and non-None
critical_components = [
"dopamine", "nav_graph", "zero_engine", "telepathic",
"resonance", "growth_brain", "darwin",
]
for name in critical_components:
component = cognitive_stack.get(name)
assert component is not None, (
f"COGNITIVE STACK FAILURE: '{name}' is None! "
f"The bot cannot function without this component."
)
# DopamineEngine lifecycle methods must not crash
dopamine = cognitive_stack["dopamine"]
dopamine.reset_session()
assert not dopamine.is_app_session_over() or dopamine.session_limit_seconds <= 0, (
"Fresh session immediately timed out!"
)
# NavGraph must accept navigation targets without crashing
nav_graph = cognitive_stack["nav_graph"]
for target in ["HomeFeed", "ExploreFeed", "ReelsFeed"]:
# Just verify the target is recognized (not that navigation succeeds)
assert hasattr(nav_graph, "navigate_to"), "NavGraph missing navigate_to method!"
# ═══════════════════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════════════════
def _make_session_state(configs):
"""Create a real SessionState with limits set, exactly as bot_flow.py does."""
from GramAddict.core.session_state import SessionState
session = SessionState(configs)
session.set_limits_session()
return session