tests/test_cloud.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Cloud tests for CarsRUS Space on Hugging Face.
Runs against: https://huggingface.co/spaces/galbendavids/CarsRUS

Requires: pip install gradio_client
Run: PYTHONPATH=. python tests/test_cloud.py
     or: bash tests/run_cloud_tests.sh
"""

import os
import sys

_tests_dir = os.path.dirname(os.path.abspath(__file__))
_project_root = os.path.dirname(_tests_dir)
if _project_root not in sys.path:
    sys.path.insert(0, _project_root)

# Space URL (public Space – no token needed for read)
SPACE_URL = os.environ.get("CARSRUS_SPACE_URL", "galbendavids/CarsRUS")


def get_client():
    """Create Gradio client for the Space. Uses HF token from env if set."""
    try:
        from gradio_client import Client
    except ImportError:
        print("❌ gradio_client not installed. Run: pip install gradio_client")
        sys.exit(1)
    hf_token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
    return Client(SPACE_URL, hf_token=hf_token or None)


def collect_chat_response(client, message: str):
    """Call the chat endpoint and return the full response (streaming collected)."""
    # Space API: predict(message, api_name="/chat") -> response (str)
    result = client.predict(message, api_name="/chat")
    return result


def test_space_reachable():
    """Space is up and client can connect."""
    print("\n[Cloud] test_space_reachable...")
    client = get_client()
    # Just ensure we can get API info (view_api or similar)
    try:
        api_info = client.view_api()
        print(f"   Endpoints: {list(api_info.get('named_endpoints', api_info) if isinstance(api_info, dict) else 'ok')}")
    except Exception as e:
        print(f"   view_api: {e}")
    print("✅ test_space_reachable passed")
    return client


def test_chat_supported_car(client):
    """Ask about a supported car (e.g. Audi RS3) – expect substantive answer, no config error."""
    print("\n[Cloud] test_chat_supported_car...")
    message = "Tell me about the Audi RS3"
    try:
        result = collect_chat_response(client, message)
    except Exception as e:
        print(f"   Call failed: {e}")
        try:
            api = client.view_api()
            print(f"   API: {api}")
        except Exception:
            pass
        raise
    text = result if isinstance(result, str) else str(result)
    # Should not be config/init error
    assert "Configuration Error" not in text and "Initialization Error" not in text, (
        f"Expected normal answer, got: {text[:200]}"
    )
    # Should mention something about the car or content
    assert len(text.strip()) > 50, f"Response too short: {text[:200]}"
    print(f"   Response length: {len(text)} chars")
    print("✅ test_chat_supported_car passed")


def test_chat_unsupported_car(client):
    """Ask about an unsupported car – expect refusal or supported list."""
    print("\n[Cloud] test_chat_unsupported_car...")
    message = "What do you think about BMW X5?"
    try:
        result = collect_chat_response(client, message)
    except Exception as e:
        print(f"   Call failed: {e}")
        raise
    text = result if isinstance(result, str) else str(result)
    assert "Configuration Error" not in text and "Initialization Error" not in text
    # Refusal or supported list
    has_refusal = (
        "not in my knowledge" in text.lower()
        or "לא נמצא" in text
        or "supported" in text.lower()
        or "נתמכים" in text
        or "Citroen" in text
        or "Audi RS3" in text
    )
    assert has_refusal or len(text) > 20, f"Expected refusal/supported list, got: {text[:300]}"
    print("✅ test_chat_unsupported_car passed")


def test_chat_comparison(client):
    """Ask to compare two supported cars – expect comparison content."""
    print("\n[Cloud] test_chat_comparison...")
    message = "Compare Audi RS3 vs Hyundai Elantra N"
    try:
        result = collect_chat_response(client, message)
    except Exception as e:
        print(f"   Call failed: {e}")
        raise
    text = result if isinstance(result, str) else str(result)
    assert "Configuration Error" not in text and "Initialization Error" not in text
    assert len(text.strip()) > 30, f"Response too short: {text[:200]}"
    print("✅ test_chat_comparison passed")


def test_chat_hebrew(client):
    """Hebrew query – app should respond (Hebrew or English)."""
    print("\n[Cloud] test_chat_hebrew...")
    message = "ספר לי על אודי RS3"
    try:
        result = collect_chat_response(client, message)
    except Exception as e:
        print(f"   Call failed: {e}")
        raise
    text = result if isinstance(result, str) else str(result)
    assert "Configuration Error" not in text and "Initialization Error" not in text
    assert len(text.strip()) > 20, f"Response too short: {text[:200]}"
    print("✅ test_chat_hebrew passed")


def test_chat_link_co_01(client):
    """Link & Co 01 is in the knowledge base – must NOT say 'not in my knowledge'."""
    print("\n[Cloud] test_chat_link_co_01...")
    for message in ["ספר על לינק אנד קו 01", "Tell me about Link and Co 01"]:
        try:
            result = collect_chat_response(client, message)
        except Exception as e:
            print(f"   Call failed for {message!r}: {e}")
            raise
        text = result if isinstance(result, str) else str(result)
        assert "Configuration Error" not in text and "Initialization Error" not in text
        assert "not in my knowledge" not in text and "לא נמצא בבסיס הידע" not in text, (
            f"Link & Co 01 is in scraped_data (link-and-co-01-2026); got refusal: {text[:300]}"
        )
        assert len(text.strip()) > 50, f"Response too short for {message!r}: {text[:200]}"
    print("✅ test_chat_link_co_01 passed")


def run_all():
    """Run all cloud tests. Exit 0 if all pass."""
    print("=" * 60)
    print("CarsRUS – Cloud tests")
    print(f"Space: {SPACE_URL}")
    print("=" * 60)
    try:
        client = test_space_reachable()
        test_chat_supported_car(client)
        test_chat_unsupported_car(client)
        test_chat_comparison(client)
        test_chat_hebrew(client)
        test_chat_link_co_01(client)
    except Exception as e:
        print(f"\n❌ Cloud test failed: {e}")
        import traceback
        traceback.print_exc()
        return 1
    print("\n✅ All cloud tests passed.")
    return 0


if __name__ == "__main__":
    sys.exit(run_all())