test: 补充知识处理、检索与 Agent 回归用例

覆盖切块、向量检索、工具调用和历史预算;模型相关用例使用桩。
This commit is contained in:
WangShen007 committed 2026-10-04 20:55:00 +08:00
1 parent 362d79a5c2
commit f4d8019829
123 files changed
+29005

No files matched your search

+214
View File
@@ -0,0 +1,214 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.models.agent import Agent, AgentType
from app.repositories.agent import AgentRepository
import json
@pytest.fixture
def agent_repo(db):
"""Create an agent repository instance"""
return AgentRepository(db)
def test_create_agent(agent_repo, test_organization_id):
"""Test creating a new agent"""
name = "Test Agent"
agent_type = AgentType.CUSTOMER_SUPPORT
instructions = ["Instruction 1", "Instruction 2"]
tools = [{"name": "tool1", "description": "Tool 1"}]
description = "Test agent description"
agent = agent_repo.create_agent(
name=name,
agent_type=agent_type,
instructions=instructions,
org_id=test_organization_id,
description=description,
tools=tools,
is_default=True,
is_active=True
)
assert agent.name == name
assert agent.agent_type == agent_type
assert agent.instructions == instructions
assert agent.organization_id == test_organization_id
assert agent.description == description
assert json.loads(agent.tools) == tools
assert agent.is_default is True
assert agent.is_active is True
def test_create_agent_with_string_instructions(agent_repo, test_organization_id):
"""Test creating an agent with string instructions"""
instruction = "Single instruction"
agent = agent_repo.create_agent(
name="Test Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=instruction,
org_id=test_organization_id
)
assert agent.instructions == [instruction]
def test_get_agent(agent_repo, test_organization_id):
"""Test retrieving an agent by ID"""
# Create test agent
agent = agent_repo.create_agent(
name="Test Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Test instruction"],
org_id=test_organization_id
)
# Retrieve agent
retrieved_agent = agent_repo.get_agent(agent.id)
assert retrieved_agent.id == agent.id
assert retrieved_agent.name == agent.name
def test_get_org_agents(agent_repo, test_organization_id):
"""Test retrieving all agents for an organization"""
# Create multiple test agents
agent1 = agent_repo.create_agent(
name="Agent 1",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_active=True
)
agent2 = agent_repo.create_agent(
name="Agent 2",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_active=False
)
# Test getting all agents
all_agents = agent_repo.get_org_agents(test_organization_id, active_only=False)
assert len(all_agents) == 2
# Test getting only active agents
active_agents = agent_repo.get_org_agents(test_organization_id, active_only=True)
assert len(active_agents) == 1
assert active_agents[0].id == agent1.id
def test_get_default_agent(agent_repo, test_organization_id):
"""Test retrieving default agent for an organization"""
# Create multiple agents, one default
default_agent = agent_repo.create_agent(
name="Default Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_default=True,
is_active=True
)
agent_repo.create_agent(
name="Non-default Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_default=False,
is_active=True
)
# Get default agent
retrieved_default = agent_repo.get_default_agents(test_organization_id)
assert retrieved_default.id == default_agent.id
assert retrieved_default.is_default is True
def test_update_agent(agent_repo, test_organization_id):
"""Test updating an agent"""
# Create test agent
agent = agent_repo.create_agent(
name="Test Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Original instruction"],
org_id=test_organization_id
)
# Update agent
new_name = "Updated Agent"
new_instructions = ["Updated instruction"]
updated_agent = agent_repo.update_agent(
agent.id,
name=new_name,
instructions=new_instructions
)
assert updated_agent.name == new_name
assert updated_agent.instructions == new_instructions
def test_delete_agent(agent_repo, test_organization_id):
"""Test soft deleting an agent"""
# Create test agent
agent = agent_repo.create_agent(
name="Test Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_active=True
)
# Delete agent
success = agent_repo.delete_agent(agent.id)
assert success is True
# Verify agent is inactive
deleted_agent = agent_repo.get_agent(agent.id)
assert deleted_agent.is_active is False
def test_get_active_agents(agent_repo, test_organization_id):
"""Test retrieving active agents"""
# Create agents with different active states
active_agent = agent_repo.create_agent(
name="Active Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_active=True
)
agent_repo.create_agent(
name="Inactive Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id,
is_active=False
)
# Get active agents
active_agents = agent_repo.get_active_agents(test_organization_id)
assert len(active_agents) == 1
assert active_agents[0].id == active_agent.id
def test_get_by_agent_id(agent_repo, test_organization_id):
"""Test retrieving agent by ID with relationships"""
# Create test agent
agent = agent_repo.create_agent(
name="Test Agent",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Instruction"],
org_id=test_organization_id
)
# Retrieve agent with relationships
retrieved_agent = agent_repo.get_by_agent_id(str(agent.id))
assert retrieved_agent.id == agent.id
assert retrieved_agent.name == agent.name
# Note: Additional relationship assertions would go here if the test database
# included related organization and group records
+207
View File
@@ -0,0 +1,207 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.models.ai_config import AIConfig, AIModelType
from app.repositories.ai_config import AIConfigRepository
from app.core.security import decrypt_api_key
from uuid import UUID
from app.models.organization import Organization
@pytest.fixture
def ai_config_repo(db):
"""Create an AI config repository instance"""
return AIConfigRepository(db)
def test_create_config(ai_config_repo, test_organization_id):
"""Test creating a new AI configuration"""
model_type = "openai"
model_name = "gpt-4"
api_key = "test-api-key-123"
config = ai_config_repo.create_config(
org_id=test_organization_id,
model_type=model_type,
model_name=model_name,
api_key=api_key
)
assert config is not None
assert config.organization_id == test_organization_id
assert config.model_type == AIModelType.OPENAI
assert config.model_name == model_name
assert config.is_active is True
# Verify API key is encrypted
assert config.encrypted_api_key != api_key
# Verify API key can be decrypted
decrypted_key = decrypt_api_key(config.encrypted_api_key)
assert decrypted_key == api_key
def test_create_config_deactivates_existing(ai_config_repo, test_organization_id):
"""Test that creating a new config deactivates existing ones"""
# Create first config
config1 = ai_config_repo.create_config(
org_id=test_organization_id,
model_type="openai",
model_name="gpt-4",
api_key="key1"
)
# Create second config
config2 = ai_config_repo.create_config(
org_id=test_organization_id,
model_type="anthropic",
model_name="claude-2",
api_key="key2"
)
# Verify first config is deactivated
assert config1.is_active is False
# Verify second config is active
assert config2.is_active is True
def test_get_active_config(ai_config_repo, test_organization_id):
"""Test retrieving active AI configuration"""
# Create a config
original_config = ai_config_repo.create_config(
org_id=test_organization_id,
model_type="openai",
model_name="gpt-4",
api_key="test-key"
)
# Get active config
active_config = ai_config_repo.get_active_config(test_organization_id)
assert active_config is not None
assert active_config.id == original_config.id
assert active_config.model_type == original_config.model_type
assert active_config.model_name == original_config.model_name
def test_get_active_config_none_exists(ai_config_repo, test_organization_id):
"""Test retrieving active config when none exists"""
config = ai_config_repo.get_active_config(test_organization_id)
assert config is None
def test_update_config(ai_config_repo, test_organization_id):
"""Test updating an AI configuration"""
# Create initial config
config = ai_config_repo.create_config(
org_id=test_organization_id,
model_type="openai",
model_name="gpt-4",
api_key="old-key"
)
# Update config
new_model_name = "gpt-4-turbo"
new_api_key = "new-key"
updated_config = ai_config_repo.update_config(
config_id=config.id,
model_name=new_model_name,
api_key=new_api_key
)
assert updated_config is not None
assert updated_config.id == config.id
assert updated_config.model_name == new_model_name
# Verify new API key is encrypted and can be decrypted
decrypted_key = decrypt_api_key(updated_config.encrypted_api_key)
assert decrypted_key == new_api_key
def test_update_nonexistent_config(ai_config_repo):
"""Test updating a non-existent configuration"""
result = ai_config_repo.update_config(
config_id=999,
model_name="new-model"
)
assert result is None
def test_deactivate_config(ai_config_repo, test_organization_id):
"""Test deactivating an AI configuration"""
# Create config
config = ai_config_repo.create_config(
org_id=test_organization_id,
model_type="openai",
model_name="gpt-4",
api_key="test-key"
)
# Deactivate config
success = ai_config_repo.deactivate_config(config.id)
assert success is True
# Verify config is deactivated
active_config = ai_config_repo.get_active_config(test_organization_id)
assert active_config is None
def test_deactivate_nonexistent_config(ai_config_repo):
"""Test deactivating a non-existent configuration"""
success = ai_config_repo.deactivate_config(999)
assert success is False
def test_multiple_organizations_configs(ai_config_repo, db):
"""Test handling configs for multiple organizations"""
# Create test organizations
org1 = Organization(
name="Test Organization 1",
domain="test1.com",
timezone="UTC"
)
db.add(org1)
org2 = Organization(
name="Test Organization 2",
domain="test2.com",
timezone="UTC"
)
db.add(org2)
db.commit()
db.refresh(org1)
db.refresh(org2)
# Create configs for both organizations
config1 = ai_config_repo.create_config(
org_id=org1.id,
model_type="openai",
model_name="gpt-4",
api_key="key1"
)
config2 = ai_config_repo.create_config(
org_id=org2.id,
model_type="anthropic",
model_name="claude-2",
api_key="key2"
)
# Verify both configs are active
active1 = ai_config_repo.get_active_config(org1.id)
active2 = ai_config_repo.get_active_config(org2.id)
assert active1.id == config1.id
assert active2.id == config2.id
assert active1.is_active is True
assert active2.is_active is True
def test_create_config_invalid_model_type(ai_config_repo, test_organization_id):
"""Test creating config with invalid model type"""
with pytest.raises(ValueError):
ai_config_repo.create_config(
org_id=test_organization_id,
model_type="invalid",
model_name="model",
api_key="key"
)
+334
View File
@@ -0,0 +1,334 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from datetime import datetime
from sqlalchemy import text
from app.repositories.chat import ChatRepository
from app.models.chat_history import ChatHistory
from app.models.customer import Customer
from app.models.agent import Agent, AgentType
from app.models.session_to_agent import SessionToAgent, SessionStatus
from app.models.user import User, UserGroup
from uuid import uuid4, UUID
from app.channels.interactive import from_options
from app.channels.sent_choices import SENT_CHOICES_ATTRIBUTE, encode_sent_choices
from app.models.schemas.chat import ChatDetailResponse
@pytest.fixture
def chat_repo(db):
return ChatRepository(db)
@pytest.fixture
def test_data(db, test_organization_id):
"""Create test data including customer, agent, user, and session"""
customer = Customer(
id=uuid4(),
email="customer@test.com",
full_name="Test Customer",
organization_id=test_organization_id
)
db.add(customer)
agent = Agent(
id=uuid4(),
name="test-agent",
display_name="Test Agent",
organization_id=test_organization_id,
description="Test agent description",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Be helpful"],
is_active=True
)
db.add(agent)
user = User(
id=uuid4(),
email="user@test.com",
full_name="Test User",
hashed_password="dummy_hash",
is_active=True,
organization_id=test_organization_id
)
db.add(user)
# Create a test group
group = UserGroup(
id=uuid4(),
name="Test Group",
description="Test group description",
organization_id=test_organization_id
)
db.add(group)
db.flush()
session_id = uuid4()
session = SessionToAgent(
session_id=session_id,
organization_id=test_organization_id,
agent_id=agent.id,
user_id=user.id,
customer_id=customer.id,
status=SessionStatus.OPEN,
group_id=group.id
)
db.add(session)
# Add some chat messages
messages = []
for i in range(3):
message = ChatHistory(
organization_id=test_organization_id,
session_id=session_id,
customer_id=customer.id,
agent_id=agent.id,
user_id=user.id,
message=f"Test message {i}",
message_type="text",
attributes={"key": f"value{i}"}
)
messages.append(message)
db.add(message)
db.commit()
return {
"org_id": test_organization_id,
"customer": customer,
"agent": agent,
"user": user,
"session_id": session_id,
"messages": messages,
"group": group
}
def test_create_message(chat_repo, test_data):
"""Test creating a new chat message"""
message_data = {
"organization_id": test_data["org_id"],
"session_id": test_data["session_id"],
"customer_id": test_data["customer"].id,
"agent_id": test_data["agent"].id,
"user_id": test_data["user"].id,
"message": "New test message",
"message_type": "text",
"attributes": {"test": "value"}
}
message = chat_repo.create_message(message_data)
assert message is not None
assert message.message == "New test message"
assert message.message_type == "text"
assert message.attributes == {"test": "value"}
@pytest.mark.asyncio
async def test_get_session_history(chat_repo, test_data):
"""Test retrieving chat history for a session"""
history = await chat_repo.get_session_history(test_data["session_id"])
assert len(history) == 3
for i, msg in enumerate(history):
assert msg.message == f"Test message {i}"
assert msg.user is not None
assert msg.agent is not None
def test_get_user_history(chat_repo, test_data):
"""Test retrieving chat history for a user"""
history = chat_repo.get_user_history(test_data["user"].id)
assert len(history) == 3
for msg in history:
assert msg.user_id == test_data["user"].id
def test_get_recent_chats(chat_repo, test_data):
"""Test retrieving recent chat overviews"""
chats = chat_repo.get_recent_chats(
organization_id=test_data["org_id"],
user_id=test_data["user"].id,
user_groups=[str(test_data["group"].id)]
)
assert len(chats) == 1
chat = chats[0]
assert chat["customer"]["email"] == "customer@test.com"
assert chat["agent"]["name"] == "test-agent"
assert chat["message_count"] == 3
assert chat["status"] == SessionStatus.OPEN
assert chat["group_id"] == str(test_data["group"].id)
def test_get_recent_chats_preview_is_the_latest_message(chat_repo, test_data, db):
"""The preview is the newest message, decrypted — not the lexicographic max,
which is what an aggregate over ciphertext would return."""
db.add(ChatHistory(
organization_id=test_data["org_id"],
session_id=test_data["session_id"],
customer_id=test_data["customer"].id,
agent_id=test_data["agent"].id,
user_id=test_data["user"].id,
# Sorts before every "Test message N" alphabetically, so a max() would miss it.
message="A final reply from the agent",
message_type="text",
))
db.commit()
chats = chat_repo.get_recent_chats(
organization_id=test_data["org_id"],
user_id=test_data["user"].id,
user_groups=[str(test_data["group"].id)]
)
assert chats[0]["last_message"] == "A final reply from the agent"
def test_get_recent_chats_preview_reads_legacy_plaintext(chat_repo, test_data, db):
"""Rows written before encryption at rest have no prefix and must still show."""
legacy = ChatHistory(
organization_id=test_data["org_id"],
session_id=test_data["session_id"],
customer_id=test_data["customer"].id,
agent_id=test_data["agent"].id,
user_id=test_data["user"].id,
message="placeholder",
message_type="text",
created_at=datetime(2030, 1, 1),
)
db.add(legacy)
db.commit()
# Overwrite in raw SQL so the value stays plaintext, as it would be in a row
# written before encryption at rest.
db.execute(
text("UPDATE chat_history SET message = :message WHERE id = :id"),
{"message": "legacy plaintext message", "id": legacy.id},
)
db.commit()
chats = chat_repo.get_recent_chats(
organization_id=test_data["org_id"],
user_id=test_data["user"].id,
user_groups=[str(test_data["group"].id)]
)
assert chats[0]["last_message"] == "legacy plaintext message"
@pytest.mark.asyncio
async def test_check_session_access(chat_repo, test_data):
"""Test checking session access"""
# Test access with user_id
has_access = await chat_repo.check_session_access(
test_data["session_id"],
test_data["user"].id,
[str(test_data["group"].id)]
)
assert has_access is True
# Test access with group
has_access = await chat_repo.check_session_access(
test_data["session_id"],
uuid4(), # Different user
[str(test_data["group"].id)]
)
assert has_access is True
# Test no access
has_access = await chat_repo.check_session_access(
test_data["session_id"],
uuid4(), # Different user
[str(uuid4())] # Different group
)
assert has_access is False
@pytest.mark.asyncio
async def test_get_chat_detail(chat_repo, test_data):
"""Test retrieving detailed chat information"""
detail = await chat_repo.get_chat_detail(
test_data["session_id"],
test_data["org_id"]
)
assert detail is not None
assert detail["customer"]["email"] == "customer@test.com"
assert detail["customer"]["meta_data"] is None
assert detail["agent"]["name"] == "test-agent"
assert detail["status"] == SessionStatus.OPEN
assert detail["group_id"] == str(test_data["group"].id)
assert detail["user_id"] == test_data["user"].id
assert detail["user_name"] == "Test User"
assert len(detail["messages"]) == 3
@pytest.mark.asyncio
async def test_get_nonexistent_chat_detail(chat_repo):
"""Test retrieving detail for nonexistent chat"""
detail = await chat_repo.get_chat_detail(uuid4(), uuid4())
assert detail is None
@pytest.mark.asyncio
async def test_get_chat_detail_includes_customer_meta_data(chat_repo, db, test_data):
"""meta_data set on the customer (e.g. via /generate-token custom_data) is
surfaced in the chat detail response so the inbox can display it."""
test_data["customer"].meta_data = {
"student_name": "Aarav Krishnan",
"center_name": "Special Academy U12"
}
db.commit()
detail = await chat_repo.get_chat_detail(
test_data["session_id"],
test_data["org_id"]
)
assert detail["customer"]["meta_data"] == {
"student_name": "Aarav Krishnan",
"center_name": "Special Academy U12"
}
@pytest.mark.asyncio
async def test_get_chat_detail_exposes_channel_delivery_fields(chat_repo, db, test_data):
"""The inbox matches receipts by id, shows ticks, and reads the sent
choices decrypted — and the response schema must not strip any of it."""
sent = chat_repo.create_message({
"message": "Confirm?", "message_type": "bot",
"session_id": test_data["session_id"], "organization_id": test_data["org_id"],
"external_message_id": "wamid.X",
"attributes": {SENT_CHOICES_ATTRIBUTE: encode_sent_choices(
from_options("Confirm?", ["Yes", "No"]), as_text=False)},
})
sent.channel_status = "read"
db.commit()
detail = await chat_repo.get_chat_detail(test_data["session_id"], test_data["org_id"])
message = ChatDetailResponse(**detail).model_dump()["messages"][-1]
assert message["id"] == sent.id
assert message["external_message_id"] == "wamid.X"
assert message["channel_status"] == "read"
assert [c["text"] for c in message["attributes"][SENT_CHOICES_ATTRIBUTE]["choices"]] == ["Yes", "No"]
# still encrypted at rest
db.refresh(sent)
assert "Yes" not in str(sent.attributes)
def test_app_sent_messages_cost_no_ai_quota(chat_repo, test_data):
common = {"session_id": test_data["session_id"], "organization_id": test_data["org_id"],
"message_type": "bot"}
before = chat_repo.get_message_count_for_period(test_data["org_id"], None, None)
chat_repo.create_message({**common, "message": "AI answer", "attributes": {"source": "ai"}})
chat_repo.create_message({**common, "message": "Old reply"})
chat_repo.create_message({**common, "message": "Fee reminder", "attributes": {"source": "app"}})
assert chat_repo.get_message_count_for_period(test_data["org_id"], None, None) == before + 2
+219
View File
@@ -0,0 +1,219 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from datetime import datetime, timedelta, timezone
import pytest
from app.models.crm import CrmConnectionStatus, CrmSyncJobStatus
from app.models.lead_capture import LeadCaptureResponse
from app.repositories.crm import (
CrmConnectionRepository, CrmSyncJobRepository, MAX_ATTEMPTS,
)
@pytest.fixture
def conn_repo(db):
return CrmConnectionRepository(db)
@pytest.fixture
def job_repo(db):
return CrmSyncJobRepository(db)
@pytest.fixture
def test_lead_response(db, test_organization, test_agent, test_customer) -> LeadCaptureResponse:
response = LeadCaptureResponse(
organization_id=test_organization.id,
agent_id=test_agent.id,
customer_id=test_customer.id,
field_values={"email": "lead@example.com", "name": "Lead Person"},
consent=True,
qualified=True,
)
db.add(response)
db.commit()
db.refresh(response)
return response
CREDS = {"access_token": "at-1", "refresh_token": "rt-1", "api_domain": None}
def _aware(dt: datetime) -> datetime:
"""SQLite loses tzinfo on refresh; normalize for comparisons (PG keeps it)."""
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
def _connect(conn_repo, org_id, provider="hubspot", external_id="hub-123", **kwargs):
return conn_repo.create_or_update(
organization_id=org_id,
provider=provider,
external_account_id=external_id,
credentials=CREDS,
display_name="Acme Portal",
**kwargs,
)
class TestCrmConnectionRepository:
def test_credentials_round_trip_encrypted(self, conn_repo, test_organization):
conn = _connect(conn_repo, test_organization.id)
# Stored blob is ciphertext, decrypts back to the original dict.
assert "at-1" not in conn.encrypted_credentials
assert conn_repo.get_credentials(conn) == CREDS
def test_create_or_update_is_an_upsert(self, conn_repo, test_organization):
first = _connect(conn_repo, test_organization.id)
conn_repo.set_status(first, CrmConnectionStatus.EXPIRED.value, last_error="dead token")
reconnected = _connect(conn_repo, test_organization.id, external_id="hub-456")
assert reconnected.id == first.id # same row, not a duplicate
assert reconnected.external_account_id == "hub-456"
assert reconnected.status == CrmConnectionStatus.ACTIVE.value
assert reconnected.last_error is None
def test_get_active_ignores_expired(self, conn_repo, test_organization):
conn = _connect(conn_repo, test_organization.id)
assert conn_repo.get_active(test_organization.id, "hubspot") is not None
conn_repo.set_status(conn, CrmConnectionStatus.EXPIRED.value)
assert conn_repo.get_active(test_organization.id, "hubspot") is None
def test_save_credentials_mirrors_expiries(self, conn_repo, test_organization):
conn = _connect(conn_repo, test_organization.id, provider="pipedrive",
external_id="pd-1")
access_exp = datetime.now(timezone.utc) + timedelta(hours=1)
refresh_exp = datetime.now(timezone.utc) + timedelta(days=60)
conn = conn_repo.save_credentials(
conn, {"access_token": "at-2", "refresh_token": "rt-2"},
access_token_expires_at=access_exp,
refresh_token_expires_at=refresh_exp,
)
assert conn_repo.get_credentials(conn)["access_token"] == "at-2"
assert conn.access_token_expires_at is not None
assert conn.refresh_token_expires_at is not None
assert conn.last_refreshed_at is not None
def test_list_needing_refresh(self, conn_repo, db, test_organization):
conn = _connect(conn_repo, test_organization.id, provider="pipedrive",
external_id="pd-2")
cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
# Freshly connected → not due.
assert conn_repo.list_needing_refresh("pipedrive", cutoff) == []
conn.last_refreshed_at = datetime.now(timezone.utc) - timedelta(days=2)
db.commit()
assert [c.id for c in conn_repo.list_needing_refresh("pipedrive", cutoff)] == [conn.id]
# Expired connections are never swept.
conn_repo.set_status(conn, CrmConnectionStatus.EXPIRED.value)
assert conn_repo.list_needing_refresh("pipedrive", cutoff) == []
class TestCrmSyncJobRepository:
def test_enqueue_is_idempotent(self, job_repo, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
assert job is not None
assert job.status == CrmSyncJobStatus.PENDING.value
# Same lead + provider → no second job.
assert job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot") is None
# Same lead, other provider → its own job.
other = job_repo.enqueue(test_organization.id, test_lead_response.id, "pipedrive")
assert other is not None and other.id != job.id
def test_claim_marks_processing_and_respects_due_time(
self, job_repo, db, test_organization, test_lead_response):
due = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
future = job_repo.enqueue(test_organization.id, test_lead_response.id, "pipedrive")
future.next_attempt_at = datetime.now(timezone.utc) + timedelta(hours=1)
db.commit()
claimed = job_repo.claim_batch(limit=10)
assert [j.id for j in claimed] == [due.id]
assert claimed[0].status == CrmSyncJobStatus.PROCESSING.value
assert claimed[0].started_at is not None
# Already-claimed jobs are not claimed again.
assert job_repo.claim_batch(limit=10) == []
def test_reclaim_stale_returns_crashed_claims(
self, job_repo, db, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
job_repo.claim_batch(limit=1)
db.refresh(job)
job.started_at = datetime.now(timezone.utc) - timedelta(minutes=30)
db.commit()
assert job_repo.reclaim_stale(older_than_minutes=10) == 1
db.refresh(job)
assert job.status == CrmSyncJobStatus.PENDING.value
assert job.started_at is None
def test_complete_stamps_result_and_connection(
self, job_repo, conn_repo, test_organization, test_lead_response):
conn = _connect(conn_repo, test_organization.id)
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
job = job_repo.complete(job, {"action": "created", "contact_id": "42"},
connection_id=conn.id)
assert job.status == CrmSyncJobStatus.COMPLETED.value
assert job.result["contact_id"] == "42"
assert job.connection_id == conn.id
def test_schedule_retry_backs_off_then_fails_terminally(
self, job_repo, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
before = datetime.now(timezone.utc)
job = job_repo.schedule_retry(job, "HTTP 502")
assert job.status == CrmSyncJobStatus.PENDING.value
assert job.attempts == 1
assert _aware(job.next_attempt_at) > before
job.attempts = MAX_ATTEMPTS - 1
job = job_repo.schedule_retry(job, "HTTP 502 again")
assert job.status == CrmSyncJobStatus.FAILED.value
assert job.attempts == MAX_ATTEMPTS
def test_retry_after_wins_over_backoff_when_longer(
self, job_repo, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
job = job_repo.schedule_retry(job, "HTTP 429", retry_after_seconds=7200)
# 7200 > capped backoff (3600) → provider's Retry-After is honored.
assert _aware(job.next_attempt_at) >= datetime.now(timezone.utc) + timedelta(seconds=7000)
def test_skip_pending_for_connection(
self, job_repo, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
count = job_repo.skip_pending_for_connection(
test_organization.id, "hubspot", "app uninstalled")
assert count == 1
assert job_repo.claim_batch(limit=10) == []
def test_count_recent_failures(self, job_repo, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
job_repo.fail(job, "boom")
assert job_repo.count_recent_failures(test_organization.id, "hubspot") == 1
assert job_repo.count_recent_failures(test_organization.id, "pipedrive") == 0
def test_reopen_failed_requeues(self, job_repo, test_organization, test_lead_response):
job = job_repo.enqueue(test_organization.id, test_lead_response.id, "hubspot")
job.attempts = 7
job_repo.fail(job, "no_connection")
assert job_repo.reopen_failed(test_organization.id, "hubspot") == 1
job_repo.db.refresh(job)
assert job.status == CrmSyncJobStatus.PENDING.value
assert job.attempts == 0
# Other providers untouched.
assert job_repo.reopen_failed(test_organization.id, "pipedrive") == 0
+481
View File
@@ -0,0 +1,481 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.models.customer import Customer
from app.repositories.customer import CustomerRepository
from uuid import UUID, uuid4
@pytest.fixture
def customer_repo(db):
"""Create a customer repository instance"""
return CustomerRepository(db)
def test_get_or_create_customer_new(customer_repo, test_organization_id):
"""Test creating a new customer when one doesn't exist"""
email = "newcustomer@example.com"
full_name = "New Customer"
customer = customer_repo.get_or_create_customer(
email=email,
organization_id=test_organization_id,
full_name=full_name
)
assert customer is not None
assert customer.email == email
assert customer.full_name == full_name
assert customer.organization_id == test_organization_id
def test_get_or_create_customer_existing(customer_repo, test_organization_id):
"""Test retrieving an existing customer"""
email = "existing@example.com"
full_name = "Existing Customer"
# Create initial customer
customer1 = customer_repo.get_or_create_customer(
email=email,
organization_id=test_organization_id,
full_name=full_name
)
# Try to create/get the same customer again
customer2 = customer_repo.get_or_create_customer(
email=email,
organization_id=test_organization_id,
full_name="Different Name" # This should be ignored as customer exists
)
assert customer2 is not None
assert customer2.id == customer1.id
assert customer2.email == email
assert customer2.full_name == full_name # Should keep original name
assert customer2.organization_id == test_organization_id
def test_get_or_create_customer_different_org(customer_repo, test_organization_id, db):
"""Test creating customers with same email but different organizations"""
email = "customer@example.com"
# Create second test organization
from app.models.organization import Organization
other_org = Organization(
id=uuid4(),
name="Other Test Org",
domain="other-test-org.com",
timezone="UTC"
)
db.add(other_org)
db.commit()
# Create customer in first organization
customer1 = customer_repo.get_or_create_customer(
email=email,
organization_id=test_organization_id,
full_name="Customer One"
)
# Create customer with same email in different organization
customer2 = customer_repo.get_or_create_customer(
email=email,
organization_id=other_org.id,
full_name="Customer Two"
)
assert customer1.id != customer2.id
assert customer1.email == customer2.email
assert customer1.organization_id != customer2.organization_id
def test_get_by_id(customer_repo, test_organization_id):
"""Test retrieving a customer by ID"""
# Create test customer
customer = customer_repo.get_or_create_customer(
email="test@example.com",
organization_id=test_organization_id,
full_name="Test Customer"
)
# Retrieve customer by ID
retrieved_customer = customer_repo.get_by_id(customer.id)
assert retrieved_customer is not None
assert retrieved_customer.id == customer.id
assert retrieved_customer.email == customer.email
def test_get_by_id_nonexistent(customer_repo):
"""Test retrieving a non-existent customer by ID"""
non_existent_id = uuid4()
customer = customer_repo.get_by_id(non_existent_id)
assert customer is None
def test_get_customer_email(customer_repo, test_organization_id):
"""Test retrieving a customer's email by ID"""
email = "test@example.com"
# Create test customer
customer = customer_repo.get_or_create_customer(
email=email,
organization_id=test_organization_id,
full_name="Test Customer"
)
# Get customer email
retrieved_email = customer_repo.get_customer_email(customer.id)
assert retrieved_email == email
def test_get_customer_email_nonexistent(customer_repo):
"""Test retrieving email for a non-existent customer"""
non_existent_id = uuid4()
email = customer_repo.get_customer_email(non_existent_id)
assert email is None
def test_get_or_create_customer_no_name(customer_repo, test_organization_id):
"""Test creating a customer without a full name"""
email = "noname@example.com"
customer = customer_repo.get_or_create_customer(
email=email,
organization_id=test_organization_id
)
assert customer is not None
assert customer.email == email
assert customer.full_name is None
assert customer.organization_id == test_organization_id
def test_create_customer_with_meta_data(customer_repo, test_organization_id):
"""Test creating a customer with integrator-supplied meta_data"""
email = "student@example.com"
meta_data = {"student_name": "Aarav Krishnan", "center_name": "Special Academy U12"}
customer = customer_repo.create_customer(
email=email,
organization_id=test_organization_id,
full_name="Parent Name",
meta_data=meta_data
)
assert customer is not None
assert customer.meta_data == meta_data
def test_create_customer_without_meta_data(customer_repo, test_organization_id):
"""Test creating a customer with no meta_data leaves the column null"""
customer = customer_repo.create_customer(
email="nometa@example.com",
organization_id=test_organization_id
)
assert customer.meta_data is None
def test_update_meta_data_merges_new_keys(customer_repo, test_organization_id):
"""A second call adds new keys and overwrites changed ones, keeping the rest"""
customer = customer_repo.create_customer(
email="merge@example.com",
organization_id=test_organization_id,
meta_data={"student_name": "Aarav", "grade": "U12"}
)
updated = customer_repo.update_meta_data(
customer.id,
{"student_name": "Aarav Krishnan", "center_name": "Special Academy U12"}
)
assert updated is not None
assert updated.meta_data == {
"student_name": "Aarav Krishnan", # overwritten
"grade": "U12", # preserved
"center_name": "Special Academy U12" # added
}
def test_update_meta_data_on_customer_with_no_existing_data(customer_repo, test_organization_id):
"""Merging onto a customer created without meta_data starts from an empty dict"""
customer = customer_repo.create_customer(
email="firstmeta@example.com",
organization_id=test_organization_id
)
updated = customer_repo.update_meta_data(customer.id, {"plan": "premium"})
assert updated.meta_data == {"plan": "premium"}
def test_update_meta_data_empty_input_is_noop(customer_repo, test_organization_id):
"""Calling with no data returns None and leaves the customer untouched"""
customer = customer_repo.create_customer(
email="noop@example.com",
organization_id=test_organization_id,
meta_data={"a": "b"}
)
result = customer_repo.update_meta_data(customer.id, {})
assert result is None
assert customer_repo.get_by_id(customer.id).meta_data == {"a": "b"}
def test_update_meta_data_nonexistent_customer(customer_repo):
"""Updating meta_data for a customer that doesn't exist returns None"""
result = customer_repo.update_meta_data(uuid4(), {"a": "b"})
assert result is None
class TestPhoneIdentity:
"""customers.phone is the second identity key: phone-first resolution,
organic backfill, and the never-silently-merge conflict rule."""
def test_phone_lookup_beats_email_creation(self, customer_repo, test_organization_id):
existing = customer_repo.create_customer(
email="priya@example.com", organization_id=test_organization_id,
full_name="Priya", phone="+911234567890")
# Same human arrives via WhatsApp: synthesized email, same phone.
resolved = customer_repo.get_or_create_customer(
email="911234567890@whatsapp.channel",
organization_id=test_organization_id,
full_name="Whatsapp user 91123456",
phone="+911234567890")
assert resolved.id == existing.id
assert resolved.full_name == "Priya" # nothing overwritten
assert resolved.email == "priya@example.com" # no junk row minted
def test_backfills_phone_when_found_by_email(self, customer_repo, test_organization_id):
existing = customer_repo.create_customer(
email="911234567890@whatsapp.channel", organization_id=test_organization_id)
assert existing.phone is None
resolved = customer_repo.get_or_create_customer(
email="911234567890@whatsapp.channel",
organization_id=test_organization_id,
phone="+911234567890")
assert resolved.id == existing.id
assert resolved.phone == "+911234567890"
def test_creates_with_phone_when_nobody_matches(self, customer_repo, test_organization_id):
customer = customer_repo.get_or_create_customer(
email="447700900123@whatsapp.channel",
organization_id=test_organization_id,
phone="+447700900123")
assert customer.phone == "+447700900123"
def test_conflict_uses_phone_match_and_keeps_both_rows(self, customer_repo, test_organization_id, db):
"""The Chatwoot trap: phone says A, email says B -> A wins the
conversation, nobody gets merged or overwritten."""
by_phone = customer_repo.create_customer(
email="a@example.com", organization_id=test_organization_id,
phone="+911234567890")
by_email = customer_repo.create_customer(
email="911234567890@whatsapp.channel", organization_id=test_organization_id)
resolved = customer_repo.get_or_create_customer(
email="911234567890@whatsapp.channel",
organization_id=test_organization_id,
phone="+911234567890")
assert resolved.id == by_phone.id
db.refresh(by_email)
assert by_email.merged_into_customer_id is None # intact
assert by_email.phone is None # not clobbered
def test_without_phone_behaviour_is_unchanged(self, customer_repo, test_organization_id):
"""Channels that declare no phone must take the exact old path."""
first = customer_repo.get_or_create_customer(
email="D7@telegram.channel", organization_id=test_organization_id,
full_name="Telegram user D7")
again = customer_repo.get_or_create_customer(
email="D7@telegram.channel", organization_id=test_organization_id)
assert again.id == first.id
assert again.phone is None
class TestUpdateContactPhone:
"""Set-if-absent with the same conflict-skip rule as email."""
def test_sets_phone_when_absent(self, customer_repo, test_organization_id):
customer = customer_repo.create_customer(
email="lead@example.com", organization_id=test_organization_id)
result = customer_repo.update_contact(customer.id, phone="+91 12345 67890")
assert result['phone_updated'] is True
assert customer_repo.get_by_id(customer.id).phone == "+911234567890"
def test_never_overwrites_an_existing_phone(self, customer_repo, test_organization_id):
customer = customer_repo.create_customer(
email="lead@example.com", organization_id=test_organization_id,
phone="+911234567890")
result = customer_repo.update_contact(customer.id, phone="+15550001111")
assert result['phone_updated'] is False
assert customer_repo.get_by_id(customer.id).phone == "+911234567890"
def test_skips_a_phone_owned_by_someone_else(self, customer_repo, test_organization_id):
customer_repo.create_customer(
email="owner@example.com", organization_id=test_organization_id,
phone="+911234567890")
other = customer_repo.create_customer(
email="other@example.com", organization_id=test_organization_id)
result = customer_repo.update_contact(other.id, phone="+911234567890")
assert result['phone_updated'] is False
assert customer_repo.get_by_id(other.id).phone is None
def test_skips_an_unresolvable_number_rather_than_guessing(self, customer_repo, test_organization_id):
customer = customer_repo.create_customer(
email="lead@example.com", organization_id=test_organization_id)
# Bare national digits: normalize_phone (strict) rejects them.
result = customer_repo.update_contact(customer.id, phone="1234567890")
assert result['phone_updated'] is False
assert customer_repo.get_by_id(customer.id).phone is None
def test_phone_update_composes_with_email_and_name(self, customer_repo, test_organization_id):
customer = customer_repo.create_customer(
email="1712345@noemail.com", organization_id=test_organization_id)
result = customer_repo.update_contact(
customer.id, email="real@example.com", full_name="Priya",
phone="+911234567890")
assert result == {'email_updated': True, 'name_updated': True,
'phone_updated': True, 'email': 'real@example.com'}
class TestPlaceholderEmails:
"""A `{id}@{channel}.channel` address is a synthesized identity key, not an
address — but it is still the ONLY key a phone-less channel has."""
def test_channel_addresses_are_placeholders(self, customer_repo):
assert CustomerRepository.is_placeholder_email("911234567890@whatsapp.channel") is True
assert CustomerRepository.is_placeholder_email("555@telegram.channel") is True
assert CustomerRepository.is_placeholder_email("1712345@noemail.com") is True
assert CustomerRepository.is_placeholder_email(None) is True
assert CustomerRepository.is_placeholder_email("priya@example.com") is False
def test_display_email_hides_stand_ins(self, customer_repo):
assert CustomerRepository.display_email("911234567890@whatsapp.channel") is None
assert CustomerRepository.display_email("1712345@noemail.com") is None
assert CustomerRepository.display_email("priya@example.com") == "priya@example.com"
def test_captured_email_lands_when_a_phone_holds_the_identity(self, customer_repo, test_organization_id):
"""WhatsApp/SMS: the phone is the key, so the email is free to become real."""
customer = customer_repo.create_customer(
email="911234567890@whatsapp.channel", organization_id=test_organization_id,
phone="+911234567890")
result = customer_repo.update_contact(customer.id, email="priya@example.com")
assert result['email_updated'] is True
assert customer_repo.get_by_id(customer.id).email == "priya@example.com"
def test_phoneless_channel_keeps_its_key_so_no_duplicate_is_minted(self, customer_repo, test_organization_id):
"""Telegram has no phone: the .channel address is how the next inbound
message finds this person. Overwriting it would create a second row."""
customer = customer_repo.create_customer(
email="555@telegram.channel", organization_id=test_organization_id)
result = customer_repo.update_contact(customer.id, email="ada@example.com")
assert result['email_updated'] is False
assert customer_repo.get_by_id(customer.id).email == "555@telegram.channel"
# And the lookup that every inbound message does still resolves.
assert customer_repo.get_customer_by_email(
"555@telegram.channel", test_organization_id).id == customer.id
def test_telegram_keeps_its_key_even_after_sharing_a_phone(self, customer_repo, test_organization_id):
"""The trap: Telegram CAN learn a phone (share-contact button), but its
inbound messages never carry one — so the address is still the lookup
key. Having a phone must not be mistaken for being found by it."""
customer = customer_repo.create_customer(
email="555@telegram.channel", organization_id=test_organization_id)
# They tapped "Share my phone number" during a handover.
customer_repo.update_contact(customer.id, phone="+447700900123")
db_customer = customer_repo.get_by_id(customer.id)
assert db_customer.phone == "+447700900123"
# Now the handoff form captures their real email.
result = customer_repo.update_contact(customer.id, email="ada@example.com")
assert result['email_updated'] is False
# The key survives, so their next Telegram message still finds them.
assert customer_repo.get_customer_by_email(
"555@telegram.channel", test_organization_id).id == customer.id
def test_widget_placeholder_still_replaceable(self, customer_repo, test_organization_id):
"""Unchanged behaviour: @noemail.com was never an identity key."""
customer = customer_repo.create_customer(
email="1712345@noemail.com", organization_id=test_organization_id)
result = customer_repo.update_contact(customer.id, email="real@example.com")
assert result['email_updated'] is True
class TestPhoneIdentityHazards:
"""The ways a phone key can attach a conversation to the wrong row."""
def test_merged_row_never_owns_a_live_number(self, customer_repo, test_organization_id):
"""A merged-away row is invisible in People and uneditable — so if it
kept a phone, the phone lookup (the FIRST thing every inbound WhatsApp
message does) would attach that conversation to a row nobody can see."""
survivor = customer_repo.create_customer(
email="real@example.com", organization_id=test_organization_id,
phone="+447700900111")
tombstone = customer_repo.create_customer(
email="old@example.com", organization_id=test_organization_id,
phone="+447700900222")
tombstone.merged_into_customer_id = survivor.id
customer_repo.db.commit()
# The number the tombstone still holds must not resolve to it.
assert customer_repo.get_customer_by_phone(
"+447700900222", test_organization_id) is None
def test_inbound_wins_a_race_instead_of_being_dropped(
self, customer_repo, test_organization_id, monkeypatch):
"""A concurrent insert on the phone index must not surface as an
exception: process_channel_message swallows it and the customer's
message is never answered. The winner's row is the right answer."""
from sqlalchemy.exc import IntegrityError
winner = customer_repo.create_customer(
email="919999900001@whatsapp.channel",
organization_id=test_organization_id, phone="+919999900001")
# Simulate losing the race: our lookup saw nothing, the insert collides.
calls = {"n": 0}
real_create = customer_repo.create_customer
def racing_create(*args, **kwargs):
calls["n"] += 1
raise IntegrityError("duplicate key", None, Exception("conflict"))
monkeypatch.setattr(customer_repo, "create_customer", racing_create)
monkeypatch.setattr(customer_repo, "_resolve_existing",
lambda e, o, p: None if calls["n"] == 0 else winner)
resolved = customer_repo.get_or_create_customer(
email="919999900001@whatsapp.channel",
organization_id=test_organization_id, phone="+919999900001")
assert resolved.id == winner.id
assert calls["n"] == 1
def test_race_with_no_winner_still_raises(
self, customer_repo, test_organization_id, monkeypatch):
"""The retry must not swallow a genuine IntegrityError — if re-resolving
finds nobody, the constraint fired for a reason we don't understand."""
from sqlalchemy.exc import IntegrityError
def always_conflicts(*args, **kwargs):
raise IntegrityError("duplicate key", None, Exception("conflict"))
monkeypatch.setattr(customer_repo, "create_customer", always_conflicts)
monkeypatch.setattr(customer_repo, "_resolve_existing", lambda e, o, p: None)
with pytest.raises(IntegrityError):
customer_repo.get_or_create_customer(
email="x@whatsapp.channel", organization_id=test_organization_id,
phone="+919999900009")
@@ -0,0 +1,563 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from datetime import datetime
from uuid import uuid4
from app.repositories.file_attachment import FileAttachmentRepository
from app.models.file_attachment import FileAttachment
from app.models.chat_history import ChatHistory
from app.models.customer import Customer
from app.models.user import User, UserGroup
from app.models.agent import Agent, AgentType
@pytest.fixture
def file_attachment_repo(db):
"""Create FileAttachmentRepository instance"""
return FileAttachmentRepository(db)
@pytest.fixture
def test_data(db, test_organization_id):
"""Create test data including customer, user, agent, and chat history"""
# Create customer
customer = Customer(
id=uuid4(),
email="customer@test.com",
full_name="Test Customer",
organization_id=test_organization_id
)
db.add(customer)
# Create user group
user_group = UserGroup(
id=uuid4(),
name="Test Group",
description="Test group description",
organization_id=test_organization_id
)
db.add(user_group)
# Create user
user = User(
id=uuid4(),
email="user@test.com",
full_name="Test User",
organization_id=test_organization_id,
hashed_password="dummy_hash",
is_active=True
)
db.add(user)
# Create agent
agent = Agent(
id=uuid4(),
name="test-agent",
display_name="Test Agent",
organization_id=test_organization_id,
description="Test agent description",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions=["Be helpful"],
is_active=True
)
db.add(agent)
# Create chat history
chat_history = ChatHistory(
session_id=uuid4(),
message="Test message",
message_type="user",
customer_id=customer.id,
agent_id=agent.id,
organization_id=test_organization_id
)
db.add(chat_history)
db.commit()
db.refresh(customer)
db.refresh(user)
db.refresh(agent)
db.refresh(chat_history)
return {
"customer": customer,
"user": user,
"agent": agent,
"chat_history": chat_history,
"organization_id": test_organization_id
}
def test_create_attachment_with_user(file_attachment_repo, test_data):
"""Test creating a file attachment uploaded by a user"""
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/test-file.pdf",
filename="test-file.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
chat_history_id=test_data["chat_history"].id,
uploaded_by_user_id=test_data["user"].id
)
assert attachment is not None
assert attachment.id is not None
assert attachment.file_url == "https://s3.amazonaws.com/bucket/test-file.pdf"
assert attachment.filename == "test-file.pdf"
assert attachment.content_type == "application/pdf"
assert attachment.file_size == 1024
assert attachment.organization_id == test_data["organization_id"]
assert attachment.chat_history_id == test_data["chat_history"].id
assert attachment.uploaded_by_user_id == test_data["user"].id
assert attachment.uploaded_by_customer_id is None
assert attachment.created_at is not None
assert attachment.updated_at is not None
def test_create_attachment_with_customer(file_attachment_repo, test_data):
"""Test creating a file attachment uploaded by a customer"""
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/customer-file.jpg",
filename="customer-file.jpg",
content_type="image/jpeg",
file_size=2048,
organization_id=test_data["organization_id"],
chat_history_id=test_data["chat_history"].id,
uploaded_by_customer_id=test_data["customer"].id
)
assert attachment is not None
assert attachment.id is not None
assert attachment.file_url == "https://s3.amazonaws.com/bucket/customer-file.jpg"
assert attachment.filename == "customer-file.jpg"
assert attachment.content_type == "image/jpeg"
assert attachment.file_size == 2048
assert attachment.organization_id == test_data["organization_id"]
assert attachment.chat_history_id == test_data["chat_history"].id
assert attachment.uploaded_by_customer_id == test_data["customer"].id
assert attachment.uploaded_by_user_id is None
def test_create_attachment_without_chat_history(file_attachment_repo, test_data):
"""Test creating a file attachment without linking to chat history"""
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/standalone-file.txt",
filename="standalone-file.txt",
content_type="text/plain",
file_size=512,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
assert attachment is not None
assert attachment.id is not None
assert attachment.file_url == "https://s3.amazonaws.com/bucket/standalone-file.txt"
assert attachment.filename == "standalone-file.txt"
assert attachment.content_type == "text/plain"
assert attachment.file_size == 512
assert attachment.organization_id == test_data["organization_id"]
assert attachment.chat_history_id is None
assert attachment.uploaded_by_user_id == test_data["user"].id
def test_create_attachment_minimal_data(file_attachment_repo, test_data):
"""Test creating a file attachment with minimal required data"""
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/minimal-file.doc",
filename="minimal-file.doc",
content_type="application/msword",
file_size=4096,
organization_id=test_data["organization_id"]
)
assert attachment is not None
assert attachment.id is not None
assert attachment.file_url == "https://s3.amazonaws.com/bucket/minimal-file.doc"
assert attachment.filename == "minimal-file.doc"
assert attachment.content_type == "application/msword"
assert attachment.file_size == 4096
assert attachment.organization_id == test_data["organization_id"]
assert attachment.chat_history_id is None
assert attachment.uploaded_by_user_id is None
assert attachment.uploaded_by_customer_id is None
def test_get_attachment_existing(file_attachment_repo, test_data):
"""Test getting an existing file attachment by ID"""
# Create attachment first
created_attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/get-test.pdf",
filename="get-test.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
# Get attachment
retrieved_attachment = file_attachment_repo.get_attachment(created_attachment.id)
assert retrieved_attachment is not None
assert retrieved_attachment.id == created_attachment.id
assert retrieved_attachment.file_url == "https://s3.amazonaws.com/bucket/get-test.pdf"
assert retrieved_attachment.filename == "get-test.pdf"
assert retrieved_attachment.content_type == "application/pdf"
assert retrieved_attachment.file_size == 1024
def test_get_attachment_non_existing(file_attachment_repo):
"""Test getting a non-existing file attachment"""
attachment = file_attachment_repo.get_attachment(99999)
assert attachment is None
def test_get_attachments_by_chat_history(file_attachment_repo, test_data):
"""Test getting all attachments for a specific chat message"""
chat_history_id = test_data["chat_history"].id
# Create multiple attachments for the same chat message
attachment1 = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/chat-file1.pdf",
filename="chat-file1.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
chat_history_id=chat_history_id,
uploaded_by_customer_id=test_data["customer"].id
)
attachment2 = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/chat-file2.jpg",
filename="chat-file2.jpg",
content_type="image/jpeg",
file_size=2048,
organization_id=test_data["organization_id"],
chat_history_id=chat_history_id,
uploaded_by_customer_id=test_data["customer"].id
)
# Create attachment for different chat message (should not be included)
file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/other-file.txt",
filename="other-file.txt",
content_type="text/plain",
file_size=512,
organization_id=test_data["organization_id"],
chat_history_id=None, # Different chat message
uploaded_by_user_id=test_data["user"].id
)
# Get attachments for the specific chat message
attachments = file_attachment_repo.get_attachments_by_chat_history(chat_history_id)
assert len(attachments) == 2
attachment_ids = [att.id for att in attachments]
assert attachment1.id in attachment_ids
assert attachment2.id in attachment_ids
def test_get_attachments_by_chat_history_empty(file_attachment_repo):
"""Test getting attachments for a chat message with no attachments"""
attachments = file_attachment_repo.get_attachments_by_chat_history(99999)
assert len(attachments) == 0
def test_get_attachments_by_organization(file_attachment_repo, test_data):
"""Test getting all attachments for an organization"""
org_id = test_data["organization_id"]
# Create multiple attachments for the organization
attachment1 = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/org-file1.pdf",
filename="org-file1.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=org_id,
uploaded_by_user_id=test_data["user"].id
)
attachment2 = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/org-file2.jpg",
filename="org-file2.jpg",
content_type="image/jpeg",
file_size=2048,
organization_id=org_id,
uploaded_by_customer_id=test_data["customer"].id
)
# Get attachments for the organization
attachments = file_attachment_repo.get_attachments_by_organization(org_id)
assert len(attachments) >= 2 # May include attachments from other tests
attachment_ids = [att.id for att in attachments]
assert attachment1.id in attachment_ids
assert attachment2.id in attachment_ids
# Verify ordering (most recent first)
assert attachments[0].created_at >= attachments[-1].created_at
def test_get_attachments_by_organization_with_limit(file_attachment_repo, test_data):
"""Test getting attachments for an organization with limit"""
org_id = test_data["organization_id"]
# Create multiple attachments
for i in range(5):
file_attachment_repo.create_attachment(
file_url=f"https://s3.amazonaws.com/bucket/limit-file{i}.txt",
filename=f"limit-file{i}.txt",
content_type="text/plain",
file_size=100 * (i + 1),
organization_id=org_id,
uploaded_by_user_id=test_data["user"].id
)
# Get attachments with limit
attachments = file_attachment_repo.get_attachments_by_organization(org_id, limit=3)
assert len(attachments) <= 3
def test_get_attachments_by_organization_empty(file_attachment_repo):
"""Test getting attachments for an organization with no attachments"""
non_existing_org_id = uuid4()
attachments = file_attachment_repo.get_attachments_by_organization(non_existing_org_id)
assert len(attachments) == 0
def test_delete_attachment_existing(file_attachment_repo, test_data):
"""Test deleting an existing file attachment"""
# Create attachment first
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/delete-test.pdf",
filename="delete-test.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
attachment_id = attachment.id
# Delete attachment
result = file_attachment_repo.delete_attachment(attachment_id)
assert result is True
# Verify attachment is deleted
deleted_attachment = file_attachment_repo.get_attachment(attachment_id)
assert deleted_attachment is None
def test_delete_attachment_non_existing(file_attachment_repo):
"""Test deleting a non-existing file attachment"""
result = file_attachment_repo.delete_attachment(99999)
assert result is False
def test_link_attachment_to_message_existing(file_attachment_repo, test_data):
"""Test linking an existing attachment to a chat message"""
# Create attachment without chat history link
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/link-test.pdf",
filename="link-test.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
assert attachment.chat_history_id is None
# Link to chat message
linked_attachment = file_attachment_repo.link_attachment_to_message(
attachment.id,
test_data["chat_history"].id
)
assert linked_attachment is not None
assert linked_attachment.id == attachment.id
assert linked_attachment.chat_history_id == test_data["chat_history"].id
def test_link_attachment_to_message_non_existing_attachment(file_attachment_repo, test_data):
"""Test linking a non-existing attachment to a chat message"""
result = file_attachment_repo.link_attachment_to_message(
99999,
test_data["chat_history"].id
)
assert result is None
def test_link_attachment_to_message_update_existing_link(file_attachment_repo, test_data, db):
"""Test updating an existing attachment's chat message link"""
# Create another chat history
chat_history2 = ChatHistory(
session_id=uuid4(),
message="Another test message",
message_type="user",
customer_id=test_data["customer"].id,
agent_id=test_data["agent"].id,
organization_id=test_data["organization_id"]
)
db.add(chat_history2)
db.commit()
db.refresh(chat_history2)
# Create attachment linked to first chat message
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/relink-test.pdf",
filename="relink-test.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
chat_history_id=test_data["chat_history"].id,
uploaded_by_user_id=test_data["user"].id
)
assert attachment.chat_history_id == test_data["chat_history"].id
# Link to second chat message
relinked_attachment = file_attachment_repo.link_attachment_to_message(
attachment.id,
chat_history2.id
)
assert relinked_attachment is not None
assert relinked_attachment.id == attachment.id
assert relinked_attachment.chat_history_id == chat_history2.id
def test_create_attachment_with_large_file_size(file_attachment_repo, test_data):
"""Test creating attachment with large file size"""
large_file_size = 1024 * 1024 * 100 # 100MB
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/large-file.zip",
filename="large-file.zip",
content_type="application/zip",
file_size=large_file_size,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
assert attachment is not None
assert attachment.file_size == large_file_size
def test_create_attachment_with_special_characters_filename(file_attachment_repo, test_data):
"""Test creating attachment with special characters in filename"""
special_filename = "test file with spaces & special chars (1).pdf"
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/encoded-filename.pdf",
filename=special_filename,
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
assert attachment is not None
assert attachment.filename == special_filename
def test_create_attachment_various_content_types(file_attachment_repo, test_data):
"""Test creating attachments with various content types"""
content_types = [
("image/png", "test.png"),
("image/jpeg", "test.jpg"),
("image/gif", "test.gif"),
("video/mp4", "test.mp4"),
("audio/mpeg", "test.mp3"),
("text/csv", "test.csv"),
("application/json", "test.json"),
("application/xml", "test.xml")
]
for content_type, filename in content_types:
attachment = file_attachment_repo.create_attachment(
file_url=f"https://s3.amazonaws.com/bucket/{filename}",
filename=filename,
content_type=content_type,
file_size=1024,
organization_id=test_data["organization_id"],
uploaded_by_user_id=test_data["user"].id
)
assert attachment is not None
assert attachment.content_type == content_type
assert attachment.filename == filename
def test_attachment_relationships(file_attachment_repo, test_data, db):
"""Test that attachment relationships are properly established"""
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/relationship-test.pdf",
filename="relationship-test.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
chat_history_id=test_data["chat_history"].id,
uploaded_by_user_id=test_data["user"].id
)
# Refresh to load relationships
db.refresh(attachment)
# Test relationships
assert attachment.chat_history is not None
assert attachment.chat_history.id == test_data["chat_history"].id
assert attachment.organization is not None
assert attachment.organization.id == test_data["organization_id"]
assert attachment.uploaded_by_user is not None
assert attachment.uploaded_by_user.id == test_data["user"].id
assert attachment.uploaded_by_customer is None
def test_attachment_cascade_delete_with_chat_history(file_attachment_repo, test_data, db):
"""Test that attachment is deleted when chat history is deleted (CASCADE)"""
attachment = file_attachment_repo.create_attachment(
file_url="https://s3.amazonaws.com/bucket/cascade-test.pdf",
filename="cascade-test.pdf",
content_type="application/pdf",
file_size=1024,
organization_id=test_data["organization_id"],
chat_history_id=test_data["chat_history"].id,
uploaded_by_user_id=test_data["user"].id
)
attachment_id = attachment.id
# Delete chat history (should cascade delete attachment)
db.delete(test_data["chat_history"])
db.commit()
# Expire all objects to force fresh queries from database
db.expire_all()
# Verify attachment is also deleted
deleted_attachment = file_attachment_repo.get_attachment(attachment_id)
# Note: In test environment (SQLite), CASCADE DELETE might not work as expected
# This test verifies the relationship exists, but cascade behavior depends on DB configuration
if deleted_attachment is not None:
# If cascade didn't work, the attachment should still exist but chat_history_id should be None
# This is because SQLite might handle CASCADE DELETE differently in test environment
assert deleted_attachment.chat_history_id is None
else:
# If cascade worked properly, the attachment should be None
assert deleted_attachment is None
+78
View File
@@ -0,0 +1,78 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.group import GroupRepository
from app.models.user import UserGroup, User
from uuid import uuid4, UUID
@pytest.fixture
def group_repo(db):
return GroupRepository(db)
@pytest.fixture
def test_group(db, test_organization_id):
"""Create a test group with users"""
group = UserGroup(
id=uuid4(),
name="Test Group",
description="A test group",
organization_id=test_organization_id
)
db.add(group)
# Create test users
users = []
for i in range(3):
user = User(
id=uuid4(),
email=f"user{i}@test.com",
full_name=f"Test User {i}",
hashed_password="dummy_hash",
is_active=True,
organization_id=test_organization_id
)
users.append(user)
db.add(user)
# Associate users with the group
group.users.extend(users)
db.commit()
db.refresh(group)
return group
def test_get_group_with_users(group_repo, test_group):
"""Test retrieving a group with its associated users"""
# Get the group with users
group = group_repo.get_group_with_users(test_group.id)
# Verify group data
assert group is not None
assert group.id == test_group.id
assert group.name == "Test Group"
assert group.description == "A test group"
# Verify associated users
assert len(group.users) == 3
for i, user in enumerate(group.users):
assert user.email == f"user{i}@test.com"
assert user.full_name == f"Test User {i}"
assert user.is_active is True
def test_get_nonexistent_group(group_repo):
"""Test retrieving a group that doesn't exist"""
group = group_repo.get_group_with_users(uuid4())
assert group is None
@@ -0,0 +1,150 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.knowledge_queue import KnowledgeQueueRepository
from app.models.knowledge_queue import KnowledgeQueue, QueueStatus
from app.models.user import User
from uuid import uuid4
@pytest.fixture
def queue_repo(db):
return KnowledgeQueueRepository(db)
@pytest.fixture
def test_user(db, test_organization_id):
"""Create a test user"""
user = User(
email="test@example.com",
hashed_password="hashed_password",
organization_id=test_organization_id,
is_active=True,
full_name="Test User"
)
db.add(user)
db.commit()
db.refresh(user)
return user
@pytest.fixture
def test_queue_item(db, test_organization_id, test_user):
"""Create a test queue item"""
queue_item = KnowledgeQueue(
organization_id=test_organization_id,
user_id=test_user.id,
source_type="pdf_file",
source="/test/path/document.pdf",
status=QueueStatus.PENDING,
error=None,
queue_metadata={"test": "metadata"}
)
db.add(queue_item)
db.commit()
db.refresh(queue_item)
return queue_item
def test_create_queue_item(queue_repo, test_organization_id, test_user):
"""Test creating a new queue item"""
queue_item = KnowledgeQueue(
organization_id=test_organization_id,
user_id=test_user.id,
source_type="pdf_file",
source="/test/path/new.pdf",
status=QueueStatus.PENDING,
queue_metadata={"test": "metadata"}
)
created_item = queue_repo.create(queue_item)
assert created_item is not None
assert created_item.source == "/test/path/new.pdf"
assert created_item.status == QueueStatus.PENDING
assert created_item.error is None
def test_get_by_id(queue_repo, test_queue_item):
"""Test retrieving a queue item by ID"""
item = queue_repo.get_by_id(test_queue_item.id)
assert item is not None
assert item.id == test_queue_item.id
assert item.source == "/test/path/document.pdf"
assert item.status == QueueStatus.PENDING
def test_get_by_id_nonexistent(queue_repo):
"""Test retrieving a nonexistent queue item"""
item = queue_repo.get_by_id(999)
assert item is None
def test_get_pending(queue_repo, test_queue_item, db, test_organization_id, test_user):
"""Test retrieving pending queue items"""
# Create another pending item
another_item = KnowledgeQueue(
organization_id=test_organization_id,
user_id=test_user.id,
source_type="pdf_file",
source="/test/path/another.pdf",
status=QueueStatus.PENDING,
queue_metadata={"test": "metadata"}
)
db.add(another_item)
# Create a completed item
completed_item = KnowledgeQueue(
organization_id=test_organization_id,
user_id=test_user.id,
source_type="pdf_file",
source="/test/path/completed.pdf",
status=QueueStatus.COMPLETED,
queue_metadata={"test": "metadata"}
)
db.add(completed_item)
db.commit()
# Get pending items
pending_items = queue_repo.get_pending()
assert len(pending_items) == 2
assert all(item.status == QueueStatus.PENDING for item in pending_items)
def test_update_status_success(queue_repo, test_queue_item):
"""Test successfully updating queue item status"""
success = queue_repo.update_status(
test_queue_item.id,
QueueStatus.COMPLETED
)
assert success is True
# Verify the update
updated_item = queue_repo.get_by_id(test_queue_item.id)
assert updated_item.status == QueueStatus.COMPLETED
assert updated_item.error is None
def test_update_status_with_error(queue_repo, test_queue_item):
"""Test updating queue item status with error"""
error_message = "Processing failed"
success = queue_repo.update_status(
test_queue_item.id,
QueueStatus.FAILED,
error=error_message
)
assert success is True
# Verify the update
updated_item = queue_repo.get_by_id(test_queue_item.id)
assert updated_item.status == QueueStatus.FAILED
assert updated_item.error == error_message
def test_update_status_nonexistent(queue_repo):
"""Test updating status of nonexistent queue item"""
success = queue_repo.update_status(999, QueueStatus.COMPLETED)
assert success is False
+262
View File
@@ -0,0 +1,262 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.knowledge import KnowledgeRepository
from app.models.knowledge import Knowledge, SourceType
from app.models.knowledge_to_agent import KnowledgeToAgent
from app.models.agent import Agent, AgentType
from uuid import uuid4
@pytest.fixture
def knowledge_repo(db):
return KnowledgeRepository(db)
@pytest.fixture
def test_knowledge(db, test_organization_id):
"""Create a test knowledge source"""
knowledge = Knowledge(
organization_id=test_organization_id,
source="/test/path/document.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(knowledge)
db.commit()
db.refresh(knowledge)
return knowledge
@pytest.fixture
def test_agent(db, test_organization_id):
"""Create a test agent"""
agent = Agent(
name="Test Agent",
display_name="Test Agent",
description="A test agent",
agent_type=AgentType.GENERAL,
instructions=["Test instruction"],
organization_id=test_organization_id,
is_active=True
)
db.add(agent)
db.commit()
db.refresh(agent)
return agent
def test_create_knowledge(knowledge_repo, test_organization_id):
"""Test creating a new knowledge source"""
knowledge = Knowledge(
organization_id=test_organization_id,
source="/test/path/new.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
created = knowledge_repo.create(knowledge)
assert created is not None
assert created.source == "/test/path/new.pdf"
assert created.source_type == SourceType.FILE
def test_create_duplicate_knowledge(knowledge_repo, test_knowledge):
"""Test creating a duplicate knowledge source returns existing one"""
duplicate = Knowledge(
organization_id=test_knowledge.organization_id,
source=test_knowledge.source,
source_type=test_knowledge.source_type,
schema=test_knowledge.schema,
table_name=test_knowledge.table_name
)
created = knowledge_repo.create(duplicate)
assert created.id == test_knowledge.id
def test_get_by_id(knowledge_repo, test_knowledge):
"""Test retrieving a knowledge source by ID"""
knowledge = knowledge_repo.get_by_id(test_knowledge.id)
assert knowledge is not None
assert knowledge.id == test_knowledge.id
assert knowledge.source == test_knowledge.source
def test_get_by_id_nonexistent(knowledge_repo):
"""Test retrieving a nonexistent knowledge source"""
knowledge = knowledge_repo.get_by_id(999)
assert knowledge is None
def test_get_by_agent(knowledge_repo, test_knowledge, test_agent, db):
"""Test retrieving knowledge sources for an agent"""
# Create a link to the test knowledge
link = KnowledgeToAgent(
knowledge_id=test_knowledge.id,
agent_id=test_agent.id
)
db.add(link)
# Create another knowledge source and link
another_knowledge = Knowledge(
organization_id=test_knowledge.organization_id,
source="/test/path/another.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(another_knowledge)
db.flush()
another_link = KnowledgeToAgent(
knowledge_id=another_knowledge.id,
agent_id=test_agent.id
)
db.add(another_link)
db.commit()
# Test pagination
knowledge_sources = knowledge_repo.get_by_agent(test_agent.id, skip=0, limit=1)
assert len(knowledge_sources) == 1
knowledge_sources = knowledge_repo.get_by_agent(test_agent.id, skip=0, limit=10)
assert len(knowledge_sources) == 2
assert all(k.id in [test_knowledge.id, another_knowledge.id] for k in knowledge_sources)
def test_count_by_agent(knowledge_repo, test_knowledge, test_agent, db):
"""Test counting knowledge sources for an agent"""
# Create a link to the test knowledge
link = KnowledgeToAgent(
knowledge_id=test_knowledge.id,
agent_id=test_agent.id
)
db.add(link)
db.commit()
count = knowledge_repo.count_by_agent(test_agent.id)
assert count == 1
def test_get_by_org(knowledge_repo, test_knowledge, db):
"""Test retrieving knowledge sources for an organization"""
# Create another knowledge source for the same org
another_knowledge = Knowledge(
organization_id=test_knowledge.organization_id,
source="/test/path/another.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(another_knowledge)
db.commit()
knowledge_sources = knowledge_repo.get_by_org(test_knowledge.organization_id)
assert len(knowledge_sources) == 2
assert all(k.organization_id == test_knowledge.organization_id for k in knowledge_sources)
def test_get_by_organization_paginated(knowledge_repo, test_knowledge, db):
"""Test retrieving paginated knowledge sources for an organization"""
# Create another knowledge source for the same org
another_knowledge = Knowledge(
organization_id=test_knowledge.organization_id,
source="/test/path/another.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(another_knowledge)
db.commit()
# Test pagination
knowledge_sources = knowledge_repo.get_by_organization(
test_knowledge.organization_id,
skip=0,
limit=1
)
assert len(knowledge_sources) == 1
knowledge_sources = knowledge_repo.get_by_organization(
test_knowledge.organization_id,
skip=0,
limit=10
)
assert len(knowledge_sources) == 2
def test_count_by_organization(knowledge_repo, test_knowledge, db):
"""Test counting knowledge sources for an organization"""
# Create another knowledge source for the same org
another_knowledge = Knowledge(
organization_id=test_knowledge.organization_id,
source="/test/path/another.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(another_knowledge)
db.commit()
count = knowledge_repo.count_by_organization(test_knowledge.organization_id)
assert count == 2
def test_get_by_sources(knowledge_repo, test_knowledge, db):
"""Test retrieving knowledge sources by source URLs"""
# Create another knowledge source
another_knowledge = Knowledge(
organization_id=test_knowledge.organization_id,
source="/test/path/another.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(another_knowledge)
db.commit()
sources = [test_knowledge.source, another_knowledge.source]
knowledge_sources = knowledge_repo.get_by_sources(test_knowledge.organization_id, sources)
assert len(knowledge_sources) == 2
assert all(k.source in sources for k in knowledge_sources)
def test_delete(knowledge_repo, test_knowledge):
"""Test deleting a knowledge source"""
success = knowledge_repo.delete(test_knowledge.id)
assert success is True
# Verify deletion
knowledge = knowledge_repo.get_by_id(test_knowledge.id)
assert knowledge is None
def test_delete_nonexistent(knowledge_repo):
"""Test deleting a nonexistent knowledge source"""
success = knowledge_repo.delete(999)
assert success is False
def test_delete_with_data(knowledge_repo, test_knowledge, test_agent, db):
"""Test deleting a knowledge source with its data"""
# Create a link to test cascade deletion
link = KnowledgeToAgent(
knowledge_id=test_knowledge.id,
agent_id=test_agent.id
)
db.add(link)
db.commit()
success = knowledge_repo.delete_with_data(test_knowledge.id)
assert success is True
# Verify knowledge source deletion
knowledge = knowledge_repo.get_by_id(test_knowledge.id)
assert knowledge is None
# Verify link deletion
link = db.query(KnowledgeToAgent).filter(
KnowledgeToAgent.knowledge_id == test_knowledge.id
).first()
assert link is None
@@ -0,0 +1,158 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.knowledge_to_agent import KnowledgeToAgentRepository
from app.models.knowledge_to_agent import KnowledgeToAgent
from app.models.knowledge import Knowledge, SourceType
from app.models.agent import Agent, AgentType
from uuid import uuid4
@pytest.fixture
def link_repo(db):
return KnowledgeToAgentRepository(db)
@pytest.fixture
def test_knowledge(db, test_organization_id):
"""Create a test knowledge source"""
knowledge = Knowledge(
organization_id=test_organization_id,
source="/test/path/document.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(knowledge)
db.commit()
db.refresh(knowledge)
return knowledge
@pytest.fixture
def test_agent(db, test_organization_id):
"""Create a test agent"""
agent = Agent(
name="Test Agent",
display_name="Test Agent",
description="A test agent",
agent_type=AgentType.GENERAL,
instructions="Test instructions",
organization_id=test_organization_id,
is_active=True
)
db.add(agent)
db.commit()
db.refresh(agent)
return agent
@pytest.fixture
def test_link(db, test_knowledge, test_agent):
"""Create a test knowledge-to-agent link"""
link = KnowledgeToAgent(
knowledge_id=test_knowledge.id,
agent_id=test_agent.id
)
db.add(link)
db.commit()
db.refresh(link)
return link
def test_create_link(link_repo, test_knowledge, test_agent):
"""Test creating a new knowledge-to-agent link"""
link = KnowledgeToAgent(
knowledge_id=test_knowledge.id,
agent_id=test_agent.id
)
created_link = link_repo.create(link)
assert created_link is not None
assert created_link.knowledge_id == test_knowledge.id
assert created_link.agent_id == test_agent.id
def test_create_duplicate_link(link_repo, test_link):
"""Test creating a duplicate link returns existing link"""
duplicate_link = KnowledgeToAgent(
knowledge_id=test_link.knowledge_id,
agent_id=test_link.agent_id
)
created_link = link_repo.create(duplicate_link)
assert created_link.id == test_link.id
def test_get_by_ids(link_repo, test_link):
"""Test retrieving a link by knowledge_id and agent_id"""
link = link_repo.get_by_ids(test_link.knowledge_id, test_link.agent_id)
assert link is not None
assert link.id == test_link.id
assert link.knowledge_id == test_link.knowledge_id
assert link.agent_id == test_link.agent_id
def test_get_by_ids_nonexistent(link_repo):
"""Test retrieving a nonexistent link"""
link = link_repo.get_by_ids(999, uuid4())
assert link is None
def test_get_by_agent(link_repo, test_link, db, test_organization_id, test_agent):
"""Test retrieving all links for an agent"""
# Create another link for the same agent
another_knowledge = Knowledge(
organization_id=test_organization_id,
source="/test/path/another.pdf",
source_type=SourceType.FILE,
schema="test_schema",
table_name="test_table"
)
db.add(another_knowledge)
db.commit()
another_link = KnowledgeToAgent(
knowledge_id=another_knowledge.id,
agent_id=test_agent.id
)
db.add(another_link)
db.commit()
# Get links for agent
links = link_repo.get_by_agent(test_agent.id)
assert len(links) == 2
assert all(link.agent_id == test_agent.id for link in links)
def test_delete_by_ids(link_repo, test_link):
"""Test deleting a link by knowledge_id and agent_id"""
success = link_repo.delete_by_ids(test_link.knowledge_id, test_link.agent_id)
assert success is True
# Verify deletion
link = link_repo.get_by_ids(test_link.knowledge_id, test_link.agent_id)
assert link is None
def test_delete_by_ids_nonexistent(link_repo):
"""Test deleting a nonexistent link"""
success = link_repo.delete_by_ids(999, uuid4())
assert success is False
def test_delete(link_repo, test_link):
"""Test deleting a link by ID"""
success = link_repo.delete(test_link.id)
assert success is True
# Verify deletion
link = link_repo.get_by_ids(test_link.knowledge_id, test_link.agent_id)
assert link is None
def test_delete_nonexistent(link_repo):
"""Test deleting a nonexistent link by ID"""
success = link_repo.delete(999)
assert success is False
@@ -0,0 +1,64 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.lead_capture import LeadCaptureConfigRepository
from app.models.schemas.lead_capture import LeadCaptureConfigUpdate, LeadField
@pytest.fixture
def repo(db):
return LeadCaptureConfigRepository(db)
def test_get_by_agent_none_initially(repo, test_agent):
assert repo.get_by_agent(test_agent.id) is None
def test_get_or_create_lazily_creates_off_config(repo, test_agent):
cfg = repo.get_or_create(test_agent.id)
assert cfg is not None
assert cfg.agent_id == test_agent.id
assert cfg.enabled is False
# Idempotent — a second call returns the same row.
again = repo.get_or_create(test_agent.id)
assert again.id == cfg.id
def test_update_replaces_config(repo, test_agent):
repo.get_or_create(test_agent.id)
data = LeadCaptureConfigUpdate(
enabled=True,
require_consent=False,
guidance="Ask after pricing questions",
fields=[
LeadField(key="email", standard=True, enabled=True, required=True),
LeadField(key="custom_team", standard=False, enabled=True, label="Team size",
options=["1-10", "11-50"]),
],
)
updated = repo.update(test_agent.id, data)
assert updated.enabled is True
assert updated.require_consent is False
assert updated.guidance == "Ask after pricing questions"
keys = {f["key"] for f in (updated.fields or [])}
assert keys == {"email", "custom_team"}
# Persisted: a fresh fetch reflects the update.
fetched = repo.get_by_agent(test_agent.id)
assert fetched.enabled is True
assert len(fetched.fields) == 2
@@ -0,0 +1,197 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.models.organization import Organization
from app.repositories.organization import OrganizationRepository
@pytest.fixture
def org_repo(db):
"""Create an organization repository instance"""
return OrganizationRepository(db)
def test_create_organization(org_repo):
"""Test creating a new organization"""
name = "Test Organization"
domain = "test.com"
timezone = "UTC"
business_hours = {
"monday": {"start": "09:00", "end": "17:00"},
"tuesday": {"start": "09:00", "end": "17:00"}
}
org = org_repo.create_organization(
name=name,
domain=domain,
timezone=timezone,
business_hours=business_hours
)
assert org.name == name
assert org.domain == domain
assert org.timezone == timezone
assert org.business_hours == business_hours
assert org.is_active is True
def test_get_organization(org_repo):
"""Test retrieving an organization by ID"""
# Create test organization
org = org_repo.create_organization(
name="Test Organization",
domain="test.com",
timezone="UTC"
)
# Retrieve organization
retrieved_org = org_repo.get_organization(org.id)
assert retrieved_org.id == org.id
assert retrieved_org.name == org.name
def test_get_organization_by_domain(org_repo):
"""Test retrieving an organization by domain"""
domain = "test.com"
org = org_repo.create_organization(
name="Test Organization",
domain=domain,
timezone="UTC"
)
retrieved_org = org_repo.get_organization_by_domain(domain)
assert retrieved_org.id == org.id
assert retrieved_org.domain == domain
def test_is_domain_taken(org_repo):
"""Domain clashes ignore case and the org being updated"""
org = org_repo.create_organization(name="Test Organization", domain="test.com", timezone="UTC")
assert org_repo.is_domain_taken("TEST.com") is True
assert org_repo.is_domain_taken("test.com", exclude_org_id=org.id) is False
assert org_repo.is_domain_taken("other.com") is False
def test_get_multi(org_repo):
"""Test retrieving multiple organizations with pagination"""
# Create multiple organizations
orgs = []
for i in range(5):
org = org_repo.create_organization(
name=f"Organization {i}",
domain=f"org{i}.com",
timezone="UTC"
)
orgs.append(org)
# Test pagination
active_orgs = org_repo.get_active_organizations()
assert len(active_orgs) == 5
def test_update_organization(org_repo):
"""Test updating an organization"""
# Create test organization
org = org_repo.create_organization(
name="Test Organization",
domain="test.com",
timezone="UTC"
)
# Update organization
new_name = "Updated Organization"
new_settings = {"new_setting": "new_value"}
updated_org = org_repo.update_organization(
org_id=org.id,
name=new_name,
settings=new_settings
)
assert updated_org.name == new_name
assert updated_org.settings == new_settings
assert updated_org.id == org.id
def test_delete_organization(org_repo):
"""Test deactivating an organization"""
# Create test organization
org = org_repo.create_organization(
name="Test Organization",
domain="test.com",
timezone="UTC"
)
# Deactivate organization
success = org_repo.deactivate_organization(org.id)
assert success is True
# Verify organization is deactivated
deactivated_org = org_repo.get_organization(org.id)
assert deactivated_org is not None
assert deactivated_org.is_active is False
def test_get_active_organizations(org_repo):
"""Test retrieving only active organizations"""
# Create active and inactive organizations
active_org = org_repo.create_organization(
name="Active Organization",
domain="active.com",
timezone="UTC"
)
inactive_org = org_repo.create_organization(
name="Inactive Organization",
domain="inactive.com",
timezone="UTC"
)
org_repo.deactivate_organization(inactive_org.id)
# Get active organizations
active_orgs = org_repo.get_active_organizations()
assert len(active_orgs) == 1
assert active_orgs[0].id == active_org.id
assert active_orgs[0].is_active is True
def test_organization_with_business_hours(org_repo):
"""Test organization with business hours"""
business_hours = {
"monday": {"start": "09:00", "end": "17:00"},
"tuesday": {"start": "09:00", "end": "17:00"},
"wednesday": {"start": "09:00", "end": "17:00"},
"thursday": {"start": "09:00", "end": "17:00"},
"friday": {"start": "09:00", "end": "17:00"}
}
org = org_repo.create_organization(
name="Test Organization",
domain="test.com",
timezone="UTC",
business_hours=business_hours
)
retrieved_org = org_repo.get_organization(org.id)
assert retrieved_org.business_hours == business_hours
def test_organization_with_settings(org_repo):
"""Test organization with custom settings"""
settings = {
"chat_timeout": 300,
"max_sessions": 10,
"welcome_message": "Welcome to our organization!"
}
org = org_repo.create_organization(
name="Test Organization",
domain="test.com",
timezone="UTC"
)
# Update settings
updated_org = org_repo.update_settings(org.id, settings)
assert updated_org.settings == settings
+299
View File
@@ -0,0 +1,299 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from uuid import uuid4
from app.models.customer import Customer, LeadStage
from app.models.lead_capture import LeadCaptureResponse
from app.models.chat_history import ChatHistory
from app.repositories.people import PeopleRepository
@pytest.fixture
def repo(db):
return PeopleRepository(db)
def _customer(db, org_id, chatted=True, **kw):
"""Create a customer. By default also adds a chat message so it counts as
'engaged' (People only shows people who actually chatted). Pass chatted=False
to simulate an empty widget-load."""
c = Customer(email=kw.pop("email", f"{uuid4().hex}@noemail.com"),
organization_id=org_id, **kw)
db.add(c); db.commit(); db.refresh(c)
if chatted:
db.add(ChatHistory(organization_id=org_id, customer_id=c.id,
message="hi", message_type="user"))
db.commit()
return c
def _response(db, org_id, customer_id, agent_id, field_values):
r = LeadCaptureResponse(
organization_id=org_id, agent_id=agent_id, customer_id=customer_id,
field_values=field_values, summary="qualified", consent=True, qualified=True,
)
db.add(r); db.commit(); db.refresh(r)
return r
def test_list_and_stats(repo, db, test_organization_id, test_agent):
lead = _customer(db, test_organization_id, email="lead@acme.com",
full_name="Lead One", lead_stage=LeadStage.LEAD)
_response(db, test_organization_id, lead.id, test_agent.id, {"email": "lead@acme.com"})
_customer(db, test_organization_id) # anonymous visitor
# Identity split: the default view holds identified people only; the
# anonymous browser session lives behind view="anonymous".
items, total = repo.list_people(test_organization_id)
assert total == 1
assert items[0]["email"] == "lead@acme.com"
anon_items, anon_total = repo.list_people(test_organization_id, view="anonymous")
assert anon_total == 1
assert anon_items[0]["is_anonymous"] is True
# stage filter
leads, lead_total = repo.list_people(test_organization_id, stage="lead")
assert lead_total == 1
assert leads[0]["qualified"] is True
# search
found, found_total = repo.list_people(test_organization_id, search="Lead One")
assert found_total == 1
stats = repo.get_stats(test_organization_id)
assert stats["total_people"] == 1
assert stats["anonymous"] == 1
def test_anonymous_falls_back_to_captured_email(repo, db, test_organization_id, test_agent):
"""A captured lead whose email couldn't be written to the customer row still
shows the captured email (not 'anonymous')."""
anon = _customer(db, test_organization_id, lead_stage=LeadStage.LEAD)
_response(db, test_organization_id, anon.id, test_agent.id, {"email": "hidden@acme.com"})
items, _ = repo.list_people(test_organization_id, stage="lead")
row = next(i for i in items if i["id"] == anon.id)
assert row["email"] == "hidden@acme.com"
assert row["is_anonymous"] is False
def test_get_detail_and_mark_customer(repo, db, test_organization_id, test_agent):
person = _customer(db, test_organization_id, email="d@acme.com",
full_name="Detail", lead_stage=LeadStage.LEAD)
_response(db, test_organization_id, person.id, test_agent.id,
{"email": "d@acme.com", "company": "Acme"})
detail = repo.get_detail(test_organization_id, person.id)
assert detail is not None
assert detail["email"] == "d@acme.com"
assert detail["captured_attributes"].get("company") == "Acme"
assert detail["summary"] == "qualified"
updated = repo.mark_customer(test_organization_id, person.id)
assert updated.lead_stage == LeadStage.CUSTOMER
def test_get_detail_missing_returns_none(repo, test_organization_id):
assert repo.get_detail(test_organization_id, uuid4()) is None
def test_unengaged_widget_loads_excluded(repo, db, test_organization_id):
"""A customer that never chatted (empty widget-load) is excluded from People."""
_customer(db, test_organization_id, email="chatted@acme.com", chatted=True,
lead_stage=LeadStage.LEAD)
_customer(db, test_organization_id, chatted=False) # loaded widget, never messaged
items, total = repo.list_people(test_organization_id)
assert total == 1
assert items[0]["email"] == "chatted@acme.com"
assert repo.get_stats(test_organization_id)["total_people"] == 1
def test_authenticated_customers_excluded(repo, db, test_organization_id):
"""Integration-authenticated people (identified via generate-token) are the
business's existing customers, not leads — excluded from People + stats.
Having meta_data must NOT by itself exclude a customer: meta_data is orthogonal
to authentication (and is stored as JSON null when empty, so the old
`meta_data IS NULL` test hid every organic visitor)."""
# Organic lead that happens to carry meta_data (e.g. UTM) — must still show.
_customer(db, test_organization_id, email="organic@acme.com",
full_name="Organic", lead_stage=LeadStage.LEAD,
meta_data={"utm": "google"})
# Integration-authenticated customer — excluded.
_customer(db, test_organization_id, email="portal@acme.com", full_name="Portal User",
is_authenticated=True,
meta_data={"student_name": "Sam", "center_name": "MMCA"})
items, total = repo.list_people(test_organization_id)
emails = {i["email"] for i in items}
assert "organic@acme.com" in emails # meta_data alone no longer excludes
assert "portal@acme.com" not in emails # authenticated flag excludes
assert total == 1
stats = repo.get_stats(test_organization_id)
assert stats["total_people"] == 1 # the authenticated customer is not counted
def test_merged_rows_hidden_and_resolved(repo, db, test_organization_id):
target = _customer(db, test_organization_id, email="t@acme.com", full_name="Target",
lead_stage=LeadStage.LEAD)
merged = _customer(db, test_organization_id, lead_stage=LeadStage.LEAD)
merged.merged_into_customer_id = target.id
db.commit()
# Merged row is excluded from the list.
items, total = repo.list_people(test_organization_id)
ids = {i["id"] for i in items}
assert merged.id not in ids
assert target.id in ids
# Requesting the merged row resolves to the surviving target.
detail = repo.get_detail(test_organization_id, merged.id)
assert detail["id"] == target.id
def test_phone_identifies_and_is_searchable(repo, db, test_organization_id):
"""A phone alone identifies a person (WhatsApp/SMS contacts have no real
email), and decorated search terms match on digits."""
person = _customer(db, test_organization_id,
email="911234567890@whatsapp.channel",
full_name="Whatsapp user 91123456", phone="+911234567890")
items, total = repo.list_people(test_organization_id)
assert total == 1 and items[0]["id"] == person.id
assert items[0]["phone"] == "+911234567890"
for term in ["+91 12345 67890", "9112345", "12345 67890"]:
_, found = repo.list_people(test_organization_id, search=term)
assert found == 1, term
# A non-numeric search must not explode on the phone clause
_, none_found = repo.list_people(test_organization_id, search="nobody")
assert none_found == 0
def test_update_person_edits_and_guards(repo, db, test_organization_id):
person = _customer(db, test_organization_id)
other = _customer(db, test_organization_id, phone="+15550001111")
# Set + correct (overwrite is allowed here — it's the explicit human path)
updated, error = repo.update_person(test_organization_id, person.id,
full_name="Priya", phone="+91 12345 67890")
assert error is None
assert updated.full_name == "Priya" and updated.phone == "+911234567890"
updated, error = repo.update_person(test_organization_id, person.id, phone="+919999999999")
assert error is None and updated.phone == "+919999999999"
# Someone else's number is refused, not reassigned
_, error = repo.update_person(test_organization_id, person.id, phone="+15550001111")
assert "another person" in error
db.refresh(person)
assert person.phone == "+919999999999"
# National-format digits are refused, not guessed at
_, error = repo.update_person(test_organization_id, person.id, phone="1234567890")
assert "international format" in error
# Clearing via empty string
updated, error = repo.update_person(test_organization_id, person.id, phone="")
assert error is None and updated.phone is None
missing, _ = repo.update_person(test_organization_id, uuid4(), full_name="x")
assert missing is None
def test_mark_customer_is_gated_on_identity(repo, db, test_organization_id):
anon = _customer(db, test_organization_id)
assert repo.is_identified(anon) is False
# Identify via phone → the gate opens
repo.update_person(test_organization_id, anon.id, phone="+911234567890")
db.refresh(anon)
assert repo.is_identified(anon) is True
def _whatsapp_conversation(db, org_id, customer, wa_id="919999900001"):
"""A WhatsApp conversation — what makes a person reachable BY PHONE, and so
what makes their phone load-bearing for inbound routing."""
from app.models.channels.channel_conversation import ChannelConversation
conv = ChannelConversation(
channel_account_id=uuid4(), channel_type="whatsapp",
external_conversation_id=wa_id, external_user_id=wa_id,
session_id=uuid4(), organization_id=org_id, customer_id=customer.id)
db.add(conv); db.commit()
return conv
class TestPhoneAsSoleIdentityKey:
"""A WhatsApp person starts keyed by both their @whatsapp.channel address
and their phone. Once a capture upgrades that address to a real email, the
phone is all that's left — nothing about a WhatsApp message carries email."""
def test_editing_the_only_key_is_refused(self, repo, db, test_organization_id):
person = _customer(db, test_organization_id, email="w@corp.com",
phone="+919999900001")
_whatsapp_conversation(db, test_organization_id, person)
updated, error = repo.update_person(
test_organization_id, person.id, phone="+919999900002")
assert error is not None and "split them into a second person" in error
db.refresh(person)
assert person.phone == "+919999900001" # unchanged
def test_clearing_the_only_key_is_refused(self, repo, db, test_organization_id):
person = _customer(db, test_organization_id, email="w@corp.com",
phone="+919999900001")
_whatsapp_conversation(db, test_organization_id, person)
_, error = repo.update_person(test_organization_id, person.id, phone="")
assert error is not None
db.refresh(person)
assert person.phone == "+919999900001"
def test_still_editable_while_the_channel_address_survives(
self, repo, db, test_organization_id):
"""Not yet captured: they're still findable by 919…@whatsapp.channel,
so the phone is a correctable detail, not their only identity."""
person = _customer(db, test_organization_id,
email="919999900001@whatsapp.channel",
phone="+919999900001")
_whatsapp_conversation(db, test_organization_id, person)
_, error = repo.update_person(
test_organization_id, person.id, phone="+919999900002")
assert error is None
db.refresh(person)
assert person.phone == "+919999900002"
def test_widget_person_phone_stays_freely_editable(
self, repo, db, test_organization_id):
"""No phone-keyed channel: a typo'd number is just a typo. This is the
common case and must not be caught by the guard."""
person = _customer(db, test_organization_id, email="w@corp.com",
phone="+919999900001")
_, error = repo.update_person(
test_organization_id, person.id, phone="+919999900002")
assert error is None
db.refresh(person)
assert person.phone == "+919999900002"
+231
View File
@@ -0,0 +1,231 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.role import RoleRepository
from app.models.role import Role
from app.models.permission import Permission
from app.models.user import User
from app.models.organization import Organization
from uuid import uuid4, UUID
@pytest.fixture
def test_organization(db):
"""Create a test organization"""
org = Organization(
name="Test Organization",
domain="test.com",
timezone="UTC"
)
db.add(org)
db.commit()
db.refresh(org)
return org
@pytest.fixture
def role_repo(db):
return RoleRepository(db)
@pytest.fixture
def test_permission(db):
"""Create a test permission"""
permission = Permission(name="test_permission", description="Test Permission")
db.add(permission)
db.commit()
db.refresh(permission)
return permission
@pytest.fixture
def test_role(db, test_organization):
"""Create a test role"""
role = Role(
name="Test Role",
description="Test Role Description",
organization_id=test_organization.id,
is_default=False
)
db.add(role)
db.commit()
db.refresh(role)
return role
def test_create_role(role_repo, test_organization):
"""Test creating a new role"""
created = role_repo.create_role(
name="New Role",
description="New Role Description",
organization_id=test_organization.id,
is_default=False
)
assert created is not None
assert created.name == "New Role"
assert created.description == "New Role Description"
assert created.organization_id == test_organization.id
assert created.is_default is False
def test_create_role_with_permissions(role_repo, test_organization, test_permission):
"""Test creating a role with permissions"""
created = role_repo.create_role(
name="Role With Permissions",
description="Role With Permissions Description",
organization_id=test_organization.id,
permission_ids=[test_permission.id]
)
assert created is not None
assert len(created.permissions) == 1
assert created.permissions[0].id == test_permission.id
def test_get_role(role_repo, test_role):
"""Test retrieving a role by ID"""
role = role_repo.get_role(test_role.id)
assert role is not None
assert role.id == test_role.id
assert role.name == test_role.name
def test_get_org_roles(role_repo, test_role):
"""Test retrieving all roles for an organization"""
roles = role_repo.get_org_roles(test_role.organization_id)
assert len(roles) == 1
assert roles[0].id == test_role.id
def test_update_role(role_repo, test_role):
"""Test updating role details"""
updated = role_repo.update_role(
test_role.id,
name="Updated Role",
description="Updated Description"
)
assert updated is not None
assert updated.name == "Updated Role"
assert updated.description == "Updated Description"
def test_update_role_permissions(role_repo, test_role, test_permission):
"""Test updating role permissions"""
updated = role_repo.update_role(
test_role.id,
permissions=[{"id": test_permission.id}]
)
assert updated is not None
assert len(updated.permissions) == 1
assert updated.permissions[0].id == test_permission.id
def test_update_nonexistent_role(role_repo):
"""Test updating a nonexistent role"""
updated = role_repo.update_role(999, name="Updated Role")
assert updated is None
def test_delete_role(role_repo, test_role):
"""Test deleting a role"""
success = role_repo.delete_role(test_role.id)
assert success is True
# Verify deletion
role = role_repo.get_role(test_role.id)
assert role is None
def test_delete_nonexistent_role(role_repo):
"""Test deleting a nonexistent role"""
success = role_repo.delete_role(999)
assert success is False
def test_add_permission(role_repo, test_role, test_permission):
"""Test adding a permission to a role"""
success = role_repo.add_permission(test_role.id, test_permission.name)
assert success is True
# Verify permission was added
role = role_repo.get_role(test_role.id)
assert len(role.permissions) == 1
assert role.permissions[0].id == test_permission.id
def test_add_nonexistent_permission(role_repo, test_role):
"""Test adding a nonexistent permission to a role"""
success = role_repo.add_permission(test_role.id, "nonexistent_permission")
assert success is False
def test_remove_permission(role_repo, test_role, test_permission, db):
"""Test removing a permission from a role"""
# First add the permission
test_role.permissions.append(test_permission)
db.commit()
success = role_repo.remove_permission(test_role.id, test_permission.name)
assert success is True
# Verify permission was removed
role = role_repo.get_role(test_role.id)
assert len(role.permissions) == 0
def test_remove_nonexistent_permission(role_repo, test_role):
"""Test removing a nonexistent permission from a role"""
success = role_repo.remove_permission(test_role.id, "nonexistent_permission")
assert success is False
def test_get_role_permissions(role_repo, test_role, test_permission, db):
"""Test getting all permissions for a role"""
# Add permission to role
test_role.permissions.append(test_permission)
db.commit()
permissions = role_repo.get_role_permissions(test_role.id)
assert len(permissions) == 1
assert permissions[0] == test_permission.name
def test_get_roles_by_organization(role_repo, test_role):
"""Test getting all roles in an organization"""
roles = role_repo.get_roles_by_organization(test_role.organization_id)
assert len(roles) == 1
assert roles[0].id == test_role.id
def test_get_default_role(role_repo, test_organization, db):
"""Test getting the default role for an organization"""
default_role = Role(
name="Default Role",
description="Default Role Description",
organization_id=test_organization.id,
is_default=True
)
db.add(default_role)
db.commit()
role = role_repo.get_default_role(test_organization.id)
assert role is not None
assert role.is_default is True
assert role.name == "Default Role"
def test_is_role_in_use(role_repo, test_role, test_organization, db):
"""Test checking if a role is assigned to any users"""
# Create a test user with the role
user = User(
email="test@example.com",
hashed_password="hashed_password",
organization_id=test_organization.id,
role_id=test_role.id
)
db.add(user)
db.commit()
in_use = role_repo.is_role_in_use(test_role.id)
assert in_use is True
def test_is_role_not_in_use(role_repo, test_role):
"""Test checking if an unused role is in use"""
in_use = role_repo.is_role_in_use(test_role.id)
assert in_use is False
@@ -0,0 +1,301 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from sqlalchemy.orm import Session
from app.models.user import User
from app.services.conversation_token import VERIFIED_SCOPE
from app.models.role import Role
from app.models.permission import Permission, role_permissions
from app.models.session_to_agent import SessionToAgent, SessionStatus
from app.models.chat_history import ChatHistory
from app.models.customer import Customer
from app.models.agent import Agent, AgentType
from uuid import UUID, uuid4
from app.core.security import get_password_hash
from app.repositories.session_to_agent import SessionToAgentRepository
@pytest.fixture
def test_role(db: Session, test_organization_id: UUID) -> Role:
"""Create a test role with required permissions"""
role = Role(
name="Test Role",
organization_id=test_organization_id
)
db.add(role)
db.commit()
# Add required permissions
permission = Permission(
name="manage_chats",
description="Can manage chats"
)
db.add(permission)
db.commit()
# Associate permission with role
db.execute(
role_permissions.insert().values(
role_id=role.id,
permission_id=permission.id
)
)
db.commit()
return role
@pytest.fixture
def test_user(db: Session, test_organization_id: UUID, test_role: Role) -> User:
"""Create a test user with required permissions"""
user = User(
id=uuid4(),
email="test@test.com",
hashed_password=get_password_hash("testpassword"),
organization_id=test_organization_id,
role_id=test_role.id,
is_active=True
)
db.add(user)
db.commit()
db.refresh(user)
return user
@pytest.fixture
def test_customer(db, test_organization_id) -> Customer:
"""Create a test customer"""
customer = Customer(
id=uuid4(),
organization_id=test_organization_id,
email="customer@example.com",
full_name="Test Customer"
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def test_agent(db, test_organization_id) -> Agent:
"""Create a test agent"""
agent = Agent(
id=uuid4(),
organization_id=test_organization_id,
name="Test Agent",
display_name="Test Agent Display",
agent_type=AgentType.CUSTOMER_SUPPORT,
instructions="Test instructions"
)
db.add(agent)
db.commit()
db.refresh(agent)
return agent
@pytest.fixture
def test_session(db: Session, test_organization_id: UUID, test_customer: Customer, test_agent: Agent) -> SessionToAgent:
"""Create a test session"""
session = SessionToAgent(
session_id=uuid4(),
organization_id=test_organization_id,
customer_id=test_customer.id,
agent_id=test_agent.id,
status=SessionStatus.OPEN
)
db.add(session)
db.commit()
# Add a test chat message
chat = ChatHistory(
organization_id=test_organization_id,
customer_id=test_customer.id,
agent_id=test_agent.id,
session_id=session.session_id,
message="Test message",
message_type="agent"
)
db.add(chat)
db.commit()
db.refresh(session)
return session
@pytest.fixture
def session_repo(db):
"""Create a session repository instance"""
return SessionToAgentRepository(db)
def test_create_session(session_repo, test_organization_id, test_customer, test_agent):
"""Test creating a new session"""
session_id = uuid4()
session = session_repo.create_session(
session_id=session_id,
agent_id=test_agent.id,
customer_id=test_customer.id,
organization_id=test_organization_id
)
assert session.session_id == session_id
assert session.agent_id == test_agent.id
assert session.customer_id == test_customer.id
assert session.organization_id == test_organization_id
assert session.status == SessionStatus.OPEN
assert session.user_id is None
def test_get_session(session_repo, test_session):
"""Test retrieving a session by ID"""
retrieved_session = session_repo.get_session(test_session.session_id)
assert retrieved_session.session_id == test_session.session_id
assert retrieved_session.agent_id == test_session.agent_id
assert retrieved_session.customer_id == test_session.customer_id
def test_assign_user(session_repo, test_session, test_user):
"""Test assigning a user to a session"""
success = session_repo.assign_user(test_session.session_id, test_user.id)
assert success is True
# Verify assignment
updated_session = session_repo.get_session(test_session.session_id)
assert updated_session.user_id == test_user.id
assert updated_session.status == SessionStatus.TRANSFERRED
def test_close_session(session_repo, test_session):
"""Test closing a session"""
success = session_repo.close_session(test_session.session_id)
assert success is True
# Verify closure
closed_session = session_repo.get_session(test_session.session_id)
assert closed_session.status == SessionStatus.CLOSED
assert closed_session.closed_at is not None
def test_get_agent_sessions(session_repo, test_agent, test_organization_id, test_customer):
"""Test retrieving all sessions for an agent"""
# Create multiple sessions
session1 = session_repo.create_session(
session_id=uuid4(),
agent_id=test_agent.id,
customer_id=test_customer.id,
organization_id=test_organization_id
)
session2 = session_repo.create_session(
session_id=uuid4(),
agent_id=test_agent.id,
customer_id=test_customer.id,
organization_id=test_organization_id
)
# Get sessions
sessions = session_repo.get_agent_sessions(test_agent.id)
assert len(sessions) == 2
assert all(s.agent_id == test_agent.id for s in sessions)
def test_get_user_sessions(session_repo, test_session, test_user):
"""Test retrieving all sessions assigned to a user"""
# Assign session to user
session_repo.assign_user(test_session.session_id, test_user.id)
# Create and assign another session
session2 = session_repo.create_session(
session_id=uuid4(),
agent_id=test_session.agent_id,
customer_id=test_session.customer_id,
organization_id=test_session.organization_id
)
session_repo.assign_user(session2.session_id, test_user.id)
# Get user sessions
sessions = session_repo.get_user_sessions(test_user.id)
assert len(sessions) == 2
assert all(s.user_id == test_user.id for s in sessions)
def test_get_open_sessions(session_repo, test_session):
"""Test retrieving all open sessions"""
# Create another session and close it
closed_session = session_repo.create_session(
session_id=uuid4(),
agent_id=test_session.agent_id,
customer_id=test_session.customer_id,
organization_id=test_session.organization_id
)
session_repo.close_session(closed_session.session_id)
# Get open sessions
sessions = session_repo.get_open_sessions()
assert len(sessions) == 1
assert sessions[0].session_id == test_session.session_id
assert sessions[0].status == SessionStatus.OPEN
def test_get_customer_sessions(session_repo, test_session, test_customer):
"""Test retrieving all sessions for a customer"""
# Create another session for the same customer
session2 = session_repo.create_session(
session_id=uuid4(),
agent_id=test_session.agent_id,
customer_id=test_customer.id,
organization_id=test_session.organization_id
)
# Get customer sessions
sessions = session_repo.get_customer_sessions(test_customer.id, SessionStatus.OPEN, VERIFIED_SCOPE)
assert len(sessions) == 2
assert all(s[0].customer_id == test_customer.id for s in sessions)
def test_get_active_widget_session(session_repo, test_session, test_customer):
"""Test retrieving the active session a vouched-for widget token resumes"""
# Create another session and close it
closed_session = session_repo.create_session(
session_id=uuid4(),
agent_id=test_session.agent_id,
customer_id=test_customer.id,
organization_id=test_session.organization_id
)
session_repo.close_session(closed_session.session_id)
# Get active session
session = session_repo.get_active_widget_session(test_customer.id, test_session.agent_id, VERIFIED_SCOPE)
assert session is not None
assert session.session_id == test_session.session_id
assert session.status == SessionStatus.OPEN
def test_an_active_widget_session_is_only_found_in_its_own_scope(session_repo, test_session, test_customer):
"""A keyless session belongs to the vouched-for scope, not to any widget key."""
assert session_repo.get_active_widget_session(
test_customer.id, test_session.agent_id, "browser-a") is None
def test_takeover_session(session_repo, test_session, test_user):
"""Test taking over a chat session"""
success = session_repo.takeover_session(str(test_session.session_id), str(test_user.id))
assert success is True
# Verify takeover
updated_session = session_repo.get_session(test_session.session_id)
assert updated_session.user_id == test_user.id
assert updated_session.group_id is None
assert updated_session.status == SessionStatus.OPEN
def test_takeover_nonexistent_session(session_repo, test_user):
"""Test taking over a nonexistent session"""
success = session_repo.takeover_session(str(uuid4()), str(test_user.id))
assert success is False
def test_takeover_already_taken_session(session_repo, test_session, test_user):
"""Test taking over an already taken session"""
# First takeover
session_repo.takeover_session(str(test_session.session_id), str(test_user.id))
# Second takeover attempt
another_user_id = uuid4()
success = session_repo.takeover_session(str(test_session.session_id), str(another_user_id))
assert success is False
+210
View File
@@ -0,0 +1,210 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.repositories.user_group import UserGroupRepository
from app.models.user import User, UserGroup
from uuid import uuid4
from app.models.organization import Organization
@pytest.fixture
def user_group_repo(db):
"""Create a UserGroupRepository instance"""
return UserGroupRepository(db)
@pytest.fixture
def test_user(db, test_organization_id):
"""Create a test user"""
user = User(
id=uuid4(),
email="test@example.com",
hashed_password="hashed_password",
is_active=True,
organization_id=test_organization_id,
full_name="Test User"
)
db.add(user)
db.commit()
db.refresh(user)
return user
@pytest.fixture
def test_group(db, test_organization_id):
"""Create a test group"""
group = UserGroup(
name="Test Group",
description="Test group description",
organization_id=test_organization_id
)
db.add(group)
db.commit()
db.refresh(group)
return group
def test_get_groups_by_organization(user_group_repo, test_organization_id, test_group):
"""Test retrieving groups by organization"""
# Create another organization for testing
other_org = Organization(
name="Other Organization",
domain="other.com",
timezone="UTC"
)
user_group_repo.db.add(other_org)
user_group_repo.db.commit()
user_group_repo.db.refresh(other_org)
# Create another group in a different organization
other_group = UserGroup(
name="Other Group",
description="Other group description",
organization_id=other_org.id
)
user_group_repo.db.add(other_group)
user_group_repo.db.commit()
# Get groups for test organization
groups = user_group_repo.get_groups_by_organization(test_organization_id)
assert len(groups) == 1
assert groups[0].id == test_group.id
assert groups[0].name == test_group.name
# Get groups for other organization
other_groups = user_group_repo.get_groups_by_organization(other_org.id)
assert len(other_groups) == 1
assert other_groups[0].name == other_group.name
assert other_groups[0].organization_id == other_org.id
def test_get_group(user_group_repo, test_group):
"""Test retrieving a group by ID"""
# Get existing group
group = user_group_repo.get_group(test_group.id)
assert group is not None
assert group.id == test_group.id
assert group.name == test_group.name
# Try to get non-existent group
non_existent_group = user_group_repo.get_group(uuid4())
assert non_existent_group is None
def test_create_group(user_group_repo, test_organization_id):
"""Test creating a new group"""
name = "New Group"
description = "New group description"
group = user_group_repo.create_group(name, description, test_organization_id)
assert group is not None
assert group.name == name
assert group.description == description
assert group.organization_id == test_organization_id
# Verify group was saved to database
saved_group = user_group_repo.get_group(group.id)
assert saved_group is not None
assert saved_group.name == name
def test_update_group(user_group_repo, test_group):
"""Test updating a group"""
new_name = "Updated Group"
new_description = "Updated description"
# Update existing group
updated_group = user_group_repo.update_group(
test_group.id,
name=new_name,
description=new_description
)
assert updated_group is not None
assert updated_group.name == new_name
assert updated_group.description == new_description
# Try to update non-existent group
non_existent_update = user_group_repo.update_group(
uuid4(),
name="Non-existent Group"
)
assert non_existent_update is None
def test_delete_group(user_group_repo, test_group):
"""Test deleting a group"""
# Delete existing group
success = user_group_repo.delete_group(test_group.id)
assert success is True
# Verify group was deleted
deleted_group = user_group_repo.get_group(test_group.id)
assert deleted_group is None
# Try to delete non-existent group
success = user_group_repo.delete_group(uuid4())
assert success is False
def test_add_user(user_group_repo, test_group, test_user):
"""Test adding a user to a group"""
# Add user to group
success = user_group_repo.add_user(test_group.id, test_user.id)
assert success is True
# Verify user was added
group = user_group_repo.get_group(test_group.id)
assert test_user in group.users
# Try to add user again (should still return True but not duplicate)
success = user_group_repo.add_user(test_group.id, test_user.id)
assert success is True
assert len(group.users) == 1
# Try to add user to non-existent group
success = user_group_repo.add_user(uuid4(), test_user.id)
assert success is False
def test_remove_user(user_group_repo, test_group, test_user):
"""Test removing a user from a group"""
# First add user to group
user_group_repo.add_user(test_group.id, test_user.id)
# Remove user from group
success = user_group_repo.remove_user(test_group.id, test_user.id)
assert success is True
# Verify user was removed
group = user_group_repo.get_group(test_group.id)
assert test_user not in group.users
# Try to remove user again (should still return True)
success = user_group_repo.remove_user(test_group.id, test_user.id)
assert success is True
# Try to remove user from non-existent group
success = user_group_repo.remove_user(uuid4(), test_user.id)
assert success is False
def test_get_user_groups(user_group_repo, test_group, test_user):
"""Test getting all groups a user belongs to"""
# Initially user should have no groups
groups = user_group_repo.get_user_groups(test_user.id)
assert len(groups) == 0
# Add user to a group
user_group_repo.add_user(test_group.id, test_user.id)
# User should now have one group
groups = user_group_repo.get_user_groups(test_user.id)
assert len(groups) == 1
assert groups[0].id == test_group.id
# Test with non-existent user
groups = user_group_repo.get_user_groups(uuid4())
assert len(groups) == 0
+183
View File
@@ -0,0 +1,183 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from app.models.user import User
from app.repositories.user import UserRepository
from uuid import UUID, uuid4
from app.models.organization import Organization
@pytest.fixture
def user_repo(db):
"""Create a user repository instance"""
return UserRepository(db)
def test_create_user(user_repo, test_organization_id):
"""Test creating a new user"""
email = "test@example.com"
password = "testpassword"
full_name = "Test User"
hashed_password = User.get_password_hash(password)
user = user_repo.create_user(
email=email,
hashed_password=hashed_password,
full_name=full_name,
organization_id=test_organization_id
)
assert user.email == email
assert user.full_name == full_name
assert user.organization_id == test_organization_id
assert user.verify_password(password) # Verify the password works
def test_get_user_by_email(user_repo, test_organization_id):
"""Test retrieving a user by email"""
# Create test user
email = "test@example.com"
hashed_password = User.get_password_hash("testpassword")
user = user_repo.create_user(
email=email,
hashed_password=hashed_password,
full_name="Test User",
organization_id=test_organization_id
)
# Retrieve user
retrieved_user = user_repo.get_user_by_email(email)
assert retrieved_user.id == user.id
assert retrieved_user.email == email
def test_get_user_by_id(user_repo, test_organization_id):
"""Test retrieving a user by ID"""
# Create test user
hashed_password = User.get_password_hash("testpassword")
user = user_repo.create_user(
email="test@example.com",
hashed_password=hashed_password,
full_name="Test User",
organization_id=test_organization_id
)
# Retrieve user
retrieved_user = user_repo.get_user(user.id)
assert retrieved_user.id == user.id
assert retrieved_user.email == user.email
def test_get_users_by_organization(user_repo, test_organization_id):
"""Test retrieving all users for an organization"""
# Create multiple users
hashed_password = User.get_password_hash("testpassword")
user1 = user_repo.create_user(
email="user1@example.com",
hashed_password=hashed_password,
full_name="User One",
organization_id=test_organization_id
)
user2 = user_repo.create_user(
email="user2@example.com",
hashed_password=hashed_password,
full_name="User Two",
organization_id=test_organization_id
)
# Create another organization for testing
other_org = Organization(
name="Other Organization",
domain="other.com",
timezone="UTC"
)
user_repo.db.add(other_org)
user_repo.db.commit()
user_repo.db.refresh(other_org)
# Create user in different organization
user_repo.create_user(
email="other@example.com",
hashed_password=hashed_password,
full_name="Other User",
organization_id=other_org.id
)
# Get users for test organization
users = user_repo.get_users_by_organization(test_organization_id)
assert len(users) == 2
assert all(u.organization_id == test_organization_id for u in users)
def test_update_user(user_repo, test_organization_id):
"""Test updating a user"""
# Create test user
hashed_password = User.get_password_hash("testpassword")
user = user_repo.create_user(
email="test@example.com",
hashed_password=hashed_password,
full_name="Test User",
organization_id=test_organization_id
)
# Update user
new_name = "Updated User"
updated_user = user_repo.update_user(
user_id=str(user.id),
full_name=new_name
)
assert updated_user.full_name == new_name
assert updated_user.id == user.id
def test_authenticate_user(user_repo, test_organization_id):
"""Test user authentication"""
email = "test@example.com"
password = "testpassword"
hashed_password = User.get_password_hash(password)
# Create test user
user = user_repo.create_user(
email=email,
hashed_password=hashed_password,
full_name="Test User",
organization_id=test_organization_id
)
# Test valid password
assert user.verify_password(password)
# Test invalid password
assert not user.verify_password("wrongpassword")
def test_get_multi(user_repo, test_organization_id):
"""Test retrieving multiple users with pagination"""
# Create multiple users
users = []
hashed_password = User.get_password_hash("testpassword")
for i in range(5):
user = user_repo.create_user(
email=f"user{i}@example.com",
hashed_password=hashed_password,
full_name=f"User {i}",
organization_id=test_organization_id
)
users.append(user)
# Test pagination
first_page = user_repo.get_users_by_organization(test_organization_id)[:2]
assert len(first_page) == 2
second_page = user_repo.get_users_by_organization(test_organization_id)[2:4]
assert len(second_page) == 2
last_page = user_repo.get_users_by_organization(test_organization_id)[4:]
assert len(last_page) == 1
+183
View File
@@ -0,0 +1,183 @@
"""
Copyright 2024-2026 ChatterMate
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models.widget import Widget
from app.models.schemas.widget import WidgetCreate
from app.repositories.widget import WidgetRepository
from app.models.agent import Agent, AgentType
from app.models.organization import Organization
from uuid import UUID, uuid4
# Test database URL
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
@pytest.fixture(scope="function")
def engine():
"""Create a test database engine"""
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
Base.metadata.create_all(bind=engine)
yield engine
Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def db_session(engine):
"""Create a test database session"""
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = TestingSessionLocal()
try:
yield session
finally:
session.close()
@pytest.fixture
def test_organization_id(db):
"""Create a test organization and return its ID"""
organization = Organization(
name="Test Organization",
domain="test.com",
timezone="UTC"
)
db.add(organization)
db.commit()
db.refresh(organization)
return organization.id
@pytest.fixture
def test_agent(db, test_organization_id):
"""Create a test agent"""
agent = Agent(
name="Test Agent",
organization_id=test_organization_id,
description="Test agent description",
agent_type=AgentType.GENERAL,
_instructions="Test instructions",
is_active=True
)
db.add(agent)
db.commit()
db.refresh(agent)
return agent
@pytest.fixture
def test_widget(db, test_organization_id, test_agent):
"""Create a test widget"""
widget_data = WidgetCreate(
name="Test Widget",
agent_id=test_agent.id
)
widget_repo = WidgetRepository(db)
widget = widget_repo.create_widget(widget_data, test_organization_id)
return widget
def test_create_widget(db, test_organization_id, test_agent):
"""Test creating a new widget"""
widget_data = WidgetCreate(
name="New Widget",
agent_id=test_agent.id
)
widget_repo = WidgetRepository(db)
widget = widget_repo.create_widget(widget_data, test_organization_id)
assert widget is not None
assert widget.name == widget_data.name
assert widget.agent_id == test_agent.id
assert widget.organization_id == test_organization_id
# Verify widget was saved to database
saved_widget = widget_repo.get_widget(str(widget.id))
assert saved_widget is not None
assert saved_widget.name == widget_data.name
def test_get_widget(db, test_widget):
"""Test retrieving a widget by ID"""
# Get existing widget
widget_repo = WidgetRepository(db)
widget = widget_repo.get_widget(str(test_widget.id))
assert widget is not None
assert widget.id == test_widget.id
assert widget.name == test_widget.name
assert widget.agent_id == test_widget.agent_id
assert widget.organization_id == test_widget.organization_id
# Try to get non-existent widget
non_existent_widget = widget_repo.get_widget(str(uuid4()))
assert non_existent_widget is None
def test_get_widgets(db, test_organization_id, test_widget):
"""Test retrieving widgets by organization"""
# Create another organization for testing
other_org = Organization(
name="Other Organization",
domain="other.com",
timezone="UTC"
)
db.add(other_org)
db.commit()
db.refresh(other_org)
# Create an agent for the other organization
other_agent = Agent(
name="Other Agent",
organization_id=other_org.id,
description="Other agent description",
agent_type=AgentType.GENERAL,
_instructions="Other instructions",
is_active=True
)
db.add(other_agent)
db.commit()
db.refresh(other_agent)
# Create another widget in a different organization
other_widget_data = WidgetCreate(
name="Other Widget",
agent_id=other_agent.id
)
widget_repo = WidgetRepository(db)
widget_repo.create_widget(other_widget_data, other_org.id)
# Get widgets for test organization
widgets = widget_repo.get_widgets(test_organization_id)
assert len(widgets) == 1
assert widgets[0].id == test_widget.id
assert widgets[0].name == test_widget.name
assert widgets[0].organization_id == test_organization_id
# Get widgets for other organization
other_widgets = widget_repo.get_widgets(other_org.id)
assert len(other_widgets) == 1
assert other_widgets[0].name == other_widget_data.name
assert other_widgets[0].organization_id == other_org.id
def test_delete_widget(db, test_widget):
"""Test deleting a widget"""
# Delete existing widget
widget_repo = WidgetRepository(db)
widget_repo.delete_widget(str(test_widget.id))
# Verify widget was deleted
deleted_widget = widget_repo.get_widget(str(test_widget.id))
assert deleted_widget is None
# Try to delete non-existent widget (should not raise error)
widget_repo.delete_widget(str(uuid4()))