sass-imobiliaria/backend/tests/test_contact_flow.py
MatheusAlves96 cf5603243c
Some checks failed
CI/CD → Deploy via SSH / Build & Push Docker Images (push) Successful in 1m0s
CI/CD → Deploy via SSH / Deploy via SSH (push) Successful in 4m35s
CI/CD → Deploy via SSH / Validate HTTPS & Endpoints (push) Failing after 46s
feat: features 025-032 - favoritos, contatos, trabalhe-conosco, area-cliente, navbar, hero-light-dark, performance-homepage
- feat(025): favoritos locais com FavoritesContext, HeartButton, PublicFavoritesPage
- feat(026): central de contatos admin (leads/contatos unificados)
- feat(027): configuração da página de contato via admin
- feat(028): trabalhe conosco - candidaturas com upload e admin
- feat(029): UX área do cliente - visitas, comparação, perfil
- feat(030): navbar UX - menu mobile, ThemeToggle, useFavorites
- feat(031): hero light/dark - imagens separadas por tema, upload, preview, seed
- feat(032): performance homepage - Promise.all parallel fetches, sessionStorage cache,
  preload hero image, loading=lazy nos cards, useInView hook, will-change carrossel,
  keyframes em index.css, AgentsCarousel e HomeScrollScene via props
- fix: light mode HomeScrollScene - gradiente, cores de texto, scroll hint

migrations: g1h2i3j4k5l6 (source em leads), h1i2j3k4l5m6 (contact_config),
            i1j2k3l4m5n6 (job_applications), j2k3l4m5n6o7 (hero theme images)
2026-04-22 22:35:17 -03:00

420 lines
18 KiB
Python

"""
Testes de integração — fluxo completo de contato.
Cobre os três caminhos de submissão de lead:
1. POST /api/v1/contact (contato geral)
2. POST /api/v1/properties/<slug>/contact (contato de imóvel)
3. POST /api/v1/contact com source=cadastro_residencia
Além da visualização admin via:
- GET /api/v1/admin/leads
- GET /api/v1/admin/leads?source=<origem>
"""
import uuid
import jwt
from datetime import datetime, timedelta, timezone
import bcrypt
import pytest
from app.extensions import db as _db
from app.models.lead import ContactLead
from app.models.property import Property
from app.models.user import ClientUser
# ──────────────────────────────────────────────────────────────────────────────
# Helpers
# ──────────────────────────────────────────────────────────────────────────────
_TEST_JWT_SECRET = "test-secret-key"
_VALID_CONTACT = {
"name": "João da Silva",
"email": "joao@example.com",
"phone": "(11) 91234-5678",
"message": "Tenho interesse em anunciar meu imóvel.",
}
def _make_admin_token(user_id: str) -> str:
payload = {
"sub": user_id,
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
"iat": datetime.now(timezone.utc),
}
return jwt.encode(payload, _TEST_JWT_SECRET, algorithm="HS256")
def _admin_headers(user_id: str) -> dict:
return {"Authorization": f"Bearer {_make_admin_token(user_id)}"}
def _make_property(slug: str) -> Property:
return Property(
id=uuid.uuid4(),
title=f"Apartamento {slug}",
slug=slug,
address="Rua Teste, 10",
price="350000.00",
type="venda",
bedrooms=2,
bathrooms=1,
area_m2=70,
is_featured=False,
is_active=True,
)
def _make_admin_user(db) -> ClientUser:
pwd_hash = bcrypt.hashpw(b"admin123", bcrypt.gensalt()).decode()
admin = ClientUser(
name="Admin Teste",
email="admin@test.com",
password_hash=pwd_hash,
role="admin",
)
db.session.add(admin)
db.session.flush()
return admin
# ──────────────────────────────────────────────────────────────────────────────
# Fixtures
# ──────────────────────────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def _patch_jwt_secret(app):
"""Garante que os testes usem o mesmo secret para gerar e validar tokens."""
app.config["JWT_SECRET_KEY"] = _TEST_JWT_SECRET
# ──────────────────────────────────────────────────────────────────────────────
# 1. POST /api/v1/contact — contato geral
# ──────────────────────────────────────────────────────────────────────────────
class TestContactGeneral:
def test_returns_201_with_id(self, client):
res = client.post("/api/v1/contact", json=_VALID_CONTACT)
assert res.status_code == 201
body = res.get_json()
assert "id" in body
assert body["message"] == "Mensagem enviada com sucesso!"
def test_lead_persisted_in_db(self, client, db):
client.post("/api/v1/contact", json=_VALID_CONTACT)
lead = db.session.query(ContactLead).filter_by(email="joao@example.com").first()
assert lead is not None
assert lead.name == "João da Silva"
assert lead.source == "contato"
def test_source_defaults_to_contato_when_absent(self, client, db):
payload = {**_VALID_CONTACT, "email": "sem_source@example.com"}
client.post("/api/v1/contact", json=payload)
lead = db.session.query(ContactLead).filter_by(email="sem_source@example.com").first()
assert lead.source == "contato"
def test_explicit_source_contato_is_preserved(self, client, db):
payload = {**_VALID_CONTACT, "email": "src_contato@example.com", "source": "contato"}
client.post("/api/v1/contact", json=payload)
lead = db.session.query(ContactLead).filter_by(email="src_contato@example.com").first()
assert lead.source == "contato"
def test_unknown_source_falls_back_to_contato(self, client, db):
payload = {**_VALID_CONTACT, "email": "bad_source@example.com", "source": "spam"}
client.post("/api/v1/contact", json=payload)
lead = db.session.query(ContactLead).filter_by(email="bad_source@example.com").first()
assert lead.source == "contato"
def test_missing_name_returns_422(self, client):
payload = {k: v for k, v in _VALID_CONTACT.items() if k != "name"}
res = client.post("/api/v1/contact", json=payload)
assert res.status_code == 422
def test_invalid_email_returns_422(self, client):
payload = {**_VALID_CONTACT, "email": "not-an-email"}
res = client.post("/api/v1/contact", json=payload)
assert res.status_code == 422
def test_missing_message_returns_422(self, client):
payload = {k: v for k, v in _VALID_CONTACT.items() if k != "message"}
res = client.post("/api/v1/contact", json=payload)
assert res.status_code == 422
# ──────────────────────────────────────────────────────────────────────────────
# 2. POST /api/v1/contact com source=cadastro_residencia
# ──────────────────────────────────────────────────────────────────────────────
class TestContactCadastroResidencia:
def test_source_cadastro_residencia_is_saved(self, client, db):
payload = {
**_VALID_CONTACT,
"email": "captacao@example.com",
"source": "cadastro_residencia",
"source_detail": "Apartamento",
}
res = client.post("/api/v1/contact", json=payload)
assert res.status_code == 201
lead = db.session.query(ContactLead).filter_by(email="captacao@example.com").first()
assert lead is not None
assert lead.source == "cadastro_residencia"
assert lead.source_detail == "Apartamento"
assert lead.property_id is None
# ──────────────────────────────────────────────────────────────────────────────
# 3. POST /api/v1/properties/<slug>/contact — contato de imóvel
# ──────────────────────────────────────────────────────────────────────────────
class TestContactProperty:
def test_returns_201_and_persists_lead(self, client, db):
prop = _make_property("apto-centro-test")
db.session.add(prop)
db.session.commit()
res = client.post("/api/v1/properties/apto-centro-test/contact", json=_VALID_CONTACT)
assert res.status_code == 201
lead = db.session.query(ContactLead).filter_by(email="joao@example.com").first()
assert lead is not None
assert lead.property_id == prop.id
assert lead.source == "imovel"
assert lead.source_detail == prop.title
def test_returns_404_for_unknown_slug(self, client):
res = client.post("/api/v1/properties/slug-inexistente/contact", json=_VALID_CONTACT)
assert res.status_code == 404
def test_inactive_property_returns_404(self, client, db):
prop = _make_property("apto-inativo")
prop.is_active = False
db.session.add(prop)
db.session.commit()
res = client.post("/api/v1/properties/apto-inativo/contact", json=_VALID_CONTACT)
assert res.status_code == 404
# ──────────────────────────────────────────────────────────────────────────────
# 4. GET /api/v1/admin/leads — visualização admin
# ──────────────────────────────────────────────────────────────────────────────
class TestAdminLeads:
def test_requires_authentication(self, client):
res = client.get("/api/v1/admin/leads")
assert res.status_code == 401
def test_non_admin_user_gets_403(self, client, db):
pwd_hash = bcrypt.hashpw(b"user123", bcrypt.gensalt()).decode()
user = ClientUser(
name="Usuário Comum",
email="comum@test.com",
password_hash=pwd_hash,
role="client",
)
db.session.add(user)
db.session.flush()
res = client.get(
"/api/v1/admin/leads",
headers=_admin_headers(user.id),
)
assert res.status_code == 403
def test_admin_sees_all_leads(self, client, db):
admin = _make_admin_user(db)
# Cria 3 leads via API para garantir persistência realista
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "a1@ex.com", "source": "contato"})
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "a2@ex.com", "source": "cadastro_residencia"})
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "a3@ex.com", "source": "cadastro_residencia"})
res = client.get("/api/v1/admin/leads", headers=_admin_headers(admin.id))
assert res.status_code == 200
body = res.get_json()
assert "items" in body
assert "total" in body
assert body["total"] >= 3
def test_response_has_required_fields(self, client, db):
admin = _make_admin_user(db)
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "fields@ex.com"})
res = client.get("/api/v1/admin/leads", headers=_admin_headers(admin.id))
item = res.get_json()["items"][0]
for field in ("id", "name", "email", "phone", "message", "source", "source_detail", "created_at"):
assert field in item, f"Campo ausente na resposta: {field}"
def test_filter_by_source_contato(self, client, db):
admin = _make_admin_user(db)
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "f_contato@ex.com", "source": "contato"})
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "f_captacao@ex.com", "source": "cadastro_residencia"})
res = client.get("/api/v1/admin/leads?source=contato", headers=_admin_headers(admin.id))
body = res.get_json()
sources = {item["source"] for item in body["items"]}
assert sources == {"contato"}
def test_filter_by_source_cadastro_residencia(self, client, db):
admin = _make_admin_user(db)
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "cr1@ex.com", "source": "cadastro_residencia"})
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "cr2@ex.com", "source": "contato"})
res = client.get("/api/v1/admin/leads?source=cadastro_residencia", headers=_admin_headers(admin.id))
body = res.get_json()
assert all(item["source"] == "cadastro_residencia" for item in body["items"])
def test_filter_by_source_imovel(self, client, db):
admin = _make_admin_user(db)
prop = _make_property("apto-filtro-imovel")
db.session.add(prop)
db.session.commit()
client.post("/api/v1/properties/apto-filtro-imovel/contact", json={**_VALID_CONTACT, "email": "imovel_f@ex.com"})
client.post("/api/v1/contact", json={**_VALID_CONTACT, "email": "contato_f@ex.com", "source": "contato"})
res = client.get("/api/v1/admin/leads?source=imovel", headers=_admin_headers(admin.id))
body = res.get_json()
assert all(item["source"] == "imovel" for item in body["items"])
def test_pagination_defaults(self, client, db):
admin = _make_admin_user(db)
res = client.get("/api/v1/admin/leads", headers=_admin_headers(admin.id))
body = res.get_json()
assert body["page"] == 1
assert body["per_page"] == 20
assert "pages" in body
def test_pagination_page_2(self, client, db):
admin = _make_admin_user(db)
# Cria 25 leads
for i in range(25):
client.post(
"/api/v1/contact",
json={**_VALID_CONTACT, "email": f"page_test_{i}@ex.com"},
)
res_p1 = client.get("/api/v1/admin/leads?per_page=10&page=1", headers=_admin_headers(admin.id))
res_p2 = client.get("/api/v1/admin/leads?per_page=10&page=2", headers=_admin_headers(admin.id))
assert len(res_p1.get_json()["items"]) == 10
assert len(res_p2.get_json()["items"]) == 10
ids_p1 = {item["id"] for item in res_p1.get_json()["items"]}
ids_p2 = {item["id"] for item in res_p2.get_json()["items"]}
assert ids_p1.isdisjoint(ids_p2), "Páginas não devem ter leads em comum"
# ──────────────────────────────────────────────────────────────────────────────
# 5. Fluxo end-to-end completo
# ──────────────────────────────────────────────────────────────────────────────
class TestEndToEndContactFlow:
"""Valida o caminho completo: submissão pública → visualização admin."""
def test_property_contact_appears_in_admin_with_correct_metadata(self, client, db):
admin = _make_admin_user(db)
prop = _make_property("apto-e2e-flow")
db.session.add(prop)
db.session.commit()
# 1. Cliente envia contato pelo imóvel
submit_res = client.post(
"/api/v1/properties/apto-e2e-flow/contact",
json={
"name": "Maria Souza",
"email": "maria@example.com",
"phone": "(21) 98765-4321",
"message": "Quero agendar uma visita.",
},
)
assert submit_res.status_code == 201
lead_id = submit_res.get_json()["id"]
# 2. Admin lista todos os leads e localiza o criado
list_res = client.get("/api/v1/admin/leads", headers=_admin_headers(admin.id))
assert list_res.status_code == 200
items = list_res.get_json()["items"]
match = next((item for item in items if item["id"] == lead_id), None)
assert match is not None, "Lead não encontrado na listagem admin"
assert match["name"] == "Maria Souza"
assert match["email"] == "maria@example.com"
assert match["phone"] == "(21) 98765-4321"
assert match["message"] == "Quero agendar uma visita."
assert match["source"] == "imovel"
assert match["source_detail"] == prop.title
assert match["property_id"] == str(prop.id)
assert match["created_at"] is not None
def test_general_contact_appears_in_admin_without_property_id(self, client, db):
admin = _make_admin_user(db)
# 1. Cliente envia contato geral
submit_res = client.post(
"/api/v1/contact",
json={
"name": "Carlos Lima",
"email": "carlos@example.com",
"phone": "(31) 97654-3210",
"message": "Quero informações sobre venda.",
"source": "contato",
},
)
assert submit_res.status_code == 201
lead_id = submit_res.get_json()["id"]
# 2. Admin filtra por source=contato e encontra o lead
list_res = client.get(
"/api/v1/admin/leads?source=contato",
headers=_admin_headers(admin.id),
)
items = list_res.get_json()["items"]
match = next((item for item in items if item["id"] == lead_id), None)
assert match is not None
assert match["property_id"] is None
assert match["source"] == "contato"
def test_cadastro_residencia_appears_in_admin_with_source_detail(self, client, db):
admin = _make_admin_user(db)
# 1. Cliente submete formulário de captação
submit_res = client.post(
"/api/v1/contact",
json={
"name": "Ana Paula",
"email": "ana@example.com",
"phone": "(41) 96543-2109",
"message": "Finalidade: Venda\nTipo: Casa\nValor: R$ 800.000",
"source": "cadastro_residencia",
"source_detail": "Casa",
},
)
assert submit_res.status_code == 201
lead_id = submit_res.get_json()["id"]
# 2. Admin filtra por source=cadastro_residencia
list_res = client.get(
"/api/v1/admin/leads?source=cadastro_residencia",
headers=_admin_headers(admin.id),
)
items = list_res.get_json()["items"]
match = next((item for item in items if item["id"] == lead_id), None)
assert match is not None
assert match["source"] == "cadastro_residencia"
assert match["source_detail"] == "Casa"
assert match["property_id"] is None
assert "Venda" in match["message"]