obsidian-plugin/lllin000_PaperForge
1
0
Fork 0
mirror of https://github.com/lllin000/PaperForge.git synced 2026-07-22 06:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: CI dispatch tests — mock at SyncService level, restore --selection/--index flags

Browse source
This commit is contained in:
Research Assistant 2026-05-09 22:42:53 +08:00
parent 64e73067fb
commit a466a42d32
3 changed files with 89 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
paperforge/commands/sync.py
View file

@ -59,7 +59,7 @@ def run(args: argparse.Namespace) -> int:
from paperforge.services.sync_service import SyncService
svc = SyncService(vault)
result = svc.run(verbose=verbose, json_output=json_output)
result = svc.run(verbose=verbose, json_output=json_output, selection_only=selection_only, index_only=index_only)
if json_output:
print(result.to_json())

97
paperforge/services/sync_service.py
View file

@ -208,7 +208,9 @@ class SyncService:
logger.error("Failed to build candidate: %s", e)
return candidates
def run(self, verbose: bool = False, json_output: bool = False) -> PFResult:
def run(
self, verbose: bool = False, json_output: bool = False, selection_only: bool = False, index_only: bool = False
) -> PFResult:
"""Full sync orchestration. Returns PFResult contract.
v2.2: Service now orchestrates the full sync lifecycle:
@ -217,10 +219,12 @@ class SyncService:
3. Build canonical index (asset_index)
4. Clean orphaned records + flat notes
5. Return PFResult
"""
# ── Phase 1: Select ──
from paperforge.worker.sync import run_selection_sync
Args:
selection_only: Skip index build + cleanup phases.
index_only: Skip selection sync phase.
"""
# ── Pre-check: BBT exports ──
_export_code = "ok"
try:
rows = self.load_exports()
@ -229,54 +233,65 @@ class SyncService:
except Exception:
_export_code = "BBT_EXPORT_INVALID"
try:
selection_result = run_selection_sync(self.vault, verbose=verbose, json_output=json_output)
except Exception as exc:
return PFResult(
ok=False,
command="sync",
version=__version__,
error=PFError(
code=ErrorCode.SYNC_FAILED,
message=str(exc),
details={"phase": "selection", "exception_type": type(exc).__name__},
),
)
selection_result: dict = {"new": 0, "updated": 0, "skipped": 0, "failed": 0, "errors": []}
# ── Phase 1: Select ──
if not index_only:
from paperforge.worker.sync import run_selection_sync
try:
selection_result = run_selection_sync(self.vault, verbose=verbose, json_output=json_output)
except Exception as exc:
return PFResult(
ok=False,
command="sync",
version=__version__,
error=PFError(
code=ErrorCode.SYNC_FAILED,
message=str(exc),
details={"phase": "selection", "exception_type": type(exc).__name__},
),
)
# ── Phase 2: Index ──
from paperforge.worker._domain import load_domain_config
from paperforge.worker.base_views import ensure_base_views
index_count = 0
orphaned = 0
flat_cleaned = 0
paths = self.resolve_paths()
config = load_domain_config(paths)
ensure_base_views(self.vault, paths, config)
domain_lookup = {entry["export_file"]: entry["domain"] for entry in config["domains"]}
if not selection_only:
from paperforge.worker._domain import load_domain_config
from paperforge.worker.base_views import ensure_base_views
from paperforge.config import load_vault_config
paths = self.resolve_paths()
config = load_domain_config(paths)
ensure_base_views(self.vault, paths, config)
domain_lookup = {entry["export_file"]: entry["domain"] for entry in config["domains"]}
cfg = load_vault_config(self.vault)
exports: dict[str, dict[str, dict]] = {}
for export_path in sorted(paths["exports"].glob("*.json")):
domain = domain_lookup.get(export_path.name, export_path.stem)
export_rows = load_export_rows(export_path)
exports[domain] = {row["key"]: row for row in export_rows}
from paperforge.config import load_vault_config
from paperforge.worker.sync import migrate_to_workspace
cfg = load_vault_config(self.vault)
exports: dict[str, dict[str, dict]] = {}
for export_path in sorted(paths["exports"].glob("*.json")):
domain = domain_lookup.get(export_path.name, export_path.stem)
export_rows = load_export_rows(export_path)
exports[domain] = {row["key"]: row for row in export_rows}
migrate_to_workspace(self.vault, paths)
from paperforge.worker.sync import migrate_to_workspace
import paperforge.worker.asset_index as asset_index
migrate_to_workspace(self.vault, paths)
index_count = asset_index.build_index(self.vault, verbose)
import paperforge.worker.asset_index as asset_index
# ── Phase 3: Clean ──
orphaned = self.clean_orphaned_records(exports, paths, json_output=json_output)
flat_cleaned = self.clean_flat_notes(paths, json_output=json_output)
index_count = asset_index.build_index(self.vault, verbose)
if not json_output:
print(f"index-refresh: {index_count} entries in index")
total = sum(1 for _ in paths["literature"].rglob("*.md")) if paths["literature"].exists() else 0
print(f"index-refresh: {total} formal note(s) in literature")
# ── Phase 3: Clean ──
orphaned = self.clean_orphaned_records(exports, paths, json_output=json_output)
flat_cleaned = self.clean_flat_notes(paths, json_output=json_output)
if not json_output:
print(f"index-refresh: {index_count} entries in index")
total = sum(1 for _ in paths["literature"].rglob("*.md")) if paths["literature"].exists() else 0
print(f"index-refresh: {total} formal note(s) in literature")
all_errors = list(selection_result.get("errors", []))
is_ok = selection_result.get("failed", 0) == 0 and len(all_errors) == 0

45
tests/test_cli_worker_dispatch.py
View file

@ -18,14 +18,16 @@ def stub_run_status(vault: Path, verbose: bool = False, json_output: bool = Fals
return 0
def stub_run_selection_sync(vault: Path, verbose: bool = False, json_output: bool = False) -> int:
def stub_run_selection_sync(vault: Path, verbose: bool = False, json_output: bool = False) -> dict:
CAPTURED_CALLS.append(("run_selection_sync", vault))
return 0
return {"new": 0, "updated": 0, "skipped": 0, "failed": 0, "errors": []}
def stub_run_index_refresh(vault: Path, verbose: bool = False, rebuild_index: bool = False, json_output: bool = False) -> int:
def stub_run_index_refresh(
vault: Path, verbose: bool = False, rebuild_index: bool = False, json_output: bool = False
) -> dict:
CAPTURED_CALLS.append(("run_index_refresh", vault))
return 0
return {"updated": 0, "failed": 0, "errors": []}
def stub_run_deep_reading(vault: Path, verbose: bool = False) -> int:
@ -82,33 +84,50 @@ def test_status_dispatch(clean_captured, mock_vault):
def test_selection_sync_dispatch(clean_captured, mock_vault):
"""main(['--vault', vault, 'selection-sync']) calls run_selection_sync."""
"""main(['--vault', vault, 'selection-sync']) runs SyncService (v2.1 contract)."""
import importlib
import paperforge.cli as cli
import paperforge.worker.sync as wsync
importlib.reload(cli)
importlib.reload(wsync)
with patch.object(cli, "run_selection_sync", stub_run_selection_sync):
with patch.object(wsync, "run_selection_sync", stub_run_selection_sync):
argv = ["--vault", str(mock_vault), "selection-sync"]
import paperforge.cli as cli
importlib.reload(cli)
cli.main(argv)
assert ("run_selection_sync", mock_vault) in CAPTURED_CALLS
def test_index_refresh_dispatch(clean_captured, mock_vault):
"""main(['--vault', vault, 'index-refresh']) calls run_index_refresh."""
"""main(['--vault', vault, 'index-refresh']) runs index-only SyncService lookup."""
# v2.1: index-refresh routes through SyncService, not directly to worker.
# Mock the worker import that SyncService uses for the selection phase skip.
import importlib
import paperforge.cli as cli
import paperforge.services.sync_service as svc_mod
importlib.reload(cli)
importlib.reload(svc_mod)
with patch.object(cli, "run_index_refresh", stub_run_index_refresh):
captured_vault = []
def mock_run(self, verbose=False, json_output=False, selection_only=False, index_only=False):
captured_vault.append(self.vault)
from paperforge.core.result import PFResult
return PFResult(ok=True, command="sync", version="1.0.0", data={})
with patch.object(svc_mod.SyncService, "run", mock_run):
argv = ["--vault", str(mock_vault), "index-refresh"]
import paperforge.cli as cli
importlib.reload(cli)
cli.main(argv)
assert ("run_index_refresh", mock_vault) in CAPTURED_CALLS
assert len(captured_vault) == 1
assert captured_vault[0] == mock_vault
def test_deep_reading_dispatch(clean_captured, mock_vault):