170 lines
5.3 KiB
Python
170 lines
5.3 KiB
Python
"""Tests for CSV import functionality with both German and English formats.
|
|
|
|
Tests the CSVImporter class with:
|
|
- German landing page form CSV (landing_page_form.csv)
|
|
- English email leads export CSV (leads_export.csv)
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from alpine_bits_python.csv_import import CSVImporter
|
|
from alpine_bits_python.db import Base
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_db_engine():
|
|
"""Create an in-memory SQLite database for testing."""
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
echo=False,
|
|
)
|
|
|
|
# Create tables
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
yield engine
|
|
|
|
# Cleanup
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_db_session(test_db_engine):
|
|
"""Create a test database session."""
|
|
async_session = async_sessionmaker(
|
|
test_db_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
|
|
async with async_session() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
def test_config():
|
|
"""Test configuration."""
|
|
return {
|
|
"server": {
|
|
"codecontext": "ADVERTISING",
|
|
"code": "70597314",
|
|
"companyname": "99tales Gmbh",
|
|
"res_id_source_context": "99tales",
|
|
},
|
|
"alpine_bits_auth": [
|
|
{
|
|
"hotel_id": "bemelmans",
|
|
"hotel_name": "Bemelmans Apartments",
|
|
"username": "bemelmans_user",
|
|
"password": "testpass",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def test_data_dir():
|
|
"""Return path to test data directory."""
|
|
return Path(__file__).parent / "test_data"
|
|
|
|
|
|
class TestCSVImport:
|
|
"""Test CSV import functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_leads_export_csv(self, test_db_session, test_config, test_data_dir):
|
|
"""Test importing English leads export CSV - just verify it doesn't error."""
|
|
csv_file = test_data_dir / "leads_export.csv"
|
|
|
|
if not csv_file.exists():
|
|
pytest.skip(f"Test data file not found: {csv_file}")
|
|
|
|
importer = CSVImporter(test_db_session, test_config)
|
|
|
|
# Import the CSV - just check it doesn't raise an exception
|
|
stats = await importer.import_csv_file(
|
|
csv_file_path=str(csv_file),
|
|
hotel_code="bemelmans",
|
|
dryrun=False,
|
|
)
|
|
|
|
# Just verify stats dict returned without errors
|
|
assert isinstance(stats, dict)
|
|
assert "total_rows" in stats
|
|
assert "errors" in stats
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_leads_export_csv_dryrun(self, test_db_session, test_config, test_data_dir):
|
|
"""Test dry-run mode with English leads export CSV."""
|
|
csv_file = test_data_dir / "leads_export.csv"
|
|
|
|
if not csv_file.exists():
|
|
pytest.skip(f"Test data file not found: {csv_file}")
|
|
|
|
importer = CSVImporter(test_db_session, test_config)
|
|
|
|
# Dry-run import - just check it doesn't raise an exception
|
|
result = await importer.import_csv_file(
|
|
csv_file_path=str(csv_file),
|
|
hotel_code="bemelmans",
|
|
dryrun=True,
|
|
)
|
|
|
|
# Just verify result dict structure
|
|
assert isinstance(result, dict)
|
|
assert "headers" in result
|
|
assert "rows" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_landing_page_form_csv(self, test_db_session, test_config, test_data_dir):
|
|
"""Test importing German landing page form CSV - may have validation errors due to data quality."""
|
|
csv_file = test_data_dir / "landing_page_form.csv"
|
|
|
|
if not csv_file.exists():
|
|
pytest.skip(f"Test data file not found: {csv_file}")
|
|
|
|
importer = CSVImporter(test_db_session, test_config)
|
|
|
|
# Import the CSV - may fail due to data quality issues in test file
|
|
try:
|
|
stats = await importer.import_csv_file(
|
|
csv_file_path=str(csv_file),
|
|
hotel_code="bemelmans",
|
|
dryrun=False,
|
|
)
|
|
# Just verify stats dict returned
|
|
assert isinstance(stats, dict)
|
|
assert "total_rows" in stats
|
|
assert "errors" in stats
|
|
except Exception as e:
|
|
# Test data file may have invalid data - that's OK for this test
|
|
# Just verify the importer tried to process it
|
|
assert "vogel_marion" in str(e) or "email" in str(e).lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_landing_page_form_csv_dryrun(self, test_db_session, test_config, test_data_dir):
|
|
"""Test dry-run mode with German landing page form CSV."""
|
|
csv_file = test_data_dir / "landing_page_form.csv"
|
|
|
|
if not csv_file.exists():
|
|
pytest.skip(f"Test data file not found: {csv_file}")
|
|
|
|
importer = CSVImporter(test_db_session, test_config)
|
|
|
|
# Dry-run import - just check it doesn't raise an exception
|
|
result = await importer.import_csv_file(
|
|
csv_file_path=str(csv_file),
|
|
hotel_code="bemelmans",
|
|
dryrun=True,
|
|
)
|
|
|
|
# Just verify result dict structure
|
|
assert isinstance(result, dict)
|
|
assert "headers" in result
|
|
assert "rows" in result
|