Test-Driven Development (TDD) Specification

CODITECT Context Intelligence Platform

Document Version: 1.0.0 Date: November 26, 2025 Status: Approved for Implementation Compliance: IEEE 829-2008 Test Documentation Standard

---

Document Control

Role	Name	Date
Author	CODITECT Engineering Team	2025-11-26
Reviewer	Technical Architect	2025-11-26
Approver	CTO	2025-11-26

Revision History

Version	Date	Author	Changes
1.0.0	2025-11-26	Engineering Team	Initial TDD specification

---

Table of Contents

1. Test Strategy
2. Test Environment
3. Unit Tests
4. Integration Tests
5. End-to-End Tests
6. Performance Tests
7. Security Tests
8. Test Automation
9. Quality Gates
10. Test Data Management

---

1. Test Strategy

1.1 Testing Approach

Philosophy: Test-Driven Development (TDD) with "Red-Green-Refactor" cycle

┌─────────────────────────────────────────────────────────────┐
│ TDD CYCLE                                                    │
│                                                              │
│  RED → Write failing test for new feature                   │
│   ↓                                                          │
│  GREEN → Write minimal code to pass test                    │
│   ↓                                                          │
│  REFACTOR → Improve code while keeping tests green          │
│   ↓                                                          │
│  REPEAT → Next feature                                      │
└─────────────────────────────────────────────────────────────┘

1.2 Test Pyramid

                   ┌─────────────┐
                   │   E2E Tests │  (10% of tests)
                   │   ~50 tests │  Manual + Automated
                   └─────────────┘
                  ┌───────────────┐
                  │  Integration  │   (30% of tests)
                  │  Tests        │   ~150 tests
                  │  ~150 tests   │   API, DB, Services
                  └───────────────┘
               ┌─────────────────────┐
               │   Unit Tests        │    (60% of tests)
               │   ~300 tests        │    ~300 tests
               │   Fast, Isolated    │    Fast, Isolated
               └─────────────────────┘

1.3 Coverage Requirements

Test Type	Target Coverage	Critical Path Coverage	Acceptance Criteria
Unit Tests	80%+	95%+	All core services covered
Integration Tests	70%+	90%+	All API endpoints covered
E2E Tests	50%+	100%	All user workflows covered
Performance Tests	N/A	N/A	Meet SLA requirements
Security Tests	N/A	100%	OWASP Top 10 coverage

1.4 Test Categories

Category	Framework	Run Frequency	Duration
Unit Tests	pytest	Every commit	<2 minutes
Integration Tests	pytest + testcontainers	Every PR	<10 minutes
E2E Tests	Playwright + Gherkin	Nightly	<30 minutes
Performance Tests	Locust	Weekly	<2 hours
Security Tests	OWASP ZAP + Bandit	Every release	<1 hour

---

2. Test Environment

2.1 Local Development Environment

# docker-compose.test.yml
version: '3.8'

services:
  postgres-test:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: context_intelligence_test
      POSTGRES_USER: test_user
      POSTGRES_PASSWORD: test_password
    ports:
      - "5433:5432"
    tmpfs:
      - /var/lib/postgresql/data  # In-memory for speed

  weaviate-test:
    image: semitechnologies/weaviate:1.23.0
    environment:
      QUERY_DEFAULTS_LIMIT: 20
      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
      PERSISTENCE_DATA_PATH: '/tmp/weaviate'
    ports:
      - "8081:8080"
    tmpfs:
      - /tmp/weaviate

  redis-test:
    image: redis:7-alpine
    ports:
      - "6380:6379"
    tmpfs:
      - /data

2.2 CI/CD Test Environment

# .github/workflows/test.yml
name: Test Suite

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-test.txt
      - name: Run unit tests
        run: pytest tests/unit/ -v --cov=core --cov-report=xml
      - name: Upload coverage
        uses: codecov/codecov-action@v3

  integration-tests:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_DB: test_db
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_password
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      redis:
        image: redis:7
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
      - uses: actions/checkout@v4
      - name: Run integration tests
        run: pytest tests/integration/ -v

2.3 Staging Environment

• Purpose: Pre-production testing with production-like data
• Infrastructure: GCP Cloud Run (sandbox project)
• Data: Anonymized production data subset (10% sample)
• Access: QA team + automated E2E tests

---

3. Unit Tests

3.1 Test Structure

# tests/unit/services/test_search_service.py

import pytest
from unittest.mock import Mock, AsyncMock
from uuid import UUID, uuid4
from core.services.search_service import SearchService
from core.models.conversation import Conversation

class TestSearchService:
    """Unit tests for SearchService using TDD methodology."""

    @pytest.fixture
    def mock_conversation_repo(self):
        """Mock ConversationRepository for isolated testing."""
        return Mock()

    @pytest.fixture
    def mock_weaviate_client(self):
        """Mock WeaviateClient for isolated testing."""
        return Mock()

    @pytest.fixture
    def search_service(self, mock_conversation_repo, mock_weaviate_client):
        """Fixture for SearchService with mocked dependencies."""
        return SearchService(
            conversation_repo=mock_conversation_repo,
            weaviate_client=mock_weaviate_client,
            rrf_k=60
        )

    @pytest.mark.asyncio
    async def test_hybrid_search_returns_fused_results(
        self,
        search_service,
        mock_conversation_repo,
        mock_weaviate_client
    ):
        """
        GIVEN: A search query with keyword and semantic results
        WHEN: hybrid_search() is called
        THEN: Results are fused using RRF and returned in correct order
        """
        # ARRANGE
        org_id = uuid4()
        query = "authentication bug"

        # Mock keyword search results
        keyword_results = [
            Conversation(id=uuid4(), title="Auth bug fix", rank=1),
            Conversation(id=uuid4(), title="Login issue", rank=2),
        ]
        mock_conversation_repo.search_by_keyword = AsyncMock(
            return_value=keyword_results
        )

        # Mock semantic search results
        semantic_results = [
            Conversation(id=uuid4(), title="Security vulnerability", rank=1),
            Conversation(id=uuid4(), title="Auth bug fix", rank=2),  # Overlap
        ]
        mock_weaviate_client.search_semantic = AsyncMock(
            return_value=semantic_results
        )

        # ACT
        results = await search_service.hybrid_search(
            organization_id=org_id,
            query=query,
            limit=10,
            alpha=0.5
        )

        # ASSERT
        assert len(results) > 0
        assert results[0].title == "Auth bug fix"  # Should rank highest
        mock_conversation_repo.search_by_keyword.assert_called_once()
        mock_weaviate_client.search_semantic.assert_called_once()

    @pytest.mark.asyncio
    async def test_hybrid_search_respects_organization_isolation(
        self,
        search_service,
        mock_conversation_repo,
        mock_weaviate_client
    ):
        """
        GIVEN: Two different organizations
        WHEN: hybrid_search() is called for each organization
        THEN: Results are isolated by organization_id (multi-tenancy)
        """
        # ARRANGE
        org_1_id = uuid4()
        org_2_id = uuid4()
        query = "test query"

        mock_conversation_repo.search_by_keyword = AsyncMock(return_value=[])
        mock_weaviate_client.search_semantic = AsyncMock(return_value=[])

        # ACT
        await search_service.hybrid_search(org_1_id, query, limit=10)
        await search_service.hybrid_search(org_2_id, query, limit=10)

        # ASSERT
        calls = mock_conversation_repo.search_by_keyword.call_args_list
        assert calls[0][0][0] == org_1_id  # First call used org_1_id
        assert calls[1][0][0] == org_2_id  # Second call used org_2_id

    def test_rrf_fusion_algorithm_correctness(self, search_service):
        """
        GIVEN: Two ranked lists with known overlap
        WHEN: _rrf_fusion() is called
        THEN: Fusion scores are calculated correctly using RRF formula

        RRF Formula: score = Σ(1 / (k + rank))
        k = 60 (default constant)
        """
        # ARRANGE
        conv_1 = Conversation(id=uuid4(), title="Conv 1")
        conv_2 = Conversation(id=uuid4(), title="Conv 2")
        conv_3 = Conversation(id=uuid4(), title="Conv 3")

        keyword_results = [(conv_1, 1), (conv_2, 2)]  # (conversation, rank)
        semantic_results = [(conv_2, 1), (conv_3, 2)]  # conv_2 appears in both

        # ACT
        fused = search_service._rrf_fusion(
            keyword_results,
            semantic_results,
            alpha=0.5
        )

        # ASSERT
        # conv_2 should rank highest (appears in both lists)
        assert fused[0].id == conv_2.id

        # Manual RRF score calculation for conv_2:
        # keyword_score = 1 / (60 + 2) = 1/62 ≈ 0.0161
        # semantic_score = 1 / (60 + 1) = 1/61 ≈ 0.0164
        # fused_score = 0.5 * 0.0161 + 0.5 * 0.0164 ≈ 0.01625
        expected_score = 0.5 * (1/62) + 0.5 * (1/61)
        assert abs(fused[0].score - expected_score) < 0.0001

    @pytest.mark.asyncio
    async def test_hybrid_search_handles_empty_results(
        self,
        search_service,
        mock_conversation_repo,
        mock_weaviate_client
    ):
        """
        GIVEN: No matching conversations in database or vector store
        WHEN: hybrid_search() is called
        THEN: Empty list is returned without errors
        """
        # ARRANGE
        org_id = uuid4()
        query = "nonexistent query"

        mock_conversation_repo.search_by_keyword = AsyncMock(return_value=[])
        mock_weaviate_client.search_semantic = AsyncMock(return_value=[])

        # ACT
        results = await search_service.hybrid_search(org_id, query, limit=10)

        # ASSERT
        assert results == []

    @pytest.mark.asyncio
    async def test_hybrid_search_alpha_weight_affects_ranking(
        self,
        search_service,
        mock_conversation_repo,
        mock_weaviate_client
    ):
        """
        GIVEN: Different alpha values (0.0 = semantic only, 1.0 = keyword only)
        WHEN: hybrid_search() is called with different alpha values
        THEN: Result ranking changes based on alpha weight
        """
        # ARRANGE
        org_id = uuid4()
        query = "test"

        keyword_top = Conversation(id=uuid4(), title="Keyword winner")
        semantic_top = Conversation(id=uuid4(), title="Semantic winner")

        mock_conversation_repo.search_by_keyword = AsyncMock(
            return_value=[(keyword_top, 1), (semantic_top, 5)]
        )
        mock_weaviate_client.search_semantic = AsyncMock(
            return_value=[(semantic_top, 1), (keyword_top, 5)]
        )

        # ACT - Alpha = 1.0 (keyword only)
        results_keyword = await search_service.hybrid_search(
            org_id, query, limit=10, alpha=1.0
        )

        # ACT - Alpha = 0.0 (semantic only)
        results_semantic = await search_service.hybrid_search(
            org_id, query, limit=10, alpha=0.0
        )

        # ASSERT
        assert results_keyword[0].id == keyword_top.id  # Keyword winner first
        assert results_semantic[0].id == semantic_top.id  # Semantic winner first

3.2 Unit Test Coverage Map

Component	Test File	Test Count	Coverage Target	Critical Tests
SearchService	test_search_service.py	15	90%+	RRF fusion, multi-tenancy
CorrelationService	test_correlation_service.py	20	90%+	Temporal/semantic scoring
AnalyticsService	test_analytics_service.py	12	85%+	Team velocity, insights
JWTAuthenticator	test_jwt_authenticator.py	10	95%+	Token validation, expiry
RBACEnforcer	test_rbac_enforcer.py	15	95%+	Role permissions
FeatureGateService	test_feature_gate.py	18	90%+	Tier-based gating
QuotaChecker	test_quota_checker.py	10	90%+	Usage limits
ConversationRepository	test_conversation_repo.py	25	85%+	CRUD operations, RLS
CommitRepository	test_commit_repo.py	15	85%+	GitHub webhook parsing
WeaviateClient	test_weaviate_client.py	12	80%+	Vector operations

Total Unit Tests: ~152 tests Estimated Execution Time: <2 minutes

3.3 Test Data Fixtures

# tests/conftest.py

import pytest
from uuid import uuid4
from datetime import datetime, timedelta
from core.models.organization import Organization
from core.models.user import User
from core.models.conversation import Conversation

@pytest.fixture
def sample_organization():
    """Sample organization for testing."""
    return Organization(
        id=uuid4(),
        name="Test Corp",
        subdomain="testcorp",
        tier="pro",
        created_at=datetime.utcnow()
    )

@pytest.fixture
def sample_user(sample_organization):
    """Sample user for testing."""
    return User(
        id=uuid4(),
        organization_id=sample_organization.id,
        email="test@testcorp.com",
        name="Test User",
        role="admin",
        created_at=datetime.utcnow()
    )

@pytest.fixture
def sample_conversation(sample_organization, sample_user):
    """Sample conversation for testing."""
    return Conversation(
        id=uuid4(),
        organization_id=sample_organization.id,
        user_id=sample_user.id,
        title="Authentication bug discussion",
        created_at=datetime.utcnow() - timedelta(hours=2),
        updated_at=datetime.utcnow(),
        message_count=15,
        token_count=3500
    )

@pytest.fixture
def sample_conversations_batch(sample_organization, sample_user):
    """Batch of 50 conversations for testing list operations."""
    return [
        Conversation(
            id=uuid4(),
            organization_id=sample_organization.id,
            user_id=sample_user.id,
            title=f"Conversation {i}",
            created_at=datetime.utcnow() - timedelta(hours=i),
            message_count=10 + i,
            token_count=1000 + (i * 100)
        )
        for i in range(50)
    ]

---

4. Integration Tests

4.1 API Integration Tests

# tests/integration/api/test_conversation_endpoints.py

import pytest
from httpx import AsyncClient
from fastapi import status
from uuid import uuid4
from core.main import app
from core.database import get_db
from tests.factories import OrganizationFactory, UserFactory

@pytest.mark.integration
class TestConversationEndpoints:
    """Integration tests for Conversation API endpoints."""

    @pytest.fixture
    async def client(self):
        """HTTP client for API testing."""
        async with AsyncClient(app=app, base_url="http://test") as client:
            yield client

    @pytest.fixture
    async def authenticated_client(self, client):
        """Authenticated HTTP client with valid JWT."""
        # Create test organization and user
        org = await OrganizationFactory.create(tier="pro")
        user = await UserFactory.create(organization_id=org.id, role="admin")

        # Login and get JWT token
        response = await client.post("/auth/login", json={
            "email": user.email,
            "password": "test_password"
        })
        assert response.status_code == status.HTTP_200_OK
        token = response.json()["access_token"]

        # Add token to client headers
        client.headers["Authorization"] = f"Bearer {token}"
        client.headers["X-Organization-ID"] = str(org.id)

        yield client

    @pytest.mark.asyncio
    async def test_create_conversation_success(self, authenticated_client):
        """
        GIVEN: Valid conversation data
        WHEN: POST /conversations is called
        THEN: Conversation is created with 201 status
        """
        # ARRANGE
        payload = {
            "title": "New conversation",
            "metadata": {"source": "claude-code"}
        }

        # ACT
        response = await authenticated_client.post("/conversations", json=payload)

        # ASSERT
        assert response.status_code == status.HTTP_201_CREATED
        data = response.json()
        assert data["title"] == "New conversation"
        assert "id" in data
        assert "created_at" in data

    @pytest.mark.asyncio
    async def test_create_conversation_requires_authentication(self, client):
        """
        GIVEN: No authentication token
        WHEN: POST /conversations is called
        THEN: 401 Unauthorized is returned
        """
        # ACT
        response = await client.post("/conversations", json={
            "title": "Unauthorized attempt"
        })

        # ASSERT
        assert response.status_code == status.HTTP_401_UNAUTHORIZED

    @pytest.mark.asyncio
    async def test_list_conversations_with_pagination(self, authenticated_client):
        """
        GIVEN: 50 conversations in database
        WHEN: GET /conversations with pagination params
        THEN: Correct page of results returned with pagination metadata
        """
        # ARRANGE - Create 50 test conversations
        for i in range(50):
            await authenticated_client.post("/conversations", json={
                "title": f"Conversation {i}"
            })

        # ACT - Get page 2 with 20 items per page
        response = await authenticated_client.get(
            "/conversations?page=2&limit=20"
        )

        # ASSERT
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        assert data["page"] == 2
        assert data["limit"] == 20
        assert len(data["items"]) == 20
        assert data["total"] == 50
        assert data["pages"] == 3

    @pytest.mark.asyncio
    async def test_get_conversation_by_id(self, authenticated_client):
        """
        GIVEN: Conversation exists in database
        WHEN: GET /conversations/{id} is called
        THEN: Conversation details returned with 200 status
        """
        # ARRANGE
        create_response = await authenticated_client.post("/conversations", json={
            "title": "Test conversation"
        })
        conversation_id = create_response.json()["id"]

        # ACT
        response = await authenticated_client.get(f"/conversations/{conversation_id}")

        # ASSERT
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        assert data["id"] == conversation_id
        assert data["title"] == "Test conversation"

    @pytest.mark.asyncio
    async def test_get_conversation_not_found(self, authenticated_client):
        """
        GIVEN: Conversation ID does not exist
        WHEN: GET /conversations/{id} is called
        THEN: 404 Not Found is returned
        """
        # ACT
        nonexistent_id = str(uuid4())
        response = await authenticated_client.get(f"/conversations/{nonexistent_id}")

        # ASSERT
        assert response.status_code == status.HTTP_404_NOT_FOUND

    @pytest.mark.asyncio
    async def test_conversation_multi_tenant_isolation(self, client):
        """
        GIVEN: Two organizations with conversations
        WHEN: User from org A tries to access org B's conversation
        THEN: 404 Not Found is returned (data isolation)
        """
        # ARRANGE - Create org A and conversation
        org_a = await OrganizationFactory.create()
        user_a = await UserFactory.create(organization_id=org_a.id)
        token_a = await self._get_auth_token(client, user_a)

        client.headers["Authorization"] = f"Bearer {token_a}"
        client.headers["X-Organization-ID"] = str(org_a.id)

        response = await client.post("/conversations", json={"title": "Org A conv"})
        conv_id = response.json()["id"]

        # ARRANGE - Create org B user
        org_b = await OrganizationFactory.create()
        user_b = await UserFactory.create(organization_id=org_b.id)
        token_b = await self._get_auth_token(client, user_b)

        # ACT - User B tries to access org A's conversation
        client.headers["Authorization"] = f"Bearer {token_b}"
        client.headers["X-Organization-ID"] = str(org_b.id)

        response = await client.get(f"/conversations/{conv_id}")

        # ASSERT
        assert response.status_code == status.HTTP_404_NOT_FOUND

4.2 Database Integration Tests

# tests/integration/database/test_conversation_repository.py

import pytest
from uuid import uuid4
from sqlalchemy.ext.asyncio import AsyncSession
from core.repositories.conversation_repository import ConversationRepository
from core.models.conversation import Conversation
from tests.factories import OrganizationFactory, UserFactory

@pytest.mark.integration
class TestConversationRepository:
    """Integration tests for ConversationRepository with real PostgreSQL."""

    @pytest.fixture
    async def db_session(self):
        """Real database session for integration testing."""
        # Use testcontainers to spin up PostgreSQL
        async with get_test_db_session() as session:
            yield session

    @pytest.fixture
    async def conversation_repo(self, db_session):
        """ConversationRepository with real database."""
        return ConversationRepository(db_session)

    @pytest.mark.asyncio
    async def test_create_and_retrieve_conversation(
        self,
        conversation_repo,
        db_session
    ):
        """
        GIVEN: Valid conversation data
        WHEN: create() and get_by_id() are called
        THEN: Conversation is persisted and retrievable
        """
        # ARRANGE
        org = await OrganizationFactory.create(db_session)
        user = await UserFactory.create(db_session, organization_id=org.id)

        conversation = Conversation(
            id=uuid4(),
            organization_id=org.id,
            user_id=user.id,
            title="Integration test conversation"
        )

        # ACT - Create
        created = await conversation_repo.create(conversation)

        # ACT - Retrieve
        retrieved = await conversation_repo.get_by_id(
            conversation_id=created.id,
            organization_id=org.id
        )

        # ASSERT
        assert retrieved is not None
        assert retrieved.id == created.id
        assert retrieved.title == "Integration test conversation"

    @pytest.mark.asyncio
    async def test_row_level_security_enforcement(
        self,
        conversation_repo,
        db_session
    ):
        """
        GIVEN: Two organizations with conversations
        WHEN: get_by_id() is called with wrong organization_id
        THEN: None is returned (RLS blocks access)
        """
        # ARRANGE - Create two organizations
        org_a = await OrganizationFactory.create(db_session)
        org_b = await OrganizFactory.create(db_session)
        user_a = await UserFactory.create(db_session, organization_id=org_a.id)

        conversation = await conversation_repo.create(Conversation(
            id=uuid4(),
            organization_id=org_a.id,
            user_id=user_a.id,
            title="Org A conversation"
        ))

        # ACT - Try to retrieve with org B's ID
        retrieved = await conversation_repo.get_by_id(
            conversation_id=conversation.id,
            organization_id=org_b.id  # Wrong org!
        )

        # ASSERT
        assert retrieved is None  # RLS blocks access

    @pytest.mark.asyncio
    async def test_keyword_search_with_postgresql_fts(
        self,
        conversation_repo,
        db_session
    ):
        """
        GIVEN: Conversations with indexed text content
        WHEN: search_by_keyword() is called
        THEN: PostgreSQL full-text search returns relevant results
        """
        # ARRANGE
        org = await OrganizationFactory.create(db_session)
        user = await UserFactory.create(db_session, organization_id=org.id)

        # Create conversations with searchable content
        await conversation_repo.create(Conversation(
            id=uuid4(),
            organization_id=org.id,
            user_id=user.id,
            title="Authentication bug in login system"
        ))
        await conversation_repo.create(Conversation(
            id=uuid4(),
            organization_id=org.id,
            user_id=user.id,
            title="Database performance optimization"
        ))

        # ACT
        results = await conversation_repo.search_by_keyword(
            organization_id=org.id,
            query="authentication login",
            limit=10
        )

        # ASSERT
        assert len(results) == 1
        assert "Authentication" in results[0].title

4.3 External Service Integration Tests

# tests/integration/external/test_github_webhook.py

import pytest
from httpx import AsyncClient
from unittest.mock import patch
from core.main import app
from tests.factories import OrganizationFactory

@pytest.mark.integration
class TestGitHubWebhookIntegration:
    """Integration tests for GitHub webhook processing."""

    @pytest.fixture
    async def client(self):
        async with AsyncClient(app=app, base_url="http://test") as client:
            yield client

    @pytest.mark.asyncio
    async def test_github_push_event_creates_commit(self, client):
        """
        GIVEN: Valid GitHub push event webhook payload
        WHEN: POST /commits/webhook/github is called
        THEN: Commit is created in database
        """
        # ARRANGE
        org = await OrganizationFactory.create()

        github_payload = {
            "ref": "refs/heads/main",
            "commits": [
                {
                    "id": "abc123def456",
                    "message": "feat: Add user authentication\n\nImplemented JWT-based auth",
                    "timestamp": "2025-11-26T10:00:00Z",
                    "author": {
                        "name": "John Doe",
                        "email": "john@example.com"
                    }
                }
            ],
            "repository": {
                "full_name": "testcorp/backend"
            }
        }

        # ACT
        response = await client.post(
            "/commits/webhook/github",
            json=github_payload,
            headers={
                "X-GitHub-Event": "push",
                "X-Hub-Signature-256": self._generate_signature(github_payload)
            }
        )

        # ASSERT
        assert response.status_code == 200
        # Verify commit was created
        commits = await client.get(
            f"/commits?repo=testcorp/backend",
            headers={"X-Organization-ID": str(org.id)}
        )
        assert len(commits.json()["items"]) == 1

4.4 Integration Test Coverage Map

Component	Test File	Test Count	Coverage Focus
API Endpoints	test_conversation_endpoints.py	25	All CRUD operations
API Endpoints	test_search_endpoints.py	15	Hybrid search API
API Endpoints	test_commit_endpoints.py	12	Commit tracking
API Endpoints	test_analytics_endpoints.py	18	Team analytics
Database	test_conversation_repo.py	20	PostgreSQL + RLS
Database	test_weaviate_integration.py	15	Vector operations
External Services	test_github_webhook.py	10	GitHub integration
External Services	test_openai_api.py	8	OpenAI embeddings
Authentication	test_auth_flow.py	12	Login, JWT, OAuth

Total Integration Tests: ~135 tests Estimated Execution Time: <10 minutes

---

5. End-to-End Tests

5.1 E2E Test Scenarios (Gherkin Format)

# tests/e2e/features/conversation_lifecycle.feature

Feature: Conversation Lifecycle Management
  As a developer using AI coding assistants
  I want to save and search my AI conversations
  So that I can reference past discussions and solutions

  Background:
    Given I am a registered user in organization "TestCorp"
    And I am logged in with role "admin"
    And I have 0 conversations in my history

  Scenario: Create and retrieve a new conversation
    Given I am on the conversation list page
    When I click "New Conversation"
    And I enter title "Authentication bug discussion"
    And I add 5 messages to the conversation
    And I click "Save Conversation"
    Then I should see "Conversation saved successfully"
    And the conversation should appear in my conversation list
    And the conversation should have 5 messages

  Scenario: Search conversations by keyword
    Given I have the following conversations:
      | Title                        | Messages | Date       |
      | Authentication bug fix       | 10       | 2025-11-20 |
      | Database optimization tips   | 15       | 2025-11-21 |
      | React component refactoring  | 8        | 2025-11-22 |
    When I enter "authentication" in the search box
    And I click "Search"
    Then I should see 1 search result
    And the result should be "Authentication bug fix"

  Scenario: Semantic search finds related concepts
    Given I have a conversation titled "JWT token expiration handling"
    When I search for "user login security"
    Then I should see the conversation in results
    Because semantic search matches related concepts

  Scenario: Multi-tenant data isolation
    Given I am in organization "TestCorp"
    And another organization "CompetitorCorp" exists with 50 conversations
    When I view my conversation list
    Then I should see only my organization's conversations
    And I should see 0 conversations from "CompetitorCorp"

  Scenario: Quota enforcement for starter tier
    Given my organization has "starter" tier
    And starter tier allows 100 conversations per month
    And I have created 99 conversations this month
    When I try to create conversation number 100
    Then it should succeed
    When I try to create conversation number 101
    Then I should see error "Monthly quota exceeded. Upgrade to Pro for unlimited conversations."
    And the conversation should not be created

# tests/e2e/features/commit_correlation.feature

Feature: Conversation-Commit Correlation
  As a development team lead
  I want to link AI conversations to git commits
  So that I can understand what discussions led to code changes

  Background:
    Given I am a registered user
    And GitHub webhook is configured for repository "testcorp/backend"

  Scenario: Automatic correlation by time proximity
    Given I had a conversation titled "Fix authentication bug" at 10:00 AM
    When a commit with message "fix: Resolve JWT expiration issue" is pushed at 10:15 AM
    Then the conversation and commit should be automatically linked
    And correlation confidence should be "high" (>80%)

  Scenario: Manual commit linking
    Given I have a conversation "Database schema migration plan"
    And I have a commit "feat: Add users table migration"
    When I view the conversation details
    And I click "Link Commit"
    And I select the commit from the list
    Then the commit should be linked to the conversation
    And correlation type should be "manual"

  Scenario: View commit history for a conversation
    Given I have a conversation with 3 linked commits
    When I view the conversation timeline
    Then I should see all 3 commits in chronological order
    And each commit should show author, message, and timestamp
    And I can click each commit to view diff

5.2 E2E Test Implementation (Playwright)

# tests/e2e/test_conversation_lifecycle.py

import pytest
from playwright.async_api import async_playwright, Page

@pytest.mark.e2e
class TestConversationLifecycle:
    """End-to-end tests for conversation lifecycle using Playwright."""

    @pytest.fixture
    async def page(self):
        """Browser page for E2E testing."""
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            context = await browser.new_context()
            page = await context.new_page()

            # Login
            await page.goto("http://localhost:3000/login")
            await page.fill('input[name="email"]', "test@testcorp.com")
            await page.fill('input[name="password"]', "test_password")
            await page.click('button[type="submit"]')
            await page.wait_for_url("**/conversations")

            yield page

            await browser.close()

    @pytest.mark.asyncio
    async def test_create_and_retrieve_conversation(self, page: Page):
        """
        GIVEN: User is on conversation list page
        WHEN: User creates a new conversation
        THEN: Conversation appears in list
        """
        # Navigate to conversations
        await page.goto("http://localhost:3000/conversations")

        # Click "New Conversation"
        await page.click('button:has-text("New Conversation")')

        # Fill conversation form
        await page.fill('input[name="title"]', "Authentication bug discussion")
        await page.click('button:has-text("Create")')

        # Wait for success message
        await page.wait_for_selector('text=Conversation saved successfully')

        # Verify conversation appears in list
        await page.goto("http://localhost:3000/conversations")
        conversation = await page.locator('text=Authentication bug discussion')
        assert await conversation.is_visible()

    @pytest.mark.asyncio
    async def test_search_conversations(self, page: Page):
        """
        GIVEN: User has multiple conversations
        WHEN: User searches by keyword
        THEN: Relevant results are displayed
        """
        # Navigate to search
        await page.goto("http://localhost:3000/conversations")

        # Enter search query
        await page.fill('input[placeholder="Search conversations"]', "authentication")
        await page.click('button:has-text("Search")')

        # Wait for results
        await page.wait_for_selector('.search-results')

        # Verify results
        results = await page.locator('.conversation-card').count()
        assert results >= 1

        # Verify "Authentication bug" appears
        assert await page.locator('text=Authentication bug').is_visible()

5.3 E2E Test Coverage Map

Feature	Scenario Count	Test File	Environment
Conversation Lifecycle	8	conversation_lifecycle.feature	Standalone
Search Functionality	6	search.feature	Standalone
Commit Correlation	5	commit_correlation.feature	Standalone
Team Analytics	4	team_analytics.feature	Standalone
CODITECT Integration	7	coditect_integration.feature	CODITECT mode
Authentication Flows	5	auth_flows.feature	Both modes

Total E2E Tests: ~35 scenarios Estimated Execution Time: <30 minutes

---

6. Performance Tests

6.1 Load Testing (Locust)

# tests/performance/locustfile.py

from locust import HttpUser, task, between
from uuid import uuid4
import random

class ConversationUser(HttpUser):
    """Simulated user for load testing conversation APIs."""

    wait_time = between(1, 3)  # Wait 1-3 seconds between tasks

    def on_start(self):
        """Login and get authentication token."""
        response = self.client.post("/auth/login", json={
            "email": "loadtest@example.com",
            "password": "test_password"
        })
        self.token = response.json()["access_token"]
        self.client.headers["Authorization"] = f"Bearer {self.token}"
        self.conversation_ids = []

    @task(3)  # Weight = 3 (most common operation)
    def list_conversations(self):
        """List conversations with pagination."""
        page = random.randint(1, 10)
        self.client.get(f"/conversations?page={page}&limit=20")

    @task(2)  # Weight = 2
    def search_conversations(self):
        """Search conversations by keyword."""
        queries = ["authentication", "database", "performance", "bug", "feature"]
        query = random.choice(queries)
        self.client.get(f"/conversations/search?q={query}")

    @task(1)  # Weight = 1
    def create_conversation(self):
        """Create a new conversation."""
        response = self.client.post("/conversations", json={
            "title": f"Load test conversation {uuid4()}"
        })
        if response.status_code == 201:
            self.conversation_ids.append(response.json()["id"])

    @task(2)
    def get_conversation_details(self):
        """Get details for a specific conversation."""
        if self.conversation_ids:
            conv_id = random.choice(self.conversation_ids)
            self.client.get(f"/conversations/{conv_id}")

Load Test Scenarios:

Scenario	Users	Spawn Rate	Duration	Success Criteria
Normal Load	100	10/sec	10 min	p95 < 100ms, 0% errors
Peak Load	500	50/sec	10 min	p95 < 200ms, <0.1% errors
Stress Test	1000	100/sec	10 min	p95 < 500ms, <1% errors
Spike Test	0→1000→0	200/sec	5 min	System recovers, no crashes

6.2 Database Performance Tests

# tests/performance/test_database_performance.py

import pytest
import asyncio
from time import time
from uuid import uuid4
from core.repositories.conversation_repository import ConversationRepository

@pytest.mark.performance
class TestDatabasePerformance:
    """Performance tests for database operations."""

    @pytest.mark.asyncio
    async def test_bulk_insert_performance(self, conversation_repo):
        """
        GIVEN: 10,000 conversations to insert
        WHEN: Bulk insert is performed
        THEN: Insertion completes in <30 seconds (>300 inserts/sec)
        """
        conversations = [
            Conversation(
                id=uuid4(),
                organization_id=uuid4(),
                user_id=uuid4(),
                title=f"Conversation {i}"
            )
            for i in range(10000)
        ]

        start_time = time()
        await conversation_repo.bulk_create(conversations)
        duration = time() - start_time

        assert duration < 30, f"Bulk insert took {duration}s (expected <30s)"
        throughput = len(conversations) / duration
        assert throughput > 300, f"Throughput {throughput:.0f}/s (expected >300/s)"

    @pytest.mark.asyncio
    async def test_search_performance_with_large_dataset(self, conversation_repo):
        """
        GIVEN: 1 million conversations in database
        WHEN: Keyword search is performed
        THEN: Results return in <50ms (p95)
        """
        # Seed database with 1M conversations (run once, cache dataset)
        await self._seed_large_dataset(conversation_repo, count=1_000_000)

        # Run 100 search queries and measure latency
        latencies = []
        for _ in range(100):
            start = time()
            await conversation_repo.search_by_keyword(
                organization_id=uuid4(),
                query="authentication bug",
                limit=20
            )
            latencies.append((time() - start) * 1000)  # Convert to ms

        p95_latency = sorted(latencies)[94]  # 95th percentile
        assert p95_latency < 50, f"p95 latency {p95_latency:.2f}ms (expected <50ms)"

6.3 Performance Test Targets

Operation	p50 Target	p95 Target	p99 Target	Throughput Target
API: List Conversations	<30ms	<50ms	<100ms	>1000 req/s
API: Create Conversation	<50ms	<100ms	<200ms	>500 req/s
API: Keyword Search	<30ms	<50ms	<100ms	>800 req/s
API: Semantic Search	<40ms	<80ms	<150ms	>500 req/s
API: Hybrid Search	<50ms	<100ms	<200ms	>400 req/s
DB: Single Insert	<5ms	<10ms	<20ms	>200 inserts/s
DB: Bulk Insert (1K)	<100ms	<200ms	<500ms	>5000 inserts/s
DB: Query by ID	<3ms	<5ms	<10ms	>2000 queries/s

---

7. Security Tests

7.1 Authentication & Authorization Tests

# tests/security/test_authentication.py

import pytest
from httpx import AsyncClient
from jose import jwt
from datetime import datetime, timedelta

@pytest.mark.security
class TestAuthenticationSecurity:
    """Security tests for authentication mechanisms."""

    @pytest.mark.asyncio
    async def test_jwt_token_expiration_enforced(self, client: AsyncClient):
        """
        GIVEN: Expired JWT token
        WHEN: API request is made with expired token
        THEN: 401 Unauthorized is returned
        """
        # Create expired token (expired 1 hour ago)
        expired_token = jwt.encode(
            {
                "sub": "user@example.com",
                "exp": datetime.utcnow() - timedelta(hours=1)
            },
            key="test_secret",
            algorithm="HS256"
        )

        response = await client.get(
            "/conversations",
            headers={"Authorization": f"Bearer {expired_token}"}
        )

        assert response.status_code == 401
        assert "Token expired" in response.json()["detail"]

    @pytest.mark.asyncio
    async def test_rbac_admin_only_endpoints(self, client: AsyncClient):
        """
        GIVEN: User with "member" role
        WHEN: Admin-only endpoint is accessed
        THEN: 403 Forbidden is returned
        """
        member_token = await self._get_token_for_role(client, role="member")

        response = await client.delete(
            "/users/some-user-id",  # Admin-only endpoint
            headers={"Authorization": f"Bearer {member_token}"}
        )

        assert response.status_code == 403
        assert "Insufficient permissions" in response.json()["detail"]

    @pytest.mark.asyncio
    async def test_password_hashing_bcrypt(self):
        """
        GIVEN: Plain text password
        WHEN: User is created
        THEN: Password is hashed with bcrypt (not stored in plaintext)
        """
        from core.security import hash_password, verify_password

        password = "SecureP@ssw0rd123"
        hashed = hash_password(password)

        # Verify hash format (bcrypt starts with $2b$)
        assert hashed.startswith("$2b$")

        # Verify password cannot be reverse-engineered
        assert password not in hashed

        # Verify password verification works
        assert verify_password(password, hashed) is True
        assert verify_password("WrongPassword", hashed) is False

7.2 OWASP Top 10 Tests

# tests/security/test_owasp_top_10.py

import pytest
from httpx import AsyncClient

@pytest.mark.security
class TestOWASPTop10:
    """Security tests for OWASP Top 10 vulnerabilities."""

    @pytest.mark.asyncio
    async def test_sql_injection_prevention(self, authenticated_client):
        """
        OWASP A03:2021 - Injection
        GIVEN: Malicious SQL in search query
        WHEN: Search is performed
        THEN: SQL injection is blocked, safe results returned
        """
        malicious_query = "'; DROP TABLE conversations; --"

        response = await authenticated_client.get(
            f"/conversations/search?q={malicious_query}"
        )

        # Should not error (parametrized queries prevent injection)
        assert response.status_code == 200

        # Verify database still intact
        response = await authenticated_client.get("/conversations")
        assert response.status_code == 200  # Table still exists

    @pytest.mark.asyncio
    async def test_xss_prevention_in_conversation_title(self, authenticated_client):
        """
        OWASP A03:2021 - Injection (XSS)
        GIVEN: Conversation title with XSS payload
        WHEN: Conversation is created and retrieved
        THEN: Script tags are escaped/sanitized
        """
        xss_payload = "<script>alert('XSS')</script>"

        response = await authenticated_client.post("/conversations", json={
            "title": xss_payload
        })
        conv_id = response.json()["id"]

        # Retrieve conversation
        response = await authenticated_client.get(f"/conversations/{conv_id}")
        title = response.json()["title"]

        # Verify script tags are escaped
        assert "<script>" not in title
        assert "&lt;script&gt;" in title or "alert" not in title

    @pytest.mark.asyncio
    async def test_broken_authentication_session_fixation(self, client):
        """
        OWASP A07:2021 - Identification and Authentication Failures
        GIVEN: User logs in
        WHEN: New session is created
        THEN: New session ID is generated (prevents session fixation)
        """
        # Login with known session ID
        response1 = await client.post("/auth/login", json={
            "email": "test@example.com",
            "password": "password"
        })
        session_id_1 = response1.cookies.get("session_id")

        # Logout
        await client.post("/auth/logout")

        # Login again
        response2 = await client.post("/auth/login", json={
            "email": "test@example.com",
            "password": "password"
        })
        session_id_2 = response2.cookies.get("session_id")

        # Verify different session IDs (prevents fixation)
        assert session_id_1 != session_id_2

    @pytest.mark.asyncio
    async def test_insecure_direct_object_references(self, client):
        """
        OWASP A01:2021 - Broken Access Control (IDOR)
        GIVEN: Conversation belongs to org A
        WHEN: User from org B tries to access by ID
        THEN: 404 Not Found (prevents IDOR)
        """
        # Create conversation in org A
        org_a_token = await self._get_token(client, org_id="org-a")
        client.headers["Authorization"] = f"Bearer {org_a_token}"

        response = await client.post("/conversations", json={"title": "Secret"})
        conv_id = response.json()["id"]

        # Try to access from org B
        org_b_token = await self._get_token(client, org_id="org-b")
        client.headers["Authorization"] = f"Bearer {org_b_token}"

        response = await client.get(f"/conversations/{conv_id}")

        assert response.status_code == 404  # Not 200 or 403!

7.3 Security Test Coverage Map

OWASP Category	Test Count	Test File	Coverage
A01: Broken Access Control	8	test_access_control.py	RBAC, IDOR, path traversal
A02: Cryptographic Failures	6	test_cryptography.py	TLS, encryption, hashing
A03: Injection	10	test_injection.py	SQL, XSS, command injection
A04: Insecure Design	5	test_insecure_design.py	Rate limiting, quotas
A05: Security Misconfiguration	7	test_misconfiguration.py	Default creds, CORS
A06: Vulnerable Components	4	test_dependencies.py	Dependency scanning
A07: Auth Failures	9	test_authentication.py	JWT, session management
A08: Data Integrity Failures	5	test_data_integrity.py	Signature verification
A09: Logging Failures	4	test_logging.py	PII redaction, audit logs
A10: SSRF	3	test_ssrf.py	Webhook validation

Total Security Tests: ~61 tests Estimated Execution Time: <15 minutes

---

8. Test Automation

8.1 CI/CD Pipeline Integration

# .github/workflows/test.yml

name: Test Suite

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  unit-tests:
    name: Unit Tests
    runs-on: ubuntu-latest
    timeout-minutes: 10

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-test.txt

      - name: Run unit tests with coverage
        run: |
          pytest tests/unit/ \
            -v \
            --cov=core \
            --cov-report=xml \
            --cov-report=html \
            --cov-fail-under=80

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml
          fail_ci_if_error: true

  integration-tests:
    name: Integration Tests
    runs-on: ubuntu-latest
    timeout-minutes: 20

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_DB: test_db
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_password
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

      redis:
        image: redis:7
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 6379:6379

      weaviate:
        image: semitechnologies/weaviate:1.23.0
        env:
          QUERY_DEFAULTS_LIMIT: 20
          AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
        ports:
          - 8080:8080

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Run integration tests
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0
          WEAVIATE_URL: http://localhost:8080
        run: |
          pytest tests/integration/ \
            -v \
            --maxfail=3 \
            --tb=short

  e2e-tests:
    name: End-to-End Tests
    runs-on: ubuntu-latest
    timeout-minutes: 30

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install Playwright
        run: |
          pip install playwright pytest-playwright
          playwright install chromium

      - name: Start application
        run: |
          docker-compose up -d
          sleep 10  # Wait for services to be ready

      - name: Run E2E tests
        run: |
          pytest tests/e2e/ \
            -v \
            --headed=false \
            --screenshot=only-on-failure

      - name: Upload test artifacts
        if: failure()
        uses: actions/upload-artifact@v3
        with:
          name: e2e-screenshots
          path: tests/e2e/screenshots/

  security-tests:
    name: Security Tests
    runs-on: ubuntu-latest
    timeout-minutes: 20

    steps:
      - uses: actions/checkout@v4

      - name: Run Bandit (SAST)
        run: |
          pip install bandit
          bandit -r core/ -f json -o bandit-report.json

      - name: Run Safety (dependency check)
        run: |
          pip install safety
          safety check --json --output safety-report.json

      - name: Run OWASP ZAP (DAST)
        uses: zaproxy/action-baseline@v0.7.0
        with:
          target: 'http://localhost:8000'
          rules_file_name: '.zap/rules.tsv'

      - name: Upload security reports
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: zap-report.sarif

  performance-tests:
    name: Performance Tests
    runs-on: ubuntu-latest
    timeout-minutes: 30
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - name: Run Locust load tests
        run: |
          pip install locust
          locust -f tests/performance/locustfile.py \
            --headless \
            --users 100 \
            --spawn-rate 10 \
            --run-time 5m \
            --host http://localhost:8000 \
            --html locust-report.html

      - name: Upload performance report
        uses: actions/upload-artifact@v3
        with:
          name: performance-report
          path: locust-report.html

8.2 Pre-commit Hooks

# .pre-commit-config.yaml

repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-added-large-files

  - repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
      - id: black
        language_version: python3.11

  - repo: https://github.com/PyCQA/flake8
    rev: 6.0.0
    hooks:
      - id: flake8
        args: ['--max-line-length=88', '--extend-ignore=E203']

  - repo: local
    hooks:
      - id: pytest-unit
        name: Run unit tests
        entry: pytest tests/unit/ --maxfail=1 --tb=short
        language: system
        pass_filenames: false
        always_run: true

---

9. Quality Gates

9.1 Quality Gate Criteria

All criteria must pass before code can be merged to main:

Quality Gate	Threshold	Blocking	Tool
Unit Test Coverage	≥80%	✅ Yes	pytest-cov
Unit Tests Pass	100%	✅ Yes	pytest
Integration Tests Pass	100%	✅ Yes	pytest
E2E Tests Pass	≥95%	✅ Yes	Playwright
Security Scan	0 high/critical	✅ Yes	Bandit, Safety
Linting	0 errors	✅ Yes	flake8, black
Type Checking	0 errors	⚠️ Warning	mypy
Performance Regression	<10% slower	⚠️ Warning	Locust
Code Complexity	Cyclomatic <10	⚠️ Warning	radon

9.2 Quality Gate Enforcement

# .github/workflows/quality-gate.yml

name: Quality Gate

on:
  pull_request:
    branches: [main]

jobs:
  quality-gate:
    name: Quality Gate
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Check unit test coverage
        run: |
          pytest tests/unit/ --cov=core --cov-fail-under=80

      - name: Check all tests pass
        run: |
          pytest tests/ -v --maxfail=1

      - name: Check security vulnerabilities
        run: |
          bandit -r core/ -ll  # Only high/critical
          safety check --exit-code 1

      - name: Check code quality
        run: |
          flake8 core/ --count --show-source --statistics
          black --check core/

      - name: Check type hints
        run: |
          mypy core/ --strict
        continue-on-error: true  # Warning only

      - name: Comment PR with results
        uses: actions/github-script@v6
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: '✅ All quality gates passed!'
            })

---

10. Test Data Management

10.1 Test Data Factories

# tests/factories.py

import factory
from factory.fuzzy import FuzzyText, FuzzyInteger, FuzzyDateTime
from datetime import datetime, timedelta
from uuid import uuid4
from core.models import Organization, User, Conversation

class OrganizationFactory(factory.Factory):
    """Factory for creating test organizations."""

    class Meta:
        model = Organization

    id = factory.LazyFunction(uuid4)
    name = factory.Sequence(lambda n: f"Test Corp {n}")
    subdomain = factory.LazyAttribute(lambda o: o.name.lower().replace(" ", "-"))
    tier = factory.Iterator(["starter", "pro", "enterprise"])
    created_at = factory.LazyFunction(datetime.utcnow)

class UserFactory(factory.Factory):
    """Factory for creating test users."""

    class Meta:
        model = User

    id = factory.LazyFunction(uuid4)
    organization_id = factory.SubFactory(OrganizationFactory)
    email = factory.LazyAttribute(lambda u: f"{u.name.lower().replace(' ', '.')}@example.com")
    name = factory.Faker('name')
    role = factory.Iterator(["owner", "admin", "member", "viewer"])
    created_at = factory.LazyFunction(datetime.utcnow)

class ConversationFactory(factory.Factory):
    """Factory for creating test conversations."""

    class Meta:
        model = Conversation

    id = factory.LazyFunction(uuid4)
    organization_id = factory.SubFactory(OrganizationFactory)
    user_id = factory.SubFactory(UserFactory)
    title = factory.Faker('sentence')
    created_at = FuzzyDateTime(datetime.utcnow() - timedelta(days=30))
    message_count = FuzzyInteger(1, 100)
    token_count = FuzzyInteger(100, 10000)

10.2 Test Data Seeding

# tests/seed_test_data.py

async def seed_test_database(session, size: str = "small"):
    """Seed test database with realistic data."""

    sizes = {
        "small": {"orgs": 3, "users_per_org": 5, "convs_per_user": 20},
        "medium": {"orgs": 10, "users_per_org": 20, "convs_per_user": 100},
        "large": {"orgs": 100, "users_per_org": 50, "convs_per_user": 500},
    }

    config = sizes[size]

    for _ in range(config["orgs"]):
        org = await OrganizationFactory.create(session)

        for _ in range(config["users_per_org"]):
            user = await UserFactory.create(session, organization_id=org.id)

            for _ in range(config["convs_per_user"]):
                await ConversationFactory.create(
                    session,
                    organization_id=org.id,
                    user_id=user.id
                )

    await session.commit()

---

11. Test Reporting

11.1 Test Report Dashboard

Generated artifacts:

• coverage.html - Code coverage report
• pytest-report.html - Detailed test results
• locust-report.html - Performance test results
• zap-report.html - Security scan results

11.2 Metrics Tracked

Metric	Tool	Target	Current
Unit Test Coverage	pytest-cov	≥80%	TBD
Test Execution Time	pytest	<2 min	TBD
Test Flakiness Rate	pytest-rerunfailures	<1%	TBD
Security Vulnerabilities	Bandit, Safety	0 high/critical	TBD
Performance Regressions	Locust	<10% slower	TBD

---

Appendix A: Test Execution Commands

# Run all tests
pytest tests/ -v

# Run unit tests only
pytest tests/unit/ -v --cov=core

# Run integration tests only
pytest tests/integration/ -v

# Run E2E tests
pytest tests/e2e/ -v --headed

# Run performance tests
locust -f tests/performance/locustfile.py --headless --users 100

# Run security tests
bandit -r core/ -ll
safety check

# Generate coverage report
pytest tests/unit/ --cov=core --cov-report=html
open htmlcov/index.html

---

Appendix B: Test Environment Setup

# Install test dependencies
pip install -r requirements-test.txt

# Start test infrastructure
docker-compose -f docker-compose.test.yml up -d

# Run database migrations
alembic upgrade head

# Seed test data
python tests/seed_test_data.py --size medium

# Run test suite
pytest tests/ -v --cov=core

---

Document Status: Approved for Implementation Next Review: 2025-12-26 Maintained By: QA Engineering Team