refactor(assets): API routes call services directly, extract upload handling

- Refactor routes.py to call service functions directly (no manager layer)
- Extract multipart upload parsing into upload.py
- Update API schemas
- Fix path traversal validation to return 400 instead of 500
- Rename test_tags.py to test_tags_api.py
- Update existing API-level tests

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019c9209-37af-757a-b6e4-af59b4267362

tests-unit/assets_test/conftest.py

@@ -259,13 +259,3 @@ def autoclean_unit_test_assets(http: requests.Session, api_base: str):
         for aid in ids:
             with contextlib.suppress(Exception):
                 http.delete(f"{api_base}/api/assets/{aid}", timeout=30)
-
-
-def trigger_sync_seed_assets(session: requests.Session, base_url: str) -> None:
-    """Force a fast sync/seed pass by calling the seed endpoint."""
-    session.post(base_url + "/api/assets/seed", json={"roots": ["models", "input", "output"]}, timeout=30)
-    time.sleep(0.2)
-
-
-def get_asset_filename(asset_hash: str, extension: str) -> str:
-    return asset_hash.removeprefix("blake3:") + extension

tests-unit/assets_test/test_assets_missing_sync.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 import pytest
 import requests
-from conftest import get_asset_filename, trigger_sync_seed_assets
+from helpers import get_asset_filename, trigger_sync_seed_assets
 
 
 

tests-unit/assets_test/test_crud.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 import pytest
 import requests
-from conftest import get_asset_filename, trigger_sync_seed_assets
+from helpers import get_asset_filename, trigger_sync_seed_assets
 
 
 def test_create_from_hash_success(
@@ -24,11 +24,11 @@ def test_create_from_hash_success(
     assert b1["created_new"] is False
     aid = b1["id"]
 
-    # Calling again with the same name should return the same AssetInfo id
+    # Calling again with the same name creates a new AssetReference (duplicates allowed)
     r2 = http.post(f"{api_base}/api/assets/from-hash", json=payload, timeout=120)
     b2 = r2.json()
     assert r2.status_code == 201, b2
-    assert b2["id"] == aid
+    assert b2["id"] != aid  # new reference, not the same one
 
 
 def test_get_and_delete_asset(http: requests.Session, api_base: str, seeded_asset: dict):
@@ -126,42 +126,52 @@ def test_head_asset_bad_hash_returns_400_and_no_body(http: requests.Session, api
     assert body == b""
 
 
-def test_delete_nonexistent_returns_404(http: requests.Session, api_base: str):
-    bogus = str(uuid.uuid4())
-    r = http.delete(f"{api_base}/api/assets/{bogus}", timeout=120)
+@pytest.mark.parametrize(
+    "method,endpoint_template,payload,expected_status,error_code",
+    [
+        # Delete nonexistent asset
+        ("delete", "/api/assets/{uuid}", None, 404, "ASSET_NOT_FOUND"),
+        # Bad hash algorithm in from-hash
+        (
+            "post",
+            "/api/assets/from-hash",
+            {"hash": "sha256:" + "0" * 64, "name": "x.bin", "tags": ["models", "checkpoints", "unit-tests"]},
+            400,
+            "INVALID_BODY",
+        ),
+        # Get with bad UUID format
+        ("get", "/api/assets/not-a-uuid", None, 404, None),
+        # Get content with bad UUID format
+        ("get", "/api/assets/not-a-uuid/content", None, 404, None),
+    ],
+    ids=["delete_nonexistent", "bad_hash_algorithm", "get_bad_uuid", "content_bad_uuid"],
+)
+def test_error_responses(
+    http: requests.Session, api_base: str, method, endpoint_template, payload, expected_status, error_code
+):
+    # Replace {uuid} placeholder with a random UUID for delete test
+    endpoint = endpoint_template.replace("{uuid}", str(uuid.uuid4()))
+    url = f"{api_base}{endpoint}"
+
+    if method == "get":
+        r = http.get(url, timeout=120)
+    elif method == "post":
+        r = http.post(url, json=payload, timeout=120)
+    elif method == "delete":
+        r = http.delete(url, timeout=120)
+
+    assert r.status_code == expected_status
+    if error_code:
+        body = r.json()
+        assert body["error"]["code"] == error_code
+
+
+def test_create_from_hash_invalid_json(http: requests.Session, api_base: str):
+    """Invalid JSON body requires special handling (data= instead of json=)."""
+    r = http.post(f"{api_base}/api/assets/from-hash", data=b"{not json}", timeout=120)
     body = r.json()
-    assert r.status_code == 404
-    assert body["error"]["code"] == "ASSET_NOT_FOUND"
-
-
-def test_create_from_hash_invalids(http: requests.Session, api_base: str):
-    # Bad hash algorithm
-    bad = {
-        "hash": "sha256:" + "0" * 64,
-        "name": "x.bin",
-        "tags": ["models", "checkpoints", "unit-tests"],
-    }
-    r1 = http.post(f"{api_base}/api/assets/from-hash", json=bad, timeout=120)
-    b1 = r1.json()
-    assert r1.status_code == 400
-    assert b1["error"]["code"] == "INVALID_BODY"
-
-    # Invalid JSON body
-    r2 = http.post(f"{api_base}/api/assets/from-hash", data=b"{not json}", timeout=120)
-    b2 = r2.json()
-    assert r2.status_code == 400
-    assert b2["error"]["code"] == "INVALID_JSON"
-
-
-def test_get_update_download_bad_ids(http: requests.Session, api_base: str):
-    # All endpoints should be not found, as we UUID regex directly in the route definition.
-    bad_id = "not-a-uuid"
-
-    r1 = http.get(f"{api_base}/api/assets/{bad_id}", timeout=120)
-    assert r1.status_code == 404
-
-    r3 = http.get(f"{api_base}/api/assets/{bad_id}/content", timeout=120)
-    assert r3.status_code == 404
+    assert r.status_code == 400
+    assert body["error"]["code"] == "INVALID_JSON"
 
 
 def test_update_requires_at_least_one_field(http: requests.Session, api_base: str, seeded_asset: dict):

tests-unit/assets_test/test_downloads.py

@@ -6,7 +6,7 @@ from typing import Optional
 
 import pytest
 import requests
-from conftest import get_asset_filename, trigger_sync_seed_assets
+from helpers import get_asset_filename, trigger_sync_seed_assets
 
 
 def test_download_attachment_and_inline(http: requests.Session, api_base: str, seeded_asset: dict):

tests-unit/assets_test/test_list_filter.py

@@ -1,6 +1,7 @@
 import time
 import uuid
 
+import pytest
 import requests
 
 
@@ -283,30 +284,21 @@ def test_list_assets_offset_beyond_total_and_limit_boundary(http, api_base, asse
     assert b2["has_more"] is False
 
 
-def test_list_assets_offset_negative_and_limit_nonint_rejected(http, api_base):
-    r1 = http.get(api_base + "/api/assets", params={"offset": "-1"}, timeout=120)
-    b1 = r1.json()
-    assert r1.status_code == 400
-    assert b1["error"]["code"] == "INVALID_QUERY"
-
-    r2 = http.get(api_base + "/api/assets", params={"limit": "abc"}, timeout=120)
-    b2 = r2.json()
-    assert r2.status_code == 400
-    assert b2["error"]["code"] == "INVALID_QUERY"
-
-
-def test_list_assets_invalid_query_rejected(http: requests.Session, api_base: str):
-    # limit too small
-    r1 = http.get(api_base + "/api/assets", params={"limit": "0"}, timeout=120)
-    b1 = r1.json()
-    assert r1.status_code == 400
-    assert b1["error"]["code"] == "INVALID_QUERY"
-
-    # bad metadata JSON
-    r2 = http.get(api_base + "/api/assets", params={"metadata_filter": "{not json"}, timeout=120)
-    b2 = r2.json()
-    assert r2.status_code == 400
-    assert b2["error"]["code"] == "INVALID_QUERY"
+@pytest.mark.parametrize(
+    "params,error_code",
+    [
+        ({"offset": "-1"}, "INVALID_QUERY"),
+        ({"limit": "abc"}, "INVALID_QUERY"),
+        ({"limit": "0"}, "INVALID_QUERY"),
+        ({"metadata_filter": "{not json"}, "INVALID_QUERY"),
+    ],
+    ids=["negative_offset", "non_int_limit", "zero_limit", "invalid_metadata_json"],
+)
+def test_list_assets_invalid_query_rejected(http: requests.Session, api_base: str, params, error_code):
+    r = http.get(api_base + "/api/assets", params=params, timeout=120)
+    body = r.json()
+    assert r.status_code == 400
+    assert body["error"]["code"] == error_code
 
 
 def test_list_assets_name_contains_literal_underscore(

tests-unit/assets_test/test_prune_orphaned_assets.py

@@ -3,7 +3,7 @@ from pathlib import Path
 
 import pytest
 import requests
-from conftest import get_asset_filename, trigger_sync_seed_assets
+from helpers import get_asset_filename, trigger_sync_seed_assets
 
 
 @pytest.fixture

tests-unit/assets_test/test_uploads.py

@@ -18,25 +18,25 @@ def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, ma
     assert r1.status_code == 201, a1
     assert a1["created_new"] is True
 
-    # Second upload with the same data and name should return created_new == False and the same asset
+    # Second upload with the same data and name creates a new AssetReference (duplicates allowed)
+    # Returns 200 because Asset already exists, but a new AssetReference is created
     files = {"file": (name, data, "application/octet-stream")}
     form = {"tags": json.dumps(tags), "name": name, "user_metadata": json.dumps(meta)}
     r2 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
     a2 = r2.json()
-    assert r2.status_code == 200, a2
-    assert a2["created_new"] is False
+    assert r2.status_code in (200, 201), a2
     assert a2["asset_hash"] == a1["asset_hash"]
-    assert a2["id"] == a1["id"]  # old reference
+    assert a2["id"] != a1["id"]  # new reference with same content
 
-    # Third upload with the same data but new name should return created_new == False and the new AssetReference
+    # Third upload with the same data but different name also creates new AssetReference
     files = {"file": (name, data, "application/octet-stream")}
     form = {"tags": json.dumps(tags), "name": name + "_d", "user_metadata": json.dumps(meta)}
-    r2 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
-    a3 = r2.json()
-    assert r2.status_code == 200, a3
-    assert a3["created_new"] is False
+    r3 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    a3 = r3.json()
+    assert r3.status_code in (200, 201), a3
     assert a3["asset_hash"] == a1["asset_hash"]
-    assert a3["id"] != a1["id"]  # old reference
+    assert a3["id"] != a1["id"]
+    assert a3["id"] != a2["id"]
 
 
 def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_base: str):
@@ -116,7 +116,7 @@ def test_concurrent_upload_identical_bytes_different_names(
 ):
     """
     Two concurrent uploads of identical bytes but different names.
-    Expect a single Asset (same hash), two AssetInfo rows, and exactly one created_new=True.
+    Expect a single Asset (same hash), two AssetReference rows, and exactly one created_new=True.
     """
     scope = f"concupload-{uuid.uuid4().hex[:6]}"
     name1, name2 = "cu_a.bin", "cu_b.bin"