get e2e tests running

Author: cbrady

poetry.lock

@@ -61,6 +61,7 @@ python-dateutil = "^2.9.0"
 types-python-dateutil = "^2.9.0.20240316"
 ruff = "==0.11.5"
 aiohttp = {version = ">=3.9,<4", markers = "python_version >= \"3.9\""}
+redis = {version = ">=5.0,<6", markers = "python_version >= \"3.10\""}
 
 [tool.pytest.ini_options]
 testpaths = [ "tests" ]

pyproject.toml

@@ -331,9 +331,12 @@ async def get_flag(self, flag_key: str) -> Optional[RulesengineFlag]:
             if raw is None:
                 self._logger.debug("Flag cache miss for key: %s", cache_key)
                 return None
-            return _validate(RulesengineFlag, raw)
+            self._logger.debug("Flag cache hit for key: %s, validating...", cache_key)
+            result = _validate(RulesengineFlag, raw)
+            self._logger.debug("Flag validated successfully: %s", cache_key)
+            return result
         except Exception as exc:
-            self._logger.warning("Failed to retrieve flag from cache: %s", exc)
+            self._logger.warning("Failed to retrieve/validate flag from cache: %s (key=%s)", exc, cache_key)
             return None
 
     async def get_all_flags(self) -> None:

src/schematic/datastream/datastream_client.py

@@ -62,6 +62,7 @@ def __init__(self, *, wasm_path: Optional[str] = None) -> None:
         self._get_result_json_fn: Any = None
         self._get_result_json_length_fn: Any = None
         self._get_version_key_fn: Any = None
+        self._version_key: Optional[str] = None
 
     async def initialize(self) -> None:
         """Load and instantiate the WASM module. Safe to call multiple times."""
@@ -110,7 +111,10 @@ async def initialize(self) -> None:
             raise RuntimeError("WASM module does not export 'checkFlagCombined'")
 
         self._initialized = True
-        logger.debug("Rules engine WASM initialized (version: %s)", self.get_version_key())
+        # Cache the version key immediately — the WASM pointer is only stable
+        # before any other WASM calls mutate the linear memory.
+        self._version_key = self._read_version_key_from_wasm()
+        logger.debug("Rules engine WASM initialized (version: %s)", self._version_key)
 
     def is_initialized(self) -> bool:
         return self._initialized
@@ -143,12 +147,15 @@ def get_version_key(self) -> str:
         """Get the version key from the WASM rules engine.
 
         Used for cache key generation to ensure cache invalidation on engine updates.
+        The value is computed once during initialization and cached.
         """
         self._ensure_initialized()
+        return self._version_key or "1"
 
+    def _read_version_key_from_wasm(self) -> str:
+        """Read the version key from WASM. Called once during init."""
         if self._get_version_key_fn is None:
             return "1"
-
         ptr = self._get_version_key_fn(self._store)
         return self._read_null_terminated_string(ptr)
 
@@ -190,4 +197,4 @@ def _read_null_terminated_string(self, ptr: int, max_length: int = 256) -> str:
         null_idx = raw.find(0)
         if null_idx >= 0:
             raw = raw[:null_idx]
-        return raw.decode("utf-8")
+        return raw.decode("utf-8").strip()

src/schematic/datastream/rules_engine.py

@@ -225,7 +225,7 @@ async def _connect_and_read(self) -> None:
                     )
                 async with websockets.connect(
                     self._url,
-                    additional_headers=self._headers,
+                    extra_headers=self._headers,
                     open_timeout=CONNECTION_TIMEOUT,
                     # Disable the library's built-in keepalive — we manage
                     # ping/pong ourselves to match the Node SDK behaviour and

src/schematic/datastream/websocket_client.py

@@ -32,6 +32,7 @@
 logger = logging.getLogger("testapp")
 
 CACHE_TTL_MS = 2000  # Short TTL for E2E — tests sleep past this to verify cache expiration
+FLAG_CACHE_TTL_MS = 30 * 24 * 60 * 60 * 1000  # 30 days — flags use a long TTL, matching the Go SDK
 
 
 # ---------------------------------------------------------------------------
@@ -135,13 +136,35 @@ async def handle_configure(request: web.Request) -> web.Response:
             ds.company_lookup_cache = RedisCache(redis_client, default_ttl_ms=CACHE_TTL_MS)
             ds.user_cache = RedisCache(redis_client, default_ttl_ms=CACHE_TTL_MS)
             ds.user_lookup_cache = RedisCache(redis_client, default_ttl_ms=CACHE_TTL_MS)
-            ds.flag_cache = RedisCache(redis_client, default_ttl_ms=CACHE_TTL_MS)
+            ds.flag_cache = RedisCache(redis_client, default_ttl_ms=FLAG_CACHE_TTL_MS)
 
         replicator_url = get_config_string("replicatorUrl")
         if replicator_url:
             ds.replicator_mode = True
             ds.replicator_health_url = replicator_url + "/ready"
 
+            # Verify the replicator is actually running before proceeding.
+            # Without this, flag checks silently fall back to the API and
+            # replicator mode is never truly exercised.
+            import httpx
+
+            try:
+                async with httpx.AsyncClient(timeout=5.0) as http:
+                    resp = await http.get(ds.replicator_health_url)
+                    resp.raise_for_status()
+                    health = resp.json()
+                    if not health.get("ready", False):
+                        return web.json_response(
+                            {"success": False, "error": f"Replicator is not ready: {health}"},
+                            status=503,
+                        )
+                    logger.info("Replicator is healthy: %s", health)
+            except Exception as e:
+                return web.json_response(
+                    {"success": False, "error": f"Replicator health check failed: {e}"},
+                    status=503,
+                )
+
         cfg.datastream = ds
 
     client = AsyncSchematic(api_key, cfg)