Unit tests for migrations and version APIs

This commit is contained in:
Cole Medin
2025-09-20 13:10:09 -05:00
parent 48e7d59db4
commit dd79254689
4 changed files with 858 additions and 0 deletions

View File

@@ -0,0 +1,206 @@
"""
Unit tests for migration_api.py
"""
from datetime import datetime
from unittest.mock import AsyncMock, patch
import pytest
from fastapi.testclient import TestClient
from src.server.config.version import ARCHON_VERSION
from src.server.main import app
from src.server.services.migration_service import MigrationRecord, PendingMigration
@pytest.fixture
def client():
"""Create test client."""
return TestClient(app)
@pytest.fixture
def mock_applied_migrations():
"""Mock applied migration data."""
return [
MigrationRecord({
"version": "0.1.0",
"migration_name": "001_initial",
"applied_at": datetime(2025, 1, 1, 0, 0, 0),
"checksum": "abc123",
}),
MigrationRecord({
"version": "0.1.0",
"migration_name": "002_add_column",
"applied_at": datetime(2025, 1, 2, 0, 0, 0),
"checksum": "def456",
}),
]
@pytest.fixture
def mock_pending_migrations():
"""Mock pending migration data."""
return [
PendingMigration(
version="0.1.0",
name="003_add_index",
sql_content="CREATE INDEX idx_test ON test_table(name);",
file_path="migration/0.1.0/003_add_index.sql"
),
PendingMigration(
version="0.1.0",
name="004_add_table",
sql_content="CREATE TABLE new_table (id INT);",
file_path="migration/0.1.0/004_add_table.sql"
),
]
@pytest.fixture
def mock_migration_status(mock_applied_migrations, mock_pending_migrations):
"""Mock complete migration status."""
return {
"pending_migrations": [
{"version": m.version, "name": m.name, "sql_content": m.sql_content, "file_path": m.file_path, "checksum": m.checksum}
for m in mock_pending_migrations
],
"applied_migrations": [
{"version": m.version, "migration_name": m.migration_name, "applied_at": m.applied_at, "checksum": m.checksum}
for m in mock_applied_migrations
],
"has_pending": True,
"bootstrap_required": False,
"current_version": ARCHON_VERSION,
"pending_count": 2,
"applied_count": 2,
}
def test_get_migration_status_success(client, mock_migration_status):
"""Test successful migration status retrieval."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_migration_status = AsyncMock(return_value=mock_migration_status)
response = client.get("/api/migrations/status")
assert response.status_code == 200
data = response.json()
assert data["current_version"] == ARCHON_VERSION
assert data["has_pending"] is True
assert data["bootstrap_required"] is False
assert data["pending_count"] == 2
assert data["applied_count"] == 2
assert len(data["pending_migrations"]) == 2
assert len(data["applied_migrations"]) == 2
def test_get_migration_status_bootstrap_required(client):
"""Test migration status when bootstrap is required."""
mock_status = {
"pending_migrations": [],
"applied_migrations": [],
"has_pending": True,
"bootstrap_required": True,
"current_version": ARCHON_VERSION,
"pending_count": 5,
"applied_count": 0,
}
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_migration_status = AsyncMock(return_value=mock_status)
response = client.get("/api/migrations/status")
assert response.status_code == 200
data = response.json()
assert data["bootstrap_required"] is True
assert data["applied_count"] == 0
def test_get_migration_status_error(client):
"""Test error handling in migration status."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_migration_status = AsyncMock(side_effect=Exception("Database error"))
response = client.get("/api/migrations/status")
assert response.status_code == 500
assert "Failed to get migration status" in response.json()["detail"]
def test_get_migration_history_success(client, mock_applied_migrations):
"""Test successful migration history retrieval."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_applied_migrations = AsyncMock(return_value=mock_applied_migrations)
response = client.get("/api/migrations/history")
assert response.status_code == 200
data = response.json()
assert data["total_count"] == 2
assert data["current_version"] == ARCHON_VERSION
assert len(data["migrations"]) == 2
assert data["migrations"][0]["migration_name"] == "001_initial"
def test_get_migration_history_empty(client):
"""Test migration history when no migrations applied."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_applied_migrations = AsyncMock(return_value=[])
response = client.get("/api/migrations/history")
assert response.status_code == 200
data = response.json()
assert data["total_count"] == 0
assert len(data["migrations"]) == 0
def test_get_migration_history_error(client):
"""Test error handling in migration history."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_applied_migrations = AsyncMock(side_effect=Exception("Database error"))
response = client.get("/api/migrations/history")
assert response.status_code == 500
assert "Failed to get migration history" in response.json()["detail"]
def test_get_pending_migrations_success(client, mock_pending_migrations):
"""Test successful pending migrations retrieval."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_pending_migrations = AsyncMock(return_value=mock_pending_migrations)
response = client.get("/api/migrations/pending")
assert response.status_code == 200
data = response.json()
assert len(data) == 2
assert data[0]["name"] == "003_add_index"
assert data[0]["sql_content"] == "CREATE INDEX idx_test ON test_table(name);"
assert data[1]["name"] == "004_add_table"
def test_get_pending_migrations_none(client):
"""Test when no pending migrations exist."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_pending_migrations = AsyncMock(return_value=[])
response = client.get("/api/migrations/pending")
assert response.status_code == 200
data = response.json()
assert len(data) == 0
def test_get_pending_migrations_error(client):
"""Test error handling in pending migrations."""
with patch("src.server.api_routes.migration_api.migration_service") as mock_service:
mock_service.get_pending_migrations = AsyncMock(side_effect=Exception("File error"))
response = client.get("/api/migrations/pending")
assert response.status_code == 500
assert "Failed to get pending migrations" in response.json()["detail"]

View File

@@ -0,0 +1,147 @@
"""
Unit tests for version_api.py
"""
from datetime import datetime
from unittest.mock import AsyncMock, patch
import pytest
from fastapi.testclient import TestClient
from src.server.config.version import ARCHON_VERSION
from src.server.main import app
@pytest.fixture
def client():
"""Create test client."""
return TestClient(app)
@pytest.fixture
def mock_version_data():
"""Mock version check data."""
return {
"current": ARCHON_VERSION,
"latest": "0.2.0",
"update_available": True,
"release_url": "https://github.com/coleam00/Archon/releases/tag/v0.2.0",
"release_notes": "New features and bug fixes",
"published_at": datetime(2025, 1, 1, 0, 0, 0),
"check_error": None,
"author": "coleam00",
"assets": [{"name": "archon.zip", "size": 1024000}],
}
def test_check_for_updates_success(client, mock_version_data):
"""Test successful version check."""
with patch("src.server.api_routes.version_api.version_service") as mock_service:
mock_service.check_for_updates = AsyncMock(return_value=mock_version_data)
response = client.get("/api/version/check")
assert response.status_code == 200
data = response.json()
assert data["current"] == ARCHON_VERSION
assert data["latest"] == "0.2.0"
assert data["update_available"] is True
assert data["release_url"] == mock_version_data["release_url"]
def test_check_for_updates_no_update(client):
"""Test when no update is available."""
mock_data = {
"current": ARCHON_VERSION,
"latest": ARCHON_VERSION,
"update_available": False,
"release_url": None,
"release_notes": None,
"published_at": None,
"check_error": None,
}
with patch("src.server.api_routes.version_api.version_service") as mock_service:
mock_service.check_for_updates = AsyncMock(return_value=mock_data)
response = client.get("/api/version/check")
assert response.status_code == 200
data = response.json()
assert data["current"] == ARCHON_VERSION
assert data["latest"] == ARCHON_VERSION
assert data["update_available"] is False
def test_check_for_updates_with_etag_modified(client, mock_version_data):
"""Test ETag handling when data has changed."""
with patch("src.server.api_routes.version_api.version_service") as mock_service:
mock_service.check_for_updates = AsyncMock(return_value=mock_version_data)
# First request
response1 = client.get("/api/version/check")
assert response1.status_code == 200
old_etag = response1.headers.get("etag")
# Modify data
modified_data = mock_version_data.copy()
modified_data["latest"] = "0.3.0"
mock_service.check_for_updates = AsyncMock(return_value=modified_data)
# Second request with old ETag
response2 = client.get("/api/version/check", headers={"If-None-Match": old_etag})
assert response2.status_code == 200 # Data changed, return new data
data = response2.json()
assert data["latest"] == "0.3.0"
def test_check_for_updates_error_handling(client):
"""Test error handling in version check."""
with patch("src.server.api_routes.version_api.version_service") as mock_service:
mock_service.check_for_updates = AsyncMock(side_effect=Exception("API error"))
response = client.get("/api/version/check")
assert response.status_code == 200 # Should still return 200
data = response.json()
assert data["current"] == ARCHON_VERSION
assert data["latest"] is None
assert data["update_available"] is False
assert data["check_error"] == "API error"
def test_get_current_version(client):
"""Test getting current version."""
response = client.get("/api/version/current")
assert response.status_code == 200
data = response.json()
assert data["version"] == ARCHON_VERSION
assert "timestamp" in data
def test_clear_version_cache_success(client):
"""Test clearing version cache."""
with patch("src.server.api_routes.version_api.version_service") as mock_service:
mock_service.clear_cache.return_value = None
response = client.post("/api/version/clear-cache")
assert response.status_code == 200
data = response.json()
assert data["success"] is True
assert data["message"] == "Version cache cleared successfully"
mock_service.clear_cache.assert_called_once()
def test_clear_version_cache_error(client):
"""Test error handling when clearing cache fails."""
with patch("src.server.api_routes.version_api.version_service") as mock_service:
mock_service.clear_cache.side_effect = Exception("Cache error")
response = client.post("/api/version/clear-cache")
assert response.status_code == 500
assert "Failed to clear cache" in response.json()["detail"]

View File

@@ -0,0 +1,271 @@
"""
Fixed unit tests for migration_service.py
"""
import hashlib
from datetime import datetime
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from src.server.config.version import ARCHON_VERSION
from src.server.services.migration_service import (
MigrationRecord,
MigrationService,
PendingMigration,
)
@pytest.fixture
def migration_service():
"""Create a migration service instance."""
with patch("src.server.services.migration_service.Path.exists") as mock_exists:
# Mock that migration directory exists locally
mock_exists.return_value = False # Docker path doesn't exist
service = MigrationService()
return service
@pytest.fixture
def mock_supabase_client():
"""Mock Supabase client."""
client = MagicMock()
return client
def test_pending_migration_init():
"""Test PendingMigration initialization and checksum calculation."""
migration = PendingMigration(
version="0.1.0",
name="001_initial",
sql_content="CREATE TABLE test (id INT);",
file_path="migration/0.1.0/001_initial.sql"
)
assert migration.version == "0.1.0"
assert migration.name == "001_initial"
assert migration.sql_content == "CREATE TABLE test (id INT);"
assert migration.file_path == "migration/0.1.0/001_initial.sql"
assert migration.checksum == hashlib.md5("CREATE TABLE test (id INT);".encode()).hexdigest()
def test_migration_record_init():
"""Test MigrationRecord initialization from database data."""
data = {
"id": "123-456",
"version": "0.1.0",
"migration_name": "001_initial",
"applied_at": "2025-01-01T00:00:00Z",
"checksum": "abc123"
}
record = MigrationRecord(data)
assert record.id == "123-456"
assert record.version == "0.1.0"
assert record.migration_name == "001_initial"
assert record.applied_at == "2025-01-01T00:00:00Z"
assert record.checksum == "abc123"
def test_migration_service_init_local():
"""Test MigrationService initialization with local path."""
with patch("src.server.services.migration_service.Path.exists") as mock_exists:
# Mock that Docker path doesn't exist
mock_exists.return_value = False
service = MigrationService()
assert service._migrations_dir == Path("migration")
def test_migration_service_init_docker():
"""Test MigrationService initialization with Docker path."""
with patch("src.server.services.migration_service.Path.exists") as mock_exists:
# Mock that Docker path exists
mock_exists.return_value = True
service = MigrationService()
assert service._migrations_dir == Path("/app/migration")
@pytest.mark.asyncio
async def test_get_applied_migrations_success(migration_service, mock_supabase_client):
"""Test successful retrieval of applied migrations."""
mock_response = MagicMock()
mock_response.data = [
{
"id": "123",
"version": "0.1.0",
"migration_name": "001_initial",
"applied_at": "2025-01-01T00:00:00Z",
"checksum": "abc123",
},
]
mock_supabase_client.table.return_value.select.return_value.order.return_value.execute.return_value = mock_response
with patch.object(migration_service, '_get_supabase_client', return_value=mock_supabase_client):
with patch.object(migration_service, 'check_migrations_table_exists', return_value=True):
result = await migration_service.get_applied_migrations()
assert len(result) == 1
assert isinstance(result[0], MigrationRecord)
assert result[0].version == "0.1.0"
assert result[0].migration_name == "001_initial"
@pytest.mark.asyncio
async def test_get_applied_migrations_table_not_exists(migration_service, mock_supabase_client):
"""Test handling when migrations table doesn't exist."""
with patch.object(migration_service, '_get_supabase_client', return_value=mock_supabase_client):
with patch.object(migration_service, 'check_migrations_table_exists', return_value=False):
result = await migration_service.get_applied_migrations()
assert result == []
@pytest.mark.asyncio
async def test_get_pending_migrations_with_files(migration_service, mock_supabase_client):
"""Test getting pending migrations from filesystem."""
# Mock scan_migration_directory to return test migrations
mock_migrations = [
PendingMigration(
version="0.1.0",
name="001_initial",
sql_content="CREATE TABLE test;",
file_path="migration/0.1.0/001_initial.sql"
),
PendingMigration(
version="0.1.0",
name="002_update",
sql_content="ALTER TABLE test ADD col TEXT;",
file_path="migration/0.1.0/002_update.sql"
)
]
# Mock no applied migrations
with patch.object(migration_service, 'scan_migration_directory', return_value=mock_migrations):
with patch.object(migration_service, 'get_applied_migrations', return_value=[]):
result = await migration_service.get_pending_migrations()
assert len(result) == 2
assert all(isinstance(m, PendingMigration) for m in result)
assert result[0].name == "001_initial"
assert result[1].name == "002_update"
@pytest.mark.asyncio
async def test_get_pending_migrations_some_applied(migration_service, mock_supabase_client):
"""Test getting pending migrations when some are already applied."""
# Mock all migrations
mock_all_migrations = [
PendingMigration(
version="0.1.0",
name="001_initial",
sql_content="CREATE TABLE test;",
file_path="migration/0.1.0/001_initial.sql"
),
PendingMigration(
version="0.1.0",
name="002_update",
sql_content="ALTER TABLE test ADD col TEXT;",
file_path="migration/0.1.0/002_update.sql"
)
]
# Mock first migration as applied
mock_applied = [
MigrationRecord({
"version": "0.1.0",
"migration_name": "001_initial",
"applied_at": "2025-01-01T00:00:00Z",
"checksum": None
})
]
with patch.object(migration_service, 'scan_migration_directory', return_value=mock_all_migrations):
with patch.object(migration_service, 'get_applied_migrations', return_value=mock_applied):
with patch.object(migration_service, 'check_migrations_table_exists', return_value=True):
result = await migration_service.get_pending_migrations()
assert len(result) == 1
assert result[0].name == "002_update"
@pytest.mark.asyncio
async def test_get_migration_status_all_applied(migration_service, mock_supabase_client):
"""Test migration status when all migrations are applied."""
# Mock one migration file
mock_all_migrations = [
PendingMigration(
version="0.1.0",
name="001_initial",
sql_content="CREATE TABLE test;",
file_path="migration/0.1.0/001_initial.sql"
)
]
# Mock migration as applied
mock_applied = [
MigrationRecord({
"version": "0.1.0",
"migration_name": "001_initial",
"applied_at": "2025-01-01T00:00:00Z",
"checksum": None
})
]
with patch.object(migration_service, 'scan_migration_directory', return_value=mock_all_migrations):
with patch.object(migration_service, 'get_applied_migrations', return_value=mock_applied):
with patch.object(migration_service, 'check_migrations_table_exists', return_value=True):
result = await migration_service.get_migration_status()
assert result["current_version"] == ARCHON_VERSION
assert result["has_pending"] is False
assert result["bootstrap_required"] is False
assert result["pending_count"] == 0
assert result["applied_count"] == 1
@pytest.mark.asyncio
async def test_get_migration_status_bootstrap_required(migration_service, mock_supabase_client):
"""Test migration status when bootstrap is required (table doesn't exist)."""
# Mock migration files
mock_all_migrations = [
PendingMigration(
version="0.1.0",
name="001_initial",
sql_content="CREATE TABLE test;",
file_path="migration/0.1.0/001_initial.sql"
),
PendingMigration(
version="0.1.0",
name="002_update",
sql_content="ALTER TABLE test ADD col TEXT;",
file_path="migration/0.1.0/002_update.sql"
)
]
with patch.object(migration_service, 'scan_migration_directory', return_value=mock_all_migrations):
with patch.object(migration_service, 'get_applied_migrations', return_value=[]):
with patch.object(migration_service, 'check_migrations_table_exists', return_value=False):
result = await migration_service.get_migration_status()
assert result["bootstrap_required"] is True
assert result["has_pending"] is True
assert result["pending_count"] == 2
assert result["applied_count"] == 0
assert len(result["pending_migrations"]) == 2
@pytest.mark.asyncio
async def test_get_migration_status_no_files(migration_service, mock_supabase_client):
"""Test migration status when no migration files exist."""
with patch.object(migration_service, 'scan_migration_directory', return_value=[]):
with patch.object(migration_service, 'get_applied_migrations', return_value=[]):
with patch.object(migration_service, 'check_migrations_table_exists', return_value=True):
result = await migration_service.get_migration_status()
assert result["has_pending"] is False
assert result["pending_count"] == 0
assert len(result["pending_migrations"]) == 0

View File

@@ -0,0 +1,234 @@
"""
Unit tests for version_service.py
"""
import json
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from src.server.config.version import ARCHON_VERSION
from src.server.services.version_service import VersionService
@pytest.fixture
def version_service():
"""Create a fresh version service instance for each test."""
service = VersionService()
# Clear any cache from previous tests
service._cache = None
service._cache_time = None
return service
@pytest.fixture
def mock_release_data():
"""Mock GitHub release data."""
return {
"tag_name": "v0.2.0",
"name": "Archon v0.2.0",
"html_url": "https://github.com/coleam00/Archon/releases/tag/v0.2.0",
"body": "## Release Notes\n\nNew features and bug fixes",
"published_at": "2025-01-01T00:00:00Z",
"author": {"login": "coleam00"},
"assets": [
{
"name": "archon-v0.2.0.zip",
"size": 1024000,
"download_count": 100,
"browser_download_url": "https://github.com/coleam00/Archon/releases/download/v0.2.0/archon-v0.2.0.zip",
"content_type": "application/zip",
}
],
}
@pytest.mark.asyncio
async def test_get_latest_release_success(version_service, mock_release_data):
"""Test successful fetching of latest release from GitHub."""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_release_data
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await version_service.get_latest_release()
assert result == mock_release_data
assert version_service._cache == mock_release_data
assert version_service._cache_time is not None
@pytest.mark.asyncio
async def test_get_latest_release_uses_cache(version_service, mock_release_data):
"""Test that cache is used when available and not expired."""
# Set up cache
version_service._cache = mock_release_data
version_service._cache_time = datetime.now()
with patch("httpx.AsyncClient") as mock_client_class:
result = await version_service.get_latest_release()
# Should not make HTTP request
mock_client_class.assert_not_called()
assert result == mock_release_data
@pytest.mark.asyncio
async def test_get_latest_release_cache_expired(version_service, mock_release_data):
"""Test that cache is refreshed when expired."""
# Set up expired cache
old_data = {"tag_name": "v0.1.0"}
version_service._cache = old_data
version_service._cache_time = datetime.now() - timedelta(hours=2)
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_release_data
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await version_service.get_latest_release()
# Should make new HTTP request
mock_client.get.assert_called_once()
assert result == mock_release_data
assert version_service._cache == mock_release_data
@pytest.mark.asyncio
async def test_get_latest_release_404(version_service):
"""Test handling of 404 (no releases)."""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 404
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await version_service.get_latest_release()
assert result is None
@pytest.mark.asyncio
async def test_get_latest_release_timeout(version_service, mock_release_data):
"""Test handling of timeout with cache fallback."""
# Set up cache
version_service._cache = mock_release_data
version_service._cache_time = datetime.now() - timedelta(hours=2) # Expired
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_client.get.side_effect = httpx.TimeoutException("Timeout")
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await version_service.get_latest_release()
# Should return cached data
assert result == mock_release_data
@pytest.mark.asyncio
async def test_check_for_updates_new_version_available(version_service, mock_release_data):
"""Test when a new version is available."""
with patch.object(version_service, "get_latest_release", return_value=mock_release_data):
result = await version_service.check_for_updates()
assert result["current"] == ARCHON_VERSION
assert result["latest"] == "0.2.0"
assert result["update_available"] is True
assert result["release_url"] == mock_release_data["html_url"]
assert result["release_notes"] == mock_release_data["body"]
assert result["published_at"] == datetime.fromisoformat("2025-01-01T00:00:00+00:00")
assert result["author"] == "coleam00"
assert len(result["assets"]) == 1
@pytest.mark.asyncio
async def test_check_for_updates_same_version(version_service):
"""Test when current version is up to date."""
mock_data = {"tag_name": f"v{ARCHON_VERSION}", "html_url": "test_url", "body": "notes"}
with patch.object(version_service, "get_latest_release", return_value=mock_data):
result = await version_service.check_for_updates()
assert result["current"] == ARCHON_VERSION
assert result["latest"] == ARCHON_VERSION
assert result["update_available"] is False
@pytest.mark.asyncio
async def test_check_for_updates_no_release(version_service):
"""Test when no releases are found."""
with patch.object(version_service, "get_latest_release", return_value=None):
result = await version_service.check_for_updates()
assert result["current"] == ARCHON_VERSION
assert result["latest"] is None
assert result["update_available"] is False
assert result["release_url"] is None
@pytest.mark.asyncio
async def test_check_for_updates_parse_version(version_service, mock_release_data):
"""Test version parsing with and without 'v' prefix."""
# Test with 'v' prefix
mock_release_data["tag_name"] = "v1.2.3"
with patch.object(version_service, "get_latest_release", return_value=mock_release_data):
result = await version_service.check_for_updates()
assert result["latest"] == "1.2.3"
# Test without 'v' prefix
mock_release_data["tag_name"] = "2.0.0"
with patch.object(version_service, "get_latest_release", return_value=mock_release_data):
result = await version_service.check_for_updates()
assert result["latest"] == "2.0.0"
@pytest.mark.asyncio
async def test_check_for_updates_missing_fields(version_service):
"""Test handling of incomplete release data."""
mock_data = {"tag_name": "v0.2.0"} # Minimal data
with patch.object(version_service, "get_latest_release", return_value=mock_data):
result = await version_service.check_for_updates()
assert result["latest"] == "0.2.0"
assert result["release_url"] is None
assert result["release_notes"] is None
assert result["published_at"] is None
assert result["author"] is None
assert result["assets"] == [] # Empty list, not None
def test_clear_cache(version_service, mock_release_data):
"""Test cache clearing."""
# Set up cache
version_service._cache = mock_release_data
version_service._cache_time = datetime.now()
# Clear cache
version_service.clear_cache()
assert version_service._cache is None
assert version_service._cache_time is None
def test_is_newer_version():
"""Test version comparison logic using the utility function."""
from src.server.utils.semantic_version import is_newer_version
# Test various version comparisons
assert is_newer_version("1.0.0", "2.0.0") is True
assert is_newer_version("2.0.0", "1.0.0") is False
assert is_newer_version("1.0.0", "1.0.0") is False
assert is_newer_version("1.0.0", "1.1.0") is True
assert is_newer_version("1.0.0", "1.0.1") is True
assert is_newer_version("1.2.3", "1.2.3") is False