feat: implementacao inicial do sistema LeadRadar

2026-08-26 17:14:35 -03:00
commit eb64f9a0f6
43 changed files with 2888 additions and 0 deletions
@@ -0,0 +1,55 @@
+import pytest
+from app import create_app
+from app.config import TestConfig
+from app.extensions import db
+from app.models.user import User
+from flask_jwt_extended import create_access_token
+
+@pytest.fixture
+def app():
+    app = create_app(TestConfig)
+    with app.app_context():
+        db.create_all()
+        yield app
+        db.session.remove()
+        db.drop_all()
+
+@pytest.fixture
+def client(app):
+    return app.test_client()
+
+@pytest.fixture
+def admin_user(app):
+    user = User(
+        nome="Admin Tester",
+        email="admin@test.com",
+        role="admin",
+        ativo=True
+    )
+    user.set_password("AdminPass123!")
+    db.session.add(user)
+    db.session.commit()
+    return user
+
+@pytest.fixture
+def regular_user(app):
+    user = User(
+        nome="User Tester",
+        email="user@test.com",
+        role="user",
+        ativo=True
+    )
+    user.set_password("UserPass123!")
+    db.session.add(user)
+    db.session.commit()
+    return user
+
+@pytest.fixture
+def admin_headers(app, admin_user):
+    token = create_access_token(identity=admin_user.id, additional_claims={'role': 'admin'})
+    return {'Authorization': f'Bearer {token}'}
+
+@pytest.fixture
+def user_headers(app, regular_user):
+    token = create_access_token(identity=regular_user.id, additional_claims={'role': 'user'})
+    return {'Authorization': f'Bearer {token}'}
@@ -0,0 +1,40 @@
+def test_login_success(client, regular_user):
+    response = client.post('/api/v1/auth/login', json={
+        'email': 'user@test.com',
+        'password': 'UserPass123!'
+    })
+    assert response.status_code == 200
+    data = response.get_json()
+    assert 'access_token' in data
+    assert data['user']['email'] == 'user@test.com'
+    assert data['user']['role'] == 'user'
+
+def test_login_invalid_password(client, regular_user):
+    response = client.post('/api/v1/auth/login', json={
+        'email': 'user@test.com',
+        'password': 'WrongPassword'
+    })
+    assert response.status_code == 401
+    data = response.get_json()
+    assert 'error' in data
+
+def test_get_me_profile(client, user_headers):
+    response = client.get('/api/v1/auth/me', headers=user_headers)
+    assert response.status_code == 200
+    data = response.get_json()
+    assert data['email'] == 'user@test.com'
+
+def test_change_password_success(client, user_headers):
+    response = client.post('/api/v1/auth/change-password', headers=user_headers, json={
+        'old_password': 'UserPass123!',
+        'new_password': 'NewSuperPass123!'
+    })
+    assert response.status_code == 200
+    assert 'message' in response.get_json()
+
+    # Tentar login com a nova senha
+    login_res = client.post('/api/v1/auth/login', json={
+        'email': 'user@test.com',
+        'password': 'NewSuperPass123!'
+    })
+    assert login_res.status_code == 200
@@ -0,0 +1,83 @@
+from unittest.mock import patch
+
+def test_search_maps_endpoint(client, user_headers):
+    with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
+        mock_geo.return_value = {
+            'cep': '01310100',
+            'logradouro': 'Avenida Paulista',
+            'bairro': 'Bela Vista',
+            'cidade': 'São Paulo',
+            'uf': 'SP',
+            'formatted_cep': '01310-100'
+        }
+        
+        response = client.post('/api/v1/leads/search-maps', headers=user_headers, json={
+            'cep': '01310-100',
+            'ramo': 'Padaria',
+            'max_results': 5
+        })
+        
+        assert response.status_code == 200
+        data = response.get_json()
+        assert 'summary' in data
+        assert data['summary']['created_count'] >= 1
+
+def test_update_lead_status_and_interaction(client, user_headers):
+    # Primeiro dispara busca para criar lead
+    with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
+        mock_geo.return_value = {
+            'cep': '01310100', 'logradouro': 'Avenida Paulista', 'bairro': 'Bela Vista',
+            'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
+        }
+        client.post('/api/v1/leads/search-maps', headers=user_headers, json={'cep': '01310-100', 'ramo': 'Padaria'})
+
+    # Listar leads
+    list_res = client.get('/api/v1/leads', headers=user_headers)
+    assert list_res.status_code == 200
+    leads = list_res.get_json()['leads']
+    assert len(leads) > 0
+    lead_id = leads[0]['id']
+
+    # Atualizar status para 'contatado'
+    update_res = client.put(f'/api/v1/leads/{lead_id}', headers=user_headers, json={
+        'status_funil': 'contatado',
+        'notas': 'Primeiro contato realizado via WhatsApp.'
+    })
+    assert update_res.status_code == 200
+    assert update_res.get_json()['status_funil'] == 'contatado'
+
+    # Verificar historico de interacoes
+    detail_res = client.get(f'/api/v1/leads/{lead_id}', headers=user_headers)
+    assert detail_res.status_code == 200
+    detail = detail_res.get_json()
+    assert len(detail['interacoes']) >= 2
+    types = [i['tipo'] for i in detail['interacoes']]
+    assert 'status_change' in types
+
+def test_opt_out_lgpd(client, user_headers):
+    # Criar lead
+    with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
+        mock_geo.return_value = {
+            'cep': '01310100', 'logradouro': 'Rua Teste', 'bairro': 'Centro',
+            'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
+        }
+        client.post('/api/v1/leads/search-maps', headers=user_headers, json={'cep': '01310-100', 'ramo': 'Farmácia'})
+
+    list_res = client.get('/api/v1/leads', headers=user_headers)
+    lead_id = list_res.get_json()['leads'][0]['id']
+
+    # Opt-out
+    opt_res = client.post(f'/api/v1/leads/{lead_id}/opt-out', headers=user_headers)
+    assert opt_res.status_code == 200
+    assert opt_res.get_json()['lead']['status_funil'] == 'opt_out'
+
+def test_export_leads(client, user_headers):
+    # Exportar CSV
+    csv_res = client.get('/api/v1/leads/export?format=csv', headers=user_headers)
+    assert csv_res.status_code == 200
+    assert 'text/csv' in csv_res.content_type
+
+    # Exportar JSON
+    json_res = client.get('/api/v1/leads/export?format=json', headers=user_headers)
+    assert json_res.status_code == 200
+    assert isinstance(json_res.get_json(), list)
@@ -0,0 +1,43 @@
+def test_admin_can_list_users(client, admin_headers):
+    response = client.get('/api/v1/users', headers=admin_headers)
+    assert response.status_code == 200
+    data = response.get_json()
+    assert isinstance(data, list)
+
+def test_user_cannot_list_users(client, user_headers):
+    response = client.get('/api/v1/users', headers=user_headers)
+    assert response.status_code == 403
+    data = response.get_json()
+    assert 'Acesso negado' in data['error']
+
+def test_user_cannot_create_user(client, user_headers):
+    response = client.post('/api/v1/users', headers=user_headers, json={
+        'nome': 'Hacker User',
+        'email': 'hacker@test.com',
+        'password': 'Password123!',
+        'role': 'user'
+    })
+    assert response.status_code == 403
+
+def test_user_cannot_delete_user(client, user_headers, admin_user):
+    response = client.delete(f'/api/v1/users/{admin_user.id}', headers=user_headers)
+    assert response.status_code == 403
+
+def test_user_cannot_expunge_database(client, user_headers):
+    response = client.delete('/api/v1/leads/bulk', headers=user_headers)
+    assert response.status_code == 403
+
+def test_admin_can_create_and_delete_user(client, admin_headers):
+    # Criar usuario
+    create_res = client.post('/api/v1/users', headers=admin_headers, json={
+        'nome': 'Novo Operador',
+        'email': 'operador@test.com',
+        'password': 'OperadorPass123!',
+        'role': 'user'
+    })
+    assert create_res.status_code == 201
+    new_user_id = create_res.get_json()['id']
+
+    # Deletar usuario
+    delete_res = client.delete(f'/api/v1/users/{new_user_id}', headers=admin_headers)
+    assert delete_res.status_code == 200
@@ -0,0 +1,22 @@
+from app.utils.sanitizers import sanitize_phone, sanitize_cep, validate_email
+from app.services.geocoding_service import GeocodingService
+
+def test_sanitize_phone():
+    assert sanitize_phone('(11) 99999-8888') == '5511999998888'
+    assert sanitize_phone('1133334444') == '551133334444'
+    assert sanitize_phone('5511988887777') == '5511988887777'
+    assert sanitize_phone('') == ''
+
+def test_sanitize_cep():
+    assert sanitize_cep('01310-100') == '01310100'
+    assert sanitize_cep('01310100') == '01310100'
+    assert sanitize_cep('invalid') == ''
+
+def test_validate_email():
+    assert validate_email('admin@leadradar.com') is True
+    assert validate_email('invalid-email') is False
+
+def test_build_search_query():
+    location = {'bairro': 'Bela Vista', 'cidade': 'São Paulo', 'uf': 'SP'}
+    query = GeocodingService.build_search_query('Restaurante', location)
+    assert query == 'Restaurante em Bela Vista, São Paulo - SP'
@@ -0,0 +1,23 @@
+from unittest.mock import patch
+
+def test_n8n_webhook_flow(client, user_headers):
+    # Criar lead
+    with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
+        mock_geo.return_value = {
+            'cep': '01310100', 'logradouro': 'Rua Teste', 'bairro': 'Centro',
+            'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
+        }
+        client.post('/api/v1/leads/search-maps', headers=user_headers, json={'cep': '01310-100', 'ramo': 'Supermercado'})
+
+    leads = client.get('/api/v1/leads', headers=user_headers).get_json()['leads']
+    lead_id = leads[0]['id']
+
+    # Disparar webhook n8n
+    webhook_res = client.post('/api/v1/webhooks/n8n', json={
+        'lead_id': lead_id,
+        'event_type': 'mensagem_enviada',
+        'new_status': 'respondeu',
+        'mensagem': 'Cliente respondeu via WhatsApp demonstrando interesse.'
+    })
+    assert webhook_res.status_code == 200
+    assert webhook_res.get_json()['data']['status_atual'] == 'respondeu'