From 8b18799ad4559b72969885da71ff958287d81b09 Mon Sep 17 00:00:00 2001 From: Abhinaysai Kamineni Date: Wed, 10 Jun 2026 14:51:47 -0400 Subject: [PATCH] Phase 2: GitHub and Jira live pipeline E2E with Kafka topic bootstrap. Add inject and verify scripts for GitHub/Jira, flush after publish, init_kafka_topics, and unified --source verification so all three connectors reach Neo4j without cold-start topic errors. Co-authored-by: Cursor --- CLAUDE.md | 10 +- DECISIONS.md | 9 ++ Makefile | 18 ++- SESSIONS.md | 25 ++++ connectors/github/producer.py | 1 + connectors/jira/producer.py | 1 + scripts/init_kafka_topics.py | 59 ++++++++++ scripts/inject_github_event.py | 96 ++++++++++++++++ scripts/inject_jira_event.py | 98 ++++++++++++++++ scripts/verify_slack_pipeline.py | 107 +++++++++++++----- tests/connectors/test_github_producer.py | 5 +- tests/connectors/test_jira_producer.py | 5 +- tests/integration/test_github_pipeline_e2e.py | 90 +++++++++++++++ tests/integration/test_jira_pipeline_e2e.py | 90 +++++++++++++++ tests/scripts/test_verify_slack_pipeline.py | 9 ++ 15 files changed, 585 insertions(+), 38 deletions(-) create mode 100644 scripts/init_kafka_topics.py create mode 100644 scripts/inject_github_event.py create mode 100644 scripts/inject_jira_event.py create mode 100644 tests/integration/test_github_pipeline_e2e.py create mode 100644 tests/integration/test_jira_pipeline_e2e.py diff --git a/CLAUDE.md b/CLAUDE.md index b0acf39..74d5283 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -93,9 +93,9 @@ LinkedIn: linkedin.com/in/abhinaysai-kamineni ## Current Phase -**Phase:** 0 — Documentation and architecture finalization -**Status:** Complete -**Next phase:** Phase 1 — Kafka infrastructure + Slack connector + decision extractor +**Phase:** 1 — Kafka + Slack connector + decision extractor +**Status:** Complete (PR #9–#10 merged, live E2E verified) +**Next phase:** Phase 2 — GitHub + Jira connectors + Neo4j graph schema **Target:** Working MVP demo in 4 weeks --- @@ -105,8 +105,8 @@ LinkedIn: linkedin.com/in/abhinaysai-kamineni | Phase | Scope | Duration | Status | |---|---|---|---| | 0 | Architecture + documentation | Complete | ✅ Done | -| 1 | Kafka + Slack connector + decision extractor | Week 1 | 🔄 Next | -| 2 | GitHub + Jira connectors + Neo4j graph schema | Week 1-2 | ⏳ | +| 1 | Kafka + Slack connector + decision extractor | Week 1 | ✅ Done | +| 2 | GitHub + Jira connectors + Neo4j graph schema | Week 1-2 | 🔄 In progress | | 3 | `cortex.query()` API + Redis cache + MCP server | Week 2 | ⏳ | | 4 | Importance scorer + trust scorer + RBAC | Week 2-3 | ⏳ | | 5 | Contradiction detector + decay engine | Week 3 | ⏳ | diff --git a/DECISIONS.md b/DECISIONS.md index 6ba50ba..513bcb7 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -23,6 +23,15 @@ Agent picks up OPEN instructions at session start, executes, marks DONE. ## ACTIVE INSTRUCTIONS +### 2026-06-10 — Phase 2: GitHub + Jira connector E2E +Priority: HIGH +Status: IN PROGRESS — inject scripts + live verify done; real webhooks pending +Detail: +- GitHub/Jira inject + verify scripts (mirror Slack Phase 1) +- Kafka topic bootstrap on cold start +- Live E2E: github + jira → extract → Neo4j +- Remaining: public webhook URLs, graph schema cross-source validation + ### 2026-05-11 — Build Phase 1: Core infrastructure + Slack connector Priority: HIGH Status: DONE — Session 1 (2026-05-11) diff --git a/Makefile b/Makefile index cb34113..95eb9da 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: demo demo-dry-run test ci stack pipeline-restart pipeline-local verify-pipeline +.PHONY: demo demo-dry-run test ci stack init-kafka pipeline-restart pipeline-local verify-pipeline verify-github verify-jira verify-connectors # Local portfolio demo: Docker infra + migrations + seed + API + worker + frontend. demo: @@ -19,6 +19,10 @@ ci: test stack: docker compose --profile api up -d +# Pre-create Kafka topics (eliminates worker UNKNOWN_TOPIC errors on cold start). +init-kafka: + python scripts/init_kafka_topics.py + # Fast worker reload after Python changes (volume-mounted code; no image rebuild). pipeline-restart: docker compose --profile api restart pipeline-worker @@ -27,6 +31,14 @@ pipeline-restart: pipeline-local: bash scripts/run_pipeline_worker_local.sh -# End-to-end: inject Slack message → Ollama extract → Neo4j Decision node. +# End-to-end: inject → Ollama extract → Neo4j Decision node. verify-pipeline: - python scripts/verify_slack_pipeline.py --timeout 120 + python scripts/verify_slack_pipeline.py --source slack --timeout 120 + +verify-github: + python scripts/verify_slack_pipeline.py --source github --timeout 120 + +verify-jira: + python scripts/verify_slack_pipeline.py --source jira --timeout 120 + +verify-connectors: verify-pipeline verify-github verify-jira diff --git a/SESSIONS.md b/SESSIONS.md index f78fd58..695907f 100644 --- a/SESSIONS.md +++ b/SESSIONS.md @@ -394,3 +394,28 @@ 1. Merge live E2E PR; configure real Slack app webhook 2. Optional: slim pipeline-worker Docker image (worker-only deps) 3. Demo recording per `docs/DEMO_RECORDING.md` + +--- + +## Session — 2026-06-10 — Phase 2 kickoff: GitHub + Jira live E2E +**Duration:** ~45m +**Phase:** Phase 2 — GitHub + Jira connectors + Neo4j graph schema + +### Built +- **`scripts/inject_github_event.py`**, **`scripts/inject_jira_event.py`** — dev inject without webhooks +- **`scripts/init_kafka_topics.py`** + **`make init-kafka`** — pre-create raw/DLQ/extracted topics +- **Unified verify:** `--source slack|github|jira` on verify script; **`make verify-github`**, **`verify-jira`**, **`verify-connectors`** +- **Flush after publish** on GitHub/Jira connectors (parity with Slack) +- Integration tests: `test_github_pipeline_e2e.py`, `test_jira_pipeline_e2e.py` + +### State at end +- **`make verify-pipeline`** PASS (13 → 14 on main after merge) +- **`make verify-github`** PASS (13 → 14) — ~23s +- **`make verify-jira`** PASS (14 → 15) — ~12s +- **298** pytest passing +- Branch **`feature/phase-2-github-jira`** ready for PR + +### Next session starts with +1. Real GitHub/Jira webhook URLs (ngrok or deploy) +2. Phase 3: `cortex.query()` hardening + MCP live inject wiring +3. Graph schema validation against multi-source decisions diff --git a/connectors/github/producer.py b/connectors/github/producer.py index 82f63fa..b258a2b 100644 --- a/connectors/github/producer.py +++ b/connectors/github/producer.py @@ -454,6 +454,7 @@ def handle_event( return {"status": "skipped", "reason": "not_processable"} self._producer.publish(raw_event) + self._producer.flush(timeout=5.0) return {"status": "ok", "event_id": raw_event.event_id} def close(self) -> None: diff --git a/connectors/jira/producer.py b/connectors/jira/producer.py index bd2701d..f0bb651 100644 --- a/connectors/jira/producer.py +++ b/connectors/jira/producer.py @@ -500,6 +500,7 @@ def handle_event( return {"status": "skipped", "reason": "not_processable"} self._producer.publish(raw_event) + self._producer.flush(timeout=5.0) return {"status": "ok", "event_id": raw_event.event_id} def close(self) -> None: diff --git a/scripts/init_kafka_topics.py b/scripts/init_kafka_topics.py new file mode 100644 index 0000000..865b6c7 --- /dev/null +++ b/scripts/init_kafka_topics.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""Pre-create Cortex Kafka topics so the pipeline worker starts cleanly.""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + +_REPO = Path(__file__).resolve().parents[1] +if str(_REPO) not in sys.path: + sys.path.insert(0, str(_REPO)) + +from dotenv import load_dotenv + +from pipeline.extraction_worker import DLQ_TOPIC, EXTRACTED_TOPIC, RAW_TOPICS + +TOPICS = [*RAW_TOPICS, EXTRACTED_TOPIC, DLQ_TOPIC] + + +def main() -> int: + load_dotenv(_REPO / ".env") + bootstrap = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092") + + try: + from confluent_kafka.admin import AdminClient, NewTopic + except ImportError: + print("FAIL: confluent-kafka not installed", file=sys.stderr) + return 1 + + admin = AdminClient({"bootstrap.servers": bootstrap}) + existing = admin.list_topics(timeout=10).topics + to_create = [ + NewTopic(topic, num_partitions=1, replication_factor=1) + for topic in TOPICS + if topic not in existing + ] + + if not to_create: + print(f"All {len(TOPICS)} Cortex topics already exist on {bootstrap}") + return 0 + + futures = admin.create_topics(to_create) + for topic, future in futures.items(): + try: + future.result() + print(f"Created topic {topic}") + except Exception as exc: # noqa: BLE001 — CLI boundary + if "TOPIC_ALREADY_EXISTS" in str(exc): + print(f"Topic {topic} already exists") + else: + print(f"FAIL: {topic}: {exc}", file=sys.stderr) + return 1 + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/inject_github_event.py b/scripts/inject_github_event.py new file mode 100644 index 0000000..46a80fd --- /dev/null +++ b/scripts/inject_github_event.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Inject a synthetic GitHub PR event into the Cortex pipeline. + +Publishes to ``cortex.raw.github.events`` for the extraction worker to consume. + +Usage: + python scripts/inject_github_event.py --dry-run + python scripts/inject_github_event.py +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +_REPO = Path(__file__).resolve().parents[1] +if str(_REPO) not in sys.path: + sys.path.insert(0, str(_REPO)) + +from dotenv import load_dotenv + +from connectors.github.producer import GitHubKafkaProducer, normalise_github_event + +DEFAULT_BODY = ( + "We decided to migrate payments to CockroachDB for multi-region scale " + "because Postgres failover gaps blocked EU launch." +) +DEFAULT_TITLE = "Decision: migrate payments to CockroachDB" + + +def build_merged_pr_payload(*, title: str, body: str, repo: str) -> dict: + """Build a GitHub ``pull_request`` closed+merged webhook payload.""" + return { + "action": "closed", + "pull_request": { + "number": 42, + "title": title, + "body": body, + "merged": True, + "user": {"login": "priya"}, + "base": {"ref": "main"}, + "head": {"ref": "feature/cockroachdb"}, + "updated_at": "2026-06-10T18:00:00Z", + "created_at": "2026-06-10T17:00:00Z", + "labels": [], + "requested_reviewers": [], + }, + "repository": {"full_name": repo, "default_branch": "main"}, + } + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Publish a synthetic GitHub PR event to cortex.raw.github.events", + ) + parser.add_argument("--workspace", default=None, help="Cortex workspace id") + parser.add_argument("--title", default=DEFAULT_TITLE, help="PR title") + parser.add_argument("--body", default=DEFAULT_BODY, help="PR body") + parser.add_argument("--repo", default="acme/payments", help="GitHub repo full name") + parser.add_argument( + "--dry-run", + action="store_true", + help="Print normalised RawEvent JSON without publishing", + ) + args = parser.parse_args() + + load_dotenv(_REPO / ".env") + workspace = args.workspace or os.environ.get("CORTEX_WORKSPACE_ID", "local-dev") + + payload = build_merged_pr_payload(title=args.title, body=args.body, repo=args.repo) + raw_event = normalise_github_event(payload, "pull_request", workspace) + if raw_event is None: + print("FAIL: payload did not normalise to a RawEvent", file=sys.stderr) + return 1 + + if args.dry_run: + print(json.dumps(json.loads(raw_event.model_dump_json()), indent=2)) + print(f"\nWould publish to cortex.raw.github.events (workspace={workspace!r})") + return 0 + + producer = GitHubKafkaProducer() + try: + producer.publish(raw_event) + producer.flush(timeout=10.0) + finally: + producer.close() + + print(f"Published event_id={raw_event.event_id} to cortex.raw.github.events") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/inject_jira_event.py b/scripts/inject_jira_event.py new file mode 100644 index 0000000..2c30395 --- /dev/null +++ b/scripts/inject_jira_event.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Inject a synthetic Jira comment event into the Cortex pipeline. + +Publishes to ``cortex.raw.jira.events`` for the extraction worker to consume. + +Usage: + python scripts/inject_jira_event.py --dry-run + python scripts/inject_jira_event.py +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +_REPO = Path(__file__).resolve().parents[1] +if str(_REPO) not in sys.path: + sys.path.insert(0, str(_REPO)) + +from dotenv import load_dotenv + +from connectors.jira.producer import JiraKafkaProducer, normalise_jira_event + +DEFAULT_COMMENT = ( + "We decided to adopt event sourcing for billing because audit requirements " + "blocked the monolith release." +) + + +def build_comment_payload(*, comment: str, issue_key: str, project_key: str) -> dict: + """Build a Jira ``jira:issue_commented`` webhook payload.""" + return { + "webhookEvent": "jira:issue_commented", + "issue": { + "key": issue_key, + "fields": { + "summary": "Billing architecture decision", + "project": {"key": project_key}, + }, + }, + "comment": { + "id": "10001", + "body": comment, + "created": "2026-06-10T18:00:00.000+0000", + "author": {"displayName": "Alex Chen", "accountId": "demo-user"}, + }, + } + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Publish a synthetic Jira comment to cortex.raw.jira.events", + ) + parser.add_argument("--workspace", default=None, help="Cortex workspace id") + parser.add_argument("--comment", default=DEFAULT_COMMENT, help="Comment body") + parser.add_argument("--issue", default="ENG-123", help="Jira issue key") + parser.add_argument("--project", default="ENG", help="Jira project key") + parser.add_argument( + "--dry-run", + action="store_true", + help="Print normalised RawEvent JSON without publishing", + ) + args = parser.parse_args() + + load_dotenv(_REPO / ".env") + workspace = args.workspace or os.environ.get("CORTEX_WORKSPACE_ID", "local-dev") + + payload = build_comment_payload( + comment=args.comment, + issue_key=args.issue, + project_key=args.project, + ) + raw_event = normalise_jira_event(payload, "jira:issue_commented", workspace) + if raw_event is None: + print("FAIL: payload did not normalise to a RawEvent", file=sys.stderr) + return 1 + + if args.dry_run: + print(json.dumps(json.loads(raw_event.model_dump_json()), indent=2)) + print(f"\nWould publish to cortex.raw.jira.events (workspace={workspace!r})") + return 0 + + producer = JiraKafkaProducer() + try: + producer.publish(raw_event) + producer.flush(timeout=10.0) + finally: + producer.close() + + print(f"Published event_id={raw_event.event_id} to cortex.raw.jira.events") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify_slack_pipeline.py b/scripts/verify_slack_pipeline.py index d5fb5f1..760a37a 100644 --- a/scripts/verify_slack_pipeline.py +++ b/scripts/verify_slack_pipeline.py @@ -1,15 +1,18 @@ #!/usr/bin/env python3 -"""Verify the Slack → Kafka → extract → Neo4j pipeline end-to-end. +"""Verify connector → Kafka → extract → Neo4j pipeline end-to-end. + +Supports slack, github, and jira sources. Default source is slack for backward +compatibility with ``make verify-pipeline``. Prerequisites: - - `make stack` or `docker compose --profile api up -d` + - `make stack` and `make init-kafka` - Ollama running on the host with the configured model pulled - After code changes: `make pipeline-restart` (seconds) — not `docker compose build` Usage: - python scripts/verify_slack_pipeline.py --dry-run - python scripts/verify_slack_pipeline.py - python scripts/verify_slack_pipeline.py --timeout 180 + python scripts/verify_slack_pipeline.py --source slack + python scripts/verify_slack_pipeline.py --source github + python scripts/verify_slack_pipeline.py --source jira --dry-run """ from __future__ import annotations @@ -22,6 +25,7 @@ import time import urllib.error import urllib.request +from collections.abc import Callable from pathlib import Path _REPO = Path(__file__).resolve().parents[1] @@ -31,7 +35,50 @@ from dotenv import load_dotenv from neo4j import GraphDatabase -_INJECT = _REPO / "scripts" / "inject_slack_message.py" +DecisionTextBuilder = Callable[[str, str], str] + + +def _slack_text(marker: str, _workspace: str) -> str: + return ( + f"We decided to adopt event sourcing for billing ({marker}) " + "because audit requirements blocked the monolith release." + ) + + +def _github_body(marker: str, _workspace: str) -> str: + return ( + f"We decided to adopt event sourcing for billing ({marker}) " + "because audit requirements blocked the monolith release." + ) + + +def _jira_comment(marker: str, _workspace: str) -> str: + return ( + f"We decided to adopt event sourcing for billing ({marker}) " + "because audit requirements blocked the monolith release." + ) + + +SOURCE_CONFIG: dict[str, dict[str, object]] = { + "slack": { + "script": "inject_slack_message.py", + "label": "Slack message", + "text_builder": _slack_text, + "extra_args": lambda text, workspace: ["--text", text, "--workspace", workspace], + }, + "github": { + "script": "inject_github_event.py", + "label": "GitHub PR", + "text_builder": _github_body, + "extra_args": lambda text, workspace: ["--body", text, "--workspace", workspace], + }, + "jira": { + "script": "inject_jira_event.py", + "label": "Jira comment", + "text_builder": _jira_comment, + "extra_args": lambda text, workspace: ["--comment", text, "--workspace", workspace], + }, +} def _fetch_json(url: str) -> dict: @@ -50,7 +97,6 @@ def check_ollama(*, base_url: str, model: str) -> tuple[bool, str]: if model in names: return True, f"Ollama model {model!r} available" - # Allow unpinned tags, e.g. llama3.1:8b vs llama3.1:8b-instruct prefix = model.split(":")[0] + ":" matches = sorted(n for n in names if n.startswith(prefix)) if matches: @@ -79,22 +125,25 @@ def count_decisions( return int(record["c"]) if record else 0 -def inject_message(*, text: str, workspace: str) -> subprocess.CompletedProcess[str]: - """Publish a synthetic Slack message via inject_slack_message.py.""" - cmd = [ - sys.executable, - str(_INJECT), - "--text", - text, - "--workspace", - workspace, - ] +def inject_event(*, source: str, text: str, workspace: str) -> subprocess.CompletedProcess[str]: + """Publish a synthetic connector event via the matching inject script.""" + config = SOURCE_CONFIG[source] + script = _REPO / "scripts" / str(config["script"]) + extra_args_fn = config["extra_args"] + assert callable(extra_args_fn) + cmd = [sys.executable, str(script), *extra_args_fn(text, workspace)] return subprocess.run(cmd, capture_output=True, text=True, check=False) def main() -> int: parser = argparse.ArgumentParser( - description="Verify Slack pipeline: inject → extract → Neo4j", + description="Verify connector pipeline: inject → extract → Neo4j", + ) + parser.add_argument( + "--source", + choices=sorted(SOURCE_CONFIG), + default="slack", + help="Connector source to verify (default: slack)", ) parser.add_argument("--workspace", default=None, help="Cortex workspace id") parser.add_argument( @@ -124,6 +173,9 @@ def main() -> int: neo4j_user = os.environ.get("NEO4J_USER", "neo4j") neo4j_password = os.environ.get("NEO4J_PASSWORD", "cortex_local") + config = SOURCE_CONFIG[args.source] + label = str(config["label"]) + ok, msg = check_ollama(base_url=ollama_url, model=ollama_model) print(f"{'PASS' if ok else 'FAIL'}: {msg}") if not ok: @@ -143,19 +195,18 @@ def main() -> int: print(f"Neo4j Decision count (workspace={workspace!r}): {before}") if args.dry_run: - print("Dry run — skipping inject and poll") + print(f"Dry run — skipping {args.source} inject and poll") return 0 - marker = f"verify-pipeline-{int(time.time())}" - text = ( - f"We decided to adopt event sourcing for billing ({marker}) " - "because audit requirements blocked the monolith release." - ) - print(f"Injecting Slack message (marker={marker!r})...") - result = inject_message(text=text, workspace=workspace) + marker = f"verify-{args.source}-{int(time.time())}" + text_builder = config["text_builder"] + assert callable(text_builder) + text = text_builder(marker, workspace) + print(f"Injecting {label} (source={args.source!r}, marker={marker!r})...") + result = inject_event(source=args.source, text=text, workspace=workspace) if result.returncode != 0: print(result.stderr or result.stdout, file=sys.stderr) - print("FAIL: inject_slack_message.py exited non-zero") + print(f"FAIL: {config['script']} exited non-zero") return 1 print(result.stdout.strip()) @@ -174,7 +225,7 @@ def main() -> int: continue if after > before: - print(f"PASS: Decision count increased {before} → {after}") + print(f"PASS: {args.source} pipeline — Decision count {before} → {after}") return 0 time.sleep(args.poll_interval) diff --git a/tests/connectors/test_github_producer.py b/tests/connectors/test_github_producer.py index 40078a9..28e466b 100644 --- a/tests/connectors/test_github_producer.py +++ b/tests/connectors/test_github_producer.py @@ -341,13 +341,16 @@ class TestGitHubConnector: def test_handle_event_publishes_and_returns_ok( self, mock_producer_cls: MagicMock ) -> None: - mock_producer_cls.return_value = MagicMock() + mock_producer = MagicMock() + mock_producer.flush.return_value = 0 + mock_producer_cls.return_value = mock_producer connector = GitHubConnector(workspace_id=WORKSPACE_ID) result = connector.handle_event( _pr_payload("opened"), event_type="pull_request" ) assert result["status"] == "ok" assert "event_id" in result + mock_producer.flush.assert_called_once() @patch("connectors.github.producer.Producer") def test_handle_unhandled_event_returns_skipped( diff --git a/tests/connectors/test_jira_producer.py b/tests/connectors/test_jira_producer.py index 58fc1ee..61fbd79 100644 --- a/tests/connectors/test_jira_producer.py +++ b/tests/connectors/test_jira_producer.py @@ -409,11 +409,14 @@ class TestJiraConnector: def test_handle_event_publishes_and_returns_ok( self, mock_producer_cls: MagicMock ) -> None: - mock_producer_cls.return_value = MagicMock() + mock_producer = MagicMock() + mock_producer.flush.return_value = 0 + mock_producer_cls.return_value = mock_producer connector = JiraConnector(workspace_id=WORKSPACE_ID) result = connector.handle_event(_issue_created_payload()) assert result["status"] == "ok" assert "event_id" in result + mock_producer.flush.assert_called_once() @patch("connectors.jira.producer.Producer") def test_unhandled_event_returns_skipped( diff --git a/tests/integration/test_github_pipeline_e2e.py b/tests/integration/test_github_pipeline_e2e.py new file mode 100644 index 0000000..4f00871 --- /dev/null +++ b/tests/integration/test_github_pipeline_e2e.py @@ -0,0 +1,90 @@ +"""GitHub-specific pipeline integration test — extract → score → graph write.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import MagicMock, patch + +from pipeline.extraction_worker import ExtractionWorker +from shared.models import DecisionEvent, Provenance, RawEvent + +NOW = datetime(2026, 5, 11, 12, 0, 0, tzinfo=UTC) + + +def _github_raw() -> RawEvent: + return RawEvent( + source="github", + source_id="acme/payments:pr:42", + workspace_id="local-dev", + event_type="github:pull_request:merged", + content="PR merged: Decision: migrate payments to CockroachDB\n\nApproved for EU launch.", + author="priya", + channel="acme/payments", + timestamp=NOW, + ) + + +def _github_decision(raw: RawEvent) -> DecisionEvent: + return DecisionEvent( + source_raw_event_id=raw.event_id, + workspace_id=raw.workspace_id, + event_type="decision", + content="Migrate payments to CockroachDB for multi-region scale.", + made_by=["priya"], + affects=["payments-service"], + rationale=["EU launch blocked by Postgres failover gaps."], + extraction_confidence=0.9, + importance_score=0.0, + trust_score=0.0, + provenance=Provenance( + source="github", + channel=raw.channel, + original_timestamp=raw.timestamp, + extractor_version="0.1.0", + extractor_model="test", + verified_by=[], + raw_event_id=raw.event_id, + ), + extracted_at=NOW, + ) + + +@patch("pipeline.extraction_worker.GraphWriter") +@patch("pipeline.extraction_worker.TrustScorer") +@patch("pipeline.extraction_worker.ImportanceScorer") +@patch("pipeline.extraction_worker.DecisionExtractor") +@patch("pipeline.extraction_worker.Producer") +@patch("pipeline.extraction_worker.Consumer") +def test_github_pr_through_worker_scores_before_write( + consumer_cls: MagicMock, + producer_cls: MagicMock, + extractor_cls: MagicMock, + importance_cls: MagicMock, + trust_cls: MagicMock, + writer_cls: MagicMock, +) -> None: + """GitHub RawEvent runs extract → importance → trust → Neo4j write in order.""" + raw_event = _github_raw() + decision = _github_decision(raw_event) + + def _apply_importance(d: DecisionEvent) -> DecisionEvent: + d.importance_score = 0.82 + return d + + def _apply_trust(d: DecisionEvent) -> DecisionEvent: + d.trust_score = 0.76 + return d + + extractor_cls.return_value.extract.return_value = decision + importance_cls.return_value.score.side_effect = _apply_importance + trust_cls.return_value.score.side_effect = _apply_trust + writer_cls.return_value.write.return_value = decision.event_id + + worker = ExtractionWorker(bootstrap_servers="localhost:9092") + event_id = worker.process_raw_event(raw_event) + + assert event_id == decision.event_id + extractor_cls.return_value.extract.assert_called_once_with(raw_event) + importance_cls.return_value.score.assert_called_once() + trust_cls.return_value.score.assert_called_once() + writer_cls.return_value.write.assert_called_once_with(decision) diff --git a/tests/integration/test_jira_pipeline_e2e.py b/tests/integration/test_jira_pipeline_e2e.py new file mode 100644 index 0000000..3887963 --- /dev/null +++ b/tests/integration/test_jira_pipeline_e2e.py @@ -0,0 +1,90 @@ +"""Jira-specific pipeline integration test — extract → score → graph write.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import MagicMock, patch + +from pipeline.extraction_worker import ExtractionWorker +from shared.models import DecisionEvent, Provenance, RawEvent + +NOW = datetime(2026, 5, 11, 12, 0, 0, tzinfo=UTC) + + +def _jira_raw() -> RawEvent: + return RawEvent( + source="jira", + source_id="ENG-123:comment:10001", + workspace_id="local-dev", + event_type="jira:issue_commented", + content="Comment on ENG-123 (Billing architecture decision):\nWe decided to adopt event sourcing.", + author="Alex Chen", + channel="ENG", + timestamp=NOW, + ) + + +def _jira_decision(raw: RawEvent) -> DecisionEvent: + return DecisionEvent( + source_raw_event_id=raw.event_id, + workspace_id=raw.workspace_id, + event_type="decision", + content="Adopt event sourcing for billing.", + made_by=["Alex Chen"], + affects=["billing"], + rationale=["Audit requirements blocked monolith release."], + extraction_confidence=0.87, + importance_score=0.0, + trust_score=0.0, + provenance=Provenance( + source="jira", + channel=raw.channel, + original_timestamp=raw.timestamp, + extractor_version="0.1.0", + extractor_model="test", + verified_by=[], + raw_event_id=raw.event_id, + ), + extracted_at=NOW, + ) + + +@patch("pipeline.extraction_worker.GraphWriter") +@patch("pipeline.extraction_worker.TrustScorer") +@patch("pipeline.extraction_worker.ImportanceScorer") +@patch("pipeline.extraction_worker.DecisionExtractor") +@patch("pipeline.extraction_worker.Producer") +@patch("pipeline.extraction_worker.Consumer") +def test_jira_comment_through_worker_scores_before_write( + consumer_cls: MagicMock, + producer_cls: MagicMock, + extractor_cls: MagicMock, + importance_cls: MagicMock, + trust_cls: MagicMock, + writer_cls: MagicMock, +) -> None: + """Jira RawEvent runs extract → importance → trust → Neo4j write in order.""" + raw_event = _jira_raw() + decision = _jira_decision(raw_event) + + def _apply_importance(d: DecisionEvent) -> DecisionEvent: + d.importance_score = 0.8 + return d + + def _apply_trust(d: DecisionEvent) -> DecisionEvent: + d.trust_score = 0.74 + return d + + extractor_cls.return_value.extract.return_value = decision + importance_cls.return_value.score.side_effect = _apply_importance + trust_cls.return_value.score.side_effect = _apply_trust + writer_cls.return_value.write.return_value = decision.event_id + + worker = ExtractionWorker(bootstrap_servers="localhost:9092") + event_id = worker.process_raw_event(raw_event) + + assert event_id == decision.event_id + extractor_cls.return_value.extract.assert_called_once_with(raw_event) + importance_cls.return_value.score.assert_called_once() + trust_cls.return_value.score.assert_called_once() + writer_cls.return_value.write.assert_called_once_with(decision) diff --git a/tests/scripts/test_verify_slack_pipeline.py b/tests/scripts/test_verify_slack_pipeline.py index c4754de..6eafbfb 100644 --- a/tests/scripts/test_verify_slack_pipeline.py +++ b/tests/scripts/test_verify_slack_pipeline.py @@ -49,3 +49,12 @@ def test_main_dry_run() -> None: patch.object(sys, "argv", ["verify_slack_pipeline.py", "--dry-run"]), ): assert verify.main() == 0 + + +def test_main_dry_run_github_source() -> None: + with ( + patch.object(verify, "check_ollama", return_value=(True, "ok")), + patch.object(verify, "count_decisions", return_value=3), + patch.object(sys, "argv", ["verify_slack_pipeline.py", "--source", "github", "--dry-run"]), + ): + assert verify.main() == 0