Add Himalaya search integration tests and fix 0 results display
- 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
This commit is contained in:
274
tests/test_himalaya_integration.py
Normal file
274
tests/test_himalaya_integration.py
Normal file
@@ -0,0 +1,274 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user