Skip to content
Hoard Hoard

Testing Connectors

This page describes how to test connectors during development.

Basic Testing

Test your connector manually by instantiating it and calling methods:

from my_connector import MyConnector


# Create connector
connector = MyConnector()


# Set configuration
config = {"export_path": "fixtures/sample_export.zip"}


# Test discover
result = connector.discover(config)
assert result.success
print(f"Found {result.entity_count_estimate} entities")


# Test scan
for entity, chunks in connector.scan(config):
    print(f"{entity.title}: {len(chunks)} chunks")

Test discover()

def test_discover_success():
    connector = MyConnector()
    config = {"export_path": "fixtures/sample.zip"}


    result = connector.discover(config)


    assert result.success
    assert result.entity_count_estimate > 0


def test_discover_missing_path():
    connector = MyConnector()
    config = {"export_path": "/nonexistent"}


    result = connector.discover(config)


    assert not result.success
    assert "not found" in result.message.lower()

Test scan()

def test_scan_produces_valid_output():
    connector = MyConnector()
    config = {"export_path": "fixtures/sample.zip"}


    results = list(connector.scan(config))


    assert len(results) > 0


    for entity, chunks in results:
        assert entity.source == connector.source_name
        assert entity.source_id  # Not empty
        assert entity.title
        assert len(chunks) > 0

Test ID Stability

def test_ids_are_stable():
    """Run scan twice, verify same IDs."""
    connector = MyConnector()
    config = {"export_path": "fixtures/sample.zip"}


    run1 = {e.source_id for e, _ in connector.scan(config)}
    run2 = {e.source_id for e, _ in connector.scan(config)}


    assert run1 == run2

Fixtures

Store test data in fixtures/ directory:

my_connector/
├── connector.py
├── manifest.yaml
├── fixtures/
│   ├── sample_export.zip      # Normal case
│   ├── empty_export.zip       # Edge case: empty
│   ├── large_export.zip       # Edge case: many files
│   └── malformed_export.zip   # Edge case: bad data
└── tests/
    └── test_connector.py

Edge Case Tests

def test_empty_export():
    """Handle empty export gracefully."""
    connector = MyConnector()
    config = {"export_path": "fixtures/empty.zip"}


    results = list(connector.scan(config))
    assert results == []


def test_malformed_content():
    """Don't crash on bad data."""
    connector = MyConnector()
    config = {"export_path": "fixtures/malformed.zip"}


    # Should complete without crashing
    results = list(connector.scan(config))
    assert isinstance(results, list)


def test_large_files():
    """Handle large files without memory issues."""
    connector = MyConnector()
    config = {"export_path": "fixtures/large.zip"}


    count = sum(1 for _ in connector.scan(config))
    assert count > 0

Running Tests

Terminal window
# Run all tests
pytest tests/


# Run with coverage
pytest --cov=connector tests/


# Run specific test
pytest tests/test_connector.py::test_discover_success

CI Integration

Example GitHub Actions workflow:

name: Test Connector
on: [push, pull_request]


jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -e ".[dev]"
      - run: pytest tests/ --cov

Checklist

Before submitting a connector:

  • test_discover_success passes
  • test_discover_missing_path passes
  • test_scan_produces_valid_output passes
  • test_ids_are_stable passes
  • Edge cases handled (empty, malformed)
  • No real user data in fixtures
  • All tests pass in CI

Next Steps