Spaces:
Sleeping
Sleeping
| import io | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from app.main import app | |
| import app.routers.projects as projects_router | |
| from app.services.project_service import ProjectService | |
| def tmp_service(tmp_path, monkeypatch): | |
| svc = ProjectService(root=str(tmp_path / "registry")) | |
| monkeypatch.setattr(projects_router, "_service", svc) | |
| from app.services.student_memory import StudentMemoryService | |
| async def _noop_ensure(self, project_id): | |
| return True | |
| monkeypatch.setattr(StudentMemoryService, "ensure_project_dataset", _noop_ensure) | |
| return svc | |
| def test_create_and_get_project(): | |
| client = TestClient(app) | |
| r = client.post("/projects", json={"name": "Thesis"}) | |
| assert r.status_code == 200 | |
| assert r.json()["document_id"] == "" | |
| pid = r.json()["project_id"] | |
| g = client.get(f"/projects/{pid}") | |
| assert g.status_code == 200 | |
| assert g.json()["name"] == "Thesis" | |
| assert g.json()["document_id"] == "" | |
| def test_list_projects(): | |
| client = TestClient(app) | |
| client.post("/projects", json={"name": "A"}) | |
| client.post("/projects", json={"name": "B"}) | |
| r = client.get("/projects") | |
| assert r.status_code == 200 | |
| assert len(r.json()["items"]) == 2 | |
| assert all("document_id" in item for item in r.json()["items"]) | |
| def test_get_missing_404(): | |
| client = TestClient(app) | |
| assert client.get("/projects/nope").status_code == 404 | |
| def test_delete_project(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| deleted: list[str] = [] | |
| monkeypatch.setattr(pr, "delete_project_uploads", lambda project_id: deleted.append(project_id)) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Gone"}).json()["project_id"] | |
| assert client.delete(f"/projects/{pid}").status_code == 200 | |
| assert client.get(f"/projects/{pid}").status_code == 404 | |
| assert deleted == [pid] | |
| def test_update_project_details(): | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Old"}).json()["project_id"] | |
| r = client.patch(f"/projects/{pid}", json={"name": "New Research Direction", "title": "Working title"}) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert body["name"] == "New Research Direction" | |
| assert body["title"] == "Working title" | |
| g = client.get(f"/projects/{pid}") | |
| assert g.json()["name"] == "New Research Direction" | |
| def test_update_project_blank_name_falls_back(): | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Old"}).json()["project_id"] | |
| r = client.patch(f"/projects/{pid}", json={"name": " "}) | |
| assert r.status_code == 200 | |
| assert r.json()["name"] == "Untitled Project" | |
| def test_valid_project_id_guard(): | |
| from app.routers.projects import _valid_project_id | |
| assert _valid_project_id("some-uuid-1234") is True | |
| assert _valid_project_id("../etc") is False | |
| assert _valid_project_id("a/b") is False | |
| assert _valid_project_id("..") is False | |
| assert _valid_project_id("") is False | |
| def test_add_file_appends_to_project(monkeypatch, tmp_path): | |
| # Avoid heavy embedding: stub the chunker path. | |
| import app.routers.projects as pr | |
| monkeypatch.setattr(pr, "ingest_file", lambda *a, **k: 3) | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| import os | |
| os.makedirs(str(tmp_path / "pdfs"), exist_ok=True) | |
| monkeypatch.setattr(pr, "_PDF_CACHE_DIR", str(tmp_path / "pdfs")) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Grow"}).json()["project_id"] | |
| r = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(b"hello world"), "application/pdf"))], | |
| ) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert len(body["files"]) == 1 | |
| assert body["files"][0]["filename"] == "paper.pdf" | |
| assert len(body["files"][0]["file_id"]) == 64 # full sha256 | |
| assert len(body["document_id"]) == 64 | |
| def test_add_same_file_twice_dedupes_and_skips_rechunk(monkeypatch, tmp_path): | |
| # Fix 1 regression: re-uploading an already-present file must not append | |
| # a duplicate project.files entry, and must not re-invoke ingest_file. | |
| import app.routers.projects as pr | |
| ingest_calls: list[str] = [] | |
| def _fake_ingest(content, filename, collection, **kwargs): | |
| ingest_calls.append(filename) | |
| return 3 | |
| import os | |
| monkeypatch.setattr(pr, "ingest_file", _fake_ingest) | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| os.makedirs(str(tmp_path / "pdfs"), exist_ok=True) | |
| monkeypatch.setattr(pr, "_PDF_CACHE_DIR", str(tmp_path / "pdfs")) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Dedup"}).json()["project_id"] | |
| payload = b"same bytes every time" | |
| r1 = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(payload), "application/pdf"))], | |
| ) | |
| assert r1.status_code == 200 | |
| r2 = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(payload), "application/pdf"))], | |
| ) | |
| assert r2.status_code == 200 | |
| body = r2.json() | |
| assert len(body["files"]) == 1 | |
| assert ingest_calls == ["paper.pdf"] | |
| def test_add_more_than_five_files_returns_400(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| import os | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| os.makedirs(str(tmp_path / "pdfs"), exist_ok=True) | |
| monkeypatch.setattr(pr, "_PDF_CACHE_DIR", str(tmp_path / "pdfs")) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Cap"}).json()["project_id"] | |
| files = [ | |
| ("files", (f"paper-{i}.pdf", io.BytesIO(f"payload-{i}".encode()), "application/pdf")) | |
| for i in range(6) | |
| ] | |
| r = client.post(f"/projects/{pid}/files", files=files) | |
| assert r.status_code == 400 | |
| assert "5 files" in r.json()["detail"] | |
| def test_same_filename_different_content_gets_unique_name(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| import os | |
| monkeypatch.setattr(pr, "ingest_file", lambda *a, **k: 3) | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| os.makedirs(str(tmp_path / "pdfs"), exist_ok=True) | |
| monkeypatch.setattr(pr, "_PDF_CACHE_DIR", str(tmp_path / "pdfs")) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Collision"}).json()["project_id"] | |
| r1 = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(b"version one"), "application/pdf"))], | |
| ) | |
| assert r1.status_code == 200 | |
| r2 = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(b"version two"), "application/pdf"))], | |
| ) | |
| assert r2.status_code == 200 | |
| names = [f["filename"] for f in r2.json()["files"]] | |
| assert len(names) == 2 | |
| assert names[0] == "paper.pdf" | |
| assert names[1].startswith("paper-") | |
| assert names[1].endswith(".pdf") | |
| def test_add_file_missing_project_404(monkeypatch): | |
| client = TestClient(app) | |
| r = client.post( | |
| "/projects/nope/files", | |
| files=[("files", ("p.pdf", io.BytesIO(b"x"), "application/pdf"))], | |
| ) | |
| assert r.status_code == 404 | |
| async def _noop_remove(project_id, filename): # module-level helper for monkeypatch | |
| return True | |
| def test_remove_file_drops_entry(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| monkeypatch.setattr(pr, "ingest_file", lambda *a, **k: 3) | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| monkeypatch.setattr(pr, "_PDF_CACHE_DIR", str(tmp_path / "pdfs")) | |
| monkeypatch.setattr(pr, "remove_file_from_project", _noop_remove) | |
| import os | |
| os.makedirs(str(tmp_path / "pdfs"), exist_ok=True) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Shrink"}).json()["project_id"] | |
| add = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(b"bytes here"), "application/pdf"))], | |
| ).json() | |
| fid = add["files"][0]["file_id"] | |
| r = client.delete(f"/projects/{pid}/files/{fid}") | |
| assert r.status_code == 200 | |
| assert r.json()["files"] == [] | |
| assert r.json()["document_id"] == "" | |
| def test_remove_file_saves_registry_before_memory_reset(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| seen_counts = [] | |
| async def fake_remove(project_id, filename): | |
| project = pr._service.load(project_id) | |
| seen_counts.append(len(project.files if project else [])) | |
| return True | |
| monkeypatch.setattr(pr, "ingest_file", lambda *a, **k: 3) | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| monkeypatch.setattr(pr, "_PDF_CACHE_DIR", str(tmp_path / "pdfs")) | |
| monkeypatch.setattr(pr, "remove_file_from_project", fake_remove) | |
| import os | |
| os.makedirs(str(tmp_path / "pdfs"), exist_ok=True) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Memory Order"}).json()["project_id"] | |
| add = client.post( | |
| f"/projects/{pid}/files", | |
| files=[("files", ("paper.pdf", io.BytesIO(b"bytes here"), "application/pdf"))], | |
| ).json() | |
| fid = add["files"][0]["file_id"] | |
| assert client.delete(f"/projects/{pid}/files/{fid}").status_code == 200 | |
| assert seen_counts == [0] | |
| def test_remove_missing_file_404(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| monkeypatch.setattr(pr, "remove_file_from_project", _noop_remove) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "X"}).json()["project_id"] | |
| assert client.delete(f"/projects/{pid}/files/doesnotexist").status_code == 404 | |
| def test_get_project_drops_registry_files_missing_on_disk(monkeypatch, tmp_path): | |
| import app.routers.projects as pr | |
| monkeypatch.setattr(pr, "project_upload_dir", lambda pid: str(tmp_path / pid)) | |
| client = TestClient(app) | |
| pid = client.post("/projects", json={"name": "Stale"}).json()["project_id"] | |
| folder = tmp_path / pid | |
| folder.mkdir(parents=True, exist_ok=True) | |
| (folder / "kept.pdf").write_bytes(b"kept") | |
| project = pr._service.load(pid) | |
| project.files = [ | |
| pr.ProjectFile(filename="missing.pdf", file_id="a" * 64), | |
| pr.ProjectFile(filename="kept.pdf", file_id="b" * 64), | |
| ] | |
| pr._service.save(project) | |
| r = client.get(f"/projects/{pid}") | |
| assert r.status_code == 200 | |
| assert [f["filename"] for f in r.json()["files"]] == ["kept.pdf"] | |
| saved = pr._service.load(pid) | |
| assert [f.filename for f in saved.files] == ["kept.pdf"] | |
| def test_session_create_persists_project(monkeypatch, tmp_path): | |
| import app.routers.session as sr | |
| import app.routers.projects as pr | |
| svc = ProjectService(root=str(tmp_path / "reg2")) | |
| monkeypatch.setattr(sr, "_project_service", svc, raising=False) | |
| monkeypatch.setattr(pr, "_service", svc) | |
| client = TestClient(app) | |
| pid = client.post("/session/create", json={"topic": "Optics", "intention": "learn"}).json()["project_id"] | |
| # The same id must now be loadable as a real project via /projects | |
| assert client.get(f"/projects/{pid}").status_code == 200 | |