Spaces:

Shaikhsarib
/

EatlyticApp

Configuration error

File size: 10,205 Bytes

import os
import pytest
from unittest.mock import patch
import httpx
import app.database.connection as db_mod
from app.database.connection import db_conn
from main import app

# ── Self-contained fixture to redirect DB to temporary space ──────────────
@pytest.fixture(autouse=True)
def use_test_db(tmp_path, monkeypatch):
    """Redirect DB to a temp file for each test."""
    db_path = str(tmp_path / "test.db")
    monkeypatch.setattr(db_mod, "DATA_DIR", str(tmp_path))
    monkeypatch.setattr(db_mod, "DB_FILE", db_path)
    db_mod.init_db()
    
    # Seed mock B2B Organization & API Key
    with db_conn() as conn:
        # Seed user
        conn.execute("""
            INSERT INTO users (id, email, name, is_pro)
            VALUES (?, ?, ?, 1)
        """, ("admin_test_123", "b2b@test.com", "Test B2B Admin"))
        
        # Seed organization
        conn.execute("""
            INSERT INTO organizations (id, name, plan, admin_id)
            VALUES (?, ?, ?, ?)
        """, ("org_test_123", "Test Corp", "business", "admin_test_123"))
        
        # Seed API Key
        conn.execute("""
            INSERT INTO api_keys (api_key, client_name, organization_id, plan, scans_this_month, active, month)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, ("eatlytic_test_api_key_xyz_789", "Test B2B Client", "org_test_123", "business", 0, 1, "2026-05"))
        
        # Seed Amul Butter for fast offline matching
        cursor = conn.execute("""
            INSERT INTO food_products (
                name, brand, category, barcode, calories_100g,
                protein_100g, carbs_100g, fat_100g, sodium_100g,
                fiber_100g, sugar_100g, eatlytic_score, verified,
                scan_count
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 0)
        """, ("Pasteurised Butter", "Amul", "Dairy", "8901262010018", 722.0, 0.6, 0.0, 80.0, 830.0, 0.0, 0.0, 4))
        amul_id = cursor.lastrowid
        conn.execute("""
            INSERT INTO food_products_extra (id, sat_fat_100g, ingredients_raw, source)
            VALUES (?, ?, ?, ?)
        """, (amul_id, 51.0, "Butter, Common Salt", "verified_seeder"))

    yield

# ══ B2B INTEGRATION TESTS ═════════════════════════════════════════════
class TestB2BApiIntegration:
    @pytest.mark.asyncio
    async def test_b2b_auth_missing_key(self):
        """Should reject request with 401 when X-Eatlytic-Key header is missing."""
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.post(
                "/api/v1/analyze",
                files=dummy_file,
                data={"language": "en"}
            )
        assert resp.status_code == 401
        assert "missing" in resp.json()["detail"].lower()

    @pytest.mark.asyncio
    async def test_b2b_auth_invalid_key(self):
        """Should reject request with 401 when key is invalid."""
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.post(
                "/api/v1/analyze",
                headers={"X-Eatlytic-Key": "invalid_api_key_abc_123"},
                files=dummy_file,
                data={"language": "en"}
            )
        assert resp.status_code == 401
        assert "invalid or inactive" in resp.json()["detail"].lower()

    @pytest.mark.asyncio
    async def test_b2b_auth_inactive_key(self):
        """Should reject request with 401 when key is suspended/inactive in DB."""
        with db_conn() as conn:
            conn.execute("UPDATE api_keys SET active=0 WHERE api_key=?", ("eatlytic_test_api_key_xyz_789",))
            
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.post(
                "/api/v1/analyze",
                headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"},
                files=dummy_file,
                data={"language": "en"}
            )
        assert resp.status_code == 401
        assert "invalid or inactive" in resp.json()["detail"].lower()

    @pytest.mark.asyncio
    @patch("app.api.v1.b2b.run_ocr")
    @patch("app.ai.llm.engine.call_llm")
    async def test_b2b_analyze_success(self, mock_call_llm, mock_run_ocr):
        """Should successfully parse label offline and return Atwater compliant JSON with $0 cost."""
        mock_run_ocr.return_value = {
            "text": "Check out Amul Pasteurised Butter. Barcode: 8901262010018",
            "word_count": 25,
            "is_valid": True,
            "avg_confidence": 0.95
        }
        
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.post(
                "/api/v1/analyze",
                headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"},
                files=dummy_file,
                data={"language": "en", "persona": "adult"}
            )
        
        # Verify status code
        assert resp.status_code == 200
        
        # Verify LLM bypass ($0 cost!)
        mock_call_llm.assert_not_called()
        
        # Verify offline DB-first result details
        result = resp.json()
        assert result["product_name"] == "Amul Pasteurised Butter"
        assert result["score"] == 4
        assert result["calories"] == 722.0
        assert result["fat"] == 80.0
        
        # Check B2B audit metadata block addition
        assert "audit" in result
        assert result["audit"]["client_ref"] == "Test B2B Client"
        
        # Verify scan count was incremented in DB
        with db_conn() as conn:
            row = conn.execute("SELECT scans_this_month FROM api_keys WHERE api_key=?", ("eatlytic_test_api_key_xyz_789",)).fetchone()
        assert row["scans_this_month"] == 1

    @pytest.mark.asyncio
    async def test_b2b_rate_limit_exceeded(self):
        """Should return 429 when B2B monthly limit is reached."""
        with db_conn() as conn:
            conn.execute("UPDATE api_keys SET scans_this_month=1000 WHERE api_key=?", ("eatlytic_test_api_key_xyz_789",))
            
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.post(
                "/api/v1/analyze",
                headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"},
                files=dummy_file,
                data={"language": "en"}
            )
        assert resp.status_code == 429
        assert "limit" in resp.json()["detail"].lower()

    @pytest.mark.asyncio
    @patch("app.api.v1.b2b.run_ocr")
    async def test_b2b_analyze_blurry_image_blocked(self, mock_run_ocr):
        """Should block B2B scan and return 400 with 'blurry_image' error when image lacks confidence/words."""
        mock_run_ocr.return_value = {
            "text": "Unreadable",
            "avg_confidence": 0.10,
            "word_count": 2
        }
        
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.post(
                "/api/v1/analyze",
                headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"},
                files=dummy_file,
                data={"language": "en", "persona": "adult"}
            )
        
        assert resp.status_code == 400
        result = resp.json()
        assert result["error"] == "blurry_image"
        assert "too low" in result["message"].lower()

    @pytest.mark.asyncio
    async def test_b2b_usage_endpoint(self):
        """Should return correct B2B usage metrics via the authenticated GET /usage endpoint without incrementing."""
        # 1. Telemetry check before any scan should be 0
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.get(
                "/api/v1/usage",
                headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"}
            )
        assert resp.status_code == 200
        assert resp.json()["scans_used_this_month"] == 0

        # 2. Run a successful scan to trigger increment
        mock_run_ocr = {
            "text": "Check out Amul Pasteurised Butter. Barcode: 8901262010018",
            "word_count": 25,
            "is_valid": True,
            "avg_confidence": 0.95
        }
        dummy_file = {"image": ("label.jpg", b"\x00\x00\x00", "image/jpeg")}
        with patch("app.api.v1.b2b.run_ocr", return_value=mock_run_ocr):
            async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
                await ac.post(
                    "/api/v1/analyze",
                    headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"},
                    files=dummy_file,
                    data={"language": "en", "persona": "adult"}
                )

        # 3. Telemetry check after scan should show 1
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac:
            resp = await ac.get(
                "/api/v1/usage",
                headers={"X-Eatlytic-Key": "eatlytic_test_api_key_xyz_789"}
            )
        assert resp.status_code == 200
        result = resp.json()
        assert result["client_name"] == "Test B2B Client"
        assert result["plan"] == "business"
        assert result["scans_limit"] == 1000
        assert result["scans_used_this_month"] == 1
        assert result["scans_remaining"] == 999