bellingcat/auto-archiver-api
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver-api.git synced 2026-06-08 03:28:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

api token archive can set the author_id explicitly

Browse Source
This commit is contained in:
msramalho
2025-02-20 15:14:13 +00:00
parent b203594c4b
commit 4c06351cb3
3 changed files with 13 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
app/tests/web/endpoints/test_url.py
View File

@@ -2,6 +2,7 @@ import json
from unittest.mock import MagicMock, patch
from app.shared.schemas import ArchiveCreate, TaskResult
from app.web.config import ALLOW_ANY_EMAIL
def test_archive_url_unauthenticated(client, test_no_auth):
@@ -107,13 +108,21 @@ def test_archive_url_with_api_token(m_celery, client_with_token):
m_signature = MagicMock()
m_signature.apply_async.return_value = TaskResult(id="123-456-789", status="PENDING", result="")
m_celery.signature.return_value = m_signature
response = client_with_token.post("/url/archive", json={"url": "https://example.com"})
response = client_with_token.post("/url/archive", json={"url": "https://example.com", "author_id": "someone@example.com"})
assert response.status_code == 201
assert response.json() == {'id': '123-456-789'}
m_celery.signature.assert_called_once()
m_signature.apply_async.assert_called_once()
called_val = m_celery.signature.call_args
assert called_val[0][0] == "create_archive_task"
assert json.loads(called_val[1]['args'][0]) == {"id": None, "url": "https://example.com", "result": None, "public": False, "author_id": "someone@example.com", "group_id": "default", "tags": None, "sheet_id": None, "store_until": None, "urls": None}
# missing id should use ALLOW_ANY_EMAIL
response = client_with_token.post("/url/archive", json={"url": "https://example.com", "author_id": None})
assert response.status_code == 201
called_val = m_celery.signature.call_args
assert called_val[0][0] == "create_archive_task"
assert json.loads(called_val[1]['args'][0]) == {"id": None, "url": "https://example.com", "result": None, "public": False, "author_id": ALLOW_ANY_EMAIL, "group_id": "default", "tags": None, "sheet_id": None, "store_until": None, "urls": None}
def test_search_by_url_unauthenticated(client, test_no_auth):

2
app/web/config.py
View File

@@ -1,4 +1,4 @@
VERSION = "0.9.1"
VERSION = "0.9.2"
API_DESCRIPTION = """
#### API for the Auto-Archiver project, a tool to archive web pages and Google Sheets.

3
app/web/endpoints/url.py
View File

@@ -27,7 +27,6 @@ def archive_url(
email=Depends(get_token_or_user_auth),
db: Session = Depends(get_db_dependency)
) -> schemas.Task:
archive.author_id = email
logger.info(f"new {archive.public=} task for {email=} and {archive.group_id=}: {archive.url}")
parsed_url = urlparse(archive.url)
@@ -36,6 +35,7 @@ def archive_url(
archive_create = schemas.ArchiveCreate(**archive.model_dump())
if email != ALLOW_ANY_EMAIL:
archive_create.author_id = email
user = UserState(db, email)
if archive.group_id and not user.in_group(archive.group_id):
raise HTTPException(status_code=403, detail="User does not have access to this group.")
@@ -45,6 +45,7 @@ def archive_url(
raise HTTPException(status_code=429, detail="User has reached their monthly MB quota.")
group_queue = user.priority_group(archive_create.group_id)
else:
archive_create.author_id = archive.author_id or email
group_queue = convert_priority_to_queue_dict("high")