Merge pull request #34 from hv0905/update_opt

Allow editing url and thumbnail_url for non-local image in update_opt API

app/Controllers/admin.py

@@ -67,6 +67,14 @@ async def update_image(image_id: Annotated[UUID, params.Path(description="The id
     except PointNotFoundError as ex:
         raise HTTPException(404, "Cannot find the image with the given ID.") from ex
 
+    if model.thumbnail_url is not None:
+        if point.local or point.local_thumbnail:
+            raise HTTPException(422, "Cannot change the thumbnail URL of a local image.")
+        point.thumbnail_url = model.thumbnail_url
+    if model.url is not None:
+        if point.local:
+            raise HTTPException(422, "Cannot change the URL of a local image.")
+        point.url = model.url
     if model.starred is not None:
         point.starred = model.starred
     if model.categories is not None:

app/Models/api_models/admin_api_model.py

@@ -10,6 +10,14 @@ class ImageOptUpdateModel(BaseModel):
     categories: Optional[list[str]] = Field(None,
                                             description="The categories of the image. Leave empty to keep the value "
                                                         "unchanged.")
+    url: Optional[str] = Field(None,
+                               description="The url of the image. Leave empty to keep the value unchanged. Changing "
+                                           "the url of a local image is not allowed.")
+
+    thumbnail_url: Optional[str] = Field(None,
+                                         description="The url of the thumbnail. Leave empty to keep the value "
+                                                     "unchanged. Changing the thumbnail_url of an image with a local "
+                                                     "thumbnail is not allowed.")
 
     def empty(self) -> bool:
-        return self.starred is None and self.categories is None
+        return all([item is None for item in self.model_dump().values()])

tests/api/integrate_test.py

@@ -12,7 +12,7 @@
 
 
 @pytest.mark.asyncio
-async def test_integrate(test_client, check_local_dir_empty, wait_for_background_task):
+async def test_search(test_client, check_local_dir_empty, wait_for_background_task):
     credentials = {'x-admin-token': TEST_ADMIN_TOKEN, 'x-access-token': TEST_ACCESS_TOKEN}
     resp = test_client.get("/", headers=credentials)
     assert resp.status_code == 200

tests/api/test_upload.py

@@ -6,6 +6,14 @@
 from .conftest import TEST_ADMIN_TOKEN, TEST_ACCESS_TOKEN
 
 
+def get_single_img_info(test_client, image_id):
+    query = test_client.get('/search/random', headers={'x-access-token': TEST_ACCESS_TOKEN})
+    assert query.status_code == 200
+    assert query.json()['result'][0]['img']['id'] == image_id
+
+    return query.json()['result'][0]['img']
+
+
 def test_upload_bad_img_file(test_client):
     bad_img_file = io.BytesIO(bytearray(random.getrandbits(8) for _ in range(1024 * 1024)))
     bad_img_file.name = 'bad_image.jpg'
@@ -31,7 +39,7 @@ def test_upload_unsupported_types(test_client):
 TEST_FAKE_URL = 'fake-url'
 TEST_FAKE_THUMBNAIL_URL = 'fake-thumbnail-url'
 
-TEST_PARAMS = [
+TEST_UPLOAD_THUMBNAILS_PARAMS = [
     (True, {'local': True}, True, 'local'),
     (True, {'local': True, 'local_thumbnail': 'never'}, True, 'none'),
     (False, {'local': True, 'local_thumbnail': 'always'}, True, 'local'),
@@ -42,10 +50,11 @@ def test_upload_unsupported_types(test_client):
 ]
 
 
-@pytest.mark.parametrize('add_trailing_bytes,params,expect_local_url,expect_thumbnail_mode', TEST_PARAMS)
+@pytest.mark.parametrize('add_trailing_bytes,params,expect_local_url,expect_thumbnail_mode',
+                         TEST_UPLOAD_THUMBNAILS_PARAMS)
 @pytest.mark.asyncio
-async def test_upload_auto_local_thumbnail(test_client, check_local_dir_empty, wait_for_background_task,
-                                           add_trailing_bytes, params, expect_local_url, expect_thumbnail_mode):
+async def test_upload_thumbnails(test_client, check_local_dir_empty, wait_for_background_task,  # Fixtures
+                                 add_trailing_bytes, params, expect_local_url, expect_thumbnail_mode):  # Parameters
     with open('tests/assets/test_images/bsn_0.jpg', 'rb') as f:
         img_bytes = f.read()
         # append 500KB to the image, to make it large enough to generate a thumbnail
@@ -60,33 +69,81 @@ async def test_upload_auto_local_thumbnail(test_client, check_local_dir_empty, w
                                 headers={'x-admin-token': TEST_ADMIN_TOKEN},
                                 params=params)
     assert resp.status_code == 200
-    id = resp.json()['image_id']
+    image_id = resp.json()['image_id']
     await wait_for_background_task(1)
 
-    query = test_client.get('/search/random', headers={'x-access-token': TEST_ACCESS_TOKEN})
-    assert query.status_code == 200
-    assert query.json()['result'][0]['img']['id'] == id
+    query = get_single_img_info(test_client, image_id)
 
     if expect_local_url:
-        assert query.json()['result'][0]['img']['url'].startswith(f'/static/{id}.')
-        img_request = test_client.get(query.json()['result'][0]['img']['url'])
+        assert query['url'].startswith(f'/static/{image_id}.')
+        img_request = test_client.get(query['url'])
         assert img_request.status_code == 200
     else:
-        assert query.json()['result'][0]['img']['url'] == TEST_FAKE_URL
+        assert query['url'] == TEST_FAKE_URL
 
     match expect_thumbnail_mode:
         case 'local':
-            assert query.json()['result'][0]['img']['thumbnail_url'] == f'/static/thumbnails/{id}.webp'
+            assert query['thumbnail_url'] == f'/static/thumbnails/{image_id}.webp'
 
-            thumbnail_request = test_client.get(query.json()['result'][0]['img']['thumbnail_url'])
+            thumbnail_request = test_client.get(query['thumbnail_url'])
             assert thumbnail_request.status_code == 200
             # IDK why starlette doesn't return the correct content type, but it works on the browser anyway
             # assert thumbnail_request.headers['Content-Type'] == 'image/webp'
         case 'fake':
-            assert query.json()['result'][0]['img']['thumbnail_url'] == TEST_FAKE_THUMBNAIL_URL
+            assert query['thumbnail_url'] == TEST_FAKE_THUMBNAIL_URL
         case 'none':
-            assert query.json()['result'][0]['img']['thumbnail_url'] is None
+            assert query['thumbnail_url'] is None
+
+    # cleanup
+    resp = test_client.delete(f'/admin/delete/{image_id}', headers={'x-admin-token': TEST_ADMIN_TOKEN})
+    assert resp.status_code == 200
+
+
+TEST_FAKE_URL_NEW = 'fake-url-new'
+TEST_FAKE_THUMBNAIL_URL_NEW = 'fake-thumbnail-url-new'
+
+TEST_UPDATE_OPT_PARAMS = [
+    ({'url': TEST_FAKE_URL}, {'url': TEST_FAKE_URL_NEW, 'thumbnail_url': TEST_FAKE_THUMBNAIL_URL_NEW},
+     {'url': TEST_FAKE_URL_NEW, 'thumbnail_url': TEST_FAKE_THUMBNAIL_URL_NEW}, 200),
+    ({'local_thumbnail': 'always', 'url': TEST_FAKE_URL}, {'url': TEST_FAKE_URL_NEW}, {'url': TEST_FAKE_URL_NEW}, 200),
+    ({'local': True}, {'categories': ['1'], 'starred': True}, {'categories': ['1'], 'starred': True}, 200),
+    ({'local': True}, {'url': TEST_FAKE_URL_NEW}, {}, 422),
+    ({'local': True}, {'thumbnail_url': TEST_FAKE_THUMBNAIL_URL_NEW}, {}, 422),
+    ({'local_thumbnail': 'always', 'url': TEST_FAKE_URL}, {'thumbnail_url': TEST_FAKE_THUMBNAIL_URL_NEW}, {}, 422),
+    ({'local': True}, {}, {}, 422),
+]
+
+
+@pytest.mark.parametrize('initial_param,update_param,expected_param,resp_code', TEST_UPDATE_OPT_PARAMS)
+@pytest.mark.asyncio
+async def test_update_opt(test_client, check_local_dir_empty, wait_for_background_task,  # Fixtures
+                          initial_param, update_param, expected_param, resp_code):  # Parameters
+    with open('tests/assets/test_images/bsn_0.jpg', 'rb') as f:
+        img_bytes = f.read()
+        resp = test_client.post('/admin/upload',
+                                files={'image_file': f},
+                                headers={'x-admin-token': TEST_ADMIN_TOKEN},
+                                params=initial_param)
+    assert resp.status_code == 200
+    image_id = resp.json()['image_id']
+    await wait_for_background_task(1)
+
+    old_info = get_single_img_info(test_client, image_id)
+
+    resp = test_client.put(f'/admin/update_opt/{image_id}', json=update_param,
+                           headers={'x-admin-token': TEST_ADMIN_TOKEN})
+    assert resp.status_code == resp_code
+
+    new_info = get_single_img_info(test_client, image_id)
+    # Ensure expected keys are updated
+    for key, value in expected_param.items():
+        assert new_info[key] == value
+        del new_info[key]
+
+    # Ensure that the other keys are kept untouched
+    for key, value in new_info.items():
+        assert old_info[key] == value
 
     # cleanup
-    resp = test_client.delete(f'/admin/delete/{id}', headers={'x-admin-token': TEST_ADMIN_TOKEN})
+    resp = test_client.delete(f'/admin/delete/{image_id}', headers={'x-admin-token': TEST_ADMIN_TOKEN})
     assert resp.status_code == 200