|
2 | 2 |
|
3 | 3 |
|
4 | 4 | import pytest |
| 5 | +import json |
| 6 | +import tempfile |
| 7 | +import os |
| 8 | + |
5 | 9 | from nbmail.egress import ( |
6 | 10 | send_email_with_redmail, |
7 | 11 | send_email_with_yagmail, |
|
11 | 15 | send_quarto_email_with_gmail, |
12 | 16 | ) |
13 | 17 | from nbmail.structs import Email |
| 18 | +from nbmail.ingress import quarto_json_to_email |
14 | 19 |
|
15 | 20 |
|
16 | 21 | def make_basic_email(): |
@@ -286,3 +291,142 @@ def test_not_implemented_functions(send_func): |
286 | 291 | email = make_basic_email() |
287 | 292 | with pytest.raises(NotImplementedError): |
288 | 293 | send_func(email) |
| 294 | + |
| 295 | + |
| 296 | +# Tests for Email.write_quarto_json() method |
| 297 | +def test_email_write_quarto_json_basic(): |
| 298 | + email = Email( |
| 299 | + html="<html><body><p>Test email</p></body></html>", |
| 300 | + subject="Test Subject", |
| 301 | + text="Plain text version", |
| 302 | + external_attachments=["file1.pdf", "file2.csv"], |
| 303 | + inline_attachments={"img1": "base64data123"}, |
| 304 | + email_suppress_report_attachment=True, |
| 305 | + email_suppress_scheduled=False, |
| 306 | + ) |
| 307 | + |
| 308 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 309 | + json_path = os.path.join(tmpdir, "test.json") |
| 310 | + email.write_quarto_json(json_path) |
| 311 | + |
| 312 | + with open(json_path, "r") as f: |
| 313 | + data = json.load(f) |
| 314 | + |
| 315 | + # Check that all expected fields are present (with prefix for Quarto compatibility) |
| 316 | + assert data["email_subject"] == "Test Subject" |
| 317 | + assert data["email_body_html"] == "<html><body><p>Test email</p></body></html>" |
| 318 | + assert data["email_body_text"] == "Plain text version" |
| 319 | + assert data["email_attachments"] == ["file1.pdf", "file2.csv"] |
| 320 | + assert data["email_images"] == {"img1": "base64data123"} |
| 321 | + assert data["email_suppress_report_attachment"] is True |
| 322 | + assert data["email_suppress_scheduled"] is False |
| 323 | + |
| 324 | + |
| 325 | +def test_email_write_quarto_json_minimal(): |
| 326 | + """Test writing a minimal email to Quarto JSON format.""" |
| 327 | + email = Email( |
| 328 | + html="<html><body>Minimal</body></html>", |
| 329 | + subject="Minimal Subject", |
| 330 | + ) |
| 331 | + |
| 332 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 333 | + json_path = os.path.join(tmpdir, "minimal.json") |
| 334 | + email.write_quarto_json(json_path) |
| 335 | + |
| 336 | + with open(json_path, "r") as f: |
| 337 | + data = json.load(f) |
| 338 | + |
| 339 | + # Check minimal fields |
| 340 | + assert data["email_subject"] == "Minimal Subject" |
| 341 | + assert data["email_body_html"] == "<html><body>Minimal</body></html>" |
| 342 | + assert data["email_attachments"] == [] |
| 343 | + |
| 344 | + # Optional fields should not be present |
| 345 | + assert "email_body_text" not in data |
| 346 | + assert "email_images" not in data |
| 347 | + assert "email_suppress_report_attachment" not in data |
| 348 | + assert "email_suppress_scheduled" not in data |
| 349 | + |
| 350 | + |
| 351 | +def test_email_write_quarto_json_round_trip(): |
| 352 | + """Test writing and reading back a Quarto JSON email.""" |
| 353 | + original_email = Email( |
| 354 | + html="<html><body><p>Quarto email</p></body></html>", |
| 355 | + subject="Quarto Test", |
| 356 | + text="Plain text version", |
| 357 | + external_attachments=["output.pdf"], |
| 358 | + inline_attachments={"img1": "base64encodedstring"}, |
| 359 | + email_suppress_report_attachment=True, |
| 360 | + email_suppress_scheduled=False, |
| 361 | + ) |
| 362 | + |
| 363 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 364 | + json_path = os.path.join(tmpdir, "roundtrip.json") |
| 365 | + original_email.write_quarto_json(json_path) |
| 366 | + |
| 367 | + # Read it back |
| 368 | + read_email = quarto_json_to_email(json_path) |
| 369 | + |
| 370 | + # Verify all fields match |
| 371 | + assert read_email.subject == original_email.subject |
| 372 | + assert read_email.html == original_email.html |
| 373 | + assert read_email.text == original_email.text |
| 374 | + assert read_email.external_attachments == original_email.external_attachments |
| 375 | + assert read_email.inline_attachments == original_email.inline_attachments |
| 376 | + assert read_email.email_suppress_report_attachment == original_email.email_suppress_report_attachment |
| 377 | + assert read_email.email_suppress_scheduled == original_email.email_suppress_scheduled |
| 378 | + |
| 379 | + |
| 380 | +def test_email_write_quarto_json_no_attachments(): |
| 381 | + """Test writing an email without attachments or images.""" |
| 382 | + email = Email( |
| 383 | + html="<html><body>No attachments</body></html>", |
| 384 | + subject="No Attachments", |
| 385 | + ) |
| 386 | + |
| 387 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 388 | + json_path = os.path.join(tmpdir, "no_attachments.json") |
| 389 | + email.write_quarto_json(json_path) |
| 390 | + |
| 391 | + with open(json_path, "r") as f: |
| 392 | + data = json.load(f) |
| 393 | + |
| 394 | + # Check that attachments and images are empty |
| 395 | + assert data["email_attachments"] == [] |
| 396 | + assert "email_images" not in data |
| 397 | + |
| 398 | + |
| 399 | +def test_email_write_quarto_json_no_text(): |
| 400 | + """Test writing an email without plain text version.""" |
| 401 | + email = Email( |
| 402 | + html="<html><body>HTML only</body></html>", |
| 403 | + subject="HTML Only", |
| 404 | + ) |
| 405 | + |
| 406 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 407 | + json_path = os.path.join(tmpdir, "html_only.json") |
| 408 | + email.write_quarto_json(json_path) |
| 409 | + |
| 410 | + with open(json_path, "r") as f: |
| 411 | + data = json.load(f) |
| 412 | + |
| 413 | + # Plain text should not be present |
| 414 | + assert "email_body_text" not in data |
| 415 | + |
| 416 | + |
| 417 | +def test_email_write_quarto_json_custom_filename(): |
| 418 | + email = Email( |
| 419 | + html="<html><body>Custom</body></html>", |
| 420 | + subject="Custom Filename", |
| 421 | + ) |
| 422 | + |
| 423 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 424 | + custom_path = os.path.join(tmpdir, "my_custom_file.json") |
| 425 | + email.write_quarto_json(custom_path) |
| 426 | + |
| 427 | + assert os.path.exists(custom_path) |
| 428 | + |
| 429 | + with open(custom_path, "r") as f: |
| 430 | + data = json.load(f) |
| 431 | + |
| 432 | + assert data["email_subject"] == "Custom Filename" |
0 commit comments