Maxim Snesarev
33d6bb7ebd
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings. - Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure. - Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic. - Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks. - Improved logging and error handling across the application to facilitate better monitoring and debugging. - Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
259 lines
8.3 KiB
Python
259 lines
8.3 KiB
Python
"""
|
|
Test API endpoints with AMO CRM data examples
|
|
"""
|
|
import pytest
|
|
from tests.fixtures.amocrm_responses import USERS_RESPONSE, DEALS_RESPONSE
|
|
|
|
|
|
def test_root_endpoint(test_client):
|
|
"""Test root endpoint"""
|
|
response = test_client.get("/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "AMO CRM Data Collection Service"
|
|
assert data["version"] == "0.1.0"
|
|
|
|
|
|
def test_health_endpoint(test_client):
|
|
"""Test health check endpoint"""
|
|
response = test_client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
|
|
|
|
def test_list_entities_empty(test_client):
|
|
"""Test listing entities when database is empty"""
|
|
response = test_client.get("/api/v1/entities/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "entities" in data
|
|
|
|
# All counts should be 0
|
|
for entity_type in ["deals", "contacts", "companies", "pipelines", "users", "events"]:
|
|
assert entity_type in data["entities"]
|
|
assert data["entities"][entity_type]["count"] == 0
|
|
|
|
|
|
def test_put_users_data(test_client):
|
|
"""Test putting users data"""
|
|
users_data = USERS_RESPONSE["_embedded"]["users"]
|
|
|
|
response = test_client.post(
|
|
"/api/v1/data/users",
|
|
json={
|
|
"data": users_data,
|
|
"sync_mode": "upsert"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["processed_count"] == 2
|
|
assert "Processed 2 users" in data["message"]
|
|
|
|
|
|
def test_put_deals_data(test_client):
|
|
"""Test putting deals data"""
|
|
deals_data = DEALS_RESPONSE["_embedded"]["leads"]
|
|
|
|
response = test_client.post(
|
|
"/api/v1/data/deals",
|
|
json={
|
|
"data": deals_data,
|
|
"sync_mode": "upsert"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["processed_count"] == 2
|
|
assert "Processed 2 deals" in data["message"]
|
|
|
|
|
|
def test_list_entities_with_data(test_client):
|
|
"""Test listing entities after adding data"""
|
|
# First add the data
|
|
users_data = USERS_RESPONSE["_embedded"]["users"]
|
|
test_client.post("/api/v1/data/users", json={"data": users_data, "sync_mode": "upsert"})
|
|
|
|
deals_data = DEALS_RESPONSE["_embedded"]["leads"]
|
|
test_client.post("/api/v1/data/deals", json={"data": deals_data, "sync_mode": "upsert"})
|
|
|
|
response = test_client.get("/api/v1/entities/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Should have users and deals now
|
|
assert data["entities"]["users"]["count"] == 2
|
|
assert data["entities"]["deals"]["count"] == 2
|
|
|
|
|
|
def test_list_deal_fields(test_client):
|
|
"""Test listing deal fields"""
|
|
# First add some deals data
|
|
deals_data = DEALS_RESPONSE["_embedded"]["leads"]
|
|
test_client.post("/api/v1/data/deals", json={"data": deals_data, "sync_mode": "upsert"})
|
|
|
|
response = test_client.get("/api/v1/entities/deals/fields")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
assert data["entity_type"] == "deals"
|
|
assert "fields" in data
|
|
|
|
# Should have standard fields
|
|
field_names = [f["name"] for f in data["fields"]]
|
|
assert "name" in field_names
|
|
assert "price" in field_names
|
|
assert "created_at" in field_names
|
|
|
|
# Should have custom fields from the test data
|
|
custom_fields = [f for f in data["fields"] if f["is_custom"]]
|
|
assert len(custom_fields) > 0
|
|
|
|
|
|
def test_export_configuration(test_client):
|
|
"""Test creating export configuration"""
|
|
config_data = {
|
|
"name": "Test Export Configuration",
|
|
"sheet_id": "test-sheet-id-123",
|
|
"date_range_start": "2024-01-01T00:00:00Z",
|
|
"date_range_end": "2024-12-31T23:59:59Z",
|
|
"entity_mappings": {
|
|
"deals": {
|
|
"sheet_name": "Deals",
|
|
"is_enabled": True,
|
|
"field_mapping": [
|
|
{"field_name": "name", "column": "A", "order": 1},
|
|
{"field_name": "price", "column": "B", "order": 2}
|
|
]
|
|
},
|
|
"users": {
|
|
"sheet_name": "Users",
|
|
"is_enabled": True,
|
|
"field_mapping": [
|
|
{"field_name": "name", "column": "A", "order": 1},
|
|
{"field_name": "email", "column": "B", "order": 2}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
|
|
response = test_client.post("/api/v1/export/configure", json=config_data)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "configuration_id" in data
|
|
assert data["message"] == "Export configuration created successfully"
|
|
|
|
return data["configuration_id"]
|
|
|
|
|
|
def test_list_export_configurations(test_client):
|
|
"""Test listing export configurations"""
|
|
# First create a configuration
|
|
config_data = {
|
|
"name": "Test Export Configuration",
|
|
"sheet_id": "test-sheet-id-123",
|
|
"date_range_start": "2024-01-01T00:00:00Z",
|
|
"date_range_end": "2024-12-31T23:59:59Z",
|
|
"entity_mappings": {
|
|
"deals": {
|
|
"sheet_name": "Deals",
|
|
"is_enabled": True,
|
|
"field_mapping": [
|
|
{"field_name": "name", "column": "A", "order": 1},
|
|
{"field_name": "price", "column": "B", "order": 2}
|
|
]
|
|
},
|
|
"users": {
|
|
"sheet_name": "Users",
|
|
"is_enabled": True,
|
|
"field_mapping": [
|
|
{"field_name": "name", "column": "A", "order": 1},
|
|
{"field_name": "email", "column": "B", "order": 2}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
test_client.post("/api/v1/export/configure", json=config_data)
|
|
|
|
response = test_client.get("/api/v1/export/configurations")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
assert "configurations" in data
|
|
assert len(data["configurations"]) >= 1
|
|
|
|
config = data["configurations"][0]
|
|
assert config["name"] == "Test Export Configuration"
|
|
assert config["sheet_id"] == "test-sheet-id-123"
|
|
assert "entity_mappings" in config
|
|
|
|
|
|
@pytest.mark.skip(reason="Broker connection not initialized in test environment")
|
|
def test_start_export_job(test_client):
|
|
"""Test starting export job"""
|
|
# First create a configuration
|
|
config_data = {
|
|
"name": "Test Export Configuration",
|
|
"sheet_id": "test-sheet-id-123",
|
|
"date_range_start": "2024-01-01T00:00:00Z",
|
|
"date_range_end": "2024-12-31T23:59:59Z",
|
|
"entity_mappings": {
|
|
"deals": {
|
|
"sheet_name": "Deals",
|
|
"is_enabled": True,
|
|
"field_mapping": [
|
|
{"field_name": "name", "column": "A", "order": 1},
|
|
{"field_name": "price", "column": "B", "order": 2}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
result = test_client.post("/api/v1/export/configure", json=config_data)
|
|
config_id = result.json()["configuration_id"]
|
|
|
|
response = test_client.post(
|
|
"/api/v1/export/start",
|
|
json={"configuration_id": config_id}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "job_id" in data
|
|
assert data["status"] == "pending"
|
|
assert data["message"] == "Export job queued successfully"
|
|
|
|
return data["job_id"]
|
|
|
|
|
|
@pytest.mark.skip(reason="Broker connection not initialized in test environment")
|
|
def test_export_job_status(test_client):
|
|
"""Test getting export job status"""
|
|
# This test depends on test_start_export_job which requires broker
|
|
response = test_client.get("/api/v1/export/status/test-job-id")
|
|
# Just verify the endpoint structure
|
|
assert response.status_code in [200, 404]
|
|
|
|
|
|
@pytest.mark.skip(reason="Broker connection not initialized in test environment")
|
|
def test_list_export_jobs(test_client):
|
|
"""Test listing export jobs"""
|
|
response = test_client.get("/api/v1/export/jobs")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "jobs" in data
|
|
|
|
|
|
def test_invalid_entity_type(test_client):
|
|
"""Test invalid entity type"""
|
|
response = test_client.get("/api/v1/entities/invalid_entity/fields")
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "Entity type not found" in data["detail"]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__])
|