Spaces:
Sleeping
Sleeping
File size: 4,344 Bytes
e9406c7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import unittest
from models.application_session import ApplicationSession
from models.base import db
from services.session_manager import SessionManager
from app import create_app
from psycopg2.errors import UniqueViolation
from datetime import datetime, timedelta
class TestApplicationSessionModel(unittest.TestCase):
def setUp(self):
self.db = db
self.app = create_app()
self.app_context = self.app.app_context()
self.app_context.push()
self.session_manager = SessionManager.instance()
self.main_nifti_path='test-sessions/test-045'
self.combined_labels_id=self.session_manager.generate_uuid()
db.create_all()
def tearDown(self):
self.db.session.remove()
self.app_context.pop()
def test_insert_into_table(self):
session_key = self.session_manager.generate_uuid()
now = datetime.now()
expire = now + timedelta(days=3)
new_session = ApplicationSession(
session_id=session_key,
main_nifti_path=self.main_nifti_path,
combined_labels_id=self.combined_labels_id,
session_created=now,
session_expire_date = expire,
)
db.session.add(new_session)
db.session.commit()
stmt = db.select(ApplicationSession).where(ApplicationSession.session_id == session_key)
res = db.session.execute(stmt)
app_session = res.scalar()
self.assertEqual(app_session.session_id, session_key)
self.assertEqual(app_session.main_nifti_path, self.main_nifti_path)
self.assertEqual(app_session.combined_labels_id, self.combined_labels_id)
self.assertEqual(app_session.session_created, now)
self.assertEqual(app_session.session_expire_date, expire)
#cleanup from DB
db.session.delete(app_session)
db.session.commit()
def test_duplicate_session_id(self):
now = datetime.now()
session_key = self.session_manager.generate_uuid()
new_session = ApplicationSession(
session_id=session_key,
main_nifti_path=self.main_nifti_path,
combined_labels_id=self.combined_labels_id,
session_created=now,
session_expire_date = now + timedelta(days=3)
)
db.session.add(new_session)
db.session.commit()
now = datetime.now()
dup_key_session = ApplicationSession(
session_id=session_key, #same session_id as new_session
main_nifti_path="test/path",
combined_labels_id="unique_combined_labels_id",
session_created=now,
session_expire_date = now + timedelta(days=3)
)
with self.assertRaises(UniqueViolation):
try:
db.session.add(dup_key_session)
db.session.commit()
except:
raise UniqueViolation
db.session.rollback()
db.session.delete(new_session)
db.session.commit()
def test_duplicate_combined_labels_id(self):
now = datetime.now()
session_key = self.session_manager.generate_uuid()
new_session = ApplicationSession(
session_id=session_key,
main_nifti_path=self.main_nifti_path,
combined_labels_id=self.combined_labels_id,
session_created=now,
session_expire_date = now + timedelta(days=3),
)
db.session.add(new_session)
db.session.commit()
session_key2 = self.session_manager.generate_uuid()
now = datetime.now()
session_2 = ApplicationSession(
session_id=session_key,
main_nifti_path=self.main_nifti_path,
combined_labels_id=self.combined_labels_id, #same as new_session
session_created=now,
session_expire_date = now + timedelta(days=3),
)
with self.assertRaises(UniqueViolation):
try:
db.session.add(session_2)
db.session.commit()
except:
raise UniqueViolation
db.session.rollback()
db.session.delete(new_session)
db.session.commit()
if __name__ == "__main__":
unittest.main()
|