feat: implement contract event export CLI, chaos testing suite, and automated changelogs (#750)

* feat: implement contract event export CLI and add chaos engineering testing suite

* feat: add automated changelog generation workflow using git-cliff

* test: add automated forward and rollback validation for ingest migrations

* test migration fix

* test: add coverage for forward and rollback migration paths in the ingest app

* test: add comprehensive forward and rollback migration coverage for the ingest app

Author: Dantama022

.github/workflows/changelog.yml

@@ -31,19 +31,10 @@ jobs:
           OUTPUT: docs/changelog.md
           GITHUB_REPO: ${{ github.repository }}
 
-      - name: Ensure GH_PAT is configured
-        env:
-          GH_PAT: ${{ secrets.GH_PAT }}
-        run: |
-          if [ -z "$GH_PAT" ]; then
-            echo "GH_PAT secret is required to create changelog pull requests."
-            exit 1
-          fi
-
       - name: Create Pull Request
         uses: peter-evans/create-pull-request@v6
         with:
-          token: ${{ secrets.GH_PAT }}
+          token: ${{ secrets.GITHUB_TOKEN }}
           commit-message: "chore(changelog): update changelog"
           title: "chore(changelog): Update Changelog"
           body: |

.github/workflows/chaos-tests.yml

@@ -0,0 +1,33 @@
+name: Chaos Tests
+
+on:
+  push:
+    branches: [main]
+    paths:
+      - "chaos-tests/**"
+      - ".github/workflows/chaos-tests.yml"
+  pull_request:
+    branches: [main]
+    paths:
+      - "chaos-tests/**"
+      - ".github/workflows/chaos-tests.yml"
+
+jobs:
+  validate:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: chaos-tests
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: python -m pip install pytest PyYAML
+
+      - name: Validate chaos scenarios
+        run: python -m pytest -q

chaos-tests/__pycache__/run_chaos.cpython-314.pyc

@@ -0,0 +1,219 @@
+"""Opt-in chaos engineering harness for SoroScan Kubernetes deployments.
+
+The runner validates scenarios by default. It executes disruptive actions only
+when SOROSCAN_CHAOS_RUN=1 is set, making it safe for CI validation.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from urllib.error import URLError
+from urllib.request import urlopen
+
+import yaml
+
+
+ROOT = Path(__file__).resolve().parent
+DEFAULT_SCENARIOS = ROOT / "scenarios.yaml"
+
+
+@dataclass(frozen=True)
+class Scenario:
+    name: str
+    description: str
+    namespace: str
+    selector: str
+    action: dict[str, Any]
+    recovery: dict[str, Any]
+
+
+class ChaosError(RuntimeError):
+    """Raised when a chaos scenario cannot be validated or executed."""
+
+
+def load_scenarios(path: Path = DEFAULT_SCENARIOS) -> list[Scenario]:
+    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+    scenarios = []
+    for raw in data.get("scenarios", []):
+        scenario = Scenario(
+            name=raw["name"],
+            description=raw.get("description", ""),
+            namespace=raw["namespace"],
+            selector=raw["selector"],
+            action=raw["action"],
+            recovery=raw["recovery"],
+        )
+        validate_scenario(scenario)
+        scenarios.append(scenario)
+    if not scenarios:
+        raise ChaosError("No chaos scenarios were defined.")
+    return scenarios
+
+
+def validate_scenario(scenario: Scenario) -> None:
+    action_type = scenario.action.get("type")
+    supported = {
+        "pod_termination",
+        "network_latency",
+        "memory_exhaustion",
+        "cpu_throttling",
+    }
+    if action_type not in supported:
+        raise ChaosError(f"Unsupported action type for {scenario.name}: {action_type}")
+    if not scenario.recovery.get("url"):
+        raise ChaosError(f"Scenario {scenario.name} requires recovery.url")
+    if int(scenario.recovery.get("timeout_seconds", 0)) <= 0:
+        raise ChaosError(f"Scenario {scenario.name} requires positive recovery.timeout_seconds")
+
+
+def build_commands(scenario: Scenario) -> list[list[str]]:
+    action = scenario.action
+    action_type = action["type"]
+    namespace = scenario.namespace
+    selector = scenario.selector
+
+    if action_type == "pod_termination":
+        return [
+            [
+                "kubectl",
+                "-n",
+                namespace,
+                "delete",
+                "pod",
+                "-l",
+                selector,
+                "--field-selector=status.phase=Running",
+            ]
+        ]
+
+    if action_type == "network_latency":
+        latency = int(action["latency_ms"])
+        duration = int(action.get("duration_seconds", 30))
+        return [
+            [
+                "kubectl",
+                "-n",
+                namespace,
+                "exec",
+                "deploy/" + action.get("deployment", "soroscan-backend"),
+                "--",
+                "sh",
+                "-c",
+                f"tc qdisc add dev eth0 root netem delay {latency}ms; "
+                f"sleep {duration}; tc qdisc del dev eth0 root || true",
+            ]
+        ]
+
+    if action_type in {"memory_exhaustion", "cpu_throttling"}:
+        deployment = action["deployment"]
+        resource = (
+            f"memory={action['memory_limit']}"
+            if action_type == "memory_exhaustion"
+            else f"cpu={action['cpu_limit']}"
+        )
+        duration = int(action.get("duration_seconds", 30))
+        return [
+            [
+                "kubectl",
+                "-n",
+                namespace,
+                "set",
+                "resources",
+                f"deployment/{deployment}",
+                "--limits",
+                resource,
+            ],
+            ["sleep", str(duration)],
+            [
+                "kubectl",
+                "-n",
+                namespace,
+                "rollout",
+                "restart",
+                f"deployment/{deployment}",
+            ],
+            [
+                "kubectl",
+                "-n",
+                namespace,
+                "rollout",
+                "status",
+                f"deployment/{deployment}",
+                "--timeout=120s",
+            ],
+        ]
+
+    raise ChaosError(f"Unsupported action type: {action_type}")
+
+
+def wait_for_recovery(url: str, timeout_seconds: int, interval_seconds: float = 2.0) -> None:
+    deadline = time.time() + timeout_seconds
+    last_error = None
+    while time.time() < deadline:
+        try:
+            with urlopen(url, timeout=5) as response:
+                if 200 <= response.status < 500:
+                    return
+        except URLError as exc:
+            last_error = exc
+        time.sleep(interval_seconds)
+    raise ChaosError(f"Recovery check failed for {url}: {last_error}")
+
+
+def run_command(command: list[str]) -> None:
+    subprocess.run(command, check=True)
+
+
+def run_scenario(scenario: Scenario, execute: bool) -> None:
+    commands = build_commands(scenario)
+    if not execute:
+        print(f"[dry-run] {scenario.name}: {len(commands)} commands validated")
+        return
+    for command in commands:
+        run_command(command)
+    wait_for_recovery(
+        scenario.recovery["url"],
+        int(scenario.recovery["timeout_seconds"]),
+    )
+    print(f"[ok] {scenario.name} recovered")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Run SoroScan chaos scenarios.")
+    parser.add_argument("--scenario", help="Run one scenario by name")
+    parser.add_argument("--scenarios-file", type=Path, default=DEFAULT_SCENARIOS)
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="Execute disruptive actions. Also requires SOROSCAN_CHAOS_RUN=1.",
+    )
+    args = parser.parse_args(argv)
+
+    try:
+        scenarios = load_scenarios(args.scenarios_file)
+        if args.scenario:
+            scenarios = [item for item in scenarios if item.name == args.scenario]
+            if not scenarios:
+                raise ChaosError(f"Unknown scenario: {args.scenario}")
+
+        execute = args.execute and os.getenv("SOROSCAN_CHAOS_RUN") == "1"
+        if args.execute and not execute:
+            raise ChaosError("Set SOROSCAN_CHAOS_RUN=1 before executing chaos actions.")
+
+        for scenario in scenarios:
+            run_scenario(scenario, execute=execute)
+    except ChaosError as exc:
+        print(f"chaos error: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

chaos-tests/run_chaos.py

@@ -0,0 +1,48 @@
+scenarios:
+  - name: pod_termination
+    description: Delete one backend pod and verify readiness recovers.
+    namespace: soroscan
+    selector: app=soroscan-backend
+    action:
+      type: pod_termination
+    recovery:
+      url: http://127.0.0.1:8000/ready/
+      timeout_seconds: 120
+
+  - name: network_latency
+    description: Inject traffic-control latency into backend pods.
+    namespace: soroscan
+    selector: app=soroscan-backend
+    action:
+      type: network_latency
+      latency_ms: 250
+      duration_seconds: 30
+    recovery:
+      url: http://127.0.0.1:8000/api/ingest/health/
+      timeout_seconds: 120
+
+  - name: memory_exhaustion
+    description: Apply a temporary low memory limit and verify graceful recovery.
+    namespace: soroscan
+    selector: app=soroscan-backend
+    action:
+      type: memory_exhaustion
+      deployment: soroscan-backend
+      memory_limit: 128Mi
+      duration_seconds: 30
+    recovery:
+      url: http://127.0.0.1:8000/ready/
+      timeout_seconds: 120
+
+  - name: cpu_throttling
+    description: Apply a temporary low CPU limit and verify graceful recovery.
+    namespace: soroscan
+    selector: app=soroscan-backend
+    action:
+      type: cpu_throttling
+      deployment: soroscan-backend
+      cpu_limit: 100m
+      duration_seconds: 30
+    recovery:
+      url: http://127.0.0.1:8000/api/ingest/health/
+      timeout_seconds: 120

chaos-tests/scenarios.yaml

@@ -0,0 +1,65 @@
+from pathlib import Path
+
+import pytest
+
+from run_chaos import ChaosError, build_commands, load_scenarios, main
+
+
+def test_loads_all_required_scenarios():
+    scenarios = load_scenarios(Path(__file__).resolve().parents[1] / "scenarios.yaml")
+
+    names = {scenario.name for scenario in scenarios}
+
+    assert names == {
+        "pod_termination",
+        "network_latency",
+        "memory_exhaustion",
+        "cpu_throttling",
+    }
+
+
+@pytest.mark.parametrize(
+    ("scenario_name", "expected_fragment"),
+    [
+        ("pod_termination", "delete"),
+        ("network_latency", "tc qdisc add"),
+        ("memory_exhaustion", "memory=128Mi"),
+        ("cpu_throttling", "cpu=100m"),
+    ],
+)
+def test_builds_expected_kubectl_commands(scenario_name, expected_fragment):
+    scenarios = load_scenarios(Path(__file__).resolve().parents[1] / "scenarios.yaml")
+    scenario = next(item for item in scenarios if item.name == scenario_name)
+
+    commands = build_commands(scenario)
+    flattened = " ".join(" ".join(command) for command in commands)
+
+    assert expected_fragment in flattened
+
+
+def test_execute_requires_explicit_environment_flag(monkeypatch):
+    monkeypatch.delenv("SOROSCAN_CHAOS_RUN", raising=False)
+
+    assert main(["--execute"]) == 1
+
+
+def test_unknown_scenario_fails_cleanly():
+    assert main(["--scenario", "missing"]) == 1
+
+
+def test_invalid_scenario_file_fails(tmp_path):
+    path = tmp_path / "bad.yaml"
+    path.write_text(
+        """
+scenarios:
+  - name: bad
+    namespace: soroscan
+    selector: app=soroscan-backend
+    action: {type: unknown}
+    recovery: {url: http://127.0.0.1:8000/ready/, timeout_seconds: 1}
+""",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ChaosError):
+        load_scenarios(path)