mirror of
https://github.com/taylanbakircioglu/flowfish.git
synced 2026-09-16 23:55:07 +00:00
taylanbakircioglu
6e503368f7
- Grafana Beyla DaemonSet for kernel-level HTTP/gRPC/DNS capture (passive, zero application changes, W3C traceparent header propagation) - flowfish-l7-collector in-cluster bridge: OTLP receiver + buffered pull API - L7 Ingestion Service: K8s service-proxy poll → enrich → RabbitMQ - ClickHouse l7_http_flows / l7_grpc_flows / l7_dns_flows + APM RED MVs - Neo4j L7Workload nodes + SAME_WORKLOAD cross-cluster bridges - New pages: Service Map, Trace Explorer, APM Services List, APM Service Detail - Analysis Wizard now supports L4 / L7 / Both modes with HTTP/gRPC/DNS picks - Integration Hub gains L7 dependency summary + tree-summary integrations - Multi-Cluster Management: dual-agent install (Inspector Gadget L4 + Beyla L7), runtime OpenShift detection so SCCs auto-install with kubectl too - ServiceMap edge → Trace Explorer drill-down with virtual_trace_id correlation - Docs: new L7 architecture diagram, README L7 sections, 3 new screenshots
84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
"""Shared pytest fixtures for the graph-query service.
|
|
|
|
The query engine talks to Neo4j through ``GraphQueryEngine.execute_query``
|
|
and constructs Cypher WHERE clauses with ``_l7_match_where``. To keep
|
|
the test surface focused on the pure-Python filter/aggregation logic
|
|
introduced for the Integration Hub L7 parity work (Audit v3), we expose
|
|
a ``mock_engine`` fixture that:
|
|
|
|
* instantiates ``GraphQueryEngine`` without ever opening a real Neo4j
|
|
driver (``_connect`` is patched to a no-op),
|
|
* replaces ``execute_query`` with a side-effectful stub the tests can
|
|
program with the rows they want to feed the aggregator.
|
|
|
|
Tests get a strongly-typed helper (``set_rows``) so individual cases
|
|
read like data tables.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Dict, Iterable, List, Optional
|
|
|
|
import pytest
|
|
|
|
# Ensure the service package is importable when pytest is invoked from
|
|
# the repo root or from inside ``services/graph-query/``. We prepend
|
|
# the service directory so ``import app.graph_query_engine`` matches the
|
|
# layout used by the Dockerfile.
|
|
_SERVICE_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(_SERVICE_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_SERVICE_ROOT))
|
|
|
|
# Pydantic-Settings will refuse to load without a Neo4j password. Provide
|
|
# placeholders for the test environment before importing ``app`` modules.
|
|
os.environ.setdefault("NEO4J_PASSWORD", "test-password")
|
|
os.environ.setdefault("NEO4J_USER", "neo4j")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_engine(monkeypatch: pytest.MonkeyPatch):
|
|
"""Return a ``GraphQueryEngine`` with a stubbed ``execute_query``.
|
|
|
|
Usage::
|
|
|
|
def test_something(mock_engine):
|
|
engine, set_rows = mock_engine
|
|
set_rows([{...}, {...}])
|
|
response = engine.get_l7_dependency_summary(analysis_id="A1")
|
|
|
|
``set_rows`` accepts either a static list (echoed for every call) or
|
|
a callable ``(query, params) -> List[dict]`` for tests that need to
|
|
inspect the Cypher parameters the engine emits.
|
|
"""
|
|
from app.graph_query_engine import GraphQueryEngine # noqa: WPS433
|
|
|
|
# Prevent the constructor from attempting to open a Bolt connection.
|
|
monkeypatch.setattr(GraphQueryEngine, "_connect", lambda self: None)
|
|
|
|
engine = GraphQueryEngine()
|
|
engine.driver = object() # truthy so execute_query short-circuits the lazy reconnect
|
|
|
|
state: Dict[str, Any] = {"rows": [], "captured": []}
|
|
|
|
def _execute_query(query: str, parameters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
rows_src = state["rows"]
|
|
if callable(rows_src):
|
|
rows: List[Dict[str, Any]] = list(rows_src(query, parameters or {}))
|
|
else:
|
|
rows = list(rows_src)
|
|
state["captured"].append({"query": query, "parameters": dict(parameters or {})})
|
|
return {"success": True, "data": rows, "count": len(rows)}
|
|
|
|
engine.execute_query = _execute_query # type: ignore[assignment]
|
|
|
|
def set_rows(rows: "Iterable[Dict[str, Any]] | Callable[[str, Dict[str, Any]], Iterable[Dict[str, Any]]]") -> None:
|
|
state["rows"] = rows
|
|
|
|
# Expose captured query metadata for assertions.
|
|
engine._captured_queries = state["captured"] # type: ignore[attr-defined]
|
|
|
|
return engine, set_rows
|