mirror of
https://github.com/lllin000/PaperForge.git
synced 2026-07-22 06:50:53 +00:00
2b552be9e6
Atomic snapshots, canonical index mutation serialization, sync post-clean truth, plugin path config-awareness, embed stop signal honesty, full snapshot bootstrap, pf_ prefix unification, workflow command/lifecycle/path corrections, mechanical/cognitive route separation with unknown-command guard.
63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
from pathlib import Path
|
|
|
|
from paperforge.commands.embed import run
|
|
from paperforge.memory.vector_db import write_vector_build_state, read_vector_build_state
|
|
|
|
|
|
def test_embed_stop_returns_ok_when_no_active_job(tmp_path):
|
|
from argparse import Namespace
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
args = Namespace(vault_path=vault, embed_subcommand="stop", json=True)
|
|
result_code = run(args)
|
|
assert result_code == 0
|
|
|
|
|
|
def test_embed_status_includes_build_state(tmp_path):
|
|
from argparse import Namespace
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
write_vector_build_state(vault, {"status": "running", "current": 3, "total": 10})
|
|
args = Namespace(vault_path=vault, embed_subcommand="status", json=True)
|
|
with patch("paperforge.commands.embed.get_embed_status") as mock_status:
|
|
mock_status.return_value = {"db_exists": True, "chunk_count": 0, "model": "test", "mode": "local"}
|
|
with patch("paperforge.commands.embed.read_vector_build_state") as mock_read:
|
|
mock_read.return_value = {"status": "running", "current": 3, "total": 10}
|
|
result_code = run(args)
|
|
assert result_code == 0
|
|
|
|
|
|
def test_embed_stop_requests_signal_for_running_job(tmp_path):
|
|
from argparse import Namespace
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
args = Namespace(vault_path=vault, embed_subcommand="stop", json=True)
|
|
|
|
with patch("paperforge.commands.embed.read_vector_build_state") as mock_read:
|
|
mock_read.return_value = {"status": "running", "pid": 12345}
|
|
with patch("paperforge.commands.embed.os.kill") as mock_kill:
|
|
with patch("paperforge.commands.embed.mark_vector_build_state") as mock_mark:
|
|
result_code = run(args)
|
|
|
|
assert result_code == 0
|
|
mock_kill.assert_called_once()
|
|
mock_mark.assert_called_once()
|
|
|
|
|
|
def test_embed_stop_returns_error_when_signal_fails(tmp_path):
|
|
from argparse import Namespace
|
|
|
|
vault = tmp_path / "vault"
|
|
vault.mkdir()
|
|
args = Namespace(vault_path=vault, embed_subcommand="stop", json=True)
|
|
|
|
with patch("paperforge.commands.embed.read_vector_build_state") as mock_read:
|
|
mock_read.return_value = {"status": "running", "pid": 12345}
|
|
with patch("paperforge.commands.embed.os.kill", side_effect=OSError("denied")):
|
|
result_code = run(args)
|
|
|
|
assert result_code == 1
|