# Ckvd Testing

> 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.

- Skill: `terrylica/ckvd-testing` (Agent Skill, multi-file: 9 files)
- Install (CLI): `npx skillmds@latest add terrylica/ckvd-testing`
- Raw SKILL.md: https://api.skillmd.com/api/skills/terrylica/ckvd-testing/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: terrylica (https://skillmd.com/u/terrylica)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/terrylica/ckvd-testing

---


# 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)

```bash
# 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

```bash
# 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

```bash
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

```python
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

```python
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`).

```python
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=42500.0),
        ]
        mock_stream_instance.__aenter__.return_value = mock_stream_instance
        mock_stream_instance.__aiter__.return_value = iter(mock_updates)

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

        updates_received = []
        async with stream:
            await stream.subscribe("BTCUSDT", "1h")
            async for update in stream:
                updates_received.append(update)

        assert len(updates_received) == 2
        assert updates_received[1].is_closed is True
        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:

```bash
# 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](./references/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