-
-
Notifications
You must be signed in to change notification settings - Fork 136
Closed
Description
Context
From PR #834 code review - the test suite covers main flows well but lacks specific edge case tests.
Suggested Tests
1. Concurrent Creation Race Condition
Verify that simultaneous calls to get_or_create_personal_corpus() from multiple threads don't create duplicates or raise unhandled errors.
def test_concurrent_personal_corpus_creation(self):
from concurrent.futures import ThreadPoolExecutor
user = User.objects.create_user(...)
Corpus.objects.filter(creator=user, is_personal=True).delete()
def create_corpus():
return Corpus.get_or_create_personal_corpus(user)
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(lambda _: create_corpus(), range(5)))
# All should return the same corpus
self.assertEqual(len(set(c.pk for c in results)), 1)
# Only one should exist
self.assertEqual(Corpus.objects.filter(creator=user, is_personal=True).count(), 1)2. Delete and Recreate Flow
Verify that deleting personal corpus and then uploading creates a new one correctly.
def test_personal_corpus_recreate_after_deletion(self):
user = User.objects.create_user(...)
original_corpus = Corpus.objects.get(creator=user, is_personal=True)
original_corpus.delete()
# Upload should create new personal corpus
new_corpus = Corpus.get_or_create_personal_corpus(user)
self.assertNotEqual(new_corpus.pk, original_corpus.pk)
self.assertTrue(new_corpus.is_personal)3. Embedding Task Queue Failure
Test behavior when ensure_embeddings_for_corpus.delay() fails to queue (e.g., Redis unavailable).
Location
opencontractserver/tests/test_personal_corpus.py
Priority
Low - existing tests cover happy paths, these are hardening tests.
Reactions are currently unavailable
Metadata
Metadata
Assignees
Labels
No labels