Spaces:
Runtime error
Runtime error
File size: 2,360 Bytes
f8f02c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.v1.data import data_router
app = FastAPI()
app.include_router(data_router, prefix="/v1")
@pytest.fixture
def data_client():
from app.api.v1.data import acccess_token_bearer
# Override authentication and authorization
app.dependency_overrides[acccess_token_bearer] = lambda: {"user_id": "test_user"}
with TestClient(app) as client:
yield client
app.dependency_overrides.clear()
@pytest.fixture
def mock_data_service():
with patch("app.api.v1.data.data_service") as mock_service:
yield mock_service
def test_read_from_url(data_client, mock_data_service):
mock_data_service.read_file = AsyncMock(return_value={"data": "parsed contents"})
response = data_client.get("/v1/data/read-file?url=http://example.com/data.csv")
assert response.status_code == 200
assert response.json()["status"] is True
assert response.json()["data"] == "parsed contents"
mock_data_service.read_file.assert_called_once_with("http://example.com/data.csv")
def test_save_metadata(data_client, mock_data_service):
mock_data_service.save_metadata = AsyncMock(return_value="mock_id_123")
payload = {
"columns": {"feature_1": "float"},
"target_columns": ["target"],
"file_url": "http://example.com/data.csv"
}
response = data_client.post("/v1/data/metadata", json=payload)
assert response.status_code == 200
assert response.json()["message"] == "Metadata saved successfully"
mock_data_service.save_metadata.assert_called_once()
def test_update_metadata_success(data_client, mock_data_service):
mock_data_service.update_metadata = AsyncMock(return_value=True)
payload = {
"metadata_id": "meta_1",
"columns": {"feature_1": "float", "feature_2": "int"}
}
response = data_client.put("/v1/data/metadata", json=payload)
assert response.status_code == 200
assert response.json()["message"] == "Metadata updated successfully"
def test_get_column_types(data_client):
response = data_client.get("/v1/data/column-types")
assert response.status_code == 200
assert response.json()["status"] is True
assert "data" in response.json()
|