import uuid from datetime import date from unittest.mock import patch from sqlmodel import Session from innercontext.api.products import ( ProductSuggestion, ShoppingSuggestionResponse, _build_shopping_context, _compute_days_since_last_used, _compute_replenishment_score, _extract_requested_product_ids, build_product_details_tool_handler, ) from innercontext.models import ( Product, ProductCategory, ProductInventory, SexAtBirth, SkinConcern, SkinConditionSnapshot, ) from innercontext.models.profile import UserProfile from innercontext.validators.shopping_validator import ( ShoppingValidationContext, ShoppingValidator, ) def test_build_shopping_context(session: Session): # Empty context ctx = _build_shopping_context(session, reference_date=date.today()) assert "USER PROFILE: no data" in ctx assert "(brak danych)" in ctx assert "POSIADANE PRODUKTY" in ctx profile = UserProfile(birth_date=date(1990, 1, 10), sex_at_birth=SexAtBirth.MALE) session.add(profile) session.commit() # Add snapshot snap = SkinConditionSnapshot( id=uuid.uuid4(), snapshot_date=date.today(), overall_state="fair", skin_type="combination", hydration_level=3, sensitivity_level=4, barrier_state="mildly_compromised", active_concerns=["redness"], priorities=["soothing"], ) session.add(snap) # Add product p = Product( id=uuid.uuid4(), short_id=str(uuid.uuid4())[:8], name="Soothing Serum", brand="BrandX", category="serum", recommended_time="both", leave_on=True, targets=["redness"], product_effect_profile={"soothing_strength": 4, "hydration_immediate": 1}, actives=[{"name": "Centella"}], ) session.add(p) session.commit() # Add inventory inv = ProductInventory( id=uuid.uuid4(), product_id=p.id, is_opened=True, remaining_level="medium", ) session.add(inv) session.commit() ctx = _build_shopping_context(session, reference_date=date(2026, 3, 5)) assert "USER PROFILE:" in ctx assert "Age: 36" in ctx assert "Sex at birth: male" in ctx assert "Typ skóry: combination" in ctx assert "Nawilżenie: 3/5" in ctx assert "Wrażliwość: 4/5" in ctx assert "Aktywne problemy: redness" in ctx assert "Priorytety: soothing" in ctx # Check product assert "[✓] id=" in ctx assert "Soothing Serum" in ctx assert f"id={p.id}" in ctx assert "BrandX" in ctx assert "targets=['redness']" in ctx assert "actives=['Centella']" in ctx assert "effects={'soothing': 4}" in ctx assert "stock_state=monitor" in ctx assert "opened_count=1" in ctx assert "sealed_backup_count=0" in ctx assert "lowest_remaining_level=medium" in ctx assert "replenishment_score=30" in ctx assert "replenishment_priority_hint=low" in ctx assert "repurchase_candidate=true" in ctx def test_build_shopping_context_flags_replenishment_signal(session: Session): product = Product( id=uuid.uuid4(), short_id=str(uuid.uuid4())[:8], name="Barrier Cleanser", brand="BrandY", category="cleanser", recommended_time="both", leave_on=False, product_effect_profile={}, ) session.add(product) session.commit() session.add( ProductInventory( id=uuid.uuid4(), product_id=product.id, is_opened=True, remaining_level="nearly_empty", ) ) session.commit() ctx = _build_shopping_context(session, reference_date=date.today()) assert "lowest_remaining_level=nearly_empty" in ctx assert "stock_state=urgent" in ctx assert "replenishment_priority_hint=high" in ctx def test_compute_replenishment_score_prefers_recent_staples_without_backup(): result = _compute_replenishment_score( has_stock=True, sealed_backup_count=0, lowest_remaining_level="low", days_since_last_used=2, category=ProductCategory.CLEANSER, ) assert result["replenishment_score"] == 95 assert result["replenishment_priority_hint"] == "high" assert result["repurchase_candidate"] is True assert result["replenishment_reason_codes"] == [ "low_opened", "recently_used", "staple_category", ] def test_compute_replenishment_score_downranks_sealed_backup_and_stale_usage(): result = _compute_replenishment_score( has_stock=True, sealed_backup_count=1, lowest_remaining_level="nearly_empty", days_since_last_used=70, category=ProductCategory.EXFOLIANT, ) assert result["replenishment_score"] == 0 assert result["replenishment_priority_hint"] == "none" assert result["repurchase_candidate"] is False assert result["replenishment_reason_codes"] == [ "has_sealed_backup", "stale_usage", "occasional_category", ] def test_compute_days_since_last_used_returns_none_without_usage(): assert _compute_days_since_last_used(None, date(2026, 3, 9)) is None assert _compute_days_since_last_used(date(2026, 3, 7), date(2026, 3, 9)) == 2 def test_suggest_shopping(client, session): with patch( "innercontext.api.products.call_gemini_with_function_tools" ) as mock_gemini: product = Product( id=uuid.uuid4(), short_id=str(uuid.uuid4())[:8], name="Owned Serum", brand="BrandX", category="serum", recommended_time="both", leave_on=True, product_effect_profile={}, ) session.add(product) session.commit() session.add( ProductInventory(id=uuid.uuid4(), product_id=product.id, is_opened=True) ) session.commit() mock_response = type( "Response", (), { "text": '{"suggestions": [{"category": "cleanser", "product_type": "cleanser", "priority": "high", "key_ingredients": ["glycerin"], "target_concerns": ["dehydration"], "recommended_time": "am", "frequency": "daily", "short_reason": "Brakuje lagodnego kroku myjacego rano.", "reason_to_buy_now": "Obecnie nie masz delikatnego produktu do porannego oczyszczania i wsparcia bariery.", "reason_not_needed_if_budget_tight": "Mozesz tymczasowo ograniczyc sie do samego splukania twarzy rano, jesli skora jest spokojna.", "fit_with_current_routine": "To domknie podstawowy krok cleanse bez dokladania agresywnych aktywow.", "usage_cautions": ["unikaj mocnego domywania przy podraznieniu"]}], "reasoning": "Test shopping"}' }, ) mock_gemini.return_value = (mock_response, None) r = client.post("/products/suggest") assert r.status_code == 200 data = r.json() assert len(data["suggestions"]) == 1 assert data["suggestions"][0]["product_type"] == "cleanser" assert data["suggestions"][0]["priority"] == "high" assert data["suggestions"][0]["short_reason"] assert data["suggestions"][0]["usage_cautions"] == [ "unikaj mocnego domywania przy podraznieniu" ] assert data["reasoning"] == "Test shopping" kwargs = mock_gemini.call_args.kwargs assert "USER PROFILE:" in kwargs["contents"] assert ( 'category: "cleanser" | "toner" | "essence"' in kwargs["config"].system_instruction ) assert ( 'recommended_time: "am" | "pm" | "both"' in kwargs["config"].system_instruction ) assert "function_handlers" in kwargs assert "get_product_details" in kwargs["function_handlers"] def test_suggest_shopping_invalid_json_returns_502(client): with patch( "innercontext.api.products.call_gemini_with_function_tools" ) as mock_gemini: mock_response = type("Response", (), {"text": "{"}) mock_gemini.return_value = (mock_response, None) r = client.post("/products/suggest") assert r.status_code == 502 assert "LLM returned invalid JSON" in r.json()["detail"] def test_suggest_shopping_invalid_schema_returns_502(client): with patch( "innercontext.api.products.call_gemini_with_function_tools" ) as mock_gemini: mock_response = type( "Response", (), { "text": '{"suggestions": [{"category": "cleanser", "product_type": "cleanser", "priority": "urgent", "key_ingredients": [], "target_concerns": [], "recommended_time": "am", "frequency": "daily", "short_reason": "x", "reason_to_buy_now": "y", "fit_with_current_routine": "z", "usage_cautions": []}], "reasoning": "Test shopping"}' }, ) mock_gemini.return_value = (mock_response, None) r = client.post("/products/suggest") assert r.status_code == 502 assert "LLM returned invalid shopping suggestion schema" in r.json()["detail"] assert "suggestions/0/priority" in r.json()["detail"] def test_suggest_shopping_invalid_target_concern_returns_502(client): with patch( "innercontext.api.products.call_gemini_with_function_tools" ) as mock_gemini: mock_response = type( "Response", (), { "text": '{"suggestions": [{"category": "cleanser", "product_type": "cleanser", "priority": "high", "key_ingredients": ["glycerin"], "target_concerns": ["inflammation"], "recommended_time": "am", "frequency": "daily", "short_reason": "Brakuje lagodnego kroku myjacego rano.", "reason_to_buy_now": "Obecnie nie masz delikatnego produktu do porannego oczyszczania i wsparcia bariery.", "fit_with_current_routine": "To domknie podstawowy krok cleanse bez dokladania agresywnych aktywow.", "usage_cautions": []}], "reasoning": "Test shopping"}' }, ) mock_gemini.return_value = (mock_response, None) r = client.post("/products/suggest") assert r.status_code == 502 assert "LLM returned invalid shopping suggestion schema" in r.json()["detail"] assert "suggestions/0/target_concerns/0" in r.json()["detail"] def test_shopping_context_medication_skip(session: Session): p = Product( id=uuid.uuid4(), short_id=str(uuid.uuid4())[:8], name="Epiduo", brand="Galderma", category="serum", recommended_time="pm", leave_on=True, is_medication=True, product_effect_profile={}, ) session.add(p) session.commit() ctx = _build_shopping_context(session, reference_date=date.today()) assert "Epiduo" not in ctx def test_extract_requested_product_ids_dedupes_and_limits(): ids = _extract_requested_product_ids( {"product_ids": ["a", "b", "a", 1, "c", "d"]}, max_ids=3, ) assert ids == ["a", "b", "c"] def test_shopping_tool_handlers_return_payloads(session: Session): product = Product( id=uuid.uuid4(), short_id=str(uuid.uuid4())[:8], name="Test Product", brand="Brand", category="serum", recommended_time="both", leave_on=True, inci=["Water", "Niacinamide"], actives=[{"name": "Niacinamide", "percent": 5, "functions": ["niacinamide"]}], context_rules={"safe_after_shaving": True}, product_effect_profile={}, ) payload = {"product_ids": [str(product.id)]} details = build_product_details_tool_handler([product])(payload) assert details["products"][0]["actives"][0]["name"] == "Niacinamide" assert "context_rules" in details["products"][0] assert details["products"][0]["last_used_on"] is None assert "inci" not in details["products"][0] def test_shopping_tool_handler_includes_last_used_on_from_mapping(session: Session): product = Product( id=uuid.uuid4(), name="Mapped Product", brand="Brand", category="serum", recommended_time="both", leave_on=True, product_effect_profile={}, ) payload = {"product_ids": [str(product.id)]} details = build_product_details_tool_handler( [product], last_used_on_by_product={str(product.id): date(2026, 3, 1)}, )(payload) assert details["products"][0]["last_used_on"] == "2026-03-01" def test_shopping_validator_accepts_freeform_product_type_and_frequency(): response = ShoppingSuggestionResponse( suggestions=[ ProductSuggestion( category="spot_treatment", product_type="Punktowy preparat na wypryski z ichtiolem lub cynkiem", priority="high", key_ingredients=["ichtiol", "cynk"], target_concerns=["acne"], recommended_time="pm", frequency="Codziennie (punktowo na zmiany)", short_reason="Pomaga opanowac aktywne zmiany bez dokladania pelnego aktywu na cala twarz.", reason_to_buy_now="Brakuje Ci dedykowanego produktu punktowego na pojedyncze wypryski.", fit_with_current_routine="Mozesz dolozyc go tylko na zmiany po serum lub zamiast mocniejszego aktywu.", usage_cautions=["stosuj tylko miejscowo"], ), ProductSuggestion( category="mask", product_type="Lagodna maska oczyszczajaca", priority="low", key_ingredients=["glinka"], target_concerns=["sebum_excess"], recommended_time="pm", frequency="1 raz w tygodniu", short_reason="To opcjonalne wsparcie przy nadmiarze sebum.", reason_to_buy_now="Moze pomoc domknac sporadyczne oczyszczanie, gdy skora jest bardziej przetluszczona.", fit_with_current_routine="Najlepiej traktowac to jako dodatkowy krok, nie zamiennik podstaw rutyny.", usage_cautions=[], ), ], reasoning="Test", ) result = ShoppingValidator().validate( response, ShoppingValidationContext( owned_product_ids=set(), valid_categories=set(ProductCategory), valid_targets=set(SkinConcern), ), ) assert not any("unusual frequency" in warning for warning in result.warnings)