stl-repo/tests/test_api.py

import os
import tempfile
import zipfile
from unittest.mock import patch, AsyncMock
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.main import app

engine = create_engine(
    "sqlite:///:memory:",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
Base.metadata.create_all(bind=engine)
client = TestClient(app)

TEST_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_DIR = os.path.dirname(TEST_DIR)

_upload_counter = 0


def _get_unique_stl():
    global _upload_counter
    _upload_counter += 1
    fd, path = tempfile.mkstemp(suffix='.stl')
    with os.fdopen(fd, 'w') as f:
        f.write(f"""solid test{_upload_counter}
facet normal 0 0 1
outer loop
vertex 0 0 0
vertex {_upload_counter} 0 0
vertex {_upload_counter} {_upload_counter} 0
endloop
endfacet
endsolid test{_upload_counter}
""")
    return path


def test_list_empty_models():
    response = client.get("/api/models/")
    assert response.status_code == 200
    assert response.json() == []


def test_list_tags_empty():
    response = client.get("/api/models/tags")
    assert response.status_code == 200
    assert response.json() == []


def _upload():
    stl_path = _get_unique_stl()
    with open(stl_path, 'rb') as f:
        response = client.post(
            "/api/models/",
            data={
                "title": "Test Cube",
                "description": "A test cube",
                "author": "Tester",
                "tags": "test, cube",
                "category": "Piezas"
            },
            files={"files": (os.path.basename(stl_path), f, "application/octet-stream")}
        )
    assert response.status_code == 200
    return response.json()['id']


def test_upload_model():
    model_id = _upload()
    assert isinstance(model_id, int)


def test_get_model():
    model_id = _upload()
    response = client.get(f"/api/models/{model_id}")
    assert response.status_code == 200
    data = response.json()
    assert data['id'] == model_id
    assert data['title'] == "Test Cube"


def test_update_model():
    model_id = _upload()
    response = client.put(
        f"/api/models/{model_id}",
        json={"title": "Updated Cube", "tag_names": ["updated"]}
    )
    assert response.status_code == 200
    data = response.json()
    assert data['title'] == "Updated Cube"
    assert len(data['tags']) == 1


def test_download_model():
    model_id = _upload()
    response = client.get(f"/api/models/{model_id}/download")
    assert response.status_code == 200


def test_thumbnail():
    model_id = _upload()
    response = client.get(f"/api/models/{model_id}/thumbnail")
    assert response.status_code == 200


def test_qr_endpoint():
    model_id = _upload()
    response = client.get(f"/api/models/{model_id}/qr")
    assert response.status_code == 200


def test_pagination():
    response = client.get("/api/models/?skip=0&limit=10")
    assert response.status_code == 200


def test_search_filter():
    response = client.get("/api/models/?search=Cube")
    assert response.status_code == 200


def test_delete_model():
    model_id = _upload()
    response = client.delete(f"/api/models/{model_id}")
    assert response.status_code == 200
    response = client.get(f"/api/models/{model_id}")
    assert response.status_code == 404


def test_backup_endpoint():
    response = client.get("/api/models/system/backup")
    assert response.status_code == 200


# Phase 5: Social features

def test_create_rating():
    model_id = _upload()
    response = client.post(f"/api/models/{model_id}/ratings?stars=5")
    assert response.status_code == 200
    data = response.json()
    assert data['stars'] == 5

    response = client.get(f"/api/models/{model_id}")
    assert response.json()['avg_rating'] == 5.0


def test_create_comment():
    model_id = _upload()
    response = client.post(f"/api/models/{model_id}/comments?text=Great%20model&author_name=User1")
    assert response.status_code == 200
    data = response.json()
    assert data['text'] == "Great model"
    assert data['author_name'] == "User1"


def test_list_comments():
    model_id = _upload()
    client.post(f"/api/models/{model_id}/comments?text=Comment1")
    client.post(f"/api/models/{model_id}/comments?text=Comment2")
    response = client.get(f"/api/models/{model_id}/comments")
    assert response.status_code == 200
    assert len(response.json()) == 2


def _create_collection():
    response = client.post("/api/models/collections", json={"name": "My Collection", "description": "Test"})
    assert response.status_code == 200
    data = response.json()
    assert data['name'] == "My Collection"
    return data['id']


def test_add_to_collection():
    model_id = _upload()
    coll_id = _create_collection()
    response = client.post(f"/api/models/collections/{coll_id}/add/{model_id}")
    assert response.status_code == 200

    response = client.get(f"/api/models/collections/{coll_id}")
    assert response.status_code == 200
    assert len(response.json()['models']) == 1


# Phase 6: Import / Bulk / Batch / Validation / Estimate

def test_import_from_url():
    stl_path = _get_unique_stl()
    with open(stl_path, 'rb') as f:
        stl_bytes = f.read()

    from unittest.mock import Mock
    mock_response = Mock()
    mock_response.content = stl_bytes
    mock_response.raise_for_status = Mock()

    async_mock_client = AsyncMock()
    async_mock_client.get = AsyncMock(return_value=mock_response)

    with patch('app.routers.models.httpx.AsyncClient', return_value=async_mock_client):
        async_mock_client.__aenter__ = AsyncMock(return_value=async_mock_client)
        async_mock_client.__aexit__ = AsyncMock(return_value=False)
        response = client.post(
            "/api/models/import-url",
            data={"url": "http://example.com/model.stl", "title": "Imported Model"}
        )
    assert response.status_code == 200
    data = response.json()
    assert data['title'] == "Imported Model"


def test_bulk_zip_upload():
    stl_path = _get_unique_stl()
    zip_path = tempfile.mktemp(suffix='.zip')
    with zipfile.ZipFile(zip_path, 'w') as zf:
        zf.write(stl_path, os.path.basename(stl_path))

    with open(zip_path, 'rb') as f:
        response = client.post(
            "/api/models/bulk-zip",
            data={"author": "BulkTester"},
            files={"zip_file": ("models.zip", f, "application/zip")}
        )
    assert response.status_code == 200
    data = response.json()
    assert len(data) >= 1
    os.remove(zip_path)


def test_batch_download():
    model_id1 = _upload()
    model_id2 = _upload()
    response = client.post(
        "/api/models/batch-download",
        json=[model_id1, model_id2]
    )
    assert response.status_code == 200
    assert response.headers['content-type'] == 'application/zip'


def test_validate_mesh():
    model_id = _upload()
    response = client.get(f"/api/models/{model_id}/validate")
    assert response.status_code == 200
    data = response.json()
    assert 'is_watertight' in data
    assert 'volume_cm3' in data


def test_estimate_print():
    model_id = _upload()
    response = client.get(f"/api/models/{model_id}/estimate")
    assert response.status_code == 200
    data = response.json()
    assert 'volume_cm3' in data
    assert 'cost' in data
    assert 'grams' in data