bbc53b4ce7
- Add test mailbox with 5 sample emails for integration testing - Add himalaya_test_config.toml for local Maildir backend testing - Create 12 integration tests covering search by from/to/subject/body - Fix search results display to clear list and show message when 0 results - Add clear_content() method to ContentContainer widget
275 lines
9.1 KiB
Python
275 lines
9.1 KiB
Python
"""Integration tests for Himalaya client with test mailbox.
|
|
|
|
These tests use a local Maildir test mailbox to verify himalaya operations
|
|
without touching real email accounts.
|
|
|
|
Run with:
|
|
pytest tests/test_himalaya_integration.py -v
|
|
|
|
Requires:
|
|
- himalaya CLI installed
|
|
- tests/fixtures/test_mailbox with sample emails
|
|
- tests/fixtures/himalaya_test_config.toml
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# Path to the test config
|
|
TEST_CONFIG = Path(__file__).parent / "fixtures" / "himalaya_test_config.toml"
|
|
TEST_MAILBOX = Path(__file__).parent / "fixtures" / "test_mailbox"
|
|
|
|
|
|
def himalaya_available() -> bool:
|
|
"""Check if himalaya CLI is installed."""
|
|
try:
|
|
result = subprocess.run(
|
|
["himalaya", "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.returncode == 0
|
|
except FileNotFoundError:
|
|
return False
|
|
|
|
|
|
# Skip all tests if himalaya is not installed
|
|
pytestmark = pytest.mark.skipif(
|
|
not himalaya_available(),
|
|
reason="himalaya CLI not installed",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def himalaya_cmd():
|
|
"""Return the base himalaya command with test config."""
|
|
# Note: -a must come after the subcommand in himalaya
|
|
return f"himalaya -c {TEST_CONFIG}"
|
|
|
|
|
|
@pytest.fixture
|
|
def account_arg():
|
|
"""Return the account argument for himalaya."""
|
|
return "-a test-account"
|
|
|
|
|
|
class TestHimalayaListEnvelopes:
|
|
"""Tests for listing envelopes."""
|
|
|
|
def test_list_all_envelopes(self, himalaya_cmd, account_arg):
|
|
"""Test listing all envelopes from test mailbox."""
|
|
result = subprocess.run(
|
|
f"{himalaya_cmd} envelope list {account_arg} -o json",
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
assert len(envelopes) == 5, f"Expected 5 emails, got {len(envelopes)}"
|
|
|
|
def test_envelope_has_required_fields(self, himalaya_cmd, account_arg):
|
|
"""Test that envelopes have all required fields."""
|
|
result = subprocess.run(
|
|
f"{himalaya_cmd} envelope list {account_arg} -o json",
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
required_fields = ["id", "subject", "from", "to", "date"]
|
|
|
|
for envelope in envelopes:
|
|
for field in required_fields:
|
|
assert field in envelope, f"Missing field: {field}"
|
|
|
|
|
|
class TestHimalayaSearch:
|
|
"""Tests for search functionality."""
|
|
|
|
def test_search_by_from_name(self, himalaya_cmd, account_arg):
|
|
"""Test searching by sender name."""
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "from edson"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
assert len(envelopes) == 1
|
|
assert "Edson" in envelopes[0]["from"]["name"]
|
|
|
|
def test_search_by_body_content(self, himalaya_cmd, account_arg):
|
|
"""Test searching by body content."""
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "body edson"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
# Should find 2: one from Edson, one mentioning Edson in body
|
|
assert len(envelopes) == 2
|
|
|
|
def test_search_by_subject(self, himalaya_cmd, account_arg):
|
|
"""Test searching by subject."""
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "subject meeting"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
assert len(envelopes) == 1
|
|
assert "Meeting" in envelopes[0]["subject"]
|
|
|
|
def test_search_compound_or_query(self, himalaya_cmd, account_arg):
|
|
"""Test compound OR search query."""
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "from edson or body edson"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
# Should find both emails mentioning Edson
|
|
assert len(envelopes) >= 2
|
|
|
|
def test_search_no_results(self, himalaya_cmd, account_arg):
|
|
"""Test search that returns no results."""
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "from nonexistent_person_xyz"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
assert len(envelopes) == 0
|
|
|
|
def test_search_full_compound_query(self, himalaya_cmd, account_arg):
|
|
"""Test the full compound query format used by our search function."""
|
|
query = "edson"
|
|
search_query = f"from {query} or to {query} or subject {query} or body {query}"
|
|
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "{search_query}"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, f"Error: {result.stderr}"
|
|
|
|
envelopes = json.loads(result.stdout)
|
|
# Should find emails where Edson is sender or mentioned in body
|
|
assert len(envelopes) >= 2
|
|
|
|
|
|
class TestHimalayaReadMessage:
|
|
"""Tests for reading message content."""
|
|
|
|
def test_read_message_by_id(self, himalaya_cmd, account_arg):
|
|
"""Test reading a message by ID."""
|
|
# First get the list to find an ID
|
|
list_result = subprocess.run(
|
|
f"{himalaya_cmd} envelope list {account_arg} -o json",
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert list_result.returncode == 0
|
|
|
|
envelopes = json.loads(list_result.stdout)
|
|
message_id = envelopes[0]["id"]
|
|
|
|
# Read the message
|
|
read_result = subprocess.run(
|
|
f"{himalaya_cmd} message read {account_arg} {message_id}",
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert read_result.returncode == 0, f"Error: {read_result.stderr}"
|
|
assert len(read_result.stdout) > 0
|
|
|
|
|
|
class TestHimalayaAsyncClient:
|
|
"""Tests for the async himalaya client module."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_envelopes_async(self):
|
|
"""Test the async search_envelopes function."""
|
|
# Import here to avoid issues if himalaya module has import errors
|
|
import sys
|
|
|
|
# Add project root to path for proper imports
|
|
project_root = str(Path(__file__).parent.parent)
|
|
if project_root not in sys.path:
|
|
sys.path.insert(0, project_root)
|
|
|
|
from src.services.himalaya import client as himalaya_client
|
|
|
|
# Note: This test would need the CLI to use our test config
|
|
# For now, just verify the function exists and has correct signature
|
|
assert hasattr(himalaya_client, "search_envelopes")
|
|
assert asyncio.iscoroutinefunction(himalaya_client.search_envelopes)
|
|
|
|
|
|
# Additional test for edge cases
|
|
class TestSearchEdgeCases:
|
|
"""Edge case tests for search."""
|
|
|
|
def test_search_with_special_characters(self, himalaya_cmd, account_arg):
|
|
"""Test searching with special characters in query."""
|
|
# This should not crash, even if no results
|
|
result = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "subject Q4"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
# May fail or succeed depending on himalaya version
|
|
# Just verify it doesn't crash catastrophically
|
|
assert result.returncode in [0, 1]
|
|
|
|
def test_search_case_insensitive(self, himalaya_cmd, account_arg):
|
|
"""Test that search is case insensitive."""
|
|
result_upper = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "from EDSON"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
result_lower = subprocess.run(
|
|
f'{himalaya_cmd} envelope list {account_arg} -o json "from edson"',
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
# Both should succeed
|
|
assert result_upper.returncode == 0
|
|
assert result_lower.returncode == 0
|
|
|
|
# Results should be the same (case insensitive)
|
|
upper_envelopes = json.loads(result_upper.stdout)
|
|
lower_envelopes = json.loads(result_lower.stdout)
|
|
assert len(upper_envelopes) == len(lower_envelopes)
|