ckvd-testing

name	ckvd-testing
description	Run tests for crypto-kline-vision-data with proper markers and coverage. TRIGGERS - write tests, run tests, pytest, test coverage, unit tests, integration tests, mocking patterns.
argument-hint	[test-pattern]
user-invocable	true
allowed-tools	Read, Bash, Grep, Glob

Testing Crypto Kline Vision Data

Run tests for: $ARGUMENTS

Test Workflow Checklist

Copy this checklist and track progress:

Test Progress:
- [ ] Step 1: Run lint check (ruff check)
- [ ] Step 2: Run unit tests (fast, no network)
- [ ] Step 3: Verify import works
- [ ] Step 4: Run integration tests (if changing APIs)
- [ ] Step 5: Check coverage (if adding new code)

Step 1: uv run -p 3.13 ruff check --fix . Step 2: uv run -p 3.13 pytest tests/unit/ -v Step 3: uv run -p 3.13 python -c "from ckvd import CryptoKlineVisionData; print('OK')" Step 4: uv run -p 3.13 pytest tests/integration/ -v (if needed) Step 5: uv run -p 3.13 pytest tests/unit/ --cov=src/ckvd

Test Organization

tests/
├── unit/                    # Fast, no network (~0.5s)
│   └── streaming/           # Async streaming tests (pytest-asyncio)
├── integration/             # External services
├── okx/                     # OKX API integration
└── fcp_pm/                  # FCP protocol tests

Running Tests

Unit Tests (Fast)

# Quick validation
uv run -p 3.13 pytest tests/unit/ -v

# With coverage
uv run -p 3.13 pytest tests/unit/ --cov=src/ckvd --cov-report=term-missing

Integration Tests

# Requires network access
uv run -p 3.13 pytest tests/integration/ -v

# OKX-specific tests
uv run -p 3.13 pytest tests/okx/ -m okx -v

All Tests

uv run -p 3.13 pytest tests/ -v

Test Markers

Marker	Purpose
@pytest.mark.integration	Tests that call external APIs
@pytest.mark.okx	OKX-specific tests
@pytest.mark.serial	Must run sequentially
@pytest.mark.asyncio	Async streaming tests

Writing New Tests

import pytest
from ckvd import CryptoKlineVisionData, DataProvider, MarketType

class TestMyFeature:
    """Tests for MyFeature."""

    def test_basic_functionality(self):
        """Verify basic operation."""
        # Arrange
        manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.SPOT)

        # Act
        result = manager.some_method()

        # Assert
        assert result is not None
        manager.close()

    @pytest.mark.integration
    def test_with_network(self):
        """Test requiring network access."""
        # Mark with @pytest.mark.integration for external calls
        pass

Mocking HTTP Calls

from unittest.mock import patch, MagicMock

@patch("ckvd.core.sync.crypto_kline_vision_data.FSSpecVisionHandler")
@patch("ckvd.core.sync.crypto_kline_vision_data.UnifiedCacheManager")
def test_with_mocks(self, mock_cache, mock_handler):
    mock_handler.return_value = MagicMock()
    mock_cache.return_value = MagicMock()
    # Test logic...

Async Streaming Tests

Streaming tests use pytest-asyncio. Ensure it's installed (uv run -p 3.13 pip install pytest-asyncio).

import pytest
from ckvd import CryptoKlineVisionData, DataProvider, MarketType, StreamConfig, KlineUpdate
from unittest.mock import AsyncMock, MagicMock, patch


@pytest.fixture
async def stream_manager():
    """Create streaming manager for tests (auto-cleanup)."""
    manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)
    yield manager
    manager.close()


@pytest.mark.asyncio
async def test_stream_creation(stream_manager):
    """Verify stream creation."""
    config = StreamConfig(market_type=MarketType.FUTURES_USDT, confirmed_only=True)
    stream = stream_manager.create_stream(config)
    assert stream is not None


@pytest.mark.asyncio
async def test_stream_subscribe_and_receive():
    """Test async streaming with mocked KlineStream."""
    manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)

    with patch("ckvd.core.streaming.kline_stream.KlineStream") as MockStream:
        # Mock the stream to emit KlineUpdate objects
        mock_stream_instance = AsyncMock()
        MockStream.return_value = mock_stream_instance

        # Simulate 2 updates: one open, one closed
        mock_updates = [
            MagicMock(spec=KlineUpdate, symbol="BTCUSDT", interval="1h", is_closed=False),
            MagicMock(spec=KlineUpdate, symbol="BTCUSDT", interval="1h", is_closed=True, close=),
        ]
        mock_stream_instance.__aenter__.return_value = mock_stream_instance
        mock_stream_instance.__aiter__.return_value = (mock_updates)

        config = StreamConfig(market_type=MarketType.FUTURES_USDT)
        stream = manager.create_stream(config)

        updates_received = []
          stream:
             stream.subscribe(, )
              update  stream:
                updates_received.append(update)

         (updates_received) == 
         updates_received[].is_closed  
        manager.close()

Examples

Practical test examples:

• @examples/unit-test-patterns.md - Basic tests, fixtures, mocking
• @examples/integration-test-patterns.md - API tests, markers, FCP testing

Helper Scripts

Quick test runner:

# Run all quick checks (lint + unit tests + import)
./docs/skills/ckvd-testing/scripts/run_quick_tests.sh

# Run with test pattern filter
./docs/skills/ckvd-testing/scripts/run_quick_tests.sh test_timestamp

TodoWrite Task Templates

Template A: Run Quick Test

1. Run ruff lint check (uv run -p 3.13 ruff check --fix .)
2. Run unit tests (uv run -p 3.13 pytest tests/unit/ -v)
3. Verify package import works
4. Report pass/fail summary

Template B: Add Unit Test for Feature

1. Identify module and function under test
2. Read existing test patterns in tests/unit/ for the module
3. Write test class following Arrange/Act/Assert pattern
4. Mock external services (FSSpecVisionHandler, UnifiedCacheManager)
5. Add appropriate markers (@pytest.mark.integration if needed)
6. Run new tests and verify they pass
7. Check coverage for new code

Template C: Check Coverage

1. Run unit tests with coverage (uv run -p 3.13 pytest tests/unit/ --cov=src/ckvd --cov-report=term-missing)
2. Identify modules below target coverage
3. Write tests for uncovered paths
4. Re-run coverage to verify improvement

Template D: Write Async Stream Test

1. Import pytest-asyncio and async test utilities (@pytest.mark.asyncio)
2. Create async fixture for stream manager (yield pattern)
3. Set up StreamConfig with desired options
4. Create stream via manager.create_stream()
5. Mock KlineStream if testing without network (AsyncMock)
6. Use async context manager (async with stream)
7. Subscribe to symbol/interval
8. Iterate updates with async for loop
9. Assert on KlineUpdate fields (is_closed, close, volume, etc.)
10. Run with: uv run -p 3.13 pytest tests/unit/streaming/ -v --asyncio-mode=auto

---

Post-Change Checklist

After modifying this skill:

• Test commands use uv run -p 3.13 and correct paths
• Test organization diagram matches actual directory structure
• Mocking examples match actual import paths
• @-links to references/ and examples/ resolve
• Append changes to evolution-log.md

---

Detailed References

For deeper information, see:

• @references/fixtures.md - Pytest fixtures and auto-cleanup patterns
• @references/coverage.md - Coverage configuration and thresholds
• @references/mocking-patterns.md - CKVD-specific mocking patterns
• @references/markers.md - Pytest markers and test categorization
• @references/evolution-log.md - Skill change history