Add v1.2 endpoint tests

src/simdb/remote/models.py

@@ -2,7 +2,18 @@
 
 from datetime import datetime as dt
 from datetime import timezone
-from typing import Annotated, Any, Generic, List, Literal, Optional, TypeVar, Union
+from pathlib import Path
+from typing import (
+    Annotated,
+    Any,
+    Dict,
+    Generic,
+    List,
+    Literal,
+    Optional,
+    TypeVar,
+    Union,
+)
 from urllib.parse import urlencode
 from uuid import UUID, uuid1
 
@@ -15,6 +26,8 @@
     model_validator,
 )
 
+from simdb.cli.manifest import DataObject
+
 HexUUID = Annotated[UUID, PlainSerializer(lambda x: x.hex, return_type=str)]
 """UUID serialized as a hex string."""
 
@@ -302,3 +315,157 @@ class SimulationTraceData(SimulationData):
     """Simulation this one replaces."""
     replaces_reason: Optional[Any] = None
     """Reason for replacement."""
+
+
+class ChunkInfo(BaseModel):
+    """Information about a single chunk in a chunked file upload."""
+
+    chunk_size: int
+    """Length of the chunk."""
+    chunk: int
+    """Index of the chunk."""
+    num_chunks: Optional[int] = 1
+    """Total amount of chunks in the file."""
+
+
+class ChunkInfoDict(RootModel):
+    """Dictionary mapping file UUID hex to chunk info."""
+
+    root: Dict[str, ChunkInfo]
+
+
+class FileUploadData(BaseModel):
+    """Data payload for file chunk upload (sent as JSON in 'data' field)."""
+
+    simulation: SimulationData
+    """The simulation the file belongs to."""
+    file_type: str
+    """Type of the file."""
+    chunk_info: Optional[Dict[str, ChunkInfo]] = None
+    """Info about the chunk."""
+
+
+class FilesGetResponse(RootModel):
+    """Response from the get files endpoint."""
+
+    root: List[FileData]
+    """List of files."""
+
+
+class FileInfo(BaseModel):
+    """Information about a single file on disk."""
+
+    path: Path
+    """Path to the file."""
+    checksum: str
+    """Checksum of the file."""
+
+
+class FileGetDataResponse(FileData):
+    """Response from the get file data endpoint, extending FileData with disk info."""
+
+    files: List[FileInfo]
+    """List of file info entries for the files on disk."""
+
+
+class FileUploadResponse(BaseModel):
+    """Response from file upload/chunk upload endpoint."""
+
+    pass
+
+
+class FileRegistrationItem(BaseModel):
+    """A single file entry in the file registration payload."""
+
+    chunks: int
+    """The amount of chunks to be processed."""
+    file_type: str
+    """The file type."""
+    file_uuid: HexUUID
+    """The UUID of the file."""
+    ids_list: Optional[List[Any]] = None
+    """List of IDS names associated with the file."""
+
+
+class FileRegistrationData(BaseModel):
+    """Payload for final file registration after chunk uploads."""
+
+    simulation: SimulationData
+    """The simulation the files belong to."""
+    obj_type: DataObject.Type
+    """The type of the data object being registered."""
+    files: List[FileRegistrationItem]
+    """List of file registration items."""
+
+
+class FileRegistrationResponse(BaseModel):
+    """Response from file registration endpoint."""
+
+    pass
+
+
+class WatcherReference(BaseModel):
+    """An watcher entry reference."""
+
+    simulation: HexUUID
+    """Simulation UUID the watcher has been added to."""
+    watcher: str
+    """Username of the added watcher."""
+
+
+class WatcherPostResponse(BaseModel):
+    """Response from the add watcher endpoint."""
+
+    added: WatcherReference
+    """The added watcher data."""
+
+
+class WatcherPostRequest(BaseModel):
+    """Payload for adding a watcher to a simulation."""
+
+    user: Optional[str]
+    """Username of the watcher, defaults to the signed in user."""
+    email: Optional[str]
+    """Email of the watcher, defaults to the signed in user."""
+    notification: Literal["VALIDATION", "REVISION", "OBSOLESCENCE", "ALL"]
+    """Notificaiton type of the watcher."""
+
+
+class WatcherData(BaseModel):
+    """Payload describing a watcher."""
+
+    username: str
+    """Username of the watcher."""
+    email: str
+    """Email address of the watcher."""
+    notification: Literal["V", "R", "O", "A"]
+    """Notification type of the watcher.
+        Types are: V(alidation), R(evision), O(bsolescence) and A(ll)
+    """
+
+
+class WatcherGetResponse(RootModel):
+    """Response from the get watchers endpoint."""
+
+    root: List[WatcherData]
+
+
+class WatcherDeleteRequest(BaseModel):
+    """Payload for deleting a watcher from a simulation."""
+
+    user: str
+    """Username to delete from the watchers."""
+
+
+class WatcherDeleteResponse(BaseModel):
+    """Response from the delete watchers endpoint."""
+
+    removed: WatcherReference
+    """Reference to the deleted wacher."""
+
+
+class StagingDirectoryResponse(BaseModel):
+    """Response from the get staging dir endpoint."""
+
+    staging_dir: Path
+    """Path to the staging dir."""

tests/remote/api/conftest.py

@@ -0,0 +1,128 @@
+import base64
+import importlib
+import os
+import shutil
+import tempfile
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from simdb.cli.manifest import Manifest
+from simdb.config import Config
+from simdb.database.models import Simulation
+from simdb.remote.app import create_app
+from simdb.remote.models import (
+    FileData,
+    SimulationData,
+    SimulationPostData,
+)
+
+has_flask = importlib.util.find_spec("flask") is not None
+
+
+TEST_PASSWORD = "test123"
+CREDENTIALS = base64.b64encode(f"admin:{TEST_PASSWORD}".encode()).decode()
+HEADERS = {"Authorization": f"Basic {CREDENTIALS}"}
+
+SIMULATIONS = []
+for _ in range(100):
+    SIMULATIONS.append(Simulation(Manifest()))
+
+
+@pytest.fixture(scope="session")
+def client():
+    if not has_flask:
+        pytest.skip("Flask not installed")
+    config = Config()
+    config.load()
+    db_fd, db_file = tempfile.mkstemp()
+    upload_dir = tempfile.mkdtemp()
+    config.set_option("database.type", "sqlite")
+    config.set_option("database.file", db_file)
+    config.set_option("server.admin_password", TEST_PASSWORD)
+    config.set_option("server.upload_folder", upload_dir)
+    config.set_option("authentication.type", "None")
+    config.set_option("server.copy_files", False)
+    config.set_option("role.admin.users", "admin,admin2")
+    app = create_app(config=config, testing=True, debug=True)
+    app.testing = True
+
+    with app.test_client() as client:
+        # with app.app_context():
+        for sim in SIMULATIONS:
+            app.db.insert_simulation(sim)
+
+        app.db.session.commit()
+        app.db.session.close()
+
+        yield client
+
+    os.close(db_fd)
+    Path(app.simdb_config.get_option("database.file")).unlink()
+    shutil.rmtree(upload_dir)
+
+
+@pytest.fixture(scope="session")
+def client_copy_files():
+    if not has_flask:
+        pytest.skip("Flask not installed")
+    config = Config()
+    config.load()
+    db_fd, db_file = tempfile.mkstemp()
+    upload_dir = tempfile.mkdtemp()
+    config.set_option("database.type", "sqlite")
+    config.set_option("database.file", db_file)
+    config.set_option("server.admin_password", TEST_PASSWORD)
+    config.set_option("server.upload_folder", upload_dir)
+    config.set_option("authentication.type", "None")
+    config.set_option("server.copy_files", True)
+    config.set_option("role.admin.users", "admin,admin2")
+    app = create_app(config=config, testing=True, debug=True)
+    app.testing = True
+
+    with app.test_client() as client:
+        # with app.app_context():
+        for sim in SIMULATIONS:
+            app.db.insert_simulation(sim)
+
+        app.db.session.commit()
+        app.db.session.close()
+
+        yield client
+
+    os.close(db_fd)
+    Path(app.simdb_config.get_option("database.file")).unlink()
+    shutil.rmtree(upload_dir)
+
+
+def generate_simulation_data(
+    add_watcher=False, uploaded_by=None, alias=None, **overrides
+) -> SimulationPostData:
+    if alias is None:
+        alias = uuid.uuid4().hex
+    simulation_data = SimulationData(alias=alias, **overrides)
+    data = SimulationPostData(
+        simulation=simulation_data, add_watcher=add_watcher, uploaded_by=uploaded_by
+    )
+    return data
+
+
+def generate_simulation_file() -> FileData:
+    return FileData(
+        type="FILE",
+        uri="file:///path/to/file",
+        checksum="fake_checksum",
+        datetime=datetime.now(timezone.utc),
+    )
+
+
+def post_simulation(client, simulation_data, headers=HEADERS):
+    rv_post = client.post(
+        "/v1.2/simulations",
+        json=simulation_data.model_dump(mode="json"),
+        headers=headers,
+        content_type="application/json",
+    )
+    return rv_post