Spaces:
Running
Running
| import pytest | |
| from httpx import AsyncClient | |
| from unittest.mock import patch, AsyncMock, MagicMock | |
| async def test_match_program_success(async_client: AsyncClient, auth_headers): | |
| with patch("endpoints.projects.get_llm") as mock_get_llm, patch( | |
| "core.ncbr_client.ncbr_client.get_active_nabory", new_callable=AsyncMock | |
| ) as mock_ncbr, patch( | |
| "core.parp_client.parp_client.get_active_nabory", new_callable=AsyncMock | |
| ) as mock_parp: | |
| mock_ncbr.return_value = [] | |
| mock_parp.return_value = [] | |
| mock_llm = MagicMock() | |
| mock_structured_llm = MagicMock() | |
| # Build fake programs list | |
| class DummyExplanation: | |
| def dict(self): | |
| return {"reason": "It matches", "criteria": ["R&D"], "risks": "None"} | |
| class DummyProgram: | |
| def __init__(self): | |
| self.id = 1 | |
| self.name = "FENG.01.01-IP.02-001/23" | |
| self.type = "smart" | |
| self.match = 95 | |
| self.chance = 80 | |
| self.amount = "10-50 mln PLN" | |
| self.shortDesc = "Ścieżka SMART" | |
| self.fullDesc = "Full description" | |
| self.explanation = DummyExplanation() | |
| self.criteria = ["R&D"] | |
| self.risks = "None" | |
| def dict(self): | |
| return { | |
| "id": self.id, | |
| "name": self.name, | |
| "type": self.type, | |
| "match": self.match, | |
| "chance": self.chance, | |
| "amount": self.amount, | |
| "shortDesc": self.shortDesc, | |
| "fullDesc": self.fullDesc, | |
| "explanation": self.explanation.dict(), | |
| "criteria": self.criteria, | |
| "risks": self.risks, | |
| } | |
| program_mock = DummyProgram() | |
| mock_output = MagicMock() | |
| mock_output.programs = [program_mock] | |
| mock_output.clarifying_questions = ["What is the TRL level?"] | |
| mock_output.model_dump.return_value = { | |
| "programs": [ | |
| { | |
| "id": 1, | |
| "name": "FENG.01.01-IP.02-001/23", | |
| "type": "smart", | |
| "match": 95, | |
| "chance": 80, | |
| "amount": "10-50 mln PLN", | |
| "shortDesc": "Ścieżka SMART", | |
| "fullDesc": "Full description", | |
| "explanation": { | |
| "reason": "It matches", | |
| "criteria": ["R&D"], | |
| "risks": "None", | |
| }, | |
| "criteria": ["R&D"], | |
| "risks": "None", | |
| } | |
| ], | |
| "clarifying_questions": ["What is the TRL level?"], | |
| } | |
| mock_structured_llm.invoke.return_value = mock_output | |
| mock_structured_llm.return_value = mock_output | |
| mock_get_llm.return_value = mock_structured_llm | |
| payload = { | |
| "description": "We are building an AI software.", | |
| "company_type": "sme", | |
| "company_size": "small", | |
| "voivodeship": "mazowieckie", | |
| "innovation_type": "product", | |
| } | |
| response = await async_client.post( | |
| "/api/projects/match-program", json=payload, headers=auth_headers | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "programs" in data | |
| assert "clarifying_questions" in data | |
| assert len(data["programs"]) == 1 | |
| assert "What is the TRL level?" in data["clarifying_questions"] | |
| async def test_match_program_empty_description(async_client: AsyncClient, auth_headers): | |
| payload = { | |
| "description": "", | |
| } | |
| with patch("endpoints.projects.get_llm") as mock_get_llm, patch( | |
| "core.ncbr_client.ncbr_client.get_active_nabory", new_callable=AsyncMock | |
| ) as mock_ncbr, patch( | |
| "core.parp_client.parp_client.get_active_nabory", new_callable=AsyncMock | |
| ) as mock_parp: | |
| mock_ncbr.return_value = [] | |
| mock_parp.return_value = [] | |
| mock_llm = MagicMock() | |
| mock_structured_llm = MagicMock() | |
| mock_output = MagicMock() | |
| mock_output.programs = [] | |
| mock_output.clarifying_questions = ["Proszę opisać swój projekt."] | |
| mock_output.model_dump.return_value = { | |
| "programs": [], | |
| "clarifying_questions": ["Proszę opisać swój projekt."], | |
| } | |
| mock_structured_llm.invoke.return_value = mock_output | |
| mock_structured_llm.return_value = mock_output | |
| mock_get_llm.return_value = mock_structured_llm | |
| response = await async_client.post( | |
| "/api/projects/match-program", json=payload, headers=auth_headers | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "programs" in data | |
| assert "clarifying_questions" in data | |
| async def test_match_program_llm_fallback(async_client: AsyncClient, auth_headers): | |
| payload = { | |
| "description": "Fallback test.", | |
| } | |
| with patch("endpoints.projects.get_llm") as mock_get_llm, patch( | |
| "core.ncbr_client.ncbr_client.get_active_nabory", new_callable=AsyncMock | |
| ) as mock_ncbr, patch( | |
| "core.parp_client.parp_client.get_active_nabory", new_callable=AsyncMock | |
| ) as mock_parp: | |
| mock_ncbr.return_value = [] | |
| mock_parp.return_value = [] | |
| mock_llm = MagicMock() | |
| mock_structured_llm = MagicMock() | |
| mock_structured_llm.invoke.side_effect = Exception("LLM Error") | |
| mock_structured_llm.side_effect = Exception("LLM Error") | |
| mock_get_llm.return_value = mock_structured_llm | |
| response = await async_client.post( | |
| "/api/projects/match-program", json=payload, headers=auth_headers | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "programs" in data | |
| assert len(data["programs"]) == 3 | |
| assert data["programs"][0]["name"] == "Ścieżka SMART (FENG)" | |