1166 lines
46 KiB
Python
1166 lines
46 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import yaml
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import delete, select
|
|
|
|
from memrelay.git_service import GitManager
|
|
from memrelay.lease_service import RuntimeLeaseManager
|
|
from memrelay.models import (
|
|
CuratedDocument,
|
|
CuratedDocumentRevision,
|
|
CurationDependencyState,
|
|
CurationJob,
|
|
CurationModelConfig,
|
|
CurationSettings,
|
|
GitConnection,
|
|
GitSyncJob,
|
|
MemoryReference,
|
|
MemoryRepository,
|
|
Project,
|
|
UsageProfile,
|
|
)
|
|
from memrelay.schemas import GitModeMigrationRequest
|
|
from tests.conftest import csrf_headers
|
|
from tests.fakes import FakeBasicMemory
|
|
|
|
pytestmark = pytest.mark.skipif(shutil.which("git") is None, reason="Git CLI is not installed")
|
|
|
|
|
|
class DiskBasicMemory(FakeBasicMemory):
|
|
def __init__(self, root: Path) -> None:
|
|
super().__init__()
|
|
self.root = root
|
|
|
|
async def write_note(self, **kwargs): # type: ignore[no-untyped-def]
|
|
result = await super().write_note(**kwargs)
|
|
path = self.root / f"{result['permalink']}.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
metadata = {
|
|
**kwargs["metadata"],
|
|
"title": kwargs["title"],
|
|
"tags": kwargs["tags"],
|
|
"type": kwargs["note_type"],
|
|
}
|
|
path.write_text(
|
|
f"---\n{yaml.safe_dump(metadata, allow_unicode=True, sort_keys=False)}---\n\n"
|
|
f"{kwargs['content']}\n",
|
|
encoding="utf-8",
|
|
)
|
|
return result
|
|
|
|
async def move_note(self, identifier: str, destination_path: str) -> dict:
|
|
source = self.root / f"{identifier}.md"
|
|
result = await super().move_note(identifier, destination_path)
|
|
destination = self.root / f"{result['permalink']}.md"
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
source.replace(destination)
|
|
return result
|
|
|
|
async def delete_note(self, identifier: str) -> None:
|
|
await super().delete_note(identifier)
|
|
(self.root / f"{identifier}.md").unlink(missing_ok=True)
|
|
|
|
|
|
def _git(path, *args: str) -> str: # type: ignore[no-untyped-def]
|
|
result = subprocess.run(
|
|
["git", "-C", str(path), *args],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def _enable_local_git(client: TestClient, remote_url: str | None = None) -> dict:
|
|
response = client.put(
|
|
"/api/v1/git/connection",
|
|
json={
|
|
"enabled": True,
|
|
"name": "Local memory history",
|
|
"mode": "single",
|
|
"remote_url": remote_url,
|
|
"author_name": "MemRelay Tests",
|
|
"author_email": "memrelay@example.test",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
repositories = client.get("/api/v1/git/repositories")
|
|
assert repositories.status_code == 200
|
|
assert len(repositories.json()) == 1
|
|
return repositories.json()[0]
|
|
|
|
|
|
def _save_memory(client: TestClient, request_id: str, title: str, content: str) -> dict:
|
|
response = client.post(
|
|
"/api/v1/memories",
|
|
json={
|
|
"request_id": request_id,
|
|
"scope": "global",
|
|
"memory_type": "fact",
|
|
"title": title,
|
|
"content": content,
|
|
"tags": ["git-test"],
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def _write_version(repository_path, memory: dict, content: str) -> str: # type: ignore[no-untyped-def]
|
|
relative = f"{memory['path']}.md"
|
|
target = repository_path / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(
|
|
"---\n"
|
|
f"stable_id: {memory['id']}\n"
|
|
"scope: global\n"
|
|
"memory_type: fact\n"
|
|
f"title: {memory['title']}\n"
|
|
"tags:\n"
|
|
" - git-test\n"
|
|
"---\n\n"
|
|
f"{content}\n",
|
|
encoding="utf-8",
|
|
)
|
|
_git(repository_path, "add", relative)
|
|
_git(repository_path, "commit", "-m", f"test: {content}")
|
|
return _git(repository_path, "rev-parse", "HEAD")
|
|
|
|
|
|
def _create_remote_memory(remote: Path, source: Path, memory_id: str) -> str:
|
|
_git(source.parent, "init", "-b", "main", str(source))
|
|
_git(source, "config", "user.name", "Remote Test")
|
|
_git(source, "config", "user.email", "remote@example.test")
|
|
(source / "global").mkdir(parents=True)
|
|
(source / "memrelay-manifest.json").write_text(
|
|
'{"schema_version":1,"repository_mode":"single","scope":"all","projects":[]}\n',
|
|
encoding="utf-8",
|
|
)
|
|
(source / "global" / "remote-memory.md").write_text(
|
|
"---\n"
|
|
f"stable_id: {memory_id}\n"
|
|
"scope: global\n"
|
|
"memory_type: fact\n"
|
|
"title: Remote memory\n"
|
|
"tags:\n"
|
|
" - remote\n"
|
|
"---\n\n"
|
|
"Imported from remote.\n",
|
|
encoding="utf-8",
|
|
)
|
|
_git(source, "add", "-A")
|
|
_git(source, "commit", "-m", "feat: add remote memory")
|
|
commit = _git(source, "rev-parse", "HEAD")
|
|
_git(source.parent, "init", "--bare", str(remote))
|
|
_git(source, "remote", "add", "origin", str(remote))
|
|
_git(source, "push", "-u", "origin", "main")
|
|
return commit
|
|
|
|
|
|
def _drain_git_jobs(client: TestClient, limit: int = 20) -> None:
|
|
for _ in range(limit):
|
|
if not asyncio.run(client.app.state.git_manager.process_once()):
|
|
return
|
|
raise AssertionError("Git sync jobs did not drain")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mode", "ai_enabled", "git_enabled", "remote_enabled"),
|
|
[
|
|
("base", False, False, False),
|
|
("ai_only", True, False, False),
|
|
("local_git", False, True, False),
|
|
("git_remote", False, True, True),
|
|
("ai_git", True, True, False),
|
|
],
|
|
)
|
|
def test_ai_and_git_optional_feature_matrix(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
mode: str,
|
|
ai_enabled: bool,
|
|
git_enabled: bool,
|
|
remote_enabled: bool,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
if ai_enabled:
|
|
with client.app.state.session_factory() as db:
|
|
settings = db.get(CurationSettings, 1)
|
|
dependency = db.get(CurationDependencyState, 1)
|
|
assert settings is not None and dependency is not None
|
|
settings.enabled = True
|
|
dependency.status = "available"
|
|
db.add(
|
|
CurationModelConfig(
|
|
revision=1,
|
|
name=f"{mode}-model",
|
|
base_url="http://model.invalid/v1",
|
|
text_model="test-model",
|
|
protocol="responses",
|
|
effective_protocol="responses",
|
|
active=True,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
remote: Path | None = None
|
|
if git_enabled:
|
|
if remote_enabled:
|
|
remote = tmp_path / "matrix-remote.git"
|
|
subprocess.run(["git", "init", "--bare", str(remote)], check=True, capture_output=True)
|
|
_enable_local_git(client, str(remote) if remote else None)
|
|
|
|
saved = _save_memory(
|
|
client,
|
|
f"matrix-{mode}",
|
|
f"Matrix {mode}",
|
|
f"The {mode} optional feature combination remains usable.",
|
|
)
|
|
assert saved["revision"] == 1
|
|
with client.app.state.session_factory() as db:
|
|
curation_jobs = db.scalars(select(CurationJob)).all()
|
|
git_jobs = db.scalars(select(GitSyncJob)).all()
|
|
repositories = db.scalars(select(MemoryRepository)).all()
|
|
assert bool(curation_jobs) is ai_enabled
|
|
assert bool(git_jobs) is git_enabled
|
|
assert bool(repositories) is git_enabled
|
|
assert (client.app.state.settings.memories_dir / ".git").exists() is git_enabled
|
|
|
|
if remote is not None:
|
|
_drain_git_jobs(client)
|
|
assert _git(remote, "rev-parse", "refs/heads/main")
|
|
|
|
|
|
def test_project_deletion_is_committed_before_repository_archive(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
project = client.post(
|
|
"/api/v1/projects",
|
|
json={"name": "Archived history"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
enabled = client.put(
|
|
"/api/v1/git/connection",
|
|
json={
|
|
"enabled": True,
|
|
"name": "Workspace history",
|
|
"mode": "per_workspace",
|
|
"author_name": "MemRelay Tests",
|
|
"author_email": "memrelay@example.test",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert enabled.status_code == 200, enabled.text
|
|
|
|
saved = client.post(
|
|
"/api/v1/memories",
|
|
json={
|
|
"request_id": "project-delete-history",
|
|
"scope": "project",
|
|
"project_id": project["id"],
|
|
"memory_type": "decision",
|
|
"title": "Recoverable project decision",
|
|
"content": "This revision must remain recoverable after project deletion.",
|
|
"tags": ["history"],
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert saved.status_code == 201, saved.text
|
|
memory = saved.json()
|
|
_drain_git_jobs(client)
|
|
|
|
with client.app.state.session_factory() as db:
|
|
repository = db.scalar(
|
|
select(MemoryRepository).where(MemoryRepository.project_id == project["id"])
|
|
)
|
|
assert repository is not None
|
|
repository_id = repository.id
|
|
original_path = Path(repository.local_path)
|
|
relative = Path(f"{memory['path']}.md").relative_to(Path("projects") / project["id"])
|
|
assert (original_path / relative).exists()
|
|
|
|
archived = client.post(
|
|
f"/api/v1/projects/{project['id']}/archive",
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert archived.status_code == 200, archived.text
|
|
with client.app.state.session_factory() as db:
|
|
repository = db.get(MemoryRepository, repository_id)
|
|
assert repository is not None
|
|
assert repository.project_id == project["id"]
|
|
assert Path(repository.local_path) == original_path
|
|
|
|
deleted = client.delete(
|
|
f"/api/v1/projects/{project['id']}",
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert deleted.status_code == 204, deleted.text
|
|
with client.app.state.session_factory() as db:
|
|
repository = db.get(MemoryRepository, repository_id)
|
|
assert repository is not None
|
|
assert repository.status == "archived"
|
|
assert repository.project_id is None
|
|
archive_path = Path(repository.local_path)
|
|
deletion_job = db.scalar(
|
|
select(GitSyncJob).where(GitSyncJob.operation_id == f"project-delete:{project['id']}")
|
|
)
|
|
assert deletion_job is not None
|
|
assert deletion_job.status == "completed"
|
|
assert archive_path.is_dir()
|
|
assert "delete: 永久删除项目及其记忆" in _git(archive_path, "log", "--format=%s")
|
|
filtered_history = client.get(
|
|
f"/api/v1/git/repositories/{repository_id}/history",
|
|
params={
|
|
"author": "MemRelay Tests",
|
|
"action": "create",
|
|
"resource_id": memory["id"],
|
|
"project_id": project["id"],
|
|
},
|
|
)
|
|
assert filtered_history.status_code == 200, filtered_history.text
|
|
assert len(filtered_history.json()) == 1
|
|
assert filtered_history.json()[0]["action"] == "create"
|
|
assert filtered_history.json()[0]["resource_id"] == memory["id"]
|
|
assert filtered_history.json()[0]["project_id"] == project["id"]
|
|
previous_commit = _git(archive_path, "rev-parse", "HEAD^")
|
|
version = client.get(
|
|
f"/api/v1/git/repositories/{repository_id}/versions/{memory['id']}",
|
|
params={"commit": previous_commit},
|
|
)
|
|
assert version.status_code == 200, version.text
|
|
assert "This revision must remain recoverable" in version.json()["content"]
|
|
previous = _git(archive_path, "show", f"HEAD^:{relative.as_posix()}")
|
|
assert "This revision must remain recoverable" in previous
|
|
missing = subprocess.run(
|
|
["git", "-C", str(archive_path), "cat-file", "-e", f"HEAD:{relative.as_posix()}"],
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert missing.returncode != 0
|
|
|
|
restored = client.post(
|
|
f"/api/v1/git/repositories/{repository_id}/restore",
|
|
json={"commit": previous_commit},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert restored.status_code == 200, restored.text
|
|
assert restored.json()["restored"] == 1
|
|
with client.app.state.session_factory() as db:
|
|
repository = db.get(MemoryRepository, repository_id)
|
|
recreated = db.get(Project, project["id"])
|
|
reference = db.get(MemoryReference, memory["id"])
|
|
assert repository is not None and repository.project_id == project["id"]
|
|
assert repository.status != "archived" and repository.archived_at is None
|
|
assert recreated is not None and recreated.archived is False
|
|
assert reference is not None and reference.status == "active"
|
|
assert Path(client.app.state.settings.memories_dir / f"{memory['path']}.md").exists()
|
|
|
|
|
|
def test_project_source_git_uses_requested_branch_and_offline_cache(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = tmp_path / "source-workspace"
|
|
_git(tmp_path, "init", "-b", "main", str(source))
|
|
_git(source, "config", "user.name", "Source Test")
|
|
_git(source, "config", "user.email", "source@example.test")
|
|
(source / "README.md").write_text("main branch\n", encoding="utf-8")
|
|
_git(source, "add", "README.md")
|
|
_git(source, "commit", "-m", "main source")
|
|
_git(source, "checkout", "-b", "release/docs")
|
|
(source / "README.md").write_text("release branch documentation\n", encoding="utf-8")
|
|
_git(source, "add", "README.md")
|
|
_git(source, "commit", "-m", "release source")
|
|
release_commit = _git(source, "rev-parse", "HEAD")
|
|
|
|
with initialized_client.app.state.session_factory() as db:
|
|
project = Project(
|
|
name="Source branch",
|
|
slug="source-branch",
|
|
git_remote=str(source),
|
|
git_remote_normalized="git://local/source-branch",
|
|
source_strategy="git",
|
|
source_branch="release/docs",
|
|
)
|
|
db.add(project)
|
|
db.commit()
|
|
db.refresh(project)
|
|
project_id = project.id
|
|
db.expunge(project)
|
|
|
|
sources = asyncio.run(initialized_client.app.state.curation._collect_git(project))
|
|
assert any("release branch documentation" in item.content for item in sources)
|
|
with initialized_client.app.state.session_factory() as db:
|
|
current = db.get(Project, project_id)
|
|
assert current is not None
|
|
assert current.source_last_branch == "release/docs"
|
|
assert current.source_last_commit == release_commit
|
|
assert current.source_cache_used is False
|
|
|
|
unavailable = tmp_path / "source-workspace-offline"
|
|
source.rename(unavailable)
|
|
cached = asyncio.run(initialized_client.app.state.curation._collect_git(project))
|
|
assert any("release branch documentation" in item.content for item in cached)
|
|
with initialized_client.app.state.session_factory() as db:
|
|
current = db.get(Project, project_id)
|
|
assert current is not None
|
|
assert current.source_cache_used is True
|
|
assert current.source_last_error == "GIT_SOURCE_REFRESH_FAILED_USING_CACHE"
|
|
|
|
|
|
def test_source_git_incremental_diff_and_unchanged_short_circuit(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = tmp_path / "diff-source"
|
|
_git(tmp_path, "init", "-b", "main", str(source))
|
|
_git(source, "config", "user.name", "Diff Test")
|
|
_git(source, "config", "user.email", "diff@example.test")
|
|
(source / "README.md").write_text("initial readme\n", encoding="utf-8")
|
|
(source / "keep.md").write_text("stable content\n", encoding="utf-8")
|
|
_git(source, "add", "-A")
|
|
_git(source, "commit", "-m", "initial")
|
|
first_commit = _git(source, "rev-parse", "HEAD")
|
|
|
|
with initialized_client.app.state.session_factory() as db:
|
|
project = Project(
|
|
name="Diff source",
|
|
slug="diff-source",
|
|
git_remote=str(source),
|
|
git_remote_normalized="git://local/diff-source",
|
|
source_strategy="git",
|
|
)
|
|
db.add(project)
|
|
db.commit()
|
|
db.refresh(project)
|
|
db.expunge(project)
|
|
|
|
manager = initialized_client.app.state.curation
|
|
|
|
def make_snapshot() -> dict:
|
|
return {
|
|
"text_file_max_bytes": 2 * 1024 * 1024,
|
|
"rich_file_max_bytes": 50 * 1024 * 1024,
|
|
"max_source_files": 1000,
|
|
"max_source_chars": 20_000_000,
|
|
}
|
|
|
|
# 无基线：全量收集整个仓库。
|
|
full_snapshot = make_snapshot()
|
|
full_sources = asyncio.run(manager._collect_git(project, full_snapshot))
|
|
assert {item.source_id for item in full_sources} == {"README.md", "keep.md"}
|
|
assert full_snapshot["git_coverage"]["mode"] == "full"
|
|
|
|
# 基线等于当前 HEAD：git 部分零来源短路。
|
|
unchanged_snapshot = make_snapshot()
|
|
unchanged = asyncio.run(
|
|
manager._collect_git(project, unchanged_snapshot, base_commit=first_commit)
|
|
)
|
|
assert unchanged == []
|
|
assert unchanged_snapshot["git_coverage"]["mode"] == "unchanged"
|
|
assert unchanged_snapshot["git_coverage"]["base_commit"] == first_commit
|
|
|
|
# 修改 + 新增 + 删除后，按基线 diff 只收集变化并声明删除。
|
|
(source / "README.md").write_text("updated readme\n", encoding="utf-8")
|
|
(source / "new.md").write_text("new file\n", encoding="utf-8")
|
|
_git(source, "rm", "keep.md")
|
|
_git(source, "add", "-A")
|
|
_git(source, "commit", "-m", "changes")
|
|
|
|
diff_snapshot = make_snapshot()
|
|
diff_sources = asyncio.run(
|
|
manager._collect_git(project, diff_snapshot, base_commit=first_commit)
|
|
)
|
|
by_id = {item.source_id: item for item in diff_sources}
|
|
assert set(by_id) == {"README.md", "new.md", "keep.md"}
|
|
assert "updated readme" in by_id["README.md"].content
|
|
assert by_id["keep.md"].reason == "SOURCE_REMOVED"
|
|
assert "was removed" in by_id["keep.md"].content
|
|
assert diff_snapshot["git_coverage"]["mode"] == "diff"
|
|
|
|
# 基线 commit 在本地不可用（如浅克隆丢失）时回退全量。
|
|
fallback_snapshot = make_snapshot()
|
|
fallback = asyncio.run(manager._collect_git(project, fallback_snapshot, base_commit="0" * 40))
|
|
assert {item.source_id for item in fallback} == {"README.md", "new.md"}
|
|
assert fallback_snapshot["git_coverage"]["mode"] == "full"
|
|
|
|
|
|
def test_version_and_snapshot_restore_create_new_state_without_rewriting_history(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
repository = _enable_local_git(client)
|
|
repository_path = client.app.state.settings.memories_dir
|
|
|
|
memory = _save_memory(client, "git-memory-1", "History memory", "Current content")
|
|
old_commit = _write_version(repository_path, memory, "Historical content")
|
|
_write_version(repository_path, memory, "Current content")
|
|
|
|
archived = client.post(
|
|
f"/api/v1/memories/{memory['id']}/archive",
|
|
params={"expected_revision": memory["revision"]},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert archived.status_code == 200, archived.text
|
|
assert archived.json()["status"] == "archived"
|
|
|
|
restored = client.post(
|
|
f"/api/v1/git/repositories/{repository['id']}/versions/{memory['id']}/restore",
|
|
json={"commit": old_commit},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert restored.status_code == 200, restored.text
|
|
assert restored.json()["revision"] == 3
|
|
assert restored.json()["status"] == "active"
|
|
assert fake.notes[memory["path"]]["content"].strip() == "Historical content"
|
|
assert f"archive/{memory['id']}" not in fake.notes
|
|
|
|
missing = _save_memory(client, "git-memory-2", "Missing from snapshot", "Later content")
|
|
with client.app.state.session_factory() as db:
|
|
db.execute(delete(GitSyncJob))
|
|
record = db.scalar(select(MemoryRepository).where(MemoryRepository.id == repository["id"]))
|
|
assert record is not None
|
|
record.pending_commits = 0
|
|
db.commit()
|
|
|
|
snapshot = client.post(
|
|
f"/api/v1/git/repositories/{repository['id']}/restore",
|
|
json={"commit": old_commit},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert snapshot.status_code == 200, snapshot.text
|
|
assert snapshot.json()["restored"] == 1
|
|
assert snapshot.json()["archived"] == 1
|
|
|
|
with client.app.state.session_factory() as db:
|
|
restored_reference = db.get(MemoryReference, memory["id"])
|
|
missing_reference = db.get(MemoryReference, missing["id"])
|
|
jobs = db.scalars(select(GitSyncJob)).all()
|
|
assert restored_reference is not None and restored_reference.status == "active"
|
|
assert restored_reference.revision == 4
|
|
assert missing_reference is not None and missing_reference.status == "archived"
|
|
assert len(jobs) == 1
|
|
assert jobs[0].action == "restore_snapshot"
|
|
|
|
assert _git(repository_path, "rev-parse", old_commit) == old_commit
|
|
|
|
|
|
def test_curated_git_restore_and_rebuild_preserve_metadata_and_profile(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
repository = _enable_local_git(client)
|
|
repository_path = client.app.state.settings.memories_dir
|
|
document_id = "33333333-3333-4333-8333-333333333333"
|
|
with client.app.state.session_factory() as db:
|
|
profile_id = db.scalar(select(UsageProfile.id).where(UsageProfile.is_shared.is_(True)))
|
|
assert profile_id is not None
|
|
base_metadata = {
|
|
"stable_id": document_id,
|
|
"scope": "global",
|
|
"project_id": None,
|
|
"workspace_type": "global",
|
|
"usage_profile_id": profile_id,
|
|
"preference_context": "global",
|
|
"document_type": "preferences",
|
|
"source_memory_ids": ["memory-evidence"],
|
|
"source_checkpoint_ids": ["checkpoint-evidence"],
|
|
"source_file_ids": ["file-evidence"],
|
|
"source_git_commit": "4" * 40,
|
|
"model_connection": "historical-connection",
|
|
"model_name": "historical-model",
|
|
"prompt_version": "historical-prompt",
|
|
"curation_job_id": None,
|
|
"source_hash": "a" * 64,
|
|
"source_cursor": 7,
|
|
"created_at": "2026-08-04T01:02:03+00:00",
|
|
"updated_at": "2026-08-04T01:02:03+00:00",
|
|
"title": "Curated history",
|
|
"tags": ["curated"],
|
|
"type": "curated",
|
|
}
|
|
historical_metadata = {**base_metadata, "revision": 1}
|
|
historical = asyncio.run(
|
|
fake.write_note(
|
|
title="Curated history",
|
|
content="Historical curated content.",
|
|
directory=f"global/curated/{profile_id}",
|
|
tags=["curated"],
|
|
note_type="curated",
|
|
metadata=historical_metadata,
|
|
overwrite=False,
|
|
)
|
|
)
|
|
relative = f"{historical['permalink']}.md"
|
|
_git(repository_path, "add", relative)
|
|
_git(repository_path, "commit", "-m", "test: historical curated document")
|
|
historical_commit = _git(repository_path, "rev-parse", "HEAD")
|
|
|
|
current_metadata = {
|
|
**base_metadata,
|
|
"revision": 2,
|
|
"source_hash": "b" * 64,
|
|
"source_cursor": 11,
|
|
"model_name": "current-model",
|
|
"updated_at": "2026-08-04T02:02:03+00:00",
|
|
}
|
|
asyncio.run(
|
|
fake.write_note(
|
|
title="Curated history",
|
|
content="Current curated content.",
|
|
directory=f"global/curated/{profile_id}",
|
|
tags=["curated"],
|
|
note_type="curated",
|
|
metadata=current_metadata,
|
|
overwrite=True,
|
|
)
|
|
)
|
|
_git(repository_path, "add", relative)
|
|
_git(repository_path, "commit", "-m", "test: current curated document")
|
|
with client.app.state.session_factory() as db:
|
|
document = CuratedDocument(
|
|
id=document_id,
|
|
scope="global",
|
|
project_id=None,
|
|
usage_profile_id=profile_id,
|
|
document_type="preferences",
|
|
title="Curated history",
|
|
path=historical["permalink"],
|
|
revision=2,
|
|
source_hash="b" * 64,
|
|
source_cursor=11,
|
|
metadata_json=json.dumps(current_metadata),
|
|
model="current-model",
|
|
prompt_version="historical-prompt",
|
|
)
|
|
db.add(document)
|
|
for revision, content, metadata in (
|
|
(1, "Historical curated content.", historical_metadata),
|
|
(2, "Current curated content.", current_metadata),
|
|
):
|
|
history_path = (
|
|
client.app.state.settings.curated_history_dir / document_id / f"{revision}.md"
|
|
)
|
|
history_path.parent.mkdir(parents=True, exist_ok=True)
|
|
history_path.write_text(content, encoding="utf-8")
|
|
db.add(
|
|
CuratedDocumentRevision(
|
|
document_id=document_id,
|
|
revision=revision,
|
|
storage_path=str(history_path.relative_to(client.app.state.settings.data_dir)),
|
|
model=str(metadata["model_name"]),
|
|
source_hash=str(metadata["source_hash"]),
|
|
metadata_json=json.dumps(metadata),
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
restored = client.post(
|
|
f"/api/v1/git/repositories/{repository['id']}/versions/{document_id}/restore",
|
|
json={"commit": historical_commit},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert restored.status_code == 200, restored.text
|
|
assert restored.json()["revision"] == 3
|
|
with client.app.state.session_factory() as db:
|
|
current = db.get(CuratedDocument, document_id)
|
|
revision = db.scalar(
|
|
select(CuratedDocumentRevision).where(
|
|
CuratedDocumentRevision.document_id == document_id,
|
|
CuratedDocumentRevision.revision == 3,
|
|
)
|
|
)
|
|
assert current is not None and revision is not None
|
|
current_metadata_after_restore = json.loads(current.metadata_json)
|
|
revision_metadata = json.loads(revision.metadata_json)
|
|
assert current.usage_profile_id == profile_id
|
|
assert current.source_hash == "a" * 64
|
|
assert current.source_cursor == 7
|
|
assert current.model == "git-restore"
|
|
assert current_metadata_after_restore["source_memory_ids"] == ["memory-evidence"]
|
|
assert current_metadata_after_restore["restored_from_commit"] == historical_commit
|
|
assert current_metadata_after_restore["revision"] == 3
|
|
assert revision_metadata == current_metadata_after_restore
|
|
|
|
db.delete(current)
|
|
db.commit()
|
|
asyncio.run(fake.delete_note(historical["permalink"]))
|
|
version = client.app.state.git_manager.version_get(
|
|
repository["id"], historical_commit, document_id
|
|
)
|
|
rebuilt = asyncio.run(client.app.state.git_manager._create_from_version(version))
|
|
assert rebuilt["kind"] == "curated"
|
|
with client.app.state.session_factory() as db:
|
|
current = db.get(CuratedDocument, document_id)
|
|
revision = db.scalar(
|
|
select(CuratedDocumentRevision).where(
|
|
CuratedDocumentRevision.document_id == document_id,
|
|
CuratedDocumentRevision.revision == 1,
|
|
)
|
|
)
|
|
assert current is not None and revision is not None
|
|
rebuilt_metadata = json.loads(current.metadata_json)
|
|
assert current.usage_profile_id == profile_id
|
|
assert current.path.startswith(f"global/curated/{profile_id}/")
|
|
assert rebuilt_metadata["source_checkpoint_ids"] == ["checkpoint-evidence"]
|
|
assert json.loads(revision.metadata_json) == rebuilt_metadata
|
|
|
|
|
|
def test_remote_inspect_import_and_unbind(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
memory_id = "11111111-1111-4111-8111-111111111111"
|
|
remote = tmp_path / "memory-remote.git"
|
|
_create_remote_memory(remote, tmp_path / "source", memory_id)
|
|
repository = _enable_local_git(client, str(remote))
|
|
|
|
inspected = client.post(
|
|
f"/api/v1/git/repositories/{repository['id']}/inspect",
|
|
json={"branch": "main"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert inspected.status_code == 200, inspected.text
|
|
assert inspected.json()["valid"] is True
|
|
assert inspected.json()["relationship"] == "unrelated"
|
|
assert inspected.json()["stable_ids"] == [memory_id]
|
|
with client.app.state.session_factory() as db:
|
|
assert db.get(MemoryReference, memory_id) is None
|
|
|
|
imported = client.post(
|
|
f"/api/v1/git/repositories/{repository['id']}/remote/import",
|
|
json={"scope": "document", "resource_id": memory_id, "branch": "main"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert imported.status_code == 200, imported.text
|
|
assert imported.json()["imported"] == 1
|
|
with client.app.state.session_factory() as db:
|
|
reference = db.get(MemoryReference, memory_id)
|
|
assert reference is not None and reference.status == "active"
|
|
assert fake.notes["global/remote-memory"]["content"].strip() == "Imported from remote."
|
|
|
|
unbound = client.delete(
|
|
f"/api/v1/git/repositories/{repository['id']}/remote",
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert unbound.status_code == 200, unbound.text
|
|
assert unbound.json()["remote_url"] is None
|
|
assert "origin" not in _git(client.app.state.settings.memories_dir, "remote").splitlines()
|
|
|
|
|
|
def test_remote_bootstrap_preserves_remote_history(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
memory_id = "22222222-2222-4222-8222-222222222222"
|
|
remote = tmp_path / "bootstrap-remote.git"
|
|
remote_commit = _create_remote_memory(remote, tmp_path / "bootstrap-source", memory_id)
|
|
repository = _enable_local_git(client, str(remote))
|
|
|
|
bootstrap = client.post(
|
|
"/api/v1/git/repositories/bootstrap",
|
|
json={
|
|
"repository_id": repository["id"],
|
|
"branch": "main",
|
|
"confirmed_empty": True,
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert bootstrap.status_code == 200, bootstrap.text
|
|
result = bootstrap.json()
|
|
assert result["remote_commit"] == remote_commit
|
|
assert result["restored"] == 1
|
|
assert (
|
|
_git(
|
|
client.app.state.settings.memories_dir,
|
|
"merge-base",
|
|
"--is-ancestor",
|
|
remote_commit,
|
|
"HEAD",
|
|
)
|
|
== ""
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
reference = db.get(MemoryReference, memory_id)
|
|
repository_record = db.get(MemoryRepository, repository["id"])
|
|
assert reference is not None and reference.status == "active"
|
|
assert repository_record is not None
|
|
assert repository_record.last_pushed_commit == remote_commit
|
|
|
|
|
|
def test_initial_baseline_is_queued_and_push_to_create_setting_is_enforced(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
client = initialized_client
|
|
remote = tmp_path / "empty-remote.git"
|
|
subprocess.run(["git", "init", "--bare", str(remote)], check=True, capture_output=True)
|
|
repository = _enable_local_git(client, str(remote))
|
|
with client.app.state.session_factory() as db:
|
|
jobs = db.scalars(
|
|
select(GitSyncJob).where(GitSyncJob.repository_id == repository["id"])
|
|
).all()
|
|
assert len(jobs) == 1
|
|
assert jobs[0].operation_id.startswith("baseline-sync:")
|
|
assert asyncio.run(client.app.state.git_manager.process_once()) is True
|
|
assert _git(remote, "rev-parse", "refs/heads/main")
|
|
synchronized = client.get(f"/api/v1/git/repositories/{repository['id']}/status").json()
|
|
assert synchronized["status"] == "synced"
|
|
|
|
client.delete("/api/v1/git/connection", headers=csrf_headers(client))
|
|
shutil.rmtree(client.app.state.settings.memories_dir / ".git")
|
|
missing_remote = tmp_path / "missing.git"
|
|
response = client.put(
|
|
"/api/v1/git/connection",
|
|
json={
|
|
"enabled": True,
|
|
"name": "No remote creation",
|
|
"mode": "single",
|
|
"remote_url": str(missing_remote),
|
|
"allow_push_to_create": False,
|
|
"author_name": "MemRelay Tests",
|
|
"author_email": "memrelay@example.test",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
current = client.get("/api/v1/git/repositories").json()[0]
|
|
assert asyncio.run(client.app.state.git_manager.process_once()) is True
|
|
state = client.get(f"/api/v1/git/repositories/{current['id']}/status").json()
|
|
assert state["status"] == "remote_repository_missing"
|
|
assert not missing_remote.exists()
|
|
|
|
|
|
def test_repository_mode_can_migrate_round_trip(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.curation.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
project_response = client.post(
|
|
"/api/v1/projects",
|
|
json={"name": "Migration project"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert project_response.status_code == 201, project_response.text
|
|
project_id = project_response.json()["id"]
|
|
global_memory = _save_memory(client, "migration-global", "Global memory", "Global data")
|
|
project_memory_response = client.post(
|
|
"/api/v1/memories",
|
|
json={
|
|
"request_id": "migration-project",
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"memory_type": "decision",
|
|
"title": "Project memory",
|
|
"content": "Project data",
|
|
"tags": ["git-test"],
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert project_memory_response.status_code == 201, project_memory_response.text
|
|
project_memory = project_memory_response.json()
|
|
|
|
single = _enable_local_git(client)
|
|
root = client.app.state.settings.memories_dir
|
|
single_head = _git(root, "rev-parse", "HEAD")
|
|
|
|
preview = client.post(
|
|
"/api/v1/git/mode-migration/preview",
|
|
json={"target_mode": "per_workspace"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert preview.status_code == 200, preview.text
|
|
assert preview.json()["can_migrate"] is True
|
|
migrated = client.post(
|
|
"/api/v1/git/mode-migration/execute",
|
|
json={"target_mode": "per_workspace", "confirmed": True},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert migrated.status_code == 200, migrated.text
|
|
assert not (root / ".git").exists()
|
|
assert (root / "global" / ".git").exists()
|
|
assert (root / "projects" / project_id / ".git").exists()
|
|
assert (root / f"{global_memory['path']}.md").exists()
|
|
assert (root / f"{project_memory['path']}.md").exists()
|
|
|
|
returned = client.post(
|
|
"/api/v1/git/mode-migration/execute",
|
|
json={"target_mode": "single", "confirmed": True},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert returned.status_code == 200, returned.text
|
|
assert (root / ".git").exists()
|
|
assert not (root / "global" / ".git").exists()
|
|
assert not (root / "projects" / project_id / ".git").exists()
|
|
|
|
with client.app.state.session_factory() as db:
|
|
connection = db.scalar(select(GitConnection))
|
|
active = db.scalars(
|
|
select(MemoryRepository).where(MemoryRepository.connection_id == connection.id)
|
|
).all()
|
|
archived = db.scalars(
|
|
select(MemoryRepository).where(MemoryRepository.status == "archived")
|
|
).all()
|
|
assert connection is not None and connection.mode == "single"
|
|
assert len(active) == 1 and active[0].local_path == str(root)
|
|
assert len(archived) == 3
|
|
original = next(item for item in archived if item.id == single["id"])
|
|
assert _git(Path(original.local_path), "rev-parse", single_head) == single_head
|
|
|
|
|
|
def test_mode_migration_failure_restores_repository_and_job_state(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
repository = _enable_local_git(client)
|
|
manager = client.app.state.git_manager
|
|
root = client.app.state.settings.memories_dir
|
|
original_head = _git(root, "rev-parse", "HEAD")
|
|
retry_at = datetime.now(UTC) + timedelta(hours=1)
|
|
lease_until = datetime.now(UTC) + timedelta(minutes=5)
|
|
with client.app.state.session_factory() as db:
|
|
record = db.get(MemoryRepository, repository["id"])
|
|
assert record is not None
|
|
record.pending_commits = 1
|
|
db.add(
|
|
GitSyncJob(
|
|
id="migration-rollback-job",
|
|
operation_id="migration-rollback-job",
|
|
repository_id=record.id,
|
|
action="sync",
|
|
summary="Preserve this state",
|
|
status="committing",
|
|
next_retry_at=retry_at,
|
|
lease_owner="original-worker",
|
|
lease_expires_at=lease_until,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
def fail_after_creating_target(repository_id: str) -> None:
|
|
with client.app.state.session_factory() as db:
|
|
record = db.get(MemoryRepository, repository_id)
|
|
assert record is not None
|
|
target_git = Path(record.local_path) / ".git"
|
|
target_git.mkdir(parents=True, exist_ok=True)
|
|
raise RuntimeError("simulated repository initialization failure")
|
|
|
|
with (
|
|
patch.object(manager, "_initialize_repository", side_effect=fail_after_creating_target),
|
|
pytest.raises(RuntimeError, match="simulated repository initialization failure"),
|
|
):
|
|
asyncio.run(
|
|
manager.migrate_mode(
|
|
GitModeMigrationRequest(target_mode="per_workspace", confirmed=True)
|
|
)
|
|
)
|
|
|
|
assert (root / ".git").exists()
|
|
assert _git(root, "rev-parse", "HEAD") == original_head
|
|
assert not (root / "global" / ".git").exists()
|
|
with client.app.state.session_factory() as db:
|
|
connection = db.scalar(select(GitConnection))
|
|
record = db.get(MemoryRepository, repository["id"])
|
|
job = db.get(GitSyncJob, "migration-rollback-job")
|
|
assert connection is not None and connection.mode == "single"
|
|
assert record is not None
|
|
assert record.connection_id == connection.id
|
|
assert record.local_path == str(root)
|
|
assert record.pending_commits == 1
|
|
assert job is not None
|
|
assert job.status == "committing"
|
|
assert job.next_retry_at == retry_at
|
|
assert job.lease_owner == "original-worker"
|
|
assert job.lease_expires_at == lease_until
|
|
|
|
migration_root = client.app.state.settings.git_archive_dir / "mode-migrations"
|
|
assert not migration_root.exists() or not any(migration_root.iterdir())
|
|
|
|
|
|
def test_remote_outage_restart_recovery_and_divergence_stop_automatic_push(
|
|
initialized_client: TestClient,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = DiskBasicMemory(client.app.state.settings.memories_dir)
|
|
client.app.state.basic_memory = fake
|
|
client.app.state.git_manager.basic_memory = fake
|
|
remote = tmp_path / "memory-remote.git"
|
|
subprocess.run(["git", "init", "--bare", str(remote)], check=True, capture_output=True)
|
|
repository = _enable_local_git(client, str(remote))
|
|
_drain_git_jobs(client)
|
|
|
|
offline = tmp_path / "memory-remote.offline"
|
|
remote.rename(offline)
|
|
_save_memory(client, "offline-one", "Offline one", "First offline change")
|
|
assert asyncio.run(client.app.state.git_manager.process_once()) is True
|
|
_save_memory(client, "offline-two", "Offline two", "Second offline change")
|
|
assert asyncio.run(client.app.state.git_manager.process_once()) is True
|
|
with client.app.state.session_factory() as db:
|
|
waiting = db.scalars(
|
|
select(GitSyncJob)
|
|
.where(GitSyncJob.status == "waiting_remote")
|
|
.order_by(GitSyncJob.created_at)
|
|
).all()
|
|
assert len(waiting) == 2
|
|
assert all(item.target_commit for item in waiting)
|
|
waiting[0].status = "pushing"
|
|
waiting[0].lease_owner = "stale-worker"
|
|
waiting[0].lease_expires_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
for item in waiting:
|
|
item.next_retry_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
db.commit()
|
|
|
|
offline.rename(remote)
|
|
restarted = GitManager(
|
|
client.app.state.settings,
|
|
client.app.state.session_factory,
|
|
client.app.state.secret_box,
|
|
client.app.state.vault,
|
|
fake,
|
|
RuntimeLeaseManager(client.app.state.session_factory, "restarted-git-worker"),
|
|
)
|
|
for _ in range(5):
|
|
if not asyncio.run(restarted.process_once()):
|
|
break
|
|
with client.app.state.session_factory() as db:
|
|
jobs = db.scalars(select(GitSyncJob)).all()
|
|
assert all(item.status == "completed" for item in jobs)
|
|
record = db.get(MemoryRepository, repository["id"])
|
|
assert record is not None and record.status == "synced"
|
|
assert record.pending_commits == 0
|
|
|
|
local = client.app.state.settings.memories_dir
|
|
local_head = _git(local, "rev-parse", "HEAD")
|
|
remote_head = subprocess.run(
|
|
["git", "--git-dir", str(remote), "rev-parse", "refs/heads/main"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
assert remote_head == local_head
|
|
assert "Offline one" in (local / "global" / "offline-one.md").read_text(encoding="utf-8")
|
|
assert "Offline two" in (local / "global" / "offline-two.md").read_text(encoding="utf-8")
|
|
|
|
external = tmp_path / "external-writer"
|
|
subprocess.run(
|
|
["git", "clone", "--branch", "main", str(remote), str(external)],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
_git(external, "config", "user.name", "External Writer")
|
|
_git(external, "config", "user.email", "external@example.test")
|
|
(external / "external.md").write_text("remote-only change\n", encoding="utf-8")
|
|
_git(external, "add", "external.md")
|
|
_git(external, "commit", "-m", "feat: external divergence")
|
|
_git(external, "push", "origin", "main")
|
|
|
|
_save_memory(client, "local-divergence", "Local divergence", "Local-only change")
|
|
assert asyncio.run(restarted.process_once()) is True
|
|
with client.app.state.session_factory() as db:
|
|
failed = db.scalar(
|
|
select(GitSyncJob)
|
|
.where(GitSyncJob.status == "failed")
|
|
.order_by(GitSyncJob.created_at.desc())
|
|
)
|
|
record = db.get(MemoryRepository, repository["id"])
|
|
assert failed is not None and "REMOTE_DIVERGED" in (failed.error_message or "")
|
|
assert record is not None and record.status == "remote_diverged"
|
|
assert failed.next_retry_at is None
|
|
remote_after_divergence = subprocess.run(
|
|
["git", "--git-dir", str(remote), "rev-parse", "refs/heads/main"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
assert remote_after_divergence != _git(local, "rev-parse", "HEAD")
|
|
|
|
|
|
def test_two_git_workers_claim_one_job_once(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
repository = _enable_local_git(client)
|
|
root = client.app.state.settings.memories_dir
|
|
(root / "worker-test.md").write_text("one logical change\n", encoding="utf-8")
|
|
with client.app.state.session_factory() as db:
|
|
record = db.get(MemoryRepository, repository["id"])
|
|
assert record is not None
|
|
record.pending_commits = 1
|
|
db.add(
|
|
GitSyncJob(
|
|
id="cross-worker-job",
|
|
operation_id="cross-worker-job",
|
|
repository_id=record.id,
|
|
action="save",
|
|
summary="Cross-worker commit",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
second = GitManager(
|
|
client.app.state.settings,
|
|
client.app.state.session_factory,
|
|
client.app.state.secret_box,
|
|
client.app.state.vault,
|
|
client.app.state.basic_memory,
|
|
RuntimeLeaseManager(client.app.state.session_factory, "second-git-worker"),
|
|
)
|
|
|
|
async def process_together() -> list[bool]:
|
|
return list(
|
|
await asyncio.gather(
|
|
client.app.state.git_manager.process_once(),
|
|
second.process_once(),
|
|
)
|
|
)
|
|
|
|
results = asyncio.run(process_together())
|
|
assert sum(results) == 1
|
|
with client.app.state.session_factory() as db:
|
|
job = db.get(GitSyncJob, "cross-worker-job")
|
|
record = db.get(MemoryRepository, repository["id"])
|
|
assert job is not None and job.status == "completed"
|
|
assert job.attempts == 1
|
|
assert job.lease_owner is None and job.lease_expires_at is None
|
|
assert record is not None and record.pending_commits == 0
|
|
assert _git(root, "log", "-1", "--pretty=%s") == "save: Cross-worker commit"
|