ldy/SurfSmart
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial Commit

Browse Source
This commit is contained in:
ldy
2025-06-09 17:53:19 +08:00
parent de861d4815
commit 2be3d00ac4
96 changed files with 13327 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
test/tests_backend/__init__.py Normal file
View File

@@ -0,0 +1 @@

169
test/tests_backend/test_activity.py Normal file
View File

@@ -0,0 +1,169 @@
import json
import pytest
from backend.app import create_app
@pytest.fixture
def client():
"""
Creates a test client from your Flask app.
"""
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 and returns an authorization header.
"""
import uuid
unique_suffix = str(uuid.uuid4())[:8]
username = f"activityuser_{unique_suffix}"
email = f"{username}@example.com"
reg_payload = {"username": username, "email": email, "password": "TestPassword123"}
reg_resp = client.post("/api/register", data=json.dumps(reg_payload), content_type="application/json")
assert reg_resp.status_code == 201, f"Registration failed: {reg_resp.data.decode()}"
data = json.loads(reg_resp.data)
token = data["token"]
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def project_id(client, auth_headers):
"""
Creates a project for the user so we can attach activity logs to it.
Returns the project ID as a string.
"""
payload = {
"name": "Activity Test Project",
"topic": "Log Testing",
"description": "Project used for testing activity logs."
}
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_list_activity_empty(client, auth_headers, project_id):
"""
No logs exist initially, so listing with ?projectId=xxx returns empty array.
"""
url = f"/api/project_activity?projectId={project_id}"
resp = client.get(url, headers=auth_headers)
assert resp.status_code == 200, f"List logs failed: {resp.data.decode()}"
data = json.loads(resp.data)
assert "activity_logs" in data
assert len(data["activity_logs"]) == 0
def test_create_activity(client, auth_headers, project_id):
"""
Create an activity log for the project, then verify it appears in the listing.
"""
payload = {
"projectId": project_id,
"activityType": "URL added",
"message": "Added a new URL to the project."
}
create_resp = client.post("/api/project_activity",
data=json.dumps(payload),
content_type="application/json",
headers=auth_headers)
assert create_resp.status_code == 201, f"Create activity failed: {create_resp.data.decode()}"
create_data = json.loads(create_resp.data)
assert "activity_id" in create_data
# Now list logs
list_resp = client.get(f"/api/project_activity?projectId={project_id}", headers=auth_headers)
assert list_resp.status_code == 200
list_data = json.loads(list_resp.data)
logs = list_data.get("activity_logs", [])
assert len(logs) == 1
assert logs[0]["activityType"] == "URL added"
assert logs[0]["message"] == "Added a new URL to the project."
def test_create_activity_invalid_project(client, auth_headers):
"""
If projectId is invalid or not found, we expect 400 or 404.
"""
# Invalid format
payload1 = {
"projectId": "not_a_valid_objectid",
"activityType": "Test",
"message": ""
}
resp1 = client.post("/api/project_activity",
data=json.dumps(payload1),
content_type="application/json",
headers=auth_headers)
assert resp1.status_code == 400, f"Expected 400 for invalid projectId format, got {resp1.status_code}"
# 404 if project doesn't exist
payload2 = {
"projectId": "64f3f000000000000000abcd", # random objectId
"activityType": "Test",
"message": ""
}
resp2 = client.post("/api/project_activity",
data=json.dumps(payload2),
content_type="application/json",
headers=auth_headers)
assert resp2.status_code == 404, f"Expected 404 for non-existent project, got {resp2.status_code}"
def test_list_activity_pagination(client, auth_headers, project_id):
"""
Create multiple logs, then retrieve them with offset/limit.
"""
# Create 5 logs
for i in range(5):
pay = {
"projectId": project_id,
"activityType": f"LogType{i}",
"message": f"Message {i}"
}
resp = client.post("/api/project_activity",
data=json.dumps(pay),
content_type="application/json",
headers=auth_headers)
assert resp.status_code == 201
# List with limit=2, offset=0
resp_list1 = client.get(f"/api/project_activity?projectId={project_id}&limit=2&offset=0", headers=auth_headers)
assert resp_list1.status_code == 200
data1 = json.loads(resp_list1.data)
logs1 = data1["activity_logs"]
assert len(logs1) == 2
# List with limit=2, offset=2
resp_list2 = client.get(f"/api/project_activity?projectId={project_id}&limit=2&offset=2", headers=auth_headers)
data2 = json.loads(resp_list2.data)
logs2 = data2["activity_logs"]
assert len(logs2) == 2
def test_delete_activity(client, auth_headers, project_id):
"""
By default, only the project owner can delete logs. We'll test that scenario.
"""
# Create a log
payload = {
"projectId": project_id,
"activityType": "DeleteCheck",
"message": "Testing delete."
}
create_resp = client.post("/api/project_activity",
data=json.dumps(payload),
content_type="application/json",
headers=auth_headers)
assert create_resp.status_code == 201
activity_id = json.loads(create_resp.data)["activity_id"]
# Delete it
del_resp = client.delete(f"/api/project_activity/{activity_id}", headers=auth_headers)
assert del_resp.status_code == 200, f"Delete log failed: {del_resp.data.decode()}"
# Confirm it's gone
list_resp = client.get(f"/api/project_activity?projectId={project_id}", headers=auth_headers)
data = json.loads(list_resp.data)
logs_left = [a for a in data["activity_logs"] if a["_id"] == activity_id]
assert len(logs_left) == 0, "Activity log was not deleted properly."

162
test/tests_backend/test_api_list.py Normal file
View File

@@ -0,0 +1,162 @@
import json
import pytest
from backend.app import create_app
@pytest.fixture
def client():
"""
Creates a Flask test client from your create_app function.
"""
app = create_app()
app.config["TESTING"] = True
with app.test_client() as client:
yield client
@pytest.fixture
def auth_headers(client):
"""
Registers and logs in a test user, returning an authorization header.
"""
import uuid
unique_suffix = str(uuid.uuid4())[:8]
username = f"apilistuser_{unique_suffix}"
email = f"{username}@example.com"
reg_payload = {
"username": username,
"email": email,
"password": "TestPassword123"
}
reg_resp = client.post("/api/register", data=json.dumps(reg_payload), content_type="application/json")
assert reg_resp.status_code == 201, f"Registration failed: {reg_resp.data.decode()}"
data = json.loads(reg_resp.data)
token = data["token"]
return {"Authorization": f"Bearer {token}"}
def test_list_api_keys_empty(client, auth_headers):
"""
Initially, the user has no API keys.
GET /api/api_list should return empty array.
"""
resp = client.get("/api/api_list", headers=auth_headers)
assert resp.status_code == 200, f"List keys failed: {resp.data.decode()}"
data = json.loads(resp.data)
assert "api_keys" in data
assert len(data["api_keys"]) == 0
def test_create_api_key(client, auth_headers):
"""
Test creating an API key for e.g. 'Gemini'.
"""
payload = {
"name": "Gemini",
"key": "gemini-secret-key",
"selected": True
}
create_resp = client.post("/api/api_list", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Create API key failed: {create_resp.data.decode()}"
data = json.loads(create_resp.data)
assert "api_id" in data
# Now list and verify we see the new key
list_resp = client.get("/api/api_list", headers=auth_headers)
assert list_resp.status_code == 200
list_data = json.loads(list_resp.data)
keys = list_data.get("api_keys", [])
assert len(keys) == 1
api_entry = keys[0]
assert api_entry["name"] == "Gemini"
assert api_entry["key"] == "gemini-secret-key"
assert api_entry["selected"] is True
def test_create_api_key_duplicate(client, auth_headers):
"""
By default, the code returns 400 if the user already has an API key for the same 'name'.
"""
# Create first
payload1 = {
"name": "Chatgpt",
"key": "chatgpt-key1",
"selected": False
}
resp1 = client.post("/api/api_list", data=json.dumps(payload1),
content_type="application/json", headers=auth_headers)
assert resp1.status_code == 201
# Try creating second with same 'name' => should fail with 400
payload2 = {
"name": "Chatgpt",
"key": "chatgpt-key2",
"selected": True
}
resp2 = client.post("/api/api_list", data=json.dumps(payload2),
content_type="application/json", headers=auth_headers)
assert resp2.status_code == 400, f"Expected 400 on duplicate provider, got {resp2.status_code}"
def test_create_api_key_invalid_name(client, auth_headers):
"""
If 'name' is not in ['Gemini','Deepseek','Chatgpt'], the code should fail with 400.
"""
invalid_payload = {
"name": "InvalidProvider",
"key": "some-key",
"selected": False
}
resp = client.post("/api/api_list", data=json.dumps(invalid_payload),
content_type="application/json", headers=auth_headers)
assert resp.status_code == 400, f"Expected 400 on invalid 'name', got {resp.status_code}"
def test_update_api_key(client, auth_headers):
"""
Test updating an existing API key fields: 'key', 'selected', or 'name'.
"""
# Create
payload = {
"name": "Deepseek",
"key": "deepseek-initial",
"selected": False
}
create_resp = client.post("/api/api_list", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201
api_id = json.loads(create_resp.data)["api_id"]
# Now update
update_payload = {"key": "deepseek-updated", "selected": True}
update_resp = client.put(f"/api/api_list/{api_id}",
data=json.dumps(update_payload),
content_type="application/json", headers=auth_headers)
assert update_resp.status_code == 200, f"Update failed: {update_resp.data.decode()}"
# List and confirm
list_resp = client.get("/api/api_list", headers=auth_headers)
data = json.loads(list_resp.data)
updated_entry = next((k for k in data["api_keys"] if k["_id"] == api_id), None)
assert updated_entry is not None, "Updated key not found in list"
assert updated_entry["key"] == "deepseek-updated"
assert updated_entry["selected"] is True
def test_delete_api_key(client, auth_headers):
"""
Test deleting an API key by _id.
"""
payload = {
"name": "Gemini",
"key": "gemini-key2",
"selected": False
}
create_resp = client.post("/api/api_list", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201
api_id = json.loads(create_resp.data)["api_id"]
# Delete
delete_resp = client.delete(f"/api/api_list/{api_id}", headers=auth_headers)
assert delete_resp.status_code == 200, f"Deletion failed: {delete_resp.data.decode()}"
# Confirm it's gone
list_resp = client.get("/api/api_list", headers=auth_headers)
data = json.loads(list_resp.data)
keys_left = [k for k in data["api_keys"] if k["_id"] == api_id]
assert len(keys_left) == 0, "API key was not deleted properly"

92
test/tests_backend/test_auth.py Normal file
View File

@@ -0,0 +1,92 @@
import json
import pytest
from backend.app import create_app # Adjust import paths based on your folder structure
@pytest.fixture
def client():
app = create_app()
app.config["TESTING"] = True
# Optionally: use a test-specific MongoDB URI.
# app.config["MONGO_URI"] = "your_test_mongodb_connection_string"
with app.test_client() as client:
yield client
def test_register_and_login(client):
# Registration test.
reg_payload = {
"username": "test",
"email": "test@example.com",
"password": "1234"
}
response = client.post(
"/api/register",
data=json.dumps(reg_payload),
content_type="application/json"
)
assert response.status_code == 201
reg_data = json.loads(response.data)
assert "token" in reg_data
user_id = reg_data["user_id"]
# Login test.
login_payload = {
"username": "test",
"password": "1234"
}
response = client.post(
"/api/login",
data=json.dumps(login_payload),
content_type="application/json"
)
assert response.status_code == 200
login_data = json.loads(response.data)
assert "token" in login_data
assert user_id == login_data["user_id"]
def test_delete_account(client):
# Step 1: Register a new user
reg_payload = {
"username": "testuse",
"email": "testuse@example.com",
"password": "TestPassword123"
}
response = client.post(
"/api/register",
data=json.dumps(reg_payload),
content_type="application/json"
)
assert response.status_code == 201
reg_data = json.loads(response.data)
token = reg_data["token"]
user_id = reg_data["user_id"]
# Step 2: Optionally, login to verify credentials and obtain a fresh token
login_payload = {
"username": "testuse",
"password": "TestPassword123"
}
response = client.post(
"/api/login",
data=json.dumps(login_payload),
content_type="application/json"
)
assert response.status_code == 200
login_data = json.loads(response.data)
assert "token" in login_data
assert user_id == login_data["user_id"]
# Use the token from login (or registration) and add the "Bearer" prefix as expected.
headers = {"Authorization": f"Bearer {login_data['token']}"}
# Step 3: Call the delete_account endpoint using the DELETE method.
response = client.delete("/api/delete_account", headers=headers)
# Expecting a successful deletion, i.e. status code 200.
assert response.status_code == 200
delete_data = json.loads(response.data)
assert "deleted successfully" in delete_data["message"]

218
test/tests_backend/test_dialog.py Normal file
View File

@@ -0,0 +1,218 @@
import os
import json
import pytest
import uuid
from bson.objectid import ObjectId
from backend.app import create_app
from backend.extensions import mongo
# Ensure a valid Gemini API key is provided for integration testing.
VALID_GEMINI_KEY = "AIzaSyAMpVRmzQPYAYRH5GiBoQLY-r95ohYmhYs"
@pytest.fixture
def client():
"""
Initializes and yields a Flask test client from create_app().
"""
app = create_app()
app.config["TESTING"] = True
with app.test_client() as test_client:
yield test_client
@pytest.fixture
def auth_headers(client):
"""
Registers and logs in a new test user.
Returns a dictionary containing the Authorization header and the user_id.
"""
unique_suffix = str(uuid.uuid4())[:8]
username = f"test_dialog_{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()}"
reg_data = reg_resp.get_json()
return {"Authorization": f"Bearer {reg_data['token']}", "user_id": reg_data["user_id"]}
@pytest.fixture
def gemini_api_config(auth_headers):
"""
Inserts a Gemini API key document for the test user into the api_list collection.
Uses the valid API key from environment variable.
"""
from datetime import datetime # Import the datetime class for utcnow()
user_id = ObjectId(auth_headers["user_id"])
api_payload = {
"uid": user_id,
"name": "Gemini",
"key": VALID_GEMINI_KEY,
"selected": True,
"createdAt": datetime.utcnow(),
"updatedAt": datetime.utcnow()
}
inserted_result = mongo.db.api_list.insert_one(api_payload)
yield
# Teardown: remove the API key document after test.
mongo.db.api_list.delete_one({"_id": inserted_result.inserted_id})
@pytest.fixture
def project_id(client, auth_headers):
"""
Creates a new project for the test user and returns its ID.
"""
payload = {
"name": "Dialog Test Project",
"topic": "Integration Testing",
"description": "Project created for testing dialog endpoints."
}
resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers={"Authorization": auth_headers["Authorization"]})
assert resp.status_code == 201, f"Project creation failed: {resp.data.decode()}"
data = resp.get_json()
return data["project_id"]
def test_create_dialog_no_start_message(client, auth_headers, gemini_api_config, project_id):
"""
Test creating a dialog session without a start message.
"""
payload = {"projectId": project_id}
resp = client.post("/api/dialog",
data=json.dumps(payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
assert resp.status_code == 201, f"Expected 201, got {resp.status_code}, {resp.data.decode()}"
data = resp.get_json()
assert "dialog_id" in data
# Retrieve the new dialog session to verify
dialog_id = data["dialog_id"]
get_resp = client.get(f"/api/dialog/{dialog_id}", headers={"Authorization": auth_headers["Authorization"]})
assert get_resp.status_code == 200, f"Failed to get dialog: {get_resp.data.decode()}"
dialog_data = get_resp.get_json()
assert "sessionStartedAt" in dialog_data
def test_create_dialog_with_start_message(client, auth_headers, gemini_api_config, project_id):
"""
Test creating a dialog session with a start message.
"""
payload = {
"projectId": project_id,
"sessionId": "testSession123",
"startMessage": "Hello, I need research guidance."
}
resp = client.post("/api/dialog",
data=json.dumps(payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
assert resp.status_code == 201, f"Expected 201, got {resp.status_code}, {resp.data.decode()}"
data = resp.get_json()
dialog_id = data["dialog_id"]
get_resp = client.get(f"/api/dialog/{dialog_id}", headers={"Authorization": auth_headers["Authorization"]})
assert get_resp.status_code == 200, f"Failed to retrieve dialog: {get_resp.data.decode()}"
dialog_data = get_resp.get_json()
msgs = dialog_data.get("messages", [])
assert len(msgs) >= 1, "Expected at least one message in the dialog."
assert msgs[0]["role"] == "user"
assert "Hello, I need research guidance." in msgs[0]["content"]
def test_list_dialogs(client, auth_headers, gemini_api_config, project_id):
"""
Creates a couple of dialogs and then lists dialogs filtered by project.
"""
for _ in range(2):
payload = {"projectId": project_id}
resp = client.post("/api/dialog",
data=json.dumps(payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
assert resp.status_code == 201
list_resp = client.get(f"/api/dialog?projectId={project_id}",
headers={"Authorization": auth_headers["Authorization"]})
assert list_resp.status_code == 200, f"Listing dialogs failed: {list_resp.data.decode()}"
data = list_resp.get_json()
dialogs = data.get("dialogs", [])
assert len(dialogs) >= 2, "Expected at least two dialog sessions."
def test_send_dialog_message_real_gemini(client, auth_headers, gemini_api_config, project_id):
"""
Test sending a message in a dialog session using the vector-based prompt.
This test interacts with the real Gemini API.
"""
# Create a new dialog session.
create_payload = {"projectId": project_id}
create_resp = client.post("/api/dialog",
data=json.dumps(create_payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
assert create_resp.status_code == 201, f"Dialog creation failed: {create_resp.data.decode()}"
dialog_id = create_resp.get_json()["dialog_id"]
# Send a message.
send_payload = {"content": "What further research should I pursue based on my current websites?"}
send_resp = client.post(f"/api/dialog/{dialog_id}/send",
data=json.dumps(send_payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
# This test makes a live call to the Gemini API.
assert send_resp.status_code == 200, f"Send message failed: {send_resp.data.decode()}"
send_data = send_resp.get_json()
assert "llmResponse" in send_data, "Response missing LLM response."
print("Gemini LLM response:", send_data["llmResponse"])
# Verify that the dialog now has additional messages.
get_resp = client.get(f"/api/dialog/{dialog_id}", headers={"Authorization": auth_headers["Authorization"]})
assert get_resp.status_code == 200, f"Retrieving dialog failed: {get_resp.data.decode()}"
dialog_data = get_resp.get_json()
messages = dialog_data.get("messages", [])
assert len(messages) >= 2, "Expected at least two messages after sending (user and system)."
def test_end_and_delete_session(client, auth_headers, gemini_api_config, project_id):
"""
Tests ending a dialog session and then deleting it.
"""
create_payload = {"projectId": project_id}
create_resp = client.post("/api/dialog",
data=json.dumps(create_payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
assert create_resp.status_code == 201, f"Dialog creation failed: {create_resp.data.decode()}"
dialog_id = create_resp.get_json()["dialog_id"]
# End the dialog session.
end_resp = client.put(f"/api/dialog/{dialog_id}/end", headers={"Authorization": auth_headers["Authorization"]})
assert end_resp.status_code == 200, f"Ending dialog failed: {end_resp.data.decode()}"
# Attempt to send another message; should fail.
send_payload = {"content": "Trying to send after end."}
send_resp = client.post(f"/api/dialog/{dialog_id}/send",
data=json.dumps(send_payload),
headers={"Authorization": auth_headers["Authorization"]},
content_type="application/json")
assert send_resp.status_code == 400, "Expected error when sending message after ending session."
# Delete the dialog session.
del_resp = client.delete(f"/api/dialog/{dialog_id}", headers={"Authorization": auth_headers["Authorization"]})
assert del_resp.status_code == 200, f"Deleting dialog failed: {del_resp.data.decode()}"
# Verify that retrieving the dialog now returns 404.
get_resp = client.get(f"/api/dialog/{dialog_id}", headers={"Authorization": auth_headers["Authorization"]})
assert get_resp.status_code == 404, "Expected 404 when retrieving a deleted dialog."

208
test/tests_backend/test_projects.py Normal file
View File

@@ -0,0 +1,208 @@
import json
import uuid
import pytest
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 and returns auth headers.
"""
unique_suffix = str(uuid.uuid4())[:8]
username = f"projtest_{unique_suffix}"
email = f"{username}@example.com"
payload = {
"username": username,
"email": email,
"password": "Password123"
}
response = client.post("/api/register", data=json.dumps(payload), content_type="application/json")
assert response.status_code == 201, f"Registration failed: {response.data.decode()}"
data = json.loads(response.data)
token = data["token"]
return {"Authorization": f"Bearer {token}"}
def test_create_project(client, auth_headers):
payload = {
"name": "Test Project",
"topic": "Testing",
"description": "A project for testing purposes."
}
response = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert response.status_code == 201, f"Create project failed: {response.data.decode()}"
data = json.loads(response.data)
assert "project_id" in data
assert "passkey" in data
def test_get_projects(client, auth_headers):
# Create two projects.
payload1 = {"name": "Test Project One", "description": "First test project."}
payload2 = {"name": "Test Project Two", "description": "Second test project."}
response1 = client.post("/api/projects", data=json.dumps(payload1),
content_type="application/json", headers=auth_headers)
response2 = client.post("/api/projects", data=json.dumps(payload2),
content_type="application/json", headers=auth_headers)
assert response1.status_code == 201, f"Project one creation failed: {response1.data.decode()}"
assert response2.status_code == 201, f"Project two creation failed: {response2.data.decode()}"
response = client.get("/api/projects", headers=auth_headers)
assert response.status_code == 200, f"Get projects failed: {response.data.decode()}"
data = json.loads(response.data)
assert "projects" in data
assert isinstance(data["projects"], list)
assert len(data["projects"]) >= 2
def test_get_project_detail(client, auth_headers):
payload = {"name": "Detail Project1111", "description": "A project for detail testing."}
create_resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Creation failed: {create_resp.data.decode()}"
project_id = json.loads(create_resp.data)["project_id"]
response = client.get(f"/api/projects/{project_id}", headers=auth_headers)
assert response.status_code == 200, f"Get detail failed: {response.data.decode()}"
data = json.loads(response.data)
assert data.get("name") == "Detail Project1111"
assert "updatedAt" in data
def test_update_project(client, auth_headers):
payload = {"name": "Update Project", "description": "Initial description."}
create_resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Creation failed: {create_resp.data.decode()}"
project_id = json.loads(create_resp.data)["project_id"]
update_payload = {"description": "Updated description.", "topic": "Updated Topic"}
update_resp = client.put(f"/api/projects/{project_id}", data=json.dumps(update_payload),
content_type="application/json", headers=auth_headers)
assert update_resp.status_code == 200, f"Update failed: {update_resp.data.decode()}"
detail_resp = client.get(f"/api/projects/{project_id}", headers=auth_headers)
detail_data = json.loads(detail_resp.data)
assert detail_data.get("description") == "Updated description."
assert detail_data.get("topic") == "Updated Topic"
def test_delete_project(client, auth_headers):
payload = {"name": "Delete Project", "description": "Project to be deleted."}
create_resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Creation failed: {create_resp.data.decode()}"
project_id = json.loads(create_resp.data)["project_id"]
delete_resp = client.delete(f"/api/projects/{project_id}", headers=auth_headers)
assert delete_resp.status_code == 200, f"Deletion failed: {delete_resp.data.decode()}"
detail_resp = client.get(f"/api/projects/{project_id}", headers=auth_headers)
assert detail_resp.status_code == 404, "Deleted project still accessible"
def test_get_project_summaries(client, auth_headers):
# Create a new project to ensure at least one exists.
payload = {"name": "Summary Project", "description": "Project for summary test."}
create_resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Project creation failed: {create_resp.data.decode()}"
response = client.get("/api/projects/summary", headers=auth_headers)
assert response.status_code == 200, f"Summary fetch failed: {response.data.decode()}"
data = json.loads(response.data)
assert "projects" in data
# Each project summary should include project_id, name, and updatedAt.
for proj in data["projects"]:
assert "project_id" in proj
assert "name" in proj
assert "updatedAt" in proj
# ------------------------------------------------------------------------------
# New Tests
# ------------------------------------------------------------------------------
def test_get_project_info(client, auth_headers):
"""
Tests the /api/projects/<project_id>/info endpoint,
which returns name, topic, description, keywords, summary.
"""
payload = {
"name": "Info Project",
"topic": "InfoTopic",
"description": "Project with an info endpoint."
}
create_resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Project creation failed: {create_resp.data.decode()}"
project_id = json.loads(create_resp.data)["project_id"]
info_resp = client.get(f"/api/projects/{project_id}/info", headers=auth_headers)
assert info_resp.status_code == 200, f"Get project info failed: {info_resp.data.decode()}"
info_data = json.loads(info_resp.data)
assert info_data.get("name") == "Info Project"
assert info_data.get("topic") == "InfoTopic"
assert info_data.get("description") == "Project with an info endpoint."
assert isinstance(info_data.get("keywords", []), list)
assert "summary" in info_data
def test_recalc_project_keywords(client, auth_headers):
"""
Tests the /api/projects/<project_id>/recalc_keywords endpoint.
It should gather all URL keywords, sum them, sort by top 20, and update the project.
"""
# 1) Create a project
payload = {"name": "Keyword Recalc Project", "description": "Test for recalc keywords."}
create_resp = client.post("/api/projects", data=json.dumps(payload),
content_type="application/json", headers=auth_headers)
assert create_resp.status_code == 201, f"Project creation failed: {create_resp.data.decode()}"
project_id = json.loads(create_resp.data)["project_id"]
# 2) Create multiple URLs with keywords
url1_keywords = [{"word": "alpha", "percentage": 50}, {"word": "beta", "percentage": 10}]
url2_keywords = [{"word": "alpha", "percentage": 20}, {"word": "gamma", "percentage": 15}]
url_create_payload1 = {"url": "https://url1.com", "keywords": url1_keywords}
url_create_payload2 = {"url": "https://url2.com", "keywords": url2_keywords}
resp1 = client.post(f"/api/projects/{project_id}/urls", data=json.dumps(url_create_payload1),
content_type="application/json", headers=auth_headers)
resp2 = client.post(f"/api/projects/{project_id}/urls", data=json.dumps(url_create_payload2),
content_type="application/json", headers=auth_headers)
assert resp1.status_code == 201, f"URL1 creation failed: {resp1.data.decode()}"
assert resp2.status_code == 201, f"URL2 creation failed: {resp2.data.decode()}"
# 3) Recalc keywords
recalc_resp = client.put(f"/api/projects/{project_id}/recalc_keywords", headers=auth_headers)
assert recalc_resp.status_code == 200, f"Keyword recalc failed: {recalc_resp.data.decode()}"
recalc_data = json.loads(recalc_resp.data)
assert "keywords" in recalc_data
top_keywords = recalc_data["keywords"]
# Check that alpha is presumably 70, gamma=15, beta=10 => alpha, gamma, beta
assert len(top_keywords) <= 20
# alpha => 70 (50 + 20)
alpha_kw = next((k for k in top_keywords if k["word"] == "alpha"), None)
assert alpha_kw is not None, "Expected alpha in combined keywords"
assert alpha_kw["percentage"] == 70.0
# 4) Fetch the project detail or info to confirm stored result
info_resp = client.get(f"/api/projects/{project_id}/info", headers=auth_headers)
assert info_resp.status_code == 200, f"Get project info after recalc failed: {info_resp.data.decode()}"
info_data = json.loads(info_resp.data)
# The project now has 'keywords' -> top 20
project_keywords = info_data.get("keywords", [])
alpha_in_project = next((k for k in project_keywords if k["word"] == "alpha"), None)
assert alpha_in_project is not None, "Project keywords missing alpha"
assert alpha_in_project["percentage"] == 70.0

219
test/tests_backend/test_urls.py Normal file
View File

@@ -0,0 +1,219 @@
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."