testing-translated-code

name	testing-translated-code
description	Generates and runs comprehensive tests for translated code using real execution history as fixtures. Covers happy paths, error branches, edge cases, and integration tests against test credentials. Activate when translation is complete and the user is ready to test, validate, or verify the translated code.

Testing Translated Code

You are generating tests that prove the translated code behaves identically to the original n8n workflows. You use real execution data as the source of truth.

---

0. Critical Rules — Read First

0.1 — TEST_USER_PREFIX MUST be ≥ 10¹³

The smoke suite seeds and tears down test users by ID range. The prefix you pick must be larger than any real ID in the production system, or cleanup will silently delete real users and the test-capture buffer will swallow real production messages.

For Telegram specifically: real user IDs are now in the 8–9 billion range (8 × 10⁹) and growing. A safe prefix is 10¹³ = 10_000_000_000_000 (ten trillion). Anything below 10¹⁰ is unsafe today; anything below 10¹² is unsafe within a few years.

# workspace/scripts/smoke_test.py
TEST_USER_PREFIX = 10_000_000_000_000  # 10^13 — above any real Telegram ID

The cleanup endpoint must filter on this prefix:

@router.delete("/admin/db/test-users")
async def delete_test_users():
    await conn.execute(
        "DELETE FROM users WHERE telegram_id >= $1",
        TEST_USER_PREFIX,
    )

And the test-capture buffer in the Telegram client must intercept on the same prefix, so production messages are never captured:

async def send_message(chat_id: int, text: str, **kwargs):
    if chat_id >= TEST_USER_PREFIX:
        TEST_CAPTURES.append({"chat_id": chat_id, "text": text, **kwargs})
        return {"ok": True, "captured": True}
    return await api_call("sendMessage", chat_id=chat_id, text=text, **kwargs)

0.2 — Admin endpoints belong in the dev app from day one

Add a fixed set of admin endpoints to the FastAPI app as part of the translation phase, not retrofitted during testing. They are invaluable for test automation, manual debugging, and benchmarking — and impossible to add later without state pollution.

Required endpoints (bearer-token-protected, dev-only):

POST /admin/trigger/<job_name>            # fire any scheduled job on demand
POST /admin/db/seed-test-user             # seed a deterministic user
POST /admin/db/seed-prayer-cache          # seed cache rows for a test user
GET  /admin/db/user/{telegram_id}         # inspect a row
DELETE /admin/db/test-users               # cleanup by TEST_USER_PREFIX
GET  /admin/test/captures                 # read the in-process capture buffer
DELETE /admin/test/captures               # clear the capture buffer

These map 1:1 to "what does the smoke test need to do?" and remove the need for direct DB access from the test harness.

0.3 — Real-LLM tests are opt-in but documented

Mock-only suites miss the entire prompt-quality dimension. The smoke test should have an explicit --with-llm (or RUN_REAL_LLM=1) flag that enables a small number of real-LLM calls — chat router happy path, daily feedback content, weekly report scenario detection — with assertions on the returned content shape (not exact text). These caught prompt-drift regressions in this migration that mocks would have missed.

---

1. Test Strategy

Three layers of testing, executed in order:

Layer	What It Tests	Data Source
Unit Tests	Each translated function in isolation	Execution history fixtures
Integration Tests	Full workflow chains end-to-end	Test credentials + fixtures
Parity Tests	Code output matches n8n output exactly	Side-by-side comparison

---

2. Unit Test Generation

For each translated function in workspace/src/services/:

Step 1 — Load fixtures from analysis phase

cat migration-state/analysis/<workflow_id>_fixtures.json | jq '
  .[] | select(.node_name == "<function_source_node>")'

Step 2 — Generate test cases

For each fixture (real input → expected output pair):

import pytest
from workspace.src.services.workflow_service import process_node_name


@pytest.mark.asyncio
async def test_process_node_name_case_1():
    """Real execution data from [date] - execution [id]."""
    input_items = [
        # Paste sanitized input from fixture
        {"field": "value", "count": 5}
    ]

    result = await process_node_name(input_items)

    assert len(result) == 1
    assert result[0]["field"] == "expected_value"
    assert result[0]["count"] == 6

Step 3 — Generate edge case tests

For every IF/Switch node, generate tests for EACH branch:

@pytest.mark.asyncio
async def test_route_command_prayer_branch():
    """Should route to prayer handler when command is /pray."""
    items = [{"command": "/pray", "user_id": "test_123"}]
    result = await route_command(items)
    assert result["branch"] == "prayer"


@pytest.mark.asyncio
async def test_route_command_unknown_branch():
    """Should route to fallback when command is unrecognized."""
    items = [{"command": "/nonexistent", "user_id": "test_123"}]
    result = await route_command(items)
    assert result["branch"] == "fallback"

Generate at minimum:

• 1 test per happy path
• 1 test per error/alternate branch
• 1 test with empty input ([])
• 1 test with null/missing fields where applicable

Step 4 — Generate mock-dependent tests

For functions using mocked integrations:

@pytest.mark.asyncio
async def test_send_telegram_dry_run(monkeypatch):
    """In DRY_RUN mode, should log but not send."""
    monkeypatch.setattr("workspace.src.config.settings.DRY_RUN", True)
    items = [{"chat_id": "123", "text": "Hello"}]
    result = await send_telegram_message(items)
    assert result == items  # passthrough in dry run

---

3. Integration Tests

After unit tests pass, test full workflow chains with test credentials:

import pytest
from httpx import AsyncClient, ASGITransport
from workspace.src.main import app


@pytest.mark.asyncio
async def test_full_webhook_to_response():
    """End-to-end: webhook receives prayer, returns confirmation."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/webhook/prayer",
            json={"user_id": "test_123", "prayer": "fajr"}
        )
        assert response.status_code == 200
        assert "accepted" in response.json()["status"]

For integrations with test credentials (strategy: test_credentials):

• Run actual calls against the test instances
• Verify the test database receives expected writes
• Verify the test Telegram bot receives expected messages

For integrations with mocks (strategy: mock):

• Verify mock was called with correct parameters
• Verify the rest of the chain handles mock responses correctly

---

4. Parity Tests

The most critical test layer. Prove that given the same input, the Python code produces the same output as n8n did.

@pytest.mark.asyncio
@pytest.mark.parametrize("fixture", load_fixtures("workflow_1"))
async def test_parity_with_n8n(fixture):
    """Output must match what n8n produced for this exact input."""
    result = await run_full_workflow(fixture["input"])

    # Compare key output fields (not timestamps or IDs)
    for key in fixture["comparison_keys"]:
        assert result[key] == fixture["expected_output"][key], \
            f"Parity failure on key '{key}': " \
            f"got {result[key]}, n8n produced {fixture['expected_output'][key]}"

What to Compare

• Business logic outputs (calculated values, routed branches, transformed data)
• Database writes (same rows, same values in test DB)
• API call payloads (same method, same body sent to external services)

What NOT to Compare

• Timestamps (will always differ)
• Auto-generated IDs
• Execution metadata
• Exact string formatting (whitespace differences are acceptable)

---

5. Run Tests and Report

cd workspace
pytest tests/ -v --tb=short 2>&1 | tee migration-state/test-results.txt

Present results:

Test Results
============
Total:   [n] tests
Passed:  [n]
Failed:  [n]
Skipped: [n]

Unit Tests:        [pass/fail]
Integration Tests: [pass/fail]
Parity Tests:      [pass/fail]

Failed Tests:
  test_route_command_edge_case — AssertionError: expected "prayer", got "fallback"
    → Source node: IF node "Check Command" in workflow "Main Bot"
    → Likely cause: expression translation error
    → Fix: [suggested fix]

If ANY parity test fails:

1. Identify the exact node where divergence occurs
2. Show the user the n8n node config vs the translated code
3. Propose a fix
4. WAIT for user approval before modifying code
5. Re-run tests after fix

Do NOT proceed to Phase 6 until ALL tests pass and the user confirms.

---

6. Required Smoke-Suite Assertions

These assertions MUST be in the smoke suite. Each one corresponds to a bug class that has actually shipped and been caught later by users. Treat this list as a regression floor — add to it, never remove from it.

1. Post-onboarding cache assertion. After running the AI-agent onboarding flow for a test user, query the cache table (e.g., prayer_times_cache) and assert the row exists with non-null typed columns. This single assertion catches the asyncpg time-codec class of bugs (see translating-workflows §0.1) the moment any ::time/::date cast is mistakenly used.

2. Capture-buffer round trip. After every test that should produce an outgoing message, assert the test-capture buffer contains a row with the right chat_id, text substring, and (where applicable) reply_markup structure. Empty-buffer-after-trigger is the most common silent regression.

3. No production-domain leaks. Assert no captured message and no log line contains the production hostname. Catches hardcoded webhook URLs that slipped through translation review.

4. Scheduled-job-via-admin-trigger. For each scheduled job (heartbeat, cache refresh, daily feedback, weekly report, cleanup), call its /admin/trigger/<job> endpoint and assert downstream state changes — not just HTTP 200. Tests that the job actually runs end-to-end, not that the endpoint exists.

5. Real-LLM smoke (opt-in, --with-llm). At least one chat happy path, one daily-feedback generation, and one weekly-report generation hitting the real LLM with content-shape assertions (Arabic characters present, expected section markers present, length within bounds).

6. DELETE-by-prefix safety. Before any test that mutates state, assert TEST_USER_PREFIX >= 10**13. A misconfigured prefix is a fast way to destroy production rows.

---

7. Known Edge Cases

Tests have already failed in the past for these specific reasons. Add cases covering them when generating the suite.

1. TEST_USER_PREFIX too low — see §0.1.

2. docker stats 1-second sample alternation — postgres routinely shows 0% then 100% in successive samples, making per-test latency measurements noisy. For benchmarking inside tests, prefer cgroup cpu.stat deltas over docker stats.

3. Mac↔server network adds ~150 ms per call — run the smoke suite ON the deployed server, not from a developer laptop. A test that times webhook handling needs server-local latency to be fair.

4. DRY_RUN=true swallows output silently — if the suite expects messages to actually flow through the test-capture buffer, it must run with DRY_RUN=false (or have the dry-run path also append to the capture buffer). Otherwise the buffer is always empty and assertions silently pass.

5. Inserts that hit JSONB columns must use json.dumps, not raw dicts — asyncpg accepts text → ::jsonb but not dict. Tests that seed JSONB columns directly will fail with TypeError: dict has no encoder.

6. Telegram callback data length cap (64 bytes) — generate at least one test that exercises the longest possible callback payload your bot constructs (longest prayer name, longest date format, longest user ID). Catches any composite payload that runs over the limit only in production.

7. is_paused filter bypass — write at least one test where a paused user exists in the seeded data and assert they receive zero messages from any scheduled job.