""" Test Suite for Phase 1: Session Management - Active Company Tests the active company functionality added to ConversationSession: - Setting active company - Getting active company - Clearing active company - Serialization/deserialization (to_dict/from_dict) - Database persistence """ import pytest import json from datetime import datetime from app.agent.session import ConversationSession, SessionManager class TestActiveCompanyBasics: """Test basic active company operations.""" def test_initial_state_no_company(self): """Test that new session has no active company.""" session = ConversationSession(telegram_user_id=123456) assert session.active_company_id is None assert session.active_company_name is None assert session.active_company_cui is None assert session.get_active_company() is None def test_set_active_company_with_cui(self): """Test setting active company with all fields.""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="RO12345678" ) assert session.active_company_id == 42 assert session.active_company_name == "ACME SRL" assert session.active_company_cui == "RO12345678" def test_set_active_company_without_cui(self): """Test setting active company without CUI (optional parameter).""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=99, company_name="Test Company" ) assert session.active_company_id == 99 assert session.active_company_name == "Test Company" assert session.active_company_cui is None def test_get_active_company_returns_dict(self): """Test that get_active_company returns correct dict structure.""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="RO12345678" ) company = session.get_active_company() assert isinstance(company, dict) assert company["id"] == 42 assert company["name"] == "ACME SRL" assert company["cui"] == "RO12345678" def test_get_active_company_returns_none_when_not_set(self): """Test that get_active_company returns None when no company set.""" session = ConversationSession(telegram_user_id=123456) company = session.get_active_company() assert company is None def test_clear_active_company(self): """Test clearing active company.""" session = ConversationSession(telegram_user_id=123456) # Set a company first session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="RO12345678" ) # Verify it's set assert session.get_active_company() is not None # Clear it session.clear_active_company() # Verify it's cleared assert session.active_company_id is None assert session.active_company_name is None assert session.active_company_cui is None assert session.get_active_company() is None def test_clear_active_company_when_not_set(self): """Test that clearing when no company set is safe (idempotent).""" session = ConversationSession(telegram_user_id=123456) # Should not raise error session.clear_active_company() assert session.get_active_company() is None def test_overwrite_active_company(self): """Test that setting company multiple times overwrites previous.""" session = ConversationSession(telegram_user_id=123456) # Set first company session.set_active_company( company_id=1, company_name="Company One" ) # Set second company (should overwrite) session.set_active_company( company_id=2, company_name="Company Two", company_cui="RO99999" ) company = session.get_active_company() assert company["id"] == 2 assert company["name"] == "Company Two" assert company["cui"] == "RO99999" class TestActiveCompanySerialization: """Test serialization/deserialization with active company.""" def test_to_dict_includes_company_fields(self): """Test that to_dict includes active company fields.""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="RO12345678" ) data = session.to_dict() assert "active_company_id" in data assert "active_company_name" in data assert "active_company_cui" in data assert data["active_company_id"] == 42 assert data["active_company_name"] == "ACME SRL" assert data["active_company_cui"] == "RO12345678" def test_to_dict_with_no_company(self): """Test that to_dict includes None values when no company set.""" session = ConversationSession(telegram_user_id=123456) data = session.to_dict() assert "active_company_id" in data assert "active_company_name" in data assert "active_company_cui" in data assert data["active_company_id"] is None assert data["active_company_name"] is None assert data["active_company_cui"] is None def test_from_dict_restores_company_fields(self): """Test that from_dict properly restores active company.""" original_session = ConversationSession(telegram_user_id=123456) original_session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="RO12345678" ) # Serialize data = original_session.to_dict() # Deserialize restored_session = ConversationSession.from_dict(data) # Verify company was restored assert restored_session.active_company_id == 42 assert restored_session.active_company_name == "ACME SRL" assert restored_session.active_company_cui == "RO12345678" company = restored_session.get_active_company() assert company["id"] == 42 assert company["name"] == "ACME SRL" assert company["cui"] == "RO12345678" def test_from_dict_backward_compatible(self): """Test that from_dict works with old session data (no company fields).""" # Simulate old session data without company fields old_data = { "telegram_user_id": 123456, "session_id": "test-session-id", "messages": [], "last_context": {}, "max_messages": 20, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat() # Note: No active_company_* fields } # Should not raise error session = ConversationSession.from_dict(old_data) # Company fields should default to None assert session.active_company_id is None assert session.active_company_name is None assert session.active_company_cui is None assert session.get_active_company() is None def test_json_serialization_roundtrip(self): """Test full JSON serialization roundtrip.""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="RO12345678" ) # Serialize to JSON string (simulates database storage) data_dict = session.to_dict() json_string = json.dumps(data_dict) # Deserialize from JSON string restored_dict = json.loads(json_string) restored_session = ConversationSession.from_dict(restored_dict) # Verify everything was restored assert restored_session.telegram_user_id == 123456 assert restored_session.active_company_id == 42 assert restored_session.active_company_name == "ACME SRL" assert restored_session.active_company_cui == "RO12345678" class TestActiveCompanyWithSessionManager: """Test active company functionality through SessionManager.""" @pytest.mark.asyncio async def test_session_manager_preserves_company_across_save_load(self): """ Test that SessionManager properly saves and loads active company. NOTE: This is an integration test that requires database access. It may be skipped in CI if database is not available. """ try: session_manager = SessionManager() test_user_id = 999888777 # Unique test ID # Create session and set company session = await session_manager.get_or_create_session(test_user_id) session.set_active_company( company_id=42, company_name="Test Company", company_cui="RO12345" ) # Save to database await session_manager.save_session(test_user_id) # Clear in-memory cache (simulate bot restart) session_manager._sessions.clear() # Load from database restored_session = await session_manager.get_or_create_session(test_user_id) # Verify company was persisted company = restored_session.get_active_company() assert company is not None assert company["id"] == 42 assert company["name"] == "Test Company" assert company["cui"] == "RO12345" # Cleanup await session_manager.delete_session(test_user_id) except Exception as e: pytest.skip(f"Database not available or error: {e}") class TestActiveCompanyEdgeCases: """Test edge cases and error handling.""" def test_company_with_none_cui(self): """Test explicitly setting CUI to None.""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=42, company_name="ACME SRL", company_cui=None ) company = session.get_active_company() assert company["id"] == 42 assert company["name"] == "ACME SRL" assert company["cui"] is None def test_company_with_empty_string_cui(self): """Test setting CUI to empty string.""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=42, company_name="ACME SRL", company_cui="" ) company = session.get_active_company() assert company["cui"] == "" def test_updated_at_changes_on_company_operations(self): """Test that updated_at timestamp changes when setting/clearing company.""" session = ConversationSession(telegram_user_id=123456) initial_time = session.updated_at # Small delay to ensure timestamp difference import time time.sleep(0.01) session.set_active_company( company_id=42, company_name="ACME SRL" ) assert session.updated_at > initial_time time.sleep(0.01) clear_time = session.updated_at session.clear_active_company() assert session.updated_at > clear_time def test_company_id_zero_is_valid(self): """Test that company_id = 0 is treated as a valid ID (not None).""" session = ConversationSession(telegram_user_id=123456) session.set_active_company( company_id=0, company_name="Zero Company" ) # Should NOT be None - 0 is a valid ID company = session.get_active_company() assert company is not None assert company["id"] == 0 assert company["name"] == "Zero Company" if __name__ == "__main__": pytest.main([__file__, "-v"])