Spaces:
Sleeping
Sleeping
| import json | |
| import pytest | |
| from app import app as flask_app | |
| # TODO: Re-enable these tests after memory monitoring is stabilized | |
| # Current issue: Memory monitoring endpoints may behave differently in CI environment | |
| # pytestmark = pytest.mark.skip( | |
| # reason="Memory monitoring endpoints disabled in CI until stabilized" | |
| # ) | |
| def app(): | |
| yield flask_app | |
| def client(app): | |
| return app.test_client() | |
| def test_health_endpoint(client): | |
| """ | |
| Tests the /health endpoint. | |
| """ | |
| response = client.get("/health") | |
| assert response.status_code == 200 | |
| # Check that required fields are present | |
| response_data = response.json | |
| assert "status" in response_data | |
| assert "memory_mb" in response_data | |
| assert "timestamp" in response_data | |
| # Check status is ok | |
| assert response_data["status"] == "ok" | |
| # Check memory_mb is a number >= 0 | |
| assert isinstance(response_data["memory_mb"], (int, float)) | |
| assert response_data["memory_mb"] >= 0 | |
| def test_memory_diagnostics_endpoint(client): | |
| """Test /memory/diagnostics basic response.""" | |
| resp = client.get("/memory/diagnostics") | |
| assert resp.status_code == 200 | |
| data = resp.get_json() | |
| assert data["status"] == "success" | |
| assert "memory" in data | |
| assert "summary" in data["memory"] | |
| assert "rss_mb" in data["memory"]["summary"] | |
| def test_memory_diagnostics_with_top(client): | |
| """Test /memory/diagnostics with include_top param (should not error).""" | |
| resp = client.get("/memory/diagnostics?include_top=1&limit=3") | |
| assert resp.status_code == 200 | |
| data = resp.get_json() | |
| assert data["status"] == "success" | |
| # top_allocations may or may not be present depending on tracemalloc flag, | |
| # just ensure no error | |
| assert "memory" in data | |
| def test_memory_force_clean_endpoint(client): | |
| """Test POST /memory/force-clean returns summary.""" | |
| resp = client.post("/memory/force-clean", json={"label": "test"}) | |
| assert resp.status_code == 200 | |
| data = resp.get_json() | |
| assert data["status"] == "success" | |
| assert data["label"] == "test" | |
| assert "summary" in data | |
| assert "rss_mb" in data["summary"] or "rss_mb" in data["summary"].get("summary", {}) | |
| def test_index_endpoint(client): | |
| """ | |
| Tests the / endpoint. | |
| """ | |
| response = client.get("/") | |
| assert response.status_code == 200 | |
| def test_ingest_endpoint_exists(): | |
| """Test that the ingest endpoint is available""" | |
| from app import app | |
| client = app.test_client() | |
| response = client.post("/ingest") | |
| # Should not be 404 (not found) | |
| assert response.status_code != 404 | |
| class TestSearchEndpoint: | |
| """Test cases for the /search endpoint""" | |
| def test_search_endpoint_valid_request(self, client): | |
| """Test search endpoint with valid request""" | |
| request_data = {"query": "remote work policy", "top_k": 3, "threshold": 0.3} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 200 | |
| data = response.get_json() | |
| assert data["status"] == "success" | |
| assert data["query"] == "remote work policy" | |
| assert "results_count" in data | |
| assert "results" in data | |
| assert isinstance(data["results"], list) | |
| def test_search_endpoint_minimal_request(self, client): | |
| """Test search endpoint with minimal request (only query)""" | |
| request_data = {"query": "employee benefits"} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 200 | |
| data = response.get_json() | |
| assert data["status"] == "success" | |
| assert data["query"] == "employee benefits" | |
| def test_search_endpoint_missing_query(self, client): | |
| """Test search endpoint with missing query parameter""" | |
| request_data = {"top_k": 5} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 400 | |
| data = response.get_json() | |
| assert data["status"] == "error" | |
| assert "Query parameter is required" in data["message"] | |
| def test_search_endpoint_empty_query(self, client): | |
| """Test search endpoint with empty query""" | |
| request_data = {"query": ""} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 400 | |
| data = response.get_json() | |
| assert data["status"] == "error" | |
| assert "non-empty string" in data["message"] | |
| def test_search_endpoint_invalid_top_k(self, client): | |
| """Test search endpoint with invalid top_k parameter""" | |
| request_data = {"query": "test query", "top_k": -1} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 400 | |
| data = response.get_json() | |
| assert data["status"] == "error" | |
| assert "positive integer" in data["message"] | |
| def test_search_endpoint_invalid_threshold(self, client): | |
| """Test search endpoint with invalid threshold parameter""" | |
| request_data = {"query": "test query", "threshold": 1.5} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 400 | |
| data = response.get_json() | |
| assert data["status"] == "error" | |
| assert "between 0 and 1" in data["message"] | |
| def test_search_endpoint_non_json_request(self, client): | |
| """Test search endpoint with non-JSON request""" | |
| response = client.post("/search", data="not json", content_type="text/plain") | |
| assert response.status_code == 400 | |
| data = response.get_json() | |
| assert data["status"] == "error" | |
| assert "application/json" in data["message"] | |
| def test_search_endpoint_result_structure(self, client): | |
| """Test that search results have the correct structure""" | |
| request_data = {"query": "policy"} | |
| response = client.post("/search", data=json.dumps(request_data), content_type="application/json") | |
| assert response.status_code == 200 | |
| data = response.get_json() | |
| if data["results_count"] > 0: | |
| result = data["results"][0] | |
| assert "chunk_id" in result | |
| assert "content" in result | |
| assert "similarity_score" in result | |
| assert "metadata" in result | |
| assert isinstance(result["similarity_score"], (int, float)) | |