220 lines
9.1 KiB
Python
220 lines
9.1 KiB
Python
import json
|
|
import pytest
|
|
from unittest.mock import patch, ANY
|
|
from backend.app import create_app
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
app = create_app()
|
|
app.config["TESTING"] = True
|
|
with app.test_client() as client:
|
|
yield client
|
|
|
|
@pytest.fixture
|
|
def auth_headers(client):
|
|
"""
|
|
Registers a test user with unique username,
|
|
logs in, and returns { "Authorization": "Bearer <token>" } headers.
|
|
"""
|
|
import uuid
|
|
unique_suffix = str(uuid.uuid4())[:8]
|
|
username = f"urltester_{unique_suffix}"
|
|
email = f"{username}@example.com"
|
|
reg_payload = {
|
|
"username": username,
|
|
"email": email,
|
|
"password": "Password123"
|
|
}
|
|
reg_resp = client.post("/api/register", data=json.dumps(reg_payload), content_type="application/json")
|
|
assert reg_resp.status_code == 201, f"User registration failed: {reg_resp.data.decode()}"
|
|
data = json.loads(reg_resp.data)
|
|
return {"Authorization": f"Bearer {data['token']}"}
|
|
|
|
@pytest.fixture
|
|
def project_id(client, auth_headers):
|
|
"""
|
|
Creates a project for the user so we can attach URLs to it.
|
|
Returns the project ID as a string.
|
|
"""
|
|
payload = {"name": "URLs Project", "description": "Project for URL tests."}
|
|
resp = client.post("/api/projects", data=json.dumps(payload),
|
|
content_type="application/json", headers=auth_headers)
|
|
assert resp.status_code == 201, f"Project creation failed: {resp.data.decode()}"
|
|
data = json.loads(resp.data)
|
|
return data["project_id"]
|
|
|
|
def test_create_url(client, auth_headers, project_id):
|
|
"""
|
|
Test creating a URL within a project.
|
|
"""
|
|
payload = {
|
|
"url": "https://example.com",
|
|
"title": "Example Site",
|
|
"note": "Some personal note."
|
|
}
|
|
resp = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert resp.status_code == 201, f"Create URL failed: {resp.data.decode()}"
|
|
resp_data = json.loads(resp.data)
|
|
assert "url_id" in resp_data
|
|
|
|
def test_list_urls(client, auth_headers, project_id):
|
|
"""
|
|
Test listing multiple URLs in a project.
|
|
"""
|
|
# Create first URL
|
|
payload1 = {"url": "https://first-url.com", "title": "First URL"}
|
|
resp1 = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload1),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert resp1.status_code == 201, f"First URL creation failed: {resp1.data.decode()}"
|
|
|
|
# Create second URL
|
|
payload2 = {"url": "https://second-url.com", "title": "Second URL"}
|
|
resp2 = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload2),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert resp2.status_code == 201, f"Second URL creation failed: {resp2.data.decode()}"
|
|
|
|
# Now list them
|
|
list_resp = client.get(f"/api/projects/{project_id}/urls", headers=auth_headers)
|
|
assert list_resp.status_code == 200, f"List URLs failed: {list_resp.data.decode()}"
|
|
data = json.loads(list_resp.data)
|
|
assert "urls" in data
|
|
assert len(data["urls"]) >= 2
|
|
|
|
def test_get_url_detail(client, auth_headers, project_id):
|
|
"""
|
|
Test retrieving URL detail.
|
|
"""
|
|
payload = {"url": "https://detail-url.com", "title": "Detail URL"}
|
|
resp = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert resp.status_code == 201
|
|
url_id = json.loads(resp.data)["url_id"]
|
|
|
|
detail_resp = client.get(f"/api/urls/{url_id}", headers=auth_headers)
|
|
assert detail_resp.status_code == 200, f"Get URL detail failed: {detail_resp.data.decode()}"
|
|
detail_data = json.loads(detail_resp.data)
|
|
assert detail_data.get("title") == "Detail URL"
|
|
|
|
def test_update_url(client, auth_headers, project_id):
|
|
"""
|
|
Test updating an existing URL's fields.
|
|
"""
|
|
payload = {"url": "https://update-url.com", "title": "ToBeUpdated"}
|
|
create_resp = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert create_resp.status_code == 201
|
|
url_id = json.loads(create_resp.data)["url_id"]
|
|
|
|
update_payload = {"title": "Updated Title", "starred": True, "note": "Updated note."}
|
|
update_resp = client.put(f"/api/urls/{url_id}",
|
|
data=json.dumps(update_payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert update_resp.status_code == 200, f"Update URL failed: {update_resp.data.decode()}"
|
|
|
|
# Confirm
|
|
detail_resp = client.get(f"/api/urls/{url_id}", headers=auth_headers)
|
|
data = json.loads(detail_resp.data)
|
|
assert data.get("title") == "Updated Title"
|
|
assert data.get("starred") is True
|
|
assert data.get("note") == "Updated note."
|
|
|
|
@patch("backend.routes.urls.async_extract_title_and_keywords.delay")
|
|
def test_extract_title_and_keywords(mock_task_delay, client, auth_headers, project_id):
|
|
"""
|
|
Test the asynchronous title/keyword extraction. We mock the Celery task's .delay() call.
|
|
"""
|
|
payload = {"url": "https://mock-url.com"}
|
|
create_resp = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert create_resp.status_code == 201
|
|
url_id = json.loads(create_resp.data)["url_id"]
|
|
|
|
# Call the asynchronous endpoint
|
|
extract_resp = client.put(f"/api/urls/{url_id}/extract_title_and_keywords", headers=auth_headers)
|
|
# We now expect 202, since it queues a Celery task
|
|
assert extract_resp.status_code == 202, f"Extraction queueing failed: {extract_resp.data.decode()}"
|
|
|
|
# Confirm the Celery task was indeed called with .delay(...)
|
|
mock_task_delay.assert_called_once_with(url_id, ANY)
|
|
|
|
@patch("backend.routes.urls.async_summarize_url.delay")
|
|
def test_summarize_url(mock_task_delay, client, auth_headers, project_id):
|
|
"""
|
|
Test the asynchronous summarization by mocking the Celery task call.
|
|
"""
|
|
payload = {"url": "https://mock-summary.com"}
|
|
create_resp = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert create_resp.status_code == 201
|
|
url_id = json.loads(create_resp.data)["url_id"]
|
|
|
|
summarize_resp = client.put(f"/api/urls/{url_id}/summarize", headers=auth_headers)
|
|
# Again, we expect 202
|
|
assert summarize_resp.status_code == 202, f"Summarization queueing failed: {summarize_resp.data.decode()}"
|
|
|
|
mock_task_delay.assert_called_once_with(url_id, ANY)
|
|
|
|
def test_search_urls(client, auth_headers, project_id):
|
|
"""
|
|
Test searching URLs by note or keywords.
|
|
"""
|
|
# Create multiple URLs
|
|
url1_payload = {
|
|
"url": "https://search-url1.com",
|
|
"note": "Unique note text",
|
|
"keywords": [{"word": "alpha", "percentage": 90}]
|
|
}
|
|
url2_payload = {
|
|
"url": "https://search-url2.com",
|
|
"note": "Another note containing alpha",
|
|
"keywords": [{"word": "beta", "percentage": 50}]
|
|
}
|
|
resp1 = client.post(f"/api/projects/{project_id}/urls", data=json.dumps(url1_payload),
|
|
content_type="application/json", headers=auth_headers)
|
|
resp2 = client.post(f"/api/projects/{project_id}/urls", data=json.dumps(url2_payload),
|
|
content_type="application/json", headers=auth_headers)
|
|
assert resp1.status_code == 201
|
|
assert resp2.status_code == 201
|
|
|
|
search_resp = client.get(f"/api/projects/{project_id}/search?q=alpha", headers=auth_headers)
|
|
assert search_resp.status_code == 200, f"Search failed: {search_resp.data.decode()}"
|
|
data = json.loads(search_resp.data)
|
|
results = data.get("results", [])
|
|
# Both URLs mention 'alpha'
|
|
assert len(results) >= 2
|
|
|
|
def test_delete_url(client, auth_headers, project_id):
|
|
"""
|
|
Test deleting a URL.
|
|
"""
|
|
payload = {"url": "https://delete-url.com", "title": "Delete-Me"}
|
|
create_resp = client.post(f"/api/projects/{project_id}/urls",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
headers=auth_headers)
|
|
assert create_resp.status_code == 201
|
|
url_id = json.loads(create_resp.data)["url_id"]
|
|
|
|
delete_resp = client.delete(f"/api/urls/{url_id}", headers=auth_headers)
|
|
assert delete_resp.status_code == 200, f"Deletion failed: {delete_resp.data.decode()}"
|
|
|
|
# Confirm it's gone
|
|
detail_resp = client.get(f"/api/urls/{url_id}", headers=auth_headers)
|
|
assert detail_resp.status_code == 404, "URL is still accessible after deletion."
|