"""Tests koda ģenerēšanai.""" import sys from pathlib import Path from typing import Any from unittest.mock import AsyncMock, patch import pytest from fastapi import HTTPException from maris_core.code.generate_code import ( CodeRequest, FixCodeRequest, ProjectFile, _detect_stack, _extract_code_block, _extract_project_files, fix_code, generate_code, ) def test_extract_code_block_with_fences() -> None: text = "Šeit ir kods:\n```\nprint('hello')\n```\nPaskaidrojums." code, explanation = _extract_code_block(text, "Python") assert "print" in code assert "Paskaidrojums" in explanation def test_extract_code_block_no_fences() -> None: text = "print('hello world')" code, explanation = _extract_code_block(text, "Python") assert "print" in code def test_extract_project_files_from_structured_json_payload() -> None: text = """```json { "explanation": "Pilns kalkulatora projekts.", "entrypoint": "index.html", "files": [ {"path": "index.html", "content": ""}, {"path": "assets/app.js", "content": "console.log('ok');"} ] } ```""" files, entrypoint, explanation = _extract_project_files(text, language="HTML/CSS/JavaScript") assert files == [ ProjectFile(path="index.html", content="", absolute_path=None), ProjectFile(path="assets/app.js", content="console.log('ok');", absolute_path=None), ] assert entrypoint == "index.html" assert explanation == "Pilns kalkulatora projekts." def test_detect_stack_prefers_nextjs_from_prompt() -> None: detected = _detect_stack( "Uztaisi Next.js dashboard ar app router un TypeScript", "Python", None, ) assert detected == "nextjs" @pytest.mark.asyncio async def test_generate_code_requires_text_model() -> None: with ( patch("maris_core.code.generate_code.get_pipeline", return_value=None), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): req = CodeRequest(prompt="Hello world skripts", language="Python") with pytest.raises(HTTPException) as exc_info: await generate_code(req) assert exc_info.value.status_code == 503 @pytest.mark.asyncio async def test_generate_code_uses_requested_hf_fallback_model_when_text_runtime_is_unavailable() -> ( None ): class FakeClient: def chat_completion( self, *, model: str, messages: list[dict[str, str]], max_tokens: int, temperature: float, ) -> dict[str, Any]: del messages, max_tokens, temperature assert model == "Qwen/Qwen2.5-Coder-32B-Instruct" return { "choices": [ { "message": { "content": "```python\ndef normalize_email(value: str) -> str:\n return value.strip().lower()\n```" } } ] } fake_hf_module = type("FakeHFModule", (), {"InferenceClient": object})() fake_hf_utils = type("FakeHFUtils", (), {"HfHubHTTPError": RuntimeError})() with ( patch("maris_core.code.generate_code.get_pipeline", return_value=None), patch("maris_core.text.generate.create_hf_inference_client", return_value=FakeClient()), patch.dict( sys.modules, {"huggingface_hub": fake_hf_module, "huggingface_hub.utils": fake_hf_utils} ), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): response = await generate_code( CodeRequest( prompt="Uzraksti Python helperi normalize_email", language="Python", fallback_model="Qwen/Qwen2.5-Coder-32B-Instruct", ) ) assert "normalize_email" in response.code @pytest.mark.asyncio async def test_fix_code_delegates() -> None: with ( patch("maris_core.code.generate_code.get_pipeline", return_value=None), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): req = FixCodeRequest(code="prit('hello')", error_message="NameError", language="Python") with pytest.raises(HTTPException) as exc_info: await fix_code(req) assert exc_info.value.status_code == 503 @pytest.mark.asyncio async def test_generate_code_passes_large_max_new_tokens_to_pipeline() -> None: captured_max_new_tokens: int | None = None def fake_pipeline( messages: list[dict[str, Any]], *, max_new_tokens: int, temperature: float ) -> list[dict[str, list[dict[str, str]]]]: nonlocal captured_max_new_tokens del messages, temperature captured_max_new_tokens = max_new_tokens return [{"generated_text": [{"role": "assistant", "content": "print('hello world')"}]}] with ( patch("maris_core.code.generate_code.get_pipeline", return_value=fake_pipeline), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): response = await generate_code( CodeRequest( prompt="Uzraksti pilnu Python servisu", language="Python", max_new_tokens=20_000, ) ) assert captured_max_new_tokens == 20_000 assert response.code == "print('hello world')" @pytest.mark.asyncio async def test_generate_code_uses_stronger_engineering_system_prompt() -> None: captured_messages: list[dict[str, Any]] = [] def fake_pipeline( messages: list[dict[str, Any]], *, max_new_tokens: int, temperature: float ) -> list[dict[str, list[dict[str, str]]]]: nonlocal captured_messages del max_new_tokens, temperature captured_messages = messages return [{"generated_text": [{"role": "assistant", "content": "print('hello world')"}]}] with ( patch("maris_core.code.generate_code.get_pipeline", return_value=fake_pipeline), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): await generate_code(CodeRequest(prompt="Uzraksti Python skriptu", language="Python")) assert "production-ready" in captured_messages[0]["content"] assert "edge cases" in captured_messages[0]["content"] assert "izpildāmu artefaktu" in captured_messages[0]["content"] @pytest.mark.asyncio async def test_generate_code_materializes_workspace_artifacts_from_structured_payload( tmp_path, ) -> None: def fake_pipeline( messages: list[dict[str, Any]], *, max_new_tokens: int, temperature: float ) -> list[dict[str, list[dict[str, str]]]]: del messages, max_new_tokens, temperature return [ { "generated_text": [ { "role": "assistant", "content": """```json { "explanation": "Pilns kalkulatora projekts.", "entrypoint": "index.html", "files": [ {"path": "index.html", "content": "Kalkulators"}, {"path": "assets/app.js", "content": "console.log('calc');"} ] } ```""", } ] } ] with ( patch("maris_core.code.generate_code.get_pipeline", return_value=fake_pipeline), patch("maris_core.code.generate_code.WORKSPACE_ARTIFACT_ROOT", tmp_path), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): response = await generate_code( CodeRequest(prompt="Uzprogrammē kalkulatora projektu", language="HTML/CSS/JavaScript") ) bundle_path = Path(response.bundle_path or "") assert response.entrypoint == "index.html" assert response.workspace_artifact_dir is not None assert response.detected_stack == "HTML/CSS/JavaScript" assert Path(response.workspace_artifact_dir).exists() assert [file.path for file in response.files] == ["index.html", "assets/app.js"] assert response.files[0].absolute_path is not None assert bundle_path.exists() assert ( Path(response.files[0].absolute_path or "") .read_text(encoding="utf-8") .startswith("") ) @pytest.mark.asyncio async def test_generate_code_auto_scaffolds_nextjs_project_from_single_code_block(tmp_path) -> None: def fake_pipeline( messages: list[dict[str, Any]], *, max_new_tokens: int, temperature: float ) -> list[dict[str, list[dict[str, str]]]]: del messages, max_new_tokens, temperature return [ { "generated_text": [ { "role": "assistant", "content": """```tsx export default function HomePage() { return
Analytics dashboard
; } ```""", } ] } ] with ( patch("maris_core.code.generate_code.get_pipeline", return_value=fake_pipeline), patch("maris_core.code.generate_code.WORKSPACE_ARTIFACT_ROOT", tmp_path), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): response = await generate_code( CodeRequest( prompt="Uztaisi Next.js app router dashboard ar TypeScript", language="Python", ) ) paths = [file.path for file in response.files] assert response.detected_stack == "Next.js (TypeScript)" assert response.language == "Next.js (TypeScript)" assert response.entrypoint == "app/page.tsx" assert "app/page.tsx" in paths assert "package.json" in paths assert Path(response.bundle_path or "").exists() @pytest.mark.asyncio async def test_generate_code_uses_repo_aware_entrypoint_for_existing_project(tmp_path) -> None: project_root = tmp_path / "existing-project" (project_root / "src").mkdir(parents=True) (project_root / "pyproject.toml").write_text( "[project]\nname = 'demo'\nversion = '0.1.0'\n", encoding="utf-8", ) (project_root / "src/main.py").write_text( "def main() -> None:\n print('old')\n", encoding="utf-8" ) captured_messages: list[dict[str, Any]] = [] def fake_pipeline( messages: list[dict[str, Any]], *, max_new_tokens: int, temperature: float ) -> list[dict[str, list[dict[str, str]]]]: nonlocal captured_messages del max_new_tokens, temperature captured_messages = messages return [ { "generated_text": [ { "role": "assistant", "content": """```python def main() -> None: print('updated') ```""", } ] } ] with ( patch("maris_core.code.generate_code.get_pipeline", return_value=fake_pipeline), patch("maris_core.code.generate_code.WORKSPACE_ARTIFACT_ROOT", tmp_path / "artifacts"), patch( "maris_core.utils.hf_integration.HFIntegration.save_generation", new_callable=AsyncMock, ), ): response = await generate_code( CodeRequest( prompt="Salabo esošo servisu un atjauno starta loģiku", language="Python", repo_path=str(project_root), ) ) assert response.repo_path == str(project_root) assert response.detected_stack == "Python" assert response.entrypoint == "src/main.py" assert response.files[0].path == "src/main.py" assert "Repo sakne" in captured_messages[1]["content"] assert "src/main.py" in captured_messages[1]["content"]