test(integration): add full stream lifecycle integration tests

backend/src/test/sseService.test.ts

@@ -0,0 +1,87 @@
+import { SseService } from "../services/sseService";
+
+type MockRes = {
+  write: jest.Mock;
+  end: jest.Mock;
+};
+
+const createMockRes = (): MockRes => ({
+  write: jest.fn(),
+  end: jest.fn(),
+});
+
+describe("SseService", () => {
+  let sse: SseService;
+
+  beforeEach(() => {
+    sse = new SseService();
+  });
+
+  test("test_subscribe_to_stream_events", () => {
+    const res = createMockRes();
+    sse.subscribeToStream("stream-1", res as any);
+
+    sse.broadcastToStream("stream-1", { msg: "hello" });
+
+    expect(res.write).toHaveBeenCalledWith(
+      expect.stringContaining("hello")
+    );
+  });
+
+  test("test_subscribe_to_user_events", () => {
+    const res = createMockRes();
+    sse.subscribeToUser("user-1", res as any);
+
+    sse.broadcastToUser("user-1", { msg: "user event" });
+
+    expect(res.write).toHaveBeenCalledWith(
+      expect.stringContaining("user event")
+    );
+  });
+
+  test("test_subscribe_all", () => {
+    const res = createMockRes();
+    sse.subscribeAll(res as any);
+
+    sse.broadcastAll({ msg: "global" });
+
+    expect(res.write).toHaveBeenCalledWith(
+      expect.stringContaining("global")
+    );
+  });
+
+  test("test_client_disconnect_cleaned_up", () => {
+    const res = createMockRes();
+    sse.subscribeToUser("user-1", res as any);
+
+    sse.disconnect(res as any);
+
+    expect(sse.getClientCount()).toBe(0);
+  });
+
+  test("test_broadcast_to_multiple_clients", () => {
+    const res1 = createMockRes();
+    const res2 = createMockRes();
+
+    sse.subscribeToStream("stream-1", res1 as any);
+    sse.subscribeToStream("stream-1", res2 as any);
+
+    sse.broadcastToStream("stream-1", { msg: "multi" });
+
+    expect(res1.write).toHaveBeenCalled();
+    expect(res2.write).toHaveBeenCalled();
+  });
+
+  test("test_no_cross_user_leakage", () => {
+    const resA = createMockRes();
+    const resB = createMockRes();
+
+    sse.subscribeToUser("user-A", resA as any);
+    sse.subscribeToUser("user-B", resB as any);
+
+    sse.broadcastToUser("user-A", { msg: "secret" });
+
+    expect(resA.write).toHaveBeenCalled();
+    expect(resB.write).not.toHaveBeenCalled();
+  });
+});

backend/tests/integration/streamInter.test.ts

@@ -0,0 +1,101 @@
+import request from "supertest";
+import { app } from "../../../test/setup";
+import { db } from "../../db/client";
+
+describe("Streams Integration", () => {
+  let streamId: string;
+  let sseMessages: any[] = [];
+
+  // 🔌 Mock SSE client
+  const mockSseClient = () => {
+    return {
+      write: (data: string) => {
+        try {
+          const parsed = JSON.parse(data.replace(/^data:\s*/, ""));
+          sseMessages.push(parsed);
+        } catch {}
+      },
+      end: jest.fn(),
+    };
+  };
+
+  beforeEach(() => {
+    sseMessages = [];
+  });
+
+  test("POST /v1/streams creates stream + broadcasts SSE", async () => {
+    const sseClient = mockSseClient();
+    app.get("sseService").subscribeAll(sseClient);
+
+    const res = await request(app)
+      .post("/v1/streams")
+      .send({
+        sender: "addr1",
+        recipient: "addr2",
+        amount: 1000,
+        rate: 1,
+      });
+
+    expect(res.status).toBe(201);
+    streamId = res.body.id;
+
+    // ✅ DB check
+    const stream = await db.stream.findUnique({ where: { id: streamId } });
+    expect(stream).toBeTruthy();
+
+    // ✅ SSE broadcast check
+    expect(sseMessages.length).toBeGreaterThan(0);
+  });
+
+  test("GET /v1/streams/{id} returns correct data", async () => {
+    const res = await request(app).get(`/v1/streams/${streamId}`);
+
+    expect(res.status).toBe(200);
+    expect(res.body.id).toBe(streamId);
+  });
+
+  test("GET /v1/streams?sender filters correctly", async () => {
+    const res = await request(app)
+      .get("/v1/streams")
+      .query({ sender: "addr1" });
+
+    expect(res.status).toBe(200);
+    expect(res.body.every((s: any) => s.sender === "addr1")).toBe(true);
+  });
+
+  test("GET /v1/streams/{id}/events paginates", async () => {
+    const res = await request(app)
+      .get(`/v1/streams/${streamId}/events`)
+      .query({ limit: 10 });
+
+    expect(res.status).toBe(200);
+    expect(Array.isArray(res.body.data)).toBe(true);
+  });
+
+  test("Indexer processes TOPPED_UP event", async () => {
+    await request(app)
+      .post(`/v1/indexer/event`)
+      .send({
+        type: "TOPPED_UP",
+        streamId,
+        amount: 500,
+      });
+
+    const stream = await db.stream.findUnique({ where: { id: streamId } });
+
+    expect(stream.depositedAmount).toBeGreaterThanOrEqual(1500);
+  });
+
+  test("Indexer processes CANCELLED event", async () => {
+    await request(app)
+      .post(`/v1/indexer/event`)
+      .send({
+        type: "CANCELLED",
+        streamId,
+      });
+
+    const stream = await db.stream.findUnique({ where: { id: streamId } });
+
+    expect(stream.status).toBe("cancelled");
+  });
+});

contracts/stream_contract/src/stream_contract/src/test.rs

@@ -0,0 +1,62 @@
+#![cfg(test)]
+
+use super::*;
+use soroban_sdk::{testutils::Address as _, Address, Env};
+
+/// Simple deterministic pseudo-random generator (no external deps)
+fn pseudo_rand(seed: &mut u64) -> u64 {
+    *seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1);
+    *seed
+}
+
+#[test]
+fn fuzz_stream_invariants() {
+    let env = Env::default();
+    let mut seed: u64 = 42;
+
+    for _ in 0..10_000 {
+        // Generate random actors
+        let _sender = Address::generate(&env);
+        let _recipient = Address::generate(&env);
+
+        // Generate random values (bounded to avoid overflow)
+        let deposited = (pseudo_rand(&mut seed) % 1_000_000) as i128;
+        let withdrawn = (pseudo_rand(&mut seed) % deposited.max(1) as u64) as i128;
+
+        let elapsed_seconds = (pseudo_rand(&mut seed) % 10_000) as i128;
+        let rate_per_second = (pseudo_rand(&mut seed) % 1_000) as i128;
+
+        let claimable = (rate_per_second * elapsed_seconds)
+            .min(deposited - withdrawn)
+            .max(0);
+
+        let withdrawn_before_cancel = withdrawn;
+        let cancel_refund = (deposited - withdrawn_before_cancel).max(0);
+
+        // 🧠 Invariants
+
+        // 1. withdrawn <= deposited
+        assert!(
+            withdrawn <= deposited,
+            "Invariant failed: withdrawn > deposited"
+        );
+
+        // 2. claimable <= (deposited - withdrawn)
+        assert!(
+            claimable <= (deposited - withdrawn),
+            "Invariant failed: claimable exceeds remaining balance"
+        );
+
+        // 3. rate_per_second * elapsed_seconds >= 0
+        assert!(
+            rate_per_second * elapsed_seconds >= 0,
+            "Invariant failed: negative accrual"
+        );
+
+        // 4. cancel_refund + withdrawn_before_cancel <= deposited
+        assert!(
+            cancel_refund + withdrawn_before_cancel <= deposited,
+            "Invariant failed: total payout exceeds deposit"
+        );
+    }
+}

frontend/src/components/streamCreationForm.ts

@@ -0,0 +1,10 @@
+import { render, screen, fireEvent } from "@testing-library/react";
+import StreamCreationForm from "../components/StreamCreationForm";
+
+test("StreamCreationForm — validation errors shown", () => {
+  render(<StreamCreationForm />);
+
+  fireEvent.click(screen.getByText(/create/i));
+
+  expect(screen.getByText(/invalid/i)).toBeInTheDocument();
+});