proffesor-for-testing__agentic-qe/tests/unit/sync/cloud/postgres-reader.test.ts

/**
 * PostgreSQL Reader Unit Tests
 *
 * Tests for cloud database read operations:
 * - Record reading with environment filtering
 * - Column dropping (source_env, embedding, sync_version)
 * - Column remapping (cloud → local names)
 * - Type transforms (boolean→int, jsonb→text, date→string)
 * - Cloud table format validation
 * - Incremental read with timestamp detection
 * - Count queries
 */

import { describe, it, expect, vi, beforeEach, afterEach} from 'vitest';
import { PostgresReader, createPostgresReader } from '../../../../src/sync/cloud/postgres-reader.js';
import type { CloudWriter, PullSource } from '../../../../src/sync/interfaces.js';

// Mock CloudWriter with query support
function createMockWriter(queryFn?: (sql: string, params?: unknown[]) => Promise<unknown[]>): CloudWriter {
  return {
    connect: vi.fn().mockResolvedValue(undefined),
    beginTransaction: vi.fn().mockResolvedValue(undefined),
    commit: vi.fn().mockResolvedValue(undefined),
    rollback: vi.fn().mockResolvedValue(undefined),
    upsert: vi.fn().mockResolvedValue(0),
    execute: vi.fn().mockResolvedValue(undefined),
    query: vi.fn().mockImplementation(queryFn || (async () => [])),
    close: vi.fn().mockResolvedValue(undefined),
  };
}

describe('PostgresReader', () => {
  let mockWriter: CloudWriter;
  let reader: PostgresReader;

  const basePullSource: PullSource = {
    name: 'test-source',
    cloudTable: 'aqe.qe_patterns',
    localTable: 'qe_patterns',
    enabled: true,
    priority: 'high',
    mode: 'incremental',
  };

  beforeEach(() => {
    mockWriter = createMockWriter();
    reader = createPostgresReader({ writer: mockWriter, environment: 'all' });
  });

  afterEach(() => {
    vi.restoreAllMocks();
  });


  describe('sanitizeCloudTable', () => {
    it('should reject cloud table without dot separator', async () => {
      const badSource = { ...basePullSource, cloudTable: 'qe_patterns' };
      await expect(reader.readAll(badSource)).rejects.toThrow(
        "Invalid cloud table format 'qe_patterns': expected 'schema.table'"
      );
    });

    it('should reject cloud table with leading dot', async () => {
      const badSource = { ...basePullSource, cloudTable: '.qe_patterns' };
      await expect(reader.readAll(badSource)).rejects.toThrow('Invalid cloud table format');
    });

    it('should reject cloud table with trailing dot', async () => {
      const badSource = { ...basePullSource, cloudTable: 'aqe.' };
      await expect(reader.readAll(badSource)).rejects.toThrow('Invalid cloud table format');
    });

    it('should accept valid schema.table format', async () => {
      mockWriter = createMockWriter(async () => []);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });
      const result = await reader.readAll(basePullSource);
      expect(result).toEqual([]);
      expect(mockWriter.query).toHaveBeenCalledWith(
        expect.stringContaining('aqe'),
        []
      );
    });
  });

  describe('readAll', () => {
    it('should read all records without env filter when environment is "all"', async () => {
      mockWriter = createMockWriter(async () => [
        { id: '1', name: 'pattern1', source_env: 'devpod' },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      await reader.readAll(basePullSource);

      expect(mockWriter.query).toHaveBeenCalledWith(
        expect.stringMatching(/SELECT \* FROM.*aqe.*qe_patterns/),
        []
      );
    });

    it('should filter by source_env when environment is not "all"', async () => {
      mockWriter = createMockWriter(async () => []);
      reader = createPostgresReader({ writer: mockWriter, environment: 'devpod' });

      await reader.readAll(basePullSource);

      expect(mockWriter.query).toHaveBeenCalledWith(
        expect.stringContaining('WHERE source_env = $1'),
        ['devpod']
      );
    });

    it('should drop specified columns from records', async () => {
      mockWriter = createMockWriter(async () => [
        { id: '1', name: 'test', source_env: 'devpod', embedding: '[0.1,0.2]', sync_version: 5 },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const source: PullSource = {
        ...basePullSource,
        dropColumns: ['source_env', 'embedding', 'sync_version'],
      };

      const result = await reader.readAll(source);
      expect(result).toHaveLength(1);
      expect(result[0]).toEqual({ id: '1', name: 'test' });
      expect(result[0]).not.toHaveProperty('source_env');
      expect(result[0]).not.toHaveProperty('embedding');
      expect(result[0]).not.toHaveProperty('sync_version');
    });

    it('should rename columns per columnMap', async () => {
      mockWriter = createMockWriter(async () => [
        { state: 'idle', action: 'test', q_value: 0.5 },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const source: PullSource = {
        ...basePullSource,
        cloudTable: 'aqe.qlearning_patterns',
        localTable: 'rl_q_values',
        columnMap: { state: 'state_key', action: 'action_key' },
      };

      const result = await reader.readAll(source);
      expect(result[0]).toHaveProperty('state_key', 'idle');
      expect(result[0]).toHaveProperty('action_key', 'test');
      expect(result[0]).toHaveProperty('q_value', 0.5);
      expect(result[0]).not.toHaveProperty('state');
      expect(result[0]).not.toHaveProperty('action');
    });

    it('should apply boolean-to-int transform', async () => {
      mockWriter = createMockWriter(async () => [
        { id: '1', reusable: true },
        { id: '2', reusable: false },
        { id: '3', reusable: 't' },
        { id: '4', reusable: 'f' },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const source: PullSource = {
        ...basePullSource,
        transforms: { reusable: 'boolean-to-int' },
      };

      const result = await reader.readAll(source);
      expect(result[0].reusable).toBe(1);
      expect(result[1].reusable).toBe(0);
      expect(result[2].reusable).toBe(1);
      expect(result[3].reusable).toBe(0);
    });

    it('should auto-convert JSONB objects to text', async () => {
      mockWriter = createMockWriter(async () => [
        { id: '1', metadata: { key: 'value' } },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.readAll(basePullSource);
      expect(result[0].metadata).toBe('{"key":"value"}');
    });

    it('should auto-convert Date objects to ISO strings', async () => {
      const date = new Date('2026-02-20T10:00:00.000Z');
      mockWriter = createMockWriter(async () => [
        { id: '1', created_at: date },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.readAll(basePullSource);
      // Date objects hit the instanceof Date check and get .toISOString()
      // But pg driver typically returns Date objects which are also typeof 'object'
      // so they may hit the JSON.stringify path first — verify it's a string either way
      expect(typeof result[0].created_at).toBe('string');
      expect(String(result[0].created_at)).toContain('2026-02-20');
    });

    it('should auto-convert PostgreSQL booleans to integers', async () => {
      mockWriter = createMockWriter(async () => [
        { id: '1', is_active: true, requires_tuning: false },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.readAll(basePullSource);
      expect(result[0].is_active).toBe(1);
      expect(result[0].requires_tuning).toBe(0);
    });

    it('should pass through null values', async () => {
      mockWriter = createMockWriter(async () => [
        { id: '1', description: null },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.readAll(basePullSource);
      expect(result[0].description).toBeNull();
    });

    it('should combine drop, rename, and transform on same record', async () => {
      mockWriter = createMockWriter(async () => [
        { agent_id: 'a1', task_id: 't1', reward: 0.9, source_env: 'devpod', sync_version: 1 },
      ]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const source: PullSource = {
        ...basePullSource,
        cloudTable: 'aqe.learning_experiences',
        localTable: 'captured_experiences',
        dropColumns: ['source_env', 'sync_version'],
        columnMap: { agent_id: 'agent', task_id: 'task', reward: 'quality' },
      };

      const result = await reader.readAll(source);
      expect(result[0]).toEqual({ agent: 'a1', task: 't1', quality: 0.9 });
    });
  });

  describe('count', () => {
    it('should return count for all environments', async () => {
      mockWriter = createMockWriter(async () => [{ count: 42 }]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.count(basePullSource);
      expect(result).toBe(42);
    });

    it('should handle PostgreSQL bigint count as string', async () => {
      mockWriter = createMockWriter(async () => [{ count: '11872' }]);
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.count(basePullSource);
      expect(result).toBe(11872);
    });

    it('should return -1 on query failure', async () => {
      mockWriter = createMockWriter(async () => { throw new Error('connection lost'); });
      reader = createPostgresReader({ writer: mockWriter, environment: 'all' });

      const result = await reader.count(basePullSource);
      expect(result).toBe(-1);
    });
  });
});