from contextlib import asynccontextmanager
from pathlib import Path
from unittest.mock import AsyncMock
import pytest
from ckan_mcp.ckan_tools import CkanToolsManager
from ckan_mcp.types import (
CkanDatastoreField,
CkanDatastoreResult,
CkanOrganization,
CkanPackage,
CkanResource,
CkanToolsConfig,
)
@pytest.mark.asyncio
async def test_download_dataset_locally_produces_artifacts(tmp_path, monkeypatch):
"""End-to-end exercise of download workflow without touching the network."""
config = CkanToolsConfig(
ckan_base_url="https://example.com/api/3/action",
ckan_site_url="https://example.com",
dataset_page_url_template="https://example.com/dataset/{name}",
)
manager = CkanToolsManager(config)
monkeypatch.setenv("CKAN_MCP_LOCAL_DATASTORE", str(tmp_path))
resource = CkanResource(
id="resource-1",
name="Sample Resource",
format="CSV",
url="https://example.com/data.csv",
datastore_active=True,
last_modified="2024-01-02T03:04:05Z",
)
package = CkanPackage(
id="pkg-1",
name="sample-package",
title="Sample Package",
notes="Mock dataset for integration testing.",
organization=CkanOrganization(id="org-1", name="org-1", title="Org"),
resources=[resource],
metadata_created="2024-01-01T00:00:00Z",
metadata_modified="2024-01-02T00:00:00Z",
)
datastore_result = CkanDatastoreResult(
total=42,
fields=[CkanDatastoreField(id="col_a", type="text")],
)
@asynccontextmanager
async def fake_session():
yield object()
async def fake_download(url: str, destination: Path, timeout_seconds: int = 120) -> None:
destination.write_text("id,value\n1,2\n", encoding="utf-8")
async def fake_mime(_: Path) -> str:
return "text/csv"
manager._create_session = fake_session # type: ignore[assignment]
manager.fetch_package = AsyncMock(return_value=package) # type: ignore[assignment]
manager.fetch_datastore_info = AsyncMock( # type: ignore[assignment]
return_value=datastore_result
)
monkeypatch.setattr("ckan_mcp.ckan_tools._download_file_with_curl", fake_download)
monkeypatch.setattr("ckan_mcp.ckan_tools._detect_file_mime_type", fake_mime)
result = await manager.download_dataset_locally(package_id="pkg-1")
dataset_info = result["dataset"]
download_path = Path(dataset_info["download_path"])
metadata_path = Path(dataset_info["metadata_path"])
how_to_path = Path(dataset_info["how_to_path"])
assert download_path.exists()
assert download_path.read_text(encoding="utf-8").startswith("id,value")
assert metadata_path.exists()
assert "sample-package" in metadata_path.read_text(encoding="utf-8")
assert how_to_path.exists()
assert "Sample Package" in how_to_path.read_text(encoding="utf-8")
assert dataset_info["record_count"] == datastore_result.total
assert dataset_info["verified_mime_type"] == "text/csv"
