208 lines
9.0 KiB
Python
208 lines
9.0 KiB
Python
import importlib
|
|
from datetime import datetime, timedelta, timezone
|
|
from services import battlecard_service
|
|
from repositories.tenant_repository import tenant_repository
|
|
from repositories.competitor_repository import competitor_repository
|
|
from repositories.product_repository import product_repository
|
|
from repositories.battlecard_repository import battlecard_repository
|
|
from repositories.price_history_repository import price_history_repository
|
|
|
|
|
|
def _install_fake_extractor(monkeypatch, fn):
|
|
# core.extractor is a real Phase 2 module — import it for real (not
|
|
# a blank fake substitute) and monkeypatch.setattr the one function
|
|
# this test needs, so the swap reverts after the test instead of
|
|
# permanently corrupting the shared module.
|
|
target = importlib.import_module('core.extractor')
|
|
monkeypatch.setattr(target, 'extract_with_openrouter', fn)
|
|
|
|
|
|
def _setup():
|
|
tenant = tenant_repository.create({
|
|
'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
|
|
'status': 'active', 'max_seats': 5,
|
|
})
|
|
competitor = competitor_repository.create({
|
|
'tenant_id': tenant['id'], 'name': 'Intrepid Travel', 'website': 'https://intrepidtravel.com',
|
|
'catalogue_url': 'https://intrepidtravel.com/adventures', 'primary_market': 'GB', 'active': True,
|
|
})
|
|
product_repository.create({
|
|
'tenant_id': tenant['id'], 'competitor_id': competitor['id'], 'trip_name': 'Peru Classic',
|
|
'majority_price_usd': 2190, 'duration_days': 15,
|
|
})
|
|
return tenant, competitor
|
|
|
|
|
|
def test_generate_battlecard_creates_record(monkeypatch):
|
|
tenant, competitor = _setup()
|
|
_install_fake_extractor(monkeypatch, lambda prompt, content: {
|
|
'positioning': 'Budget-mid, group focused',
|
|
'strengths': ['Strong brand'], 'weaknesses': ['Basic accommodation'],
|
|
'pricing_observation': 'Discounting Q3',
|
|
})
|
|
|
|
card = battlecard_service.generate_battlecard(competitor['id'], tenant['id'])
|
|
|
|
assert card is not None
|
|
assert card['content_json']['positioning'] == 'Budget-mid, group focused'
|
|
assert 'Strong brand' in card['content']
|
|
assert battlecard_repository.get_for_competitor(competitor['id']) is not None
|
|
|
|
|
|
def test_generate_battlecard_returns_none_on_extraction_failure(monkeypatch):
|
|
tenant, competitor = _setup()
|
|
|
|
def _raise(prompt, content):
|
|
raise RuntimeError('all models exhausted')
|
|
|
|
_install_fake_extractor(monkeypatch, _raise)
|
|
|
|
result = battlecard_service.generate_battlecard(competitor['id'], tenant['id'])
|
|
assert result is None
|
|
|
|
|
|
def test_generate_battlecard_handles_no_products_gracefully(monkeypatch):
|
|
tenant = tenant_repository.create({
|
|
'name': 'G Adventures', 'domain': 'gadventures.com', 'tier': 'analyse',
|
|
'status': 'active', 'max_seats': 5,
|
|
})
|
|
competitor = competitor_repository.create({
|
|
'tenant_id': tenant['id'], 'name': 'New Competitor', 'website': 'https://new.com',
|
|
'catalogue_url': 'https://new.com/trips', 'primary_market': 'US', 'active': True,
|
|
})
|
|
_install_fake_extractor(monkeypatch, lambda prompt, content: {
|
|
'positioning': 'Unknown', 'strengths': [], 'weaknesses': [], 'pricing_observation': 'n/a',
|
|
})
|
|
|
|
card = battlecard_service.generate_battlecard(competitor['id'], tenant['id'])
|
|
assert card is not None
|
|
|
|
|
|
# ── _top_values ─────────────────────────────────────────────────────
|
|
|
|
def test_top_values_dedupes_and_preserves_first_seen_order():
|
|
result = battlecard_service._top_values(['Mid-range', 'Budget', 'Mid-range', 'Premium', 'Budget'])
|
|
assert result == 'Mid-range, Budget, Premium'
|
|
|
|
|
|
def test_top_values_skips_null_and_empty_entries():
|
|
result = battlecard_service._top_values([None, 'Families', '', None, 'Couples'])
|
|
assert result == 'Families, Couples'
|
|
|
|
|
|
def test_top_values_respects_limit():
|
|
result = battlecard_service._top_values(['A', 'B', 'C', 'D'], limit=2)
|
|
assert result == 'A, B'
|
|
|
|
|
|
def test_top_values_empty_input_returns_empty_string():
|
|
assert battlecard_service._top_values([]) == ''
|
|
|
|
|
|
# ── _summarise_products (richer fields) ────────────────────────────
|
|
|
|
def test_summarise_products_includes_trip_count_and_new_fields():
|
|
products = [
|
|
{'trip_name': 'Peru Classic', 'majority_price_usd': 2190, 'duration_days': 15,
|
|
'service_level': 'Mid-range', 'target_audience': 'Couples', 'activities': 'Trekking'},
|
|
{'trip_name': 'Peru Premium', 'majority_price_usd': 3200, 'duration_days': 15,
|
|
'service_level': 'Premium', 'target_audience': 'Couples', 'activities': 'Trekking, Rafting'},
|
|
]
|
|
summary = battlecard_service._summarise_products(products)
|
|
|
|
assert summary['trip_count'] == 2
|
|
assert summary['service_levels'] == 'Mid-range, Premium'
|
|
assert summary['target_audiences'] == 'Couples'
|
|
assert 'Trekking' in summary['activities_summary']
|
|
|
|
|
|
def test_summarise_products_new_fields_default_to_unknown_when_missing():
|
|
products = [{'trip_name': 'Peru Classic'}]
|
|
summary = battlecard_service._summarise_products(products)
|
|
|
|
assert summary['service_levels'] == 'unknown'
|
|
assert summary['target_audiences'] == 'unknown'
|
|
assert summary['activities_summary'] == 'unknown'
|
|
|
|
|
|
# ── _summarise_recent_price_changes ────────────────────────────────
|
|
|
|
def _iso_days_ago(days):
|
|
return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
|
|
|
|
|
|
def test_summarise_recent_price_changes_includes_change_within_window():
|
|
product = product_repository.create({'tenant_id': 't1', 'competitor_id': 'c1', 'trip_name': 'Peru Classic'})
|
|
price_history_repository.create({
|
|
'tenant_id': 't1', 'product_id': product['id'], 'majority_price_usd': 1990,
|
|
'change_percent': -12.0, 'scraped_at': _iso_days_ago(5),
|
|
})
|
|
|
|
result = battlecard_service._summarise_recent_price_changes([product])
|
|
|
|
assert 'Peru Classic dropped 12.0%' in result
|
|
assert '$1990' in result
|
|
|
|
|
|
def test_summarise_recent_price_changes_ignores_entries_outside_window():
|
|
product = product_repository.create({'tenant_id': 't1', 'competitor_id': 'c1', 'trip_name': 'Peru Classic'})
|
|
price_history_repository.create({
|
|
'tenant_id': 't1', 'product_id': product['id'], 'majority_price_usd': 1990,
|
|
'change_percent': -12.0, 'scraped_at': _iso_days_ago(45),
|
|
})
|
|
|
|
result = battlecard_service._summarise_recent_price_changes([product])
|
|
|
|
assert result == 'No significant price changes in the last 30 days.'
|
|
|
|
|
|
def test_summarise_recent_price_changes_sorts_by_magnitude_descending():
|
|
product_a = product_repository.create({'tenant_id': 't1', 'competitor_id': 'c1', 'trip_name': 'Small Move'})
|
|
product_b = product_repository.create({'tenant_id': 't1', 'competitor_id': 'c1', 'trip_name': 'Big Move'})
|
|
price_history_repository.create({
|
|
'tenant_id': 't1', 'product_id': product_a['id'], 'majority_price_usd': 1000,
|
|
'change_percent': 2.0, 'scraped_at': _iso_days_ago(1),
|
|
})
|
|
price_history_repository.create({
|
|
'tenant_id': 't1', 'product_id': product_b['id'], 'majority_price_usd': 2000,
|
|
'change_percent': -25.0, 'scraped_at': _iso_days_ago(2),
|
|
})
|
|
|
|
result = battlecard_service._summarise_recent_price_changes([product_a, product_b])
|
|
|
|
assert result.index('Big Move') < result.index('Small Move')
|
|
|
|
|
|
def test_summarise_recent_price_changes_falls_back_when_no_price_history_at_all():
|
|
product = product_repository.create({'tenant_id': 't1', 'competitor_id': 'c1', 'trip_name': 'Peru Classic'})
|
|
|
|
result = battlecard_service._summarise_recent_price_changes([product])
|
|
|
|
assert result == 'No significant price changes in the last 30 days.'
|
|
|
|
|
|
# ── generate_battlecard() end-to-end prompt content ────────────────
|
|
|
|
def test_generate_battlecard_prompt_uses_real_data_not_placeholders(monkeypatch):
|
|
tenant, competitor = _setup()
|
|
product = product_repository.get_latest_for_competitor(competitor['id'])[0]
|
|
product_repository.update(product['id'], {'service_level': 'Premium', 'target_audience': 'Couples'})
|
|
price_history_repository.create({
|
|
'tenant_id': tenant['id'], 'product_id': product['id'], 'majority_price_usd': 1990,
|
|
'change_percent': -9.1, 'scraped_at': _iso_days_ago(2),
|
|
})
|
|
|
|
captured = {}
|
|
_install_fake_extractor(monkeypatch, lambda prompt, content: captured.update(prompt=prompt) or {
|
|
'positioning': 'x', 'strengths': [], 'weaknesses': [], 'pricing_observation': 'x',
|
|
})
|
|
|
|
battlecard_service.generate_battlecard(competitor['id'], tenant['id'])
|
|
|
|
assert 'Premium' in captured['prompt']
|
|
assert 'Couples' in captured['prompt']
|
|
assert 'dropped 9.1%' in captured['prompt']
|
|
assert 'See price_history for this competitor.' not in captured['prompt']
|
|
assert 'Derived from current product catalogue.' not in captured['prompt']
|
|
assert 'Positioning:' not in captured['prompt'] # positioning_summary input field removed entirely
|