- API returns {"vendors": [...]} not a direct list
- Update test to access vendors_response["vendors"] correctly
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
# tests/integration/workflows/test_integration.py
|
|
"""
|
|
End-to-end workflow integration tests.
|
|
|
|
Tests complete workflows using actual API endpoints:
|
|
- Admin vendor management workflow
|
|
- Admin product listing workflow
|
|
- Marketplace import workflow
|
|
"""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.e2e
|
|
class TestIntegrationFlows:
|
|
def test_admin_vendor_workflow(self, client, admin_headers, test_company):
|
|
"""Test admin vendor creation and management workflow"""
|
|
unique_id = str(uuid.uuid4())[:8]
|
|
|
|
# 1. Create a vendor
|
|
vendor_data = {
|
|
"company_id": test_company.id,
|
|
"vendor_code": f"FLOW_{unique_id.upper()}",
|
|
"subdomain": f"flow{unique_id}",
|
|
"name": "Integration Flow Vendor",
|
|
"description": "Test vendor for integration",
|
|
}
|
|
|
|
response = client.post(
|
|
"/api/v1/admin/vendors", headers=admin_headers, json=vendor_data
|
|
)
|
|
assert response.status_code == 200
|
|
vendor = response.json()
|
|
|
|
# 2. Get vendor details
|
|
response = client.get(
|
|
f"/api/v1/admin/vendors/{vendor['vendor_code']}", headers=admin_headers
|
|
)
|
|
assert response.status_code == 200
|
|
vendor_detail = response.json()
|
|
assert vendor_detail["name"] == "Integration Flow Vendor"
|
|
|
|
# 3. List all vendors
|
|
response = client.get("/api/v1/admin/vendors", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
vendors_response = response.json()
|
|
assert any(
|
|
v["vendor_code"] == vendor["vendor_code"]
|
|
for v in vendors_response["vendors"]
|
|
)
|
|
|
|
def test_admin_product_listing_workflow(
|
|
self, client, admin_headers, test_marketplace_product
|
|
):
|
|
"""Test admin product listing and search workflow"""
|
|
# 1. List all products
|
|
response = client.get("/api/v1/admin/products", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "products" in data or "items" in data or isinstance(data, list)
|
|
|
|
# 2. Get specific product
|
|
response = client.get(
|
|
f"/api/v1/admin/products/{test_marketplace_product.id}",
|
|
headers=admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# 3. Search for product (if search is supported)
|
|
response = client.get(
|
|
"/api/v1/admin/products?search=test",
|
|
headers=admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_admin_import_jobs_workflow(self, client, admin_headers):
|
|
"""Test admin import job listing workflow"""
|
|
# 1. List all import jobs
|
|
response = client.get(
|
|
"/api/v1/admin/marketplace-import-jobs", headers=admin_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# Should return jobs list (may be empty)
|
|
assert "jobs" in data or "items" in data or isinstance(data, list)
|
|
|
|
def test_admin_user_management_workflow(self, client, admin_headers, test_admin):
|
|
"""Test admin user management workflow"""
|
|
# 1. List all users
|
|
response = client.get("/api/v1/admin/users", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
|
|
# 2. Get specific user details
|
|
response = client.get(
|
|
f"/api/v1/admin/users/{test_admin.id}", headers=admin_headers
|
|
)
|
|
assert response.status_code == 200
|
|
user_detail = response.json()
|
|
assert user_detail["role"] == "admin"
|