Testing Guide¶

Comprehensive guide to testing Machineuse.

Test Suite Overview¶

tests/
├── unit/                  # Unit tests
│   ├── test_config.py
│   ├── test_storage.py
│   └── test_scheduler.py
├── integration/           # Integration tests
│   ├── test_api.py
│   └── test_messaging.py
└── conftest.py           # Shared fixtures

Running Tests¶

All Tests¶

poetry run pytest

With Coverage¶

poetry run pytest --cov=machineuse --cov-report=html
open htmlcov/index.html

Specific Tests¶

# Single file
poetry run pytest tests/unit/test_config.py

# Single test
poetry run pytest tests/unit/test_config.py::test_load_config

# By pattern
poetry run pytest -k "config"

Test Categories¶

# Unit tests only
poetry run pytest tests/unit/

# Integration tests
poetry run pytest -m integration

# Exclude slow tests
poetry run pytest -m "not slow"

# Tests requiring root
sudo poetry run pytest -m requires_root

Test Markers¶

Marker	Description
`@pytest.mark.unit`	Unit tests
`@pytest.mark.integration`	Integration tests
`@pytest.mark.slow`	Slow-running tests
`@pytest.mark.requires_root`	Requires root privileges

Writing Tests¶

Unit Test Example¶

import pytest
from machineuse.core.config import Config

class TestConfig:
    def test_load_default_config(self):
        config = Config()
        assert config.deployment_mode == "single_node"

    def test_load_from_file(self, tmp_path):
        config_file = tmp_path / "config.json"
        config_file.write_text('{"deployment_mode": "control_plane"}')

        config = Config.from_file(config_file)
        assert config.deployment_mode == "control_plane"

    def test_invalid_config_raises(self):
        with pytest.raises(ValueError, match="Invalid deployment mode"):
            Config(deployment_mode="invalid")

Integration Test Example¶

import pytest
from httpx import AsyncClient
from machineuse.api.server import app

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

    async def test_health_endpoint(self, client):
        response = await client.get("/health")
        assert response.status_code == 200
        assert response.json()["status"] == "healthy"

    async def test_create_instance(self, client):
        response = await client.post(
            "/v2/instances",
            json={"image": "ubuntu:22.04"}
        )
        assert response.status_code == 201
        assert "id" in response.json()

Async Test Example¶

import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_operation():
    result = await some_async_function()
    assert result == expected_value

Fixtures¶

Shared Fixtures (conftest.py)¶

import pytest
import tempfile
from pathlib import Path

@pytest.fixture
def temp_dir():
    """Temporary directory for tests."""
    with tempfile.TemporaryDirectory() as tmpdir:
        yield Path(tmpdir)

@pytest.fixture
def mock_config(temp_dir):
    """Test configuration."""
    return {
        "deployment_mode": "single_node",
        "storage": {
            "backend": "sqlite",
            "database_path": str(temp_dir / "test.db")
        }
    }

@pytest.fixture
async def storage_provider(mock_config):
    """Storage provider for tests."""
    from machineuse.core.storage import StorageFactory
    provider = StorageFactory.create(mock_config)
    await provider.initialize()
    yield provider
    await provider.close()

Database Fixtures¶

@pytest.fixture
async def db_session(temp_dir):
    """Isolated database session."""
    db_path = temp_dir / "test.db"
    async with aiosqlite.connect(db_path) as db:
        await db.execute("""
            CREATE TABLE instances (
                id TEXT PRIMARY KEY,
                status TEXT
            )
        """)
        await db.commit()
        yield db

Mocking¶

Mock External Services¶

from unittest.mock import patch, MagicMock, AsyncMock

@patch('machineuse.core.containers.subprocess.run')
def test_container_creation(mock_run):
    mock_run.return_value = MagicMock(returncode=0, stdout="")

    container = create_container("test-1")

    assert container.id == "test-1"
    mock_run.assert_called_once()

@patch('machineuse.core.messaging.MessagingClient')
async def test_with_mock_messaging(mock_client):
    mock_client.send_request = AsyncMock(return_value={"status": "ok"})

    result = await send_command("test")

    assert result["status"] == "ok"

Mock Time¶

from freezegun import freeze_time

@freeze_time("2026-03-19 10:00:00")
def test_time_dependent():
    from datetime import datetime
    assert datetime.now().year == 2026

Test Database¶

In-Memory SQLite¶

@pytest.fixture
async def memory_db():
    async with aiosqlite.connect(":memory:") as db:
        yield db

Test Data Factories¶

from dataclasses import dataclass
import uuid

@dataclass
class InstanceFactory:
    @staticmethod
    def create(**kwargs):
        defaults = {
            "id": str(uuid.uuid4())[:8],
            "status": "running",
            "node_id": "worker-1",
            "image": "ubuntu:22.04"
        }
        defaults.update(kwargs)
        return defaults

Performance Testing¶

Benchmark Tests¶

import pytest

@pytest.mark.benchmark
def test_scheduler_performance(benchmark):
    nodes = [create_mock_node() for _ in range(100)]

    result = benchmark(lambda: scheduler.select_node(nodes))

    assert result is not None

Load Testing¶

import asyncio
import aiohttp

async def test_concurrent_requests():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.post("http://localhost:8000/v2/instances", json={})
            for _ in range(100)
        ]
        responses = await asyncio.gather(*tasks)

        success_count = sum(1 for r in responses if r.status == 201)
        assert success_count >= 95  # 95% success rate

CI/CD Integration¶

GitHub Actions¶

name: Tests

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'
      - name: Install dependencies
        working-directory: machineuse-api
        run: |
          pip install poetry
          poetry install --with dev
      - name: Run tests
        working-directory: machineuse-api
        run: poetry run pytest --cov=machineuse

Troubleshooting Tests¶

Debug Failing Tests¶

# Verbose output
poetry run pytest -v

# Show print statements
poetry run pytest -s

# Stop on first failure
poetry run pytest -x

# Debug with pdb
poetry run pytest --pdb

Common Issues¶

Issue	Solution
Import errors	Check `PYTHONPATH`
Async test hangs	Use `pytest-asyncio`
Database locked	Use separate test databases
Fixture not found	Check `conftest.py` location