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

1import os 

2import io 

3from werkzeug.datastructures import FileStorage 

4from app.models import Document, User 

5from app import db 

6from sqlalchemy import select 

7 

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 

19 

20 return client 

21 

22 

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 

29 

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') 

34 

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 ) 

42 

43 # Prepare the data dictionary with the file for upload 

44 data = {"pdf_file": (pdf_file, "test.pdf")} 

45 

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 

50 

51 

52 return db.session.query(Document).filter_by(document_name="test").first() 

53 

54 

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) 

60 

61 item_id = uploaded_doc.id 

62 

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" 

66 

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" 

71 

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" 

76 

77 

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}") 

85 

86 assert response.status_code == 404, f"Expected 404, but got {response.status_code}" 

87 assert not response.json["success"], "Response should indicate failure" 

88 

89 

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 """ 

95 

96 def faulty_commit(): 

97 raise Exception("Simulated DB failure") 

98 

99 uploaded_doc = upload_test_document(client, app) 

100 assert uploaded_doc is not None, "Failed to retrieve uploaded document" 

101 

102 item_id = uploaded_doc.id 

103 

104 with app.app_context(): 

105 monkeypatch.setattr(db.session, "commit", faulty_commit) # Force DB commit to fail 

106 

107 response = client.delete(f"/delete/{item_id}") 

108 

109 assert response.status_code == 500, f"Expected 500, but got {response.status_code}" 

110 assert not response.json["success"], "Response should indicate failure" 

111 

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"