"""BrandForge caps how many products a catalogue holds, and meters nothing else.""" import pytest from app.services.infra import quotas from app.services.infra.quotas import ( AIVORA, BRANDFORGE, QuotaExceededError, QuotaValidationError, check_quota, _map_limits_from_features, get_mapped_limits, ) TENANT = "org_test" BRANDFORGE_PLAN = [{ "productName": "BrandForge", "features": [ {"featureKey": "Product_Count", "featureName": "Product count", "cappingValue": "100"}, {"featureKey": "Brand_Analysis", "featureName": "BrandForge Analysis", "cappingValue": "3"}, ], }] AIVORA_PLAN = [{ "productName": "Aivora", "features": [ {"featureName": "Aivora AI Token", "cappingValue": "500000"}, {"featureName": "Aivora Ticket Count", "cappingValue": "100"}, {"featureName": "Aivora Vector DB", "cappingValue": "50"}, {"featureName": "Aivora Conversation Count", "cappingValue": "200"}, ], }] def _plan(monkeypatch, features): monkeypatch.setattr(quotas, "fetch_tenant_limits", lambda t: (features, None)) monkeypatch.setattr(quotas, "get_quota_start_date", lambda *a, **kw: None) monkeypatch.setattr(quotas, "send_quota_notification", lambda *a, **kw: None) monkeypatch.setattr(quotas, "redis_client", None) def _holding(monkeypatch, count): monkeypatch.setattr(quotas, "get_catalogue_product_count", lambda t: count) # --- reading BrandForge's own numbers ----------------------------------------- def test_a_product_count_feature_is_recognised(): # It matched no category before, which is why a BrandForge plan mapped to # nothing and every build was refused. mapping = _map_limits_from_features(BRANDFORGE_PLAN[0]["features"], BRANDFORGE) assert mapping == {"product_count": 100} def test_brandforge_is_not_judged_by_aivoras_keys(): # Demanding ai_tokens, tickets, vector DB and conversations of a plan that # has none of them is what made BrandForge unusable. assert get_mapped_limits(BRANDFORGE_PLAN, BRANDFORGE) == {"product_count": 100} def test_aivora_still_needs_all_four_of_its_own(): limits = get_mapped_limits(AIVORA_PLAN, AIVORA) assert limits["ai_tokens"] == 500000 assert limits["ai_conversations"] == 200 def test_one_product_never_reads_anothers_numbers(): both = AIVORA_PLAN + BRANDFORGE_PLAN assert get_mapped_limits(both, BRANDFORGE) == {"product_count": 100} assert get_mapped_limits(both, AIVORA)["ai_tokens"] == 500000 # --- the cap ------------------------------------------------------------------ def test_adding_within_the_cap_is_allowed(monkeypatch): _plan(monkeypatch, BRANDFORGE_PLAN) _holding(monkeypatch, 80) check_quota(TENANT, "product_count", requested_amount=20, product_name=BRANDFORGE) def test_going_over_the_cap_is_refused(monkeypatch): _plan(monkeypatch, BRANDFORGE_PLAN) _holding(monkeypatch, 80) with pytest.raises(QuotaExceededError) as raised: check_quota(TENANT, "product_count", requested_amount=50, product_name=BRANDFORGE) # The merchant can act on this one, so it says the number rather than # hiding behind "temporarily unavailable". assert "100 products" in str(raised.value) def test_a_full_catalogue_still_accepts_a_refresh(monkeypatch): # A re-sync adds nothing: every product already exists. Refusing here would # freeze a merchant's prices the moment they reached their limit. _plan(monkeypatch, BRANDFORGE_PLAN) _holding(monkeypatch, 100) check_quota(TENANT, "product_count", requested_amount=0, product_name=BRANDFORGE) def test_an_over_limit_catalogue_is_not_broken_by_the_cap(monkeypatch): # Grandfathered: 1,264 products against a cap of 100 keeps working, it just # cannot grow. Deleting to fit would be the worst possible behaviour. _plan(monkeypatch, BRANDFORGE_PLAN) _holding(monkeypatch, 1264) check_quota(TENANT, "product_count", requested_amount=0, product_name=BRANDFORGE) # --- what BrandForge does NOT meter ------------------------------------------- def test_brandforge_does_not_meter_tokens(monkeypatch): # The whole point: building a catalogue spends embeddings, and a BrandForge # plan has no token allowance to spend from. That is unmetered, not zero. _plan(monkeypatch, BRANDFORGE_PLAN) check_quota(TENANT, "ai_tokens", requested_amount=999_999, product_name=BRANDFORGE) def test_a_plan_without_brandforge_does_not_block_the_catalogue(monkeypatch): # The live tenants: 218 and 1,264 products on an Aivora-only plan. Refusing # because no limit is configured would stop customers who work today. _plan(monkeypatch, AIVORA_PLAN) _holding(monkeypatch, 218) check_quota(TENANT, "product_count", requested_amount=50, product_name=BRANDFORGE) # --- Aivora's own behaviour is unchanged -------------------------------------- def test_aivora_still_fails_closed_when_it_cannot_be_read(monkeypatch): # Unmetered AI spend is unbounded cost, so the chat path must keep # refusing when limits cannot be established. _plan(monkeypatch, BRANDFORGE_PLAN) with pytest.raises(QuotaValidationError): check_quota(TENANT, "ai_tokens", requested_amount=1, product_name=AIVORA) BRANDFORGE_NO_CAP = [{ "productName": "BrandForge", # feature test plan, on 12 subscriptions: BrandForge is on the plan but # carries no Product count. "features": [{"featureKey": "Brand_Analysis", "featureName": "BrandForge Analysis", "cappingValue": "3"}], }] def test_brandforge_without_a_cap_does_not_refuse(monkeypatch): # This raised rather than falling open: the deterministic mapper found no # product_count, the LLM fallback was asked to invent one, and validation # rejected what came back. _plan(monkeypatch, BRANDFORGE_NO_CAP) _holding(monkeypatch, 195) check_quota(TENANT, "product_count", requested_amount=10, product_name=BRANDFORGE) def test_a_capless_brandforge_plan_never_calls_the_model(monkeypatch): # Asking a model to read a limit that is not there costs money and can only # hallucinate or fail. _plan(monkeypatch, BRANDFORGE_NO_CAP) def fail(*a, **kw): raise AssertionError("the model must not be consulted here") monkeypatch.setattr(quotas.openai_client.chat.completions, "create", fail) assert get_mapped_limits(BRANDFORGE_NO_CAP, BRANDFORGE) == {} def test_a_tenant_with_no_plan_can_still_ingest(monkeypatch): # No subscription row, or a master database that cannot be reached: # catalogue ingestion must not depend on the control plane. monkeypatch.setattr(quotas, "fetch_tenant_limits", lambda t: (None, None)) _holding(monkeypatch, 10) check_quota(TENANT, "product_count", requested_amount=5, product_name=BRANDFORGE) def test_a_tenant_with_no_plan_still_blocks_the_chat(monkeypatch): monkeypatch.setattr(quotas, "fetch_tenant_limits", lambda t: (None, None)) with pytest.raises(QuotaValidationError): check_quota(TENANT, "ai_tokens", product_name=AIVORA) def test_no_features_at_all_still_fails_closed_for_aivora(monkeypatch): _plan(monkeypatch, []) with pytest.raises(QuotaValidationError): check_quota(TENANT, "ai_tokens", product_name=AIVORA)