Coverage for tests/test_delete.py: 100%
56 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-20 21:23 +0000
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-20 21:23 +0000
1import os
2import io
3from werkzeug.datastructures import FileStorage
4from app.models import Document, User
5from app import db
6from sqlalchemy import select
8def login_user(client):
9 test_user = User(username="testuser")
10 test_user.set_password("password")
11 db.session.add(test_user)
12 db.session.commit()
13 response = client.post(
14 "/index",
15 data={"username": "testuser", "password": "password"},
16 follow_redirects=True,
17 )
18 assert response.status_code == 200
20 return client
23def upload_test_document(client, app):
24 """
25 Helper function to upload a test document.
26 Returns the uploaded document object.
27 """
28 app.config['WTF_CSRF_ENABLED'] = False
30 with app.app_context():
31 client = login_user(client)
32 # Path to the test file inside the test_data folder
33 file_path = os.path.join(os.path.dirname(__file__), 'test_data', 'test.pdf')
35 # Open the file in binary read mode and create a FileStorage object
36 with open(file_path, 'rb') as f:
37 pdf_file = FileStorage(
38 stream=io.BytesIO(f.read()),
39 filename="test.pdf",
40 content_type="application/pdf",
41 )
43 # Prepare the data dictionary with the file for upload
44 data = {"pdf_file": (pdf_file, "test.pdf")}
46 # Make the POST request to the admin route
47 response = client.post("/admin", data=data, content_type='multipart/form-data')
48 assert response.status_code == 200, f"Upload failed: {response.json}"
49 assert "document" in response.json
52 return db.session.query(Document).filter_by(document_name="test").first()
55def test_delete_valid_document(client, app):
56 """
57 ✅ Tests successful deletion of an existing document and its embeddings.
58 """
59 uploaded_doc = upload_test_document(client, app)
61 item_id = uploaded_doc.id
63 response = client.delete(f"/delete/{item_id}")
64 assert response.status_code == 200, f"Delete failed: {response.json}"
65 assert response.json["success"], "Delete request did not return success"
67 with app.app_context():
68 vector_db = app.vector_db
69 deleted_doc = db.session.query(Document).filter_by(id=item_id).first()
70 assert deleted_doc is None, "Document was not deleted"
72 embedding_exists_after = db.session.execute(
73 select(vector_db.EmbeddingStore).where(vector_db.EmbeddingStore.id.like(f"%{uploaded_doc.document_name}.{uploaded_doc.document_type}%"))
74 ).fetchone()
75 assert embedding_exists_after is None, "Embeddings were not deleted"
78def test_delete_non_existent_document(client):
79 """
80 ❌ Tests deletion of a non-existent document.
81 Expected: 404 Not Found
82 """
83 non_existent_id = 9999
84 response = client.delete(f"/delete/{non_existent_id}")
86 assert response.status_code == 404, f"Expected 404, but got {response.status_code}"
87 assert not response.json["success"], "Response should indicate failure"
90def test_delete_database_error(client, app, monkeypatch):
91 """
92 ❌ Simulates a database error to ensure proper rollback and 500 response.
93 Expected: 500 Internal Server Error
94 """
96 def faulty_commit():
97 raise Exception("Simulated DB failure")
99 uploaded_doc = upload_test_document(client, app)
100 assert uploaded_doc is not None, "Failed to retrieve uploaded document"
102 item_id = uploaded_doc.id
104 with app.app_context():
105 monkeypatch.setattr(db.session, "commit", faulty_commit) # Force DB commit to fail
107 response = client.delete(f"/delete/{item_id}")
109 assert response.status_code == 500, f"Expected 500, but got {response.status_code}"
110 assert not response.json["success"], "Response should indicate failure"
112 # Ensure document was not deleted due to rollback
113 doc_exists = db.session.query(Document).filter_by(id=item_id).first()
114 assert doc_exists is not None, "Document should not be deleted after rollback"