77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
import sys
|
|
import types
|
|
from services import embedding_service
|
|
|
|
|
|
def test_cosine_similarity_identical_vectors_is_one():
|
|
assert embedding_service._cosine_similarity([1, 0, 0], [1, 0, 0]) == 1.0
|
|
|
|
|
|
def test_cosine_similarity_orthogonal_vectors_is_zero():
|
|
assert embedding_service._cosine_similarity([1, 0], [0, 1]) == 0.0
|
|
|
|
|
|
def test_cosine_similarity_zero_vector_returns_zero():
|
|
assert embedding_service._cosine_similarity([0, 0], [1, 1]) == 0.0
|
|
|
|
|
|
def test_find_comparable_trips_filters_by_threshold_and_sorts():
|
|
client_trips = [
|
|
{'trip_name': 'Inca Trail Classic', 'price_usd': 1999, 'duration_days': 15, 'embedding': [1, 0, 0]},
|
|
]
|
|
competitor_trips = [
|
|
{'id': 'p1', 'majority_price_usd': 2190, 'duration_days': 15, 'embedding': [1, 0, 0]}, # perfect match
|
|
{'id': 'p2', 'majority_price_usd': 1800, 'duration_days': 10, 'embedding': [0, 1, 0]}, # orthogonal, filtered out
|
|
]
|
|
|
|
matches = embedding_service.find_comparable_trips(client_trips, competitor_trips, threshold=0.75)
|
|
|
|
assert len(matches) == 1
|
|
assert matches[0]['competitor_product'] == 'p1'
|
|
assert matches[0]['price_difference'] == 2190 - 1999
|
|
assert matches[0]['duration_difference'] == 0
|
|
|
|
|
|
def test_find_comparable_trips_skips_records_without_embeddings():
|
|
client_trips = [{'trip_name': 'x', 'embedding': None}]
|
|
competitor_trips = [{'id': 'p1', 'embedding': [1, 0]}]
|
|
assert embedding_service.find_comparable_trips(client_trips, competitor_trips) == []
|
|
|
|
|
|
def test_safe_diff_returns_none_when_either_side_missing():
|
|
assert embedding_service._safe_diff(None, 5) is None
|
|
assert embedding_service._safe_diff(5, None) is None
|
|
assert embedding_service._safe_diff(10, 4) == 6
|
|
|
|
|
|
def test_embed_trip_calls_openrouter_embeddings_api(monkeypatch):
|
|
captured = {}
|
|
|
|
class _FakeEmbeddingResponse:
|
|
class _Data:
|
|
embedding = [0.1, 0.2, 0.3]
|
|
data = [_Data()]
|
|
|
|
class _FakeEmbeddings:
|
|
def create(self, model, input):
|
|
captured['model'] = model
|
|
captured['input'] = input
|
|
return _FakeEmbeddingResponse()
|
|
|
|
class _FakeOpenAI:
|
|
def __init__(self, api_key, base_url):
|
|
self.embeddings = _FakeEmbeddings()
|
|
|
|
fake_openai_module = types.ModuleType('openai')
|
|
fake_openai_module.OpenAI = _FakeOpenAI
|
|
monkeypatch.setitem(sys.modules, 'openai', fake_openai_module)
|
|
|
|
vector = embedding_service.embed_trip({
|
|
'trip_name': 'Inca Trail Classic', 'destination': 'Peru', 'duration_days': 15,
|
|
'activities': 'Trekking', 'start_location': 'Cusco',
|
|
})
|
|
|
|
assert vector == [0.1, 0.2, 0.3]
|
|
assert captured['model'] == embedding_service.EMBEDDING_MODEL
|
|
assert 'Peru' in captured['input']
|