Spaces:
Sleeping
Sleeping
| """ | |
| tests/test_database.py | |
| ====================== | |
| Unit tests for database/db.py — SQLite persistence layer. | |
| """ | |
| import sqlite3 | |
| import pytest | |
| from database.db import DatabaseManager | |
| def db(): | |
| """Provide a fresh in-memory database for each test.""" | |
| manager = DatabaseManager(":memory:") | |
| yield manager | |
| manager.close() | |
| # --------------------------------------------------------------------------- | |
| # User operations | |
| # --------------------------------------------------------------------------- | |
| class TestUserOperations: | |
| def test_create_and_get_user(self, db): | |
| uid = db.create_user("alice", "$2b$12$fakehash") | |
| assert uid is not None | |
| user = db.get_user("alice") | |
| assert user is not None | |
| assert user["username"] == "alice" | |
| assert user["password_hash"] == "$2b$12$fakehash" | |
| assert user["encrypted_history"] is None | |
| assert user["created_at"] > 0 | |
| def test_user_exists(self, db): | |
| assert db.user_exists("alice") is False | |
| db.create_user("alice", "$2b$12$hash") | |
| assert db.user_exists("alice") is True | |
| def test_duplicate_username_raises(self, db): | |
| db.create_user("alice", "$2b$12$hash1") | |
| with pytest.raises(sqlite3.IntegrityError): | |
| db.create_user("alice", "$2b$12$hash2") | |
| def test_get_nonexistent_user_returns_none(self, db): | |
| assert db.get_user("nobody") is None | |
| def test_update_encrypted_history(self, db): | |
| db.create_user("alice", "$2b$12$hash") | |
| assert db.get_user("alice")["encrypted_history"] is None | |
| db.update_encrypted_history("alice", "encrypted-blob-123") | |
| user = db.get_user("alice") | |
| assert user["encrypted_history"] == "encrypted-blob-123" | |
| def test_multiple_users_independent(self, db): | |
| db.create_user("alice", "$2b$hash1") | |
| db.create_user("bob", "$2b$hash2") | |
| alice = db.get_user("alice") | |
| bob = db.get_user("bob") | |
| assert alice["id"] != bob["id"] | |
| assert alice["username"] == "alice" | |
| assert bob["username"] == "bob" | |
| # --------------------------------------------------------------------------- | |
| # Session operations | |
| # --------------------------------------------------------------------------- | |
| class TestSessionOperations: | |
| def _create_user(self, db, username="testuser"): | |
| db.create_user(username, "$2b$12$fakehash") | |
| def test_save_and_get_session(self, db): | |
| self._create_user(db) | |
| sid = db.save_session( | |
| username="testuser", | |
| stress_score=0.75, | |
| stress_label="stress", | |
| temporal_data={"velocity": 0.1, "threshold": 0.6}, | |
| interventions=[{"title": "Deep breathing", "category": "breathing"}], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=["work", "stress"], | |
| attention_weights=[0.1, 0.5, 0.3], | |
| ) | |
| assert sid is not None | |
| sessions = db.get_sessions("testuser") | |
| assert len(sessions) == 1 | |
| s = sessions[0] | |
| assert s["stress_score"] == 0.75 | |
| assert s["stress_label"] == "stress" | |
| assert s["temporal_data"] == {"velocity": 0.1, "threshold": 0.6} | |
| assert s["interventions"] == [ | |
| {"title": "Deep breathing", "category": "breathing"} | |
| ] | |
| assert s["is_crisis"] is False | |
| assert s["crisis_message"] is None | |
| assert s["matched_triggers"] == ["work", "stress"] | |
| assert s["attention_weights"] == [0.1, 0.5, 0.3] | |
| assert s["created_at"] > 0 | |
| def test_sessions_ordered_newest_first(self, db): | |
| self._create_user(db) | |
| for i in range(3): | |
| db.save_session( | |
| username="testuser", | |
| stress_score=i * 0.1, | |
| stress_label="no_stress", | |
| temporal_data={}, | |
| interventions=[], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=[], | |
| attention_weights=[], | |
| ) | |
| sessions = db.get_sessions("testuser") | |
| assert len(sessions) == 3 | |
| # newest first (higher id = newer insertion) | |
| assert sessions[0]["id"] > sessions[1]["id"] | |
| assert sessions[1]["id"] > sessions[2]["id"] | |
| def test_get_sessions_pagination(self, db): | |
| self._create_user(db) | |
| for i in range(5): | |
| db.save_session( | |
| username="testuser", | |
| stress_score=i * 0.1, | |
| stress_label="no_stress", | |
| temporal_data={}, | |
| interventions=[], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=[], | |
| attention_weights=[], | |
| ) | |
| page1 = db.get_sessions("testuser", limit=2, offset=0) | |
| assert len(page1) == 2 | |
| page2 = db.get_sessions("testuser", limit=2, offset=2) | |
| assert len(page2) == 2 | |
| page3 = db.get_sessions("testuser", limit=2, offset=4) | |
| assert len(page3) == 1 | |
| def test_get_session_count(self, db): | |
| self._create_user(db) | |
| assert db.get_session_count("testuser") == 0 | |
| db.save_session( | |
| username="testuser", | |
| stress_score=0.5, | |
| stress_label="stress", | |
| temporal_data={}, | |
| interventions=[], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=[], | |
| attention_weights=[], | |
| ) | |
| assert db.get_session_count("testuser") == 1 | |
| def test_sessions_isolated_per_user(self, db): | |
| self._create_user(db, "alice") | |
| self._create_user(db, "bob") | |
| db.save_session( | |
| username="alice", | |
| stress_score=0.9, | |
| stress_label="stress", | |
| temporal_data={}, | |
| interventions=[], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=[], | |
| attention_weights=[], | |
| ) | |
| assert db.get_session_count("alice") == 1 | |
| assert db.get_session_count("bob") == 0 | |
| assert db.get_sessions("bob") == [] | |
| def test_save_session_nonexistent_user_raises(self, db): | |
| with pytest.raises(ValueError, match="not found"): | |
| db.save_session( | |
| username="nobody", | |
| stress_score=0.5, | |
| stress_label="stress", | |
| temporal_data={}, | |
| interventions=[], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=[], | |
| attention_weights=[], | |
| ) | |
| def test_get_sessions_nonexistent_user_returns_empty(self, db): | |
| assert db.get_sessions("nobody") == [] | |
| def test_get_session_count_nonexistent_user_returns_zero(self, db): | |
| assert db.get_session_count("nobody") == 0 | |
| def test_session_crisis_data(self, db): | |
| self._create_user(db) | |
| db.save_session( | |
| username="testuser", | |
| stress_score=0.99, | |
| stress_label="stress", | |
| temporal_data={}, | |
| interventions=[], | |
| is_crisis=True, | |
| crisis_message="Please call 988", | |
| matched_triggers=["suicide"], | |
| attention_weights=[0.9], | |
| ) | |
| sessions = db.get_sessions("testuser") | |
| assert len(sessions) == 1 | |
| assert sessions[0]["is_crisis"] is True | |
| assert sessions[0]["crisis_message"] == "Please call 988" | |
| # --------------------------------------------------------------------------- | |
| # Database lifecycle | |
| # --------------------------------------------------------------------------- | |
| class TestDatabaseLifecycle: | |
| def test_file_based_database(self, tmp_path): | |
| db_path = str(tmp_path / "test.db") | |
| db1 = DatabaseManager(db_path) | |
| db1.create_user("alice", "$2b$hash") | |
| db1.save_session( | |
| username="alice", | |
| stress_score=0.5, | |
| stress_label="stress", | |
| temporal_data={"key": "value"}, | |
| interventions=[], | |
| is_crisis=False, | |
| crisis_message=None, | |
| matched_triggers=[], | |
| attention_weights=[], | |
| ) | |
| db1.close() | |
| # Re-open the same database — data should persist | |
| db2 = DatabaseManager(db_path) | |
| user = db2.get_user("alice") | |
| assert user is not None | |
| sessions = db2.get_sessions("alice") | |
| assert len(sessions) == 1 | |
| assert sessions[0]["stress_score"] == 0.5 | |
| db2.close() | |