fixes tests

app/tests/web/endpoints/test_sheet.py

@@ -151,14 +151,14 @@ class TestArchiveUserSheetEndpoint:
         db_session.commit()
 
         m_signature = MagicMock()
-        m_signature.delay.return_value = TaskResult(id="123-taskid", status="PENDING", result="")
+        m_signature.apply_async.return_value = TaskResult(id="123-taskid", status="PENDING", result="")
         m_celery.signature.return_value = m_signature
 
         r = client_with_auth.post("/sheet/123-sheet-id/archive")
         assert r.status_code == 201
         assert r.json() == {"id": "123-taskid"}
         m_celery.signature.assert_called_once()
-        m_signature.delay.assert_called_once()
+        m_signature.apply_async.assert_called_once()
 
     def test_token_auth(self, client_with_token, test_no_auth):
         test_no_auth(client_with_token.post, "/sheet/123-sheet-id/archive")

app/tests/web/endpoints/test_url.py

@@ -12,7 +12,7 @@ def test_archive_url_unauthenticated(client, test_no_auth):
 @patch("app.web.endpoints.url.celery", return_value=MagicMock())
 def test_archive_url(m_celery, m2, client_with_auth):
     m_signature = MagicMock()
-    m_signature.delay.return_value = TaskResult(id="123-456-789", status="PENDING", result="")
+    m_signature.apply_async.return_value = TaskResult(id="123-456-789", status="PENDING", result="")
     m_celery.signature.return_value = m_signature
 
     m_user_state = MagicMock()
@@ -36,7 +36,7 @@ def test_archive_url(m_celery, m2, client_with_auth):
     assert response.status_code == 201
     assert response.json() == {'id': '123-456-789'}
     m_celery.signature.assert_called_once()
-    m_signature.delay.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": "rick@example.com", "group_id": "default", "tags": None, "sheet_id": None, "store_until": None, "urls": None}
@@ -57,7 +57,7 @@ def test_archive_url(m_celery, m2, client_with_auth):
     assert response.status_code == 201
     assert response.json() == {'id': '123-456-789'}
     assert m_celery.signature.call_count == 2
-    assert m_signature.delay.call_count == 2
+    assert m_signature.apply_async.call_count == 2
     called_val = m_celery.signature.call_args
     assert json.loads(called_val[1]['args'][0])["group_id"] == "spaceship"
     m_user_state.in_group.assert_called_with("spaceship")
@@ -78,7 +78,7 @@ def test_archive_url(m_celery, m2, client_with_auth):
     assert response.json()["detail"] == "User has reached their monthly MB quota."
     m_user_state.has_quota_max_monthly_mbs.assert_called_with("spacesuit")
     assert m_celery.signature.call_count == 2
-    assert m_signature.delay.call_count == 2
+    assert m_signature.apply_async.call_count == 2
 
 
 @patch("app.web.endpoints.url.UserState")
@@ -105,13 +105,13 @@ def test_archive_url_quotas(m1, client_with_auth):
 @patch("app.web.endpoints.url.celery", return_value=MagicMock())
 def test_archive_url_with_api_token(m_celery, client_with_token):
     m_signature = MagicMock()
-    m_signature.delay.return_value = TaskResult(id="123-456-789", status="PENDING", result="")
+    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"})
     assert response.status_code == 201
     assert response.json() == {'id': '123-456-789'}
     m_celery.signature.assert_called_once()
-    m_signature.delay.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"
 

app/web/endpoints/url.py

@@ -15,6 +15,8 @@ from app.shared.db.database import get_db_dependency
 
 from urllib.parse import urlparse
 
+from app.web.utils.misc import convert_priority_to_queue_dict
+
 url_router = APIRouter(prefix="/url", tags=["Single URL operations"])
 
 celery = get_celery()
@@ -32,6 +34,7 @@ def archive_url(
     if not all([parsed_url.scheme, parsed_url.netloc]):
         raise HTTPException(status_code=400, detail="Invalid URL received.")
 
+    archive_create = schemas.ArchiveCreate(**archive.model_dump())
     if email != ALLOW_ANY_EMAIL:
         user = UserState(db, email)
         if archive.group_id and not user.in_group(archive.group_id):
@@ -40,10 +43,11 @@ def archive_url(
             raise HTTPException(status_code=429, detail="User has reached their monthly URL quota.")
         if not user.has_quota_max_monthly_mbs(archive.group_id):
             raise HTTPException(status_code=429, detail="User has reached their monthly MB quota.")
+        group_queue = user.priority_group(archive_create.group_id)
+    else:
+        group_queue = convert_priority_to_queue_dict("high")
     
-    archive_create = schemas.ArchiveCreate(**archive.model_dump())
 
-    group_queue = user.priority_group(archive_create.group_id)
     task = celery.signature("create_archive_task", args=[archive_create.model_dump_json()]).apply_async(**group_queue)
     task_response = schemas.Task(id=task.id)
     return JSONResponse(task_response.model_dump(), status_code=201)

app/web/events.py

@@ -176,3 +176,10 @@ async def delete_stale_sheets():
         )
         await fastmail.send_message(message)
         logger.info(f"[CRON] Email sent to {email} about stale sheets deletion.")
+
+
+# @repeat_at
+async def generate_users_export_csv():
+    #TODO: implement a cronjob that regularly requested user data to a CSV file
+    # see https://colab.research.google.com/drive/1QDbo3QXHPBdiTuANlA1AWVvN-rqxuCPa?authuser=0#scrollTo=4nPXeSdK8RBT
+    pass

app/worker/main.py

@@ -48,7 +48,7 @@ def create_archive_task(self, archive_json: str):
 @celery.task(name="create_sheet_task", bind=True)
 def create_sheet_task(self, sheet_json: str):
     sheet = schemas.SubmitSheet.model_validate_json(sheet_json)
-    queue_name = create_sheet_task.request.delivery_info.get('routing_key', 'No queue info')
+    queue_name = (create_sheet_task.request.delivery_info or {}).get('routing_key', 'unknown')
     logger.info(f"[queue={queue_name}] SHEET START {sheet=}")
 
     orchestrator = load_orchestrator(sheet.group_id, True, {"configurations": {"gsheet_feeder": {"sheet_id": sheet.sheet_id}}})