wrongmove/frontend/src/services/__tests__/streamingService.test.ts

import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import { mockUser, createMockFeature } from '@/__tests__/helpers';
import { streamListingGeoJSON } from '@/services/streamingService';
import { ApiError } from '@/types';
import type { ParameterValues } from '@/components/FilterPanel';

const defaultParams: ParameterValues = {
  metric: 'qmprice',
  listing_type: 'RENT',
  min_bedrooms: 1,
  max_bedrooms: 3,
  min_price: 1000,
  max_price: 5000,
  district: '',
};

function createMockResponse(lines: string[]) {
  const body = lines.join('\n') + '\n';
  const encoder = new TextEncoder();
  const stream = new ReadableStream({
    start(controller) {
      controller.enqueue(encoder.encode(body));
      controller.close();
    },
  });
  return new Response(stream, {
    status: 200,
    headers: { 'Content-Type': 'application/x-ndjson' },
  });
}

describe('streamingService', () => {
  const originalFetch = globalThis.fetch;

  afterEach(() => {
    globalThis.fetch = originalFetch;
  });

  it('yields feature batches', async () => {
    const features = [createMockFeature(), createMockFeature({ rooms: 3 })];

    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 2 }),
        JSON.stringify({ type: 'batch', features }),
        JSON.stringify({ type: 'complete', total: 2 }),
      ]),
    );

    const batches: unknown[][] = [];
    for await (const batch of streamListingGeoJSON(mockUser(), defaultParams)) {
      batches.push(batch);
    }

    expect(batches).toHaveLength(1);
    expect(batches[0]).toHaveLength(2);
  });

  it('calls onProgress with metadata', async () => {
    const onProgress = vi.fn();

    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 42 }),
        JSON.stringify({ type: 'complete', total: 0 }),
      ]),
    );

    const gen = streamListingGeoJSON(mockUser(), defaultParams, onProgress);
    // Consume the generator
    for await (const _ of gen) { /* drain */ }

    expect(onProgress).toHaveBeenCalledWith({ count: 0, total: 42 });
  });

  it('calls onProgress with incrementing count on batches', async () => {
    const features1 = [createMockFeature()];
    const features2 = [createMockFeature(), createMockFeature()];
    const onProgress = vi.fn();

    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 3 }),
        JSON.stringify({ type: 'batch', features: features1 }),
        JSON.stringify({ type: 'batch', features: features2 }),
        JSON.stringify({ type: 'complete', total: 3 }),
      ]),
    );

    for await (const _ of streamListingGeoJSON(mockUser(), defaultParams, onProgress)) { /* drain */ }

    // metadata: count=0, batch1: count=1, batch2: count=3, complete: count=3
    expect(onProgress).toHaveBeenCalledWith({ count: 1 });
    expect(onProgress).toHaveBeenCalledWith({ count: 3 });
  });

  it('calls onProgress on complete message', async () => {
    const onProgress = vi.fn();

    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 5 }),
        JSON.stringify({ type: 'complete', total: 5 }),
      ]),
    );

    for await (const _ of streamListingGeoJSON(mockUser(), defaultParams, onProgress)) { /* drain */ }

    expect(onProgress).toHaveBeenCalledWith({ count: 5, total: 5 });
  });

  it('handles multiple batches and yields each separately', async () => {
    const batch1 = [createMockFeature({ rooms: 1 })];
    const batch2 = [createMockFeature({ rooms: 2 })];

    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 2 }),
        JSON.stringify({ type: 'batch', features: batch1 }),
        JSON.stringify({ type: 'batch', features: batch2 }),
        JSON.stringify({ type: 'complete', total: 2 }),
      ]),
    );

    const batches: unknown[][] = [];
    for await (const batch of streamListingGeoJSON(mockUser(), defaultParams)) {
      batches.push(batch);
    }

    expect(batches).toHaveLength(2);
    expect(batches[0]).toHaveLength(1);
    expect(batches[1]).toHaveLength(1);
  });

  it('handles abort signal', async () => {
    const controller = new AbortController();
    controller.abort();

    globalThis.fetch = vi.fn().mockRejectedValue(new DOMException('Aborted', 'AbortError'));

    const batches: unknown[][] = [];
    try {
      for await (const batch of streamListingGeoJSON(mockUser(), defaultParams, undefined, { signal: controller.signal })) {
        batches.push(batch);
      }
    } catch {
      // Expected: fetch throws on aborted signal
    }

    expect(batches).toHaveLength(0);
  });

  it('handles malformed JSON lines without crashing', async () => {
    const consoleSpy = vi.spyOn(console, 'error').mockImplementation(() => {});

    const features = [createMockFeature()];
    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 1 }),
        'this is not valid json{{{',
        JSON.stringify({ type: 'batch', features }),
        JSON.stringify({ type: 'complete', total: 1 }),
      ]),
    );

    const batches: unknown[][] = [];
    for await (const batch of streamListingGeoJSON(mockUser(), defaultParams)) {
      batches.push(batch);
    }

    // Should still yield the valid batch
    expect(batches).toHaveLength(1);
    expect(consoleSpy).toHaveBeenCalled();
    consoleSpy.mockRestore();
  });

  it('handles empty stream with no batches', async () => {
    globalThis.fetch = vi.fn().mockResolvedValue(
      createMockResponse([
        JSON.stringify({ type: 'metadata', total_expected: 0 }),
        JSON.stringify({ type: 'complete', total: 0 }),
      ]),
    );

    const batches: unknown[][] = [];
    for await (const batch of streamListingGeoJSON(mockUser(), defaultParams)) {
      batches.push(batch);
    }

    expect(batches).toHaveLength(0);
  });

  it('throws ApiError on HTTP error response', async () => {
    globalThis.fetch = vi.fn().mockResolvedValue(
      new Response(null, { status: 401 }),
    );

    await expect(async () => {
      for await (const _ of streamListingGeoJSON(mockUser(), defaultParams)) { /* drain */ }
    }).rejects.toThrow(ApiError);
  });

  it('throws when response body is null', async () => {
    const response = new Response(null, { status: 200 });
    // Force body to null by creating a response without a body that reports ok
    Object.defineProperty(response, 'body', { value: null });

    globalThis.fetch = vi.fn().mockResolvedValue(response);

    await expect(async () => {
      for await (const _ of streamListingGeoJSON(mockUser(), defaultParams)) { /* drain */ }
    }).rejects.toThrow('No response body');
  });
});