# retoor import json import pytest import requests from devplacepy.database import get_table, refresh_snapshot, set_setting from tests.conftest import BASE_URL @pytest.fixture(scope="module", autouse=True) def _planning_settings(app_server): set_setting("rate_limit_per_minute", "1000000") set_setting("gitea_base_url", "https://gitea.test") set_setting("gitea_owner", "retoor") set_setting("gitea_repo", "pydevplace") set_setting("gitea_token", "planning-test-token") set_setting("issue_ai_enhance", "0") yield set_setting("gitea_base_url", "") set_setting("gitea_token", "") def _admin(seeded_db): refresh_snapshot() key = get_table("users").find_one(username="alice_test")["api_key"] session = requests.Session() session.headers.update({"X-API-KEY": key}) return session def _member(seeded_db): refresh_snapshot() key = get_table("users").find_one(username="bob_test")["api_key"] session = requests.Session() session.headers.update({"X-API-KEY": key}) return session def test_planning_requires_admin(seeded_db): member = _member(seeded_db) response = member.post( f"{BASE_URL}/issues/planning", data={"numbers": "1"}, allow_redirects=False ) assert response.status_code in (302, 303) def test_planning_status_unknown_uid_is_404(seeded_db): admin = _admin(seeded_db) response = admin.get(f"{BASE_URL}/issues/planning/does-not-exist") assert response.status_code == 404 def test_planning_enqueue_stores_selected_numbers(seeded_db): admin = _admin(seeded_db) response = admin.post( f"{BASE_URL}/issues/planning", data={"numbers": "4, 7,7, x"} ) assert response.status_code == 200, response.text[:300] body = response.json() uid = body["uid"] assert body["status_url"] == f"/issues/planning/{uid}" refresh_snapshot() job = get_table("jobs").find_one(uid=uid) assert job is not None assert json.loads(job["payload"])["numbers"] == [4, 7] status = admin.get(f"{BASE_URL}/issues/planning/{uid}") assert status.status_code == 200 payload = status.json() assert payload["uid"] == uid assert payload["kind"] == "planning" assert payload["status"] in ("pending", "running", "done", "failed") def test_planning_enqueue_without_numbers_is_all_open(seeded_db): admin = _admin(seeded_db) response = admin.post(f"{BASE_URL}/issues/planning", data={}) assert response.status_code == 200, response.text[:300] uid = response.json()["uid"] refresh_snapshot() job = get_table("jobs").find_one(uid=uid) assert json.loads(job["payload"])["numbers"] == [] def _mark_done(uid, result): get_table("jobs").update( {"uid": uid, "status": "done", "result": json.dumps(result)}, ["uid"] ) refresh_snapshot() def test_planning_download_serves_the_report_file(seeded_db): from devplacepy.config import PLANNING_REPORTS_DIR from devplacepy.services.jobs import queue admin = _admin(seeded_db) uid = queue.enqueue( "planning", {"numbers": []}, "user", "planning-download-owner", "open-tickets-plan" ) report_dir = PLANNING_REPORTS_DIR / "ab" / "cd" report_dir.mkdir(parents=True, exist_ok=True) report_path = report_dir / f"{uid}.open-tickets-plan.md" report_path.write_text("# Plan\n\nDo the thing.\n", encoding="utf-8") try: _mark_done( uid, { "download_url": f"/issues/planning/{uid}/download", "local_path": str(report_path), "final_name": "open-tickets-plan.md", "markdown": "# Plan\n\nDo the thing.\n", "ai_used": False, "item_count": 0, "bytes_out": report_path.stat().st_size, }, ) response = admin.get(f"{BASE_URL}/issues/planning/{uid}/download") assert response.status_code == 200, response.text[:300] assert response.headers["content-type"].startswith("text/markdown") assert "Do the thing." in response.text finally: get_table("jobs").delete(uid=uid) report_path.unlink(missing_ok=True) def test_planning_download_rejects_a_local_path_outside_the_reports_dir(seeded_db): from devplacepy.config import DATA_DIR from devplacepy.services.jobs import queue admin = _admin(seeded_db) uid = queue.enqueue( "planning", {"numbers": []}, "user", "planning-traversal-owner", "open-tickets-plan" ) escape_path = DATA_DIR / f"escaped-{uid}.md" escape_path.write_text("secret", encoding="utf-8") try: _mark_done( uid, { "download_url": f"/issues/planning/{uid}/download", "local_path": str(escape_path), "final_name": "escaped.md", "markdown": "secret", "ai_used": False, "item_count": 0, "bytes_out": escape_path.stat().st_size, }, ) response = admin.get(f"{BASE_URL}/issues/planning/{uid}/download") assert response.status_code == 404 finally: get_table("jobs").delete(uid=uid) escape_path.unlink(missing_ok=True) def test_planning_download_pending_job_is_404(seeded_db): from devplacepy.services.jobs import queue admin = _admin(seeded_db) uid = queue.enqueue( "planning", {"numbers": []}, "user", "planning-pending-owner", "open-tickets-plan" ) try: response = admin.get(f"{BASE_URL}/issues/planning/{uid}/download") assert response.status_code == 404 finally: get_table("jobs").delete(uid=uid)