Files

239 lines
8.2 KiB
Python

from fastapi.testclient import TestClient
from memrelay.projects import normalize_git_remote
from tests.conftest import csrf_headers
from tests.fakes import FakeBasicMemory
def test_git_remote_normalization() -> None:
assert normalize_git_remote("git@GitHub.com:Owner/Repo.git") == "git://github.com/owner/repo"
assert (
normalize_git_remote("https://user@example.com/Team/Repo/") == "git://example.com/team/repo"
)
def test_create_list_resolve_and_archive_project(initialized_client: TestClient) -> None:
client = initialized_client
created = client.post(
"/api/v1/projects",
json={
"name": "MemRelay",
"git_remote": "git@github.com:nixevol/MemRelay.git",
"aliases": ["E:\\NIXProject\\MemRelay"],
},
headers=csrf_headers(client),
)
assert created.status_code == 201
project = created.json()
assert project["slug"] == "memrelay"
assert project["git_remote"] == "git@github.com:nixevol/MemRelay.git"
assert project["git_remote_normalized"] == "git://github.com/nixevol/memrelay"
assert project["workspace_type"] == "development"
assert len(client.get("/api/v1/projects").json()) == 1
by_remote = client.post(
"/api/v1/projects/resolve",
json={"git_remote": "https://github.com/NIXEVOL/MemRelay.git"},
)
assert by_remote.json()["id"] == project["id"]
by_path = client.post("/api/v1/projects/resolve", json={"directory": "e:/nixproject/memrelay/"})
assert by_path.json()["id"] == project["id"]
def test_resolve_remembers_a_new_device_directory(initialized_client: TestClient) -> None:
client = initialized_client
project = client.post(
"/api/v1/projects",
json={
"name": "Cross-device project",
"git_remote": "https://example.test/team/cross-device.git",
},
headers=csrf_headers(client),
).json()
first = client.post(
"/api/v1/projects/resolve",
json={
"git_remote": "git@example.test:team/cross-device.git",
"directory": "D:\\Work\\CrossDevice",
},
)
assert first.status_code == 200
assert first.json()["id"] == project["id"]
assert "d:/work/crossdevice" in first.json()["aliases"]
second = client.post(
"/api/v1/projects/resolve",
json={"directory": "d:/WORK/crossdevice/"},
)
assert second.status_code == 200
assert second.json()["id"] == project["id"]
renamed = client.patch(
f"/api/v1/projects/{project['id']}",
json={"name": "MemRelay Renamed"},
headers=csrf_headers(client),
)
assert renamed.status_code == 200
assert renamed.json()["name"] == "MemRelay Renamed"
assert renamed.json()["slug"] == "memrelay-renamed"
assert (
client.delete(f"/api/v1/projects/{project['id']}", headers=csrf_headers(client)).status_code
== 409
)
archived = client.post(
f"/api/v1/projects/{project['id']}/archive", headers=csrf_headers(client)
)
assert archived.json()["archived"] is True
deleted = client.delete(f"/api/v1/projects/{project['id']}", headers=csrf_headers(client))
assert deleted.status_code == 204
assert client.get("/api/v1/projects").json() == []
def test_duplicate_project_is_rejected(initialized_client: TestClient) -> None:
client = initialized_client
payload = {"name": "Same Project", "git_remote": "git@example.com:team/repo.git"}
assert (
client.post("/api/v1/projects", json=payload, headers=csrf_headers(client)).status_code
== 201
)
assert (
client.post("/api/v1/projects", json=payload, headers=csrf_headers(client)).status_code
== 409
)
def test_project_without_remote_defaults_to_general_workspace(
initialized_client: TestClient,
) -> None:
response = initialized_client.post(
"/api/v1/projects",
json={"name": "General notes"},
headers=csrf_headers(initialized_client),
)
assert response.status_code == 201
assert response.json()["workspace_type"] == "general"
def test_projects_without_git_remote_do_not_conflict_and_source_settings_round_trip(
initialized_client: TestClient,
) -> None:
client = initialized_client
first = client.post(
"/api/v1/projects",
json={"name": "Notes A", "curation_enabled": False, "source_strategy": "mcp"},
headers=csrf_headers(client),
)
second = client.post(
"/api/v1/projects",
json={"name": "Notes B"},
headers=csrf_headers(client),
)
assert first.status_code == 201, first.text
assert second.status_code == 201, second.text
assert first.json()["git_remote"] is None
assert first.json()["git_remote_normalized"] is None
assert first.json()["curation_enabled"] is False
assert first.json()["source_strategy"] == "mcp"
updated = client.patch(
f"/api/v1/projects/{first.json()['id']}",
json={
"git_remote": "https://example.test/Owner/Notes.git",
"curation_enabled": True,
"source_strategy": "git",
"source_branch": "release/docs",
"source_credential_item_id": "vault-item-1",
},
headers=csrf_headers(client),
)
assert updated.status_code == 200, updated.text
project = updated.json()
assert project["git_remote"] == "https://example.test/Owner/Notes.git"
assert project["git_remote_normalized"] == "git://example.test/owner/notes"
assert project["source_strategy"] == "git"
assert project["source_branch"] == "release/docs"
assert project["source_credential_item_id"] == "vault-item-1"
assert project["source_last_commit"] is None
assert project["last_agent_sync_at"] is None
def test_project_custom_curation_template_can_be_updated_and_cleared(
initialized_client: TestClient,
) -> None:
client = initialized_client
created = client.post(
"/api/v1/projects",
json={
"name": "Custom Curation",
"custom_curation_template": "Always preserve experiment identifiers.",
},
headers=csrf_headers(client),
)
assert created.status_code == 201
project = created.json()
assert project["custom_curation_template"] == "Always preserve experiment identifiers."
prompt = client.app.state.curation._document_prompt(
"Evidence",
["overview"],
project["name"],
project["workspace_type"],
project["custom_curation_template"],
)
assert "Always preserve experiment identifiers." in prompt
updated = client.patch(
f"/api/v1/projects/{project['id']}",
json={"custom_curation_template": "Prefer a dated laboratory timeline."},
headers=csrf_headers(client),
)
assert updated.status_code == 200
assert updated.json()["custom_curation_template"] == "Prefer a dated laboratory timeline."
cleared = client.patch(
f"/api/v1/projects/{project['id']}",
json={"custom_curation_template": None},
headers=csrf_headers(client),
)
assert cleared.status_code == 200
assert cleared.json()["custom_curation_template"] is None
def test_deleting_archived_project_removes_its_memories(initialized_client: TestClient) -> None:
client = initialized_client
fake = FakeBasicMemory()
client.app.state.basic_memory = fake
project = client.post(
"/api/v1/projects",
json={"name": "Disposable Project"},
headers=csrf_headers(client),
).json()
memory = client.post(
"/api/v1/memories",
json={
"request_id": "project-delete-memory",
"scope": "project",
"project_id": project["id"],
"memory_type": "fact",
"title": "Disposable memory",
"content": "Removed with its project.",
},
headers=csrf_headers(client),
)
assert memory.status_code == 201
assert fake.notes
assert (
client.post(
f"/api/v1/projects/{project['id']}/archive", headers=csrf_headers(client)
).status_code
== 200
)
assert (
client.delete(f"/api/v1/projects/{project['id']}", headers=csrf_headers(client)).status_code
== 204
)
assert fake.notes == {}
assert client.get("/api/v1/memories", params={"project_id": project["id"]}).json() == []