Hola {{nombre}}, adjunto detalles.
", cc="copia@ejemplo.com", base_url="http://localhost:8765", ) self.assertEqual(log["status"], "simulated") self.assertEqual(log["send_type"], "individual") self.assertIn("Estudio Creativo Sur", log["subject"]) self.assertIn("Hola Ana", log["html_body"]) self.assertIn(f"/t/o/{log['tracking_id']}.gif", log["html_body"]) @patch("mailer.smtplib.SMTP") def test_send_individual_real_smtp_with_attachment(self, mock_smtp_cls): mock_server = MagicMock() mock_smtp_cls.return_value.__enter__.return_value = mock_server smtp_cfg = self.db.upsert_smtp_config( { "name": "Brevo Test", "mode": "smtp", "host": "smtp-relay.brevo.com", "port": 587, "security": "tls", "username": "user@brevo.com", "password": "secret", "from_name": "Mi Empresa", "from_email": "noreply@miempresa.com", "is_default": True, } ) att_b64 = base64.b64encode(b"Hola archivo").decode("ascii") log = self.mailer.send_single_email( to_email="destino@cliente.com", to_name="Juan Pérez", cc="jefe@cliente.com", bcc="archivo@miempresa.com", subject="Factura {{nombre}}", html_content="Estimado {{nombre}}, adjuntamos su documento.
", sender_id=smtp_cfg["id"], attachments=[{"filename": "doc.txt", "mime_type": "text/plain", "content_base64": att_b64}], ) self.assertEqual(log["status"], "sent") mock_server.starttls.assert_called_once() mock_server.login.assert_called_once_with("user@brevo.com", "secret") args, _ = mock_server.sendmail.call_args self.assertEqual(args[0], "noreply@miempresa.com") self.assertEqual(args[1], ["destino@cliente.com", "jefe@cliente.com", "archivo@miempresa.com"]) self.assertIn("doc.txt", args[2]) class _FakeSocket: """In-memory socket adapter for testing BaseHTTPRequestHandler without TCP.""" def __init__(self, raw_request: bytes): self._rfile = io.BytesIO(raw_request) self._wfile = io.BytesIO() def makefile(self, mode, *args, **kwargs): return self._rfile if "r" in mode else self._wfile def sendall(self, data): self._wfile.write(data) def close(self): pass class TestMailPulseHTTP(unittest.TestCase): def setUp(self): os.environ["MAILPULSE_QUIET"] = "1" self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) self.tmp.close() self.base_url = "http://127.0.0.1:8765" self.app_ctx = AppContext(db_path=self.tmp.name, base_url=self.base_url, require_auth=False) self.handler_cls = make_handler_class(self.app_ctx) def tearDown(self): if os.path.exists(self.tmp.name): os.unlink(self.tmp.name) def _req(self, method: str, path: str, body: dict = None): payload = json.dumps(body).encode("utf-8") if body is not None else b"" headers = f"{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:8765\r\n" if body is not None: headers += f"Content-Type: application/json\r\nContent-Length: {len(payload)}\r\n" headers += "\r\n" sock = _FakeSocket(headers.encode("utf-8") + payload) self.handler_cls(sock, ("127.0.0.1", 12345), None) raw_resp = sock._wfile.getvalue() head, _, resp_body = raw_resp.partition(b"\r\n\r\n") status_line = head.splitlines()[0].decode("ascii") status_code = int(status_line.split()[1]) return status_code, resp_body def test_full_campaign_and_1to1_flow(self): # 1. Static HTML served status, html_bytes = self._req("GET", "/") self.assertEqual(status, 200) self.assertIn(b"MailPulse", html_bytes) # 2. Import CSV into a new list _, lst_raw = self._req("POST", "/api/lists", {"name": "Lista Prueba HTTP"}) list_id = json.loads(lst_raw)["list"]["id"] _, imp_raw = self._req( "POST", "/api/contacts/import", { "list_id": list_id, "csv_text": "email,nombre,empresa\nuno@prueba.com,Uno,Alpha\ndos@prueba.com,Dos,Beta\n", }, ) self.assertEqual(json.loads(imp_raw)["imported"], 2) # 3. Create and send mass campaign _, camp_raw = self._req( "POST", "/api/campaigns", { "name": "Campaña Test", "subject": "Hola {{nombre}}", "list_id": list_id, "html_content": "Bienvenido {{nombre}} de {{empresa}}
", "throttle_ms": 0, }, ) camp_id = json.loads(camp_raw)["campaign"]["id"] status, _ = self._req("POST", f"/api/campaigns/{camp_id}/send", {}) self.assertEqual(status, 200) # Wait briefly for background campaign thread for _ in range(30): _, detail_raw = self._req("GET", f"/api/campaigns/{camp_id}") detail = json.loads(detail_raw) if detail["campaign"]["status"] == "sent": break time.sleep(0.05) self.assertEqual(detail["campaign"]["status"], "sent") self.assertEqual(detail["campaign"]["sent_count"], 2) self.assertEqual(len(detail["logs"]), 2) # 4. Trigger tracking pixel open tracking_id = detail["logs"][0]["tracking_id"] px_status, px_bytes = self._req("GET", f"/t/o/{tracking_id}.gif") self.assertEqual(px_status, 200) self.assertTrue(px_bytes.startswith(b"GIF89a")) _, after_open_raw = self._req("GET", f"/api/campaigns/{camp_id}") self.assertEqual(json.loads(after_open_raw)["campaign"]["open_count"], 1) # 5. Send individual 1-to-1 email ind_status, ind_raw = self._req( "POST", "/api/send-individual", { "to_email": "uno@prueba.com", "subject": "Seguimiento directo {{nombre}}", "html_content": "Hola {{nombre}}, este es un correo 1 a 1.
", }, ) self.assertEqual(ind_status, 200) ind_log = json.loads(ind_raw)["log"] self.assertEqual(ind_log["send_type"], "individual") self.assertEqual(ind_log["subject"], "Seguimiento directo Uno") # 6. Dashboard stats reflect 3 total sent (2 campaign + 1 individual) and 1 open _, dash_raw = self._req("GET", "/api/dashboard") dash = json.loads(dash_raw) self.assertEqual(dash["total_sent"], 3) self.assertEqual(dash["total_individual"], 1) self.assertEqual(dash["total_campaign_emails"], 2) self.assertEqual(dash["total_opened"], 1) # 7. Unsubscribe endpoint contact_id = detail["logs"][0]["contact_id"] unsub_status, unsub_html = self._req("GET", f"/unsubscribe/{contact_id}") self.assertEqual(unsub_status, 200) self.assertIn(b"cancelada correctamente", unsub_html) # 8. Studio assets and Outlook template export st_assets, raw_assets = self._req("GET", "/api/studio-assets") self.assertEqual(st_assets, 200) assets = json.loads(raw_assets) self.assertGreaterEqual(len(assets["blocks"]), 4) self.assertGreaterEqual(len(assets["photos"]), 6) st_tpls, raw_tpls = self._req("GET", "/api/templates") self.assertEqual(st_tpls, 200) tpls = json.loads(raw_tpls)["templates"] self.assertEqual(len(tpls), 6) st_exp, raw_exp = self._req( "POST", "/api/templates/export-outlook", { "template_id": "tpl-pro-launch", "extra_vars": {"nombre": "Sofía", "empresa": "Norte Tech"}, }, ) self.assertEqual(st_exp, 200) exp = json.loads(raw_exp) self.assertIn("Sofía", exp["body_html"]) self.assertIn("Norte Tech", exp["subject"]) self.assertIn("urn:schemas-microsoft-com:office:office", exp["outlook_html"]) self.assertIn("X-Unsent: 1", exp["eml_content"]) self.assertIn("https://images.unsplash.com/", exp["body_html"]) class TestMailPulseAuthAndWSGI(unittest.TestCase): def setUp(self): os.environ["MAILPULSE_QUIET"] = "1" self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) self.tmp.close() os.environ["MAILPULSE_DB_PATH"] = self.tmp.name import wsgi wsgi._app_ctx = None wsgi._handler_cls = None self.wsgi = wsgi def tearDown(self): os.environ.pop("MAILPULSE_DB_PATH", None) self.wsgi._app_ctx = None self.wsgi._handler_cls = None if os.path.exists(self.tmp.name): os.unlink(self.tmp.name) def _call_wsgi(self, method: str, path: str, body: dict = None, cookie: str = ""): payload = json.dumps(body).encode("utf-8") if body is not None else b"" environ = { "REQUEST_METHOD": method, "PATH_INFO": path, "QUERY_STRING": "", "HTTP_HOST": "correos.tudominio.cl", "wsgi.url_scheme": "https", "wsgi.input": io.BytesIO(payload), "CONTENT_LENGTH": str(len(payload)), } if body is not None: environ["CONTENT_TYPE"] = "application/json" if cookie: environ["HTTP_COOKIE"] = cookie captured = {} def start_response(status, headers): captured["status"] = int(status.split()[0]) captured["headers"] = headers chunks = self.wsgi.application(environ, start_response) return captured["status"], captured["headers"], b"".join(chunks) def test_private_portal_login_and_wsgi(self): # 1. Unauthenticated root redirects to /login st, hdrs, _ = self._call_wsgi("GET", "/") self.assertEqual(st, 302) self.assertTrue(any(k.lower() == "location" and v == "/login" for k, v in hdrs)) # 2. Unauthenticated API returns 401 st_api, _, _ = self._call_wsgi("GET", "/api/dashboard") self.assertEqual(st_api, 401) # 3. /login renders private portal st_login, _, login_html = self._call_wsgi("GET", "/login") self.assertEqual(st_login, 200) self.assertIn(b"PORTADA PRIVADA", login_html) # 4. Wrong password rejected st_bad, _, _ = self._call_wsgi("POST", "/api/auth/login", {"username": "maryori.vargas", "password": "wrong"}) self.assertEqual(st_bad, 401) # 5. Valid login for Maryori Vargas sets session cookie st_ok, hdrs_ok, body_ok = self._call_wsgi( "POST", "/api/auth/login", {"username": "maryori.vargas", "password": "Maryori2026!"} ) self.assertEqual(st_ok, 200) self.assertIn("Maryori Vargas", json.loads(body_ok)["user"]["display_name"]) cookie_hdr = next(v for k, v in hdrs_ok if k.lower() == "set-cookie") cookie_kv = cookie_hdr.split(";", 1)[0] # 6. Authenticated API call succeeds st_dash, _, dash_raw = self._call_wsgi("GET", "/api/dashboard", cookie=cookie_kv) self.assertEqual(st_dash, 200) self.assertIn("total_contacts", json.loads(dash_raw)) if __name__ == "__main__": unittest.main()