108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
from unittest.mock import MagicMock, patch
|
|
import os
|
|
import sys
|
|
|
|
# Add backend to sys.path to resolve imports
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), "../.."))
|
|
|
|
from backend.app import app, get_db, authenticate_user
|
|
from backend.database import Base
|
|
|
|
# --- Test Database Setup ---
|
|
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
|
|
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
def override_get_db():
|
|
try:
|
|
db = TestingSessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
# Override Auth to bypass password check for tests
|
|
def override_authenticate_user():
|
|
return "test_user"
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
app.dependency_overrides[authenticate_user] = override_authenticate_user
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
# Create tables
|
|
Base.metadata.create_all(bind=engine)
|
|
with TestClient(app) as c:
|
|
yield c
|
|
# Drop tables
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
# --- Tests ---
|
|
|
|
def test_health_check(client):
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
@patch("backend.app.run_discovery_task") # Mock the background task function directly
|
|
def test_provision_new_company(mock_discovery_task, client):
|
|
# Data simulating a webhook from Connector
|
|
payload = {
|
|
"so_contact_id": 12345,
|
|
"so_person_id": 999,
|
|
"crm_name": "Test Firma GmbH",
|
|
"crm_website": "www.test-firma.de",
|
|
"job_title": "CTO",
|
|
"campaign_tag": "standard"
|
|
}
|
|
|
|
# Call the API
|
|
response = client.post("/api/provision/superoffice-contact", json=payload)
|
|
|
|
# Assertions
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["company_name"] == "Test Firma GmbH"
|
|
|
|
# Verify Background Task was triggered
|
|
# Note: FastAPI TestClient runs background tasks synchronously by default!
|
|
# So patching the function in app.py is the right way to intercept it.
|
|
mock_discovery_task.assert_called()
|
|
|
|
def test_create_contact_manual(client):
|
|
# First create a company to attach contact to
|
|
# We can use the DB session directly or the API
|
|
from backend.database import Company
|
|
db = TestingSessionLocal()
|
|
company = Company(name="Manual Corp", status="NEW")
|
|
db.add(company)
|
|
db.commit()
|
|
db.refresh(company)
|
|
company_id = company.id
|
|
db.close()
|
|
|
|
payload = {
|
|
"company_id": company_id,
|
|
"first_name": "Hans",
|
|
"last_name": "Tester",
|
|
"email": "hans@manual.corp",
|
|
"job_title": "CEO",
|
|
"role": "Wirtschaftlicher Entscheider",
|
|
"is_primary": True
|
|
}
|
|
|
|
response = client.post("/api/contacts", json=payload)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["email"] == "hans@manual.corp"
|
|
assert data["role"] == "Wirtschaftlicher Entscheider"
|