wehub-resource-sync
2860fb5d18
Security / Dependency review (push) Has been skipped
Scorecard / Scorecard analysis (push) Failing after 0s
Validate / eval (push) Failing after 0s
Security / Dependency audit (push) Failing after 1s
Security / Secret scan (push) Failing after 1s
Validate / tests (push) Failing after 0s
Validate / mcp-tests (push) Failing after 1s
GitHub Actions Security Analysis with zizmor 🌈 / zizmor (push) Failing after 1s
Security / SAST scan (push) Failing after 13m52s
869 lines
32 KiB
Python
869 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import last30days as cli
|
|
import store
|
|
from lib import corpus, env, health, html_render, library, library_index, pipeline, render, schema
|
|
|
|
|
|
def _set_mtime(path: Path, value: str) -> None:
|
|
timestamp = datetime.fromisoformat(value).replace(tzinfo=timezone.utc).timestamp()
|
|
os.utime(path, (timestamp, timestamp))
|
|
|
|
|
|
def _scan(tmp_path: Path, *, all_time: bool = False, limit: int = 12):
|
|
return corpus.search(
|
|
"MCP servers",
|
|
[tmp_path],
|
|
from_date="2026-06-10",
|
|
to_date="2026-07-10",
|
|
all_time=all_time,
|
|
limit=limit,
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
|
|
|
|
def test_scans_matching_text_and_markdown_with_path_titles(tmp_path):
|
|
note = tmp_path / "mcp-server-notes.md"
|
|
note.write_text("MCP servers expose tools to local coding agents.", encoding="utf-8")
|
|
other = tmp_path / "groceries.txt"
|
|
other.write_text("milk eggs bread", encoding="utf-8")
|
|
_set_mtime(note, "2026-07-05T12:00:00")
|
|
_set_mtime(other, "2026-07-05T12:00:00")
|
|
|
|
result = _scan(tmp_path)
|
|
|
|
assert [item.title for item in result.items] == ["mcp server notes"]
|
|
assert result.items[0].source == "corpus"
|
|
assert result.items[0].published_at == "2026-07-05"
|
|
assert result.items[0].metadata["local_only"] is True
|
|
assert result.items[0].url.startswith("corpus://")
|
|
assert str(note) not in result.items[0].url
|
|
|
|
|
|
def test_multilingual_matching_reuses_shared_cjk_tokenizer(tmp_path):
|
|
note = tmp_path / "模型记录.md"
|
|
note.write_text("国产大模型的最新测评和部署记录", encoding="utf-8")
|
|
_set_mtime(note, "2026-07-05T12:00:00")
|
|
|
|
result = corpus.search(
|
|
"国产大模型 测评",
|
|
[tmp_path],
|
|
from_date="2026-06-10",
|
|
to_date="2026-07-10",
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
|
|
assert [item.title for item in result.items] == ["模型记录"]
|
|
|
|
|
|
def test_recency_window_and_all_time_override(tmp_path):
|
|
old = tmp_path / "old-mcp-plan.md"
|
|
old.write_text("MCP servers and local tool protocols", encoding="utf-8")
|
|
_set_mtime(old, "2025-01-01T00:00:00")
|
|
|
|
assert _scan(tmp_path).items == []
|
|
assert [item.title for item in _scan(tmp_path, all_time=True).items] == ["old mcp plan"]
|
|
|
|
|
|
def test_hidden_git_and_node_modules_directories_are_ignored(tmp_path):
|
|
visible = tmp_path / "visible.md"
|
|
visible.write_text("MCP servers are visible", encoding="utf-8")
|
|
_set_mtime(visible, "2026-07-05T00:00:00")
|
|
for directory in (tmp_path / ".git", tmp_path / ".hidden", tmp_path / "node_modules"):
|
|
directory.mkdir()
|
|
path = directory / "private.md"
|
|
path.write_text("MCP servers hidden secret", encoding="utf-8")
|
|
_set_mtime(path, "2026-07-05T00:00:00")
|
|
|
|
result = _scan(tmp_path)
|
|
|
|
assert [item.metadata["relative_path"] for item in result.items] == ["visible.md"]
|
|
|
|
|
|
def test_pdf_is_skipped_with_one_note_when_pdftotext_is_absent(tmp_path, monkeypatch):
|
|
pdf = tmp_path / "mcp.pdf"
|
|
pdf.write_bytes(b"not a real pdf")
|
|
_set_mtime(pdf, "2026-07-05T00:00:00")
|
|
monkeypatch.setattr(corpus, "which", lambda _name: None)
|
|
|
|
result = _scan(tmp_path)
|
|
|
|
assert result.items == []
|
|
assert result.notes == ["Skipped PDF files because pdftotext is not on PATH"]
|
|
|
|
|
|
def test_mtime_cache_reuses_text_and_is_private(tmp_path, monkeypatch):
|
|
note = tmp_path / "mcp.md"
|
|
note.write_text("MCP servers cache this local note", encoding="utf-8")
|
|
_set_mtime(note, "2026-07-05T00:00:00")
|
|
first = _scan(tmp_path)
|
|
assert first.cache_hits == 0
|
|
|
|
monkeypatch.setattr(corpus, "_extract_text", mock.Mock(side_effect=AssertionError("cache miss")))
|
|
second = _scan(tmp_path)
|
|
|
|
assert second.cache_hits == 1
|
|
cache_path = tmp_path / "cache" / corpus.CACHE_FILENAME
|
|
assert cache_path.stat().st_mode & 0o777 == 0o600
|
|
assert cache_path.parent.stat().st_mode & 0o777 == 0o700
|
|
|
|
|
|
def test_cache_for_500_large_documents_is_bounded_by_total_bytes(tmp_path, monkeypatch):
|
|
for index in range(500):
|
|
(tmp_path / f"document-{index:03d}.txt").touch()
|
|
|
|
monkeypatch.setattr(corpus, "MAX_CACHE_BYTES", 1_000_000)
|
|
monkeypatch.setattr(corpus, "MAX_CACHE_TEXT_CHARS", 100_000)
|
|
monkeypatch.setattr(corpus, "_extract_text", lambda *_args, **_kwargs: "x" * 1_000_000)
|
|
monkeypatch.setattr(corpus, "_match_score", lambda *_args, **_kwargs: 0.0)
|
|
|
|
result = corpus.search(
|
|
"anything",
|
|
[tmp_path],
|
|
from_date="2026-06-10",
|
|
to_date="2026-07-10",
|
|
all_time=True,
|
|
limit=0,
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
|
|
cache_path = tmp_path / "cache" / corpus.CACHE_FILENAME
|
|
payload = json.loads(cache_path.read_text(encoding="utf-8"))
|
|
assert result.files_scanned == 500
|
|
assert cache_path.stat().st_size <= corpus.MAX_CACHE_BYTES
|
|
assert len(payload["entries"]) < 500
|
|
assert all(
|
|
len(entry["text"]) <= corpus.MAX_CACHE_TEXT_CHARS
|
|
for entry in payload["entries"].values()
|
|
)
|
|
|
|
|
|
def test_result_budget_caps_one_corpus_stream(tmp_path):
|
|
for index in range(10):
|
|
path = tmp_path / f"mcp-{index}.md"
|
|
path.write_text(f"MCP servers local note {index}", encoding="utf-8")
|
|
_set_mtime(path, f"2026-07-{index + 1:02d}T00:00:00")
|
|
|
|
assert len(_scan(tmp_path, limit=3).items) == 3
|
|
|
|
|
|
def test_file_cap_is_shared_fairly_across_corpus_roots(tmp_path, monkeypatch):
|
|
archive = tmp_path / "archive"
|
|
relevant = tmp_path / "relevant"
|
|
archive.mkdir()
|
|
relevant.mkdir()
|
|
for index in range(4):
|
|
(archive / f"mcp-archive-{index}.md").write_text(
|
|
f"MCP servers archived note {index}", encoding="utf-8"
|
|
)
|
|
later = relevant / "mcp-current.md"
|
|
later.write_text("MCP servers relevant current note", encoding="utf-8")
|
|
monkeypatch.setattr(corpus, "MAX_FILES", 4)
|
|
|
|
result = corpus.search(
|
|
"MCP servers",
|
|
[archive, relevant],
|
|
from_date="2026-06-10",
|
|
to_date="2026-07-10",
|
|
all_time=True,
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
|
|
assert result.files_scanned <= corpus.MAX_FILES
|
|
assert any(item.metadata["path"].startswith(str(archive)) for item in result.items)
|
|
assert any(item.metadata["path"] == str(later) for item in result.items)
|
|
|
|
|
|
def test_pipeline_runs_corpus_outside_network_executor(tmp_path, monkeypatch):
|
|
note = tmp_path / "mcp-private.md"
|
|
note.write_text("MCP servers use a secret local transport", encoding="utf-8")
|
|
_set_mtime(note, "2026-07-05T00:00:00")
|
|
retrieve = mock.Mock(return_value=([], {}))
|
|
monkeypatch.setattr(pipeline, "_retrieve_stream", retrieve)
|
|
|
|
report = pipeline.run(
|
|
topic="MCP servers",
|
|
config={"_CORPUS_DIRS": [str(tmp_path)], "EXCLUDE_SOURCES": ""},
|
|
depth="quick",
|
|
requested_sources=["corpus"],
|
|
mock=True,
|
|
as_of_date="2026-07-10",
|
|
external_plan={
|
|
"intent": "concept",
|
|
"freshness_mode": "balanced_recent",
|
|
"cluster_mode": "none",
|
|
"source_weights": {"corpus": 1.0},
|
|
"subqueries": [{
|
|
"label": "primary",
|
|
"search_query": "MCP servers",
|
|
"ranking_query": "MCP servers",
|
|
"sources": ["corpus"],
|
|
}],
|
|
},
|
|
)
|
|
|
|
assert report.source_status["corpus"].state == health.OK
|
|
assert len(report.items_by_source["corpus"]) == 1
|
|
assert len(report.items_by_source["corpus"]) <= pipeline.DEPTH_SETTINGS["quick"]["per_stream_limit"]
|
|
assert all(call.kwargs["source"] != "corpus" for call in retrieve.call_args_list)
|
|
|
|
|
|
def test_corpus_never_enters_remote_rerank_or_fun_prompts(tmp_path, monkeypatch):
|
|
note = tmp_path / "private.md"
|
|
note.write_text("Model context protocol servers PRIVATE-RERANK-SENTINEL", encoding="utf-8")
|
|
_set_mtime(note, "2026-07-05T00:00:00")
|
|
remote = mock.Mock()
|
|
remote.generate_json = mock.Mock(side_effect=AssertionError("private prompt left machine"))
|
|
runtime = schema.ProviderRuntime("local", "remote-planner", "remote-reranker")
|
|
monkeypatch.setattr(pipeline.providers, "resolve_runtime", lambda *_args, **_kwargs: (runtime, remote))
|
|
monkeypatch.setattr(pipeline, "available_sources", lambda *_args, **_kwargs: ["corpus"])
|
|
|
|
report = pipeline.run(
|
|
topic="how do model context protocol servers work today?",
|
|
config={"_CORPUS_DIRS": [str(tmp_path)], "EXCLUDE_SOURCES": ""},
|
|
depth="quick",
|
|
requested_sources=["corpus"],
|
|
mock=False,
|
|
as_of_date="2026-07-10",
|
|
external_plan={
|
|
"intent": "how_to",
|
|
"freshness_mode": "evergreen_ok",
|
|
"cluster_mode": "workflow",
|
|
"source_weights": {"corpus": 1.0},
|
|
"subqueries": [{
|
|
"label": "primary",
|
|
"search_query": "model context protocol servers",
|
|
"ranking_query": "How do model context protocol servers work?",
|
|
"sources": ["corpus"],
|
|
}],
|
|
},
|
|
)
|
|
|
|
assert report.items_by_source["corpus"]
|
|
remote.generate_json.assert_not_called()
|
|
|
|
|
|
def test_explicit_unconfigured_corpus_records_skipped_outcome():
|
|
report = pipeline.run(
|
|
topic="how do local model context protocol servers work?",
|
|
config={"EXCLUDE_SOURCES": ""},
|
|
depth="quick",
|
|
requested_sources=["corpus"],
|
|
mock=True,
|
|
as_of_date="2026-07-10",
|
|
external_plan={
|
|
"intent": "how_to",
|
|
"freshness_mode": "evergreen_ok",
|
|
"cluster_mode": "workflow",
|
|
"source_weights": {"corpus": 1.0},
|
|
"subqueries": [{
|
|
"label": "primary",
|
|
"search_query": "model context protocol servers",
|
|
"ranking_query": "How do model context protocol servers work?",
|
|
"sources": ["corpus"],
|
|
}],
|
|
},
|
|
)
|
|
|
|
outcome = report.source_status["corpus"]
|
|
assert outcome.state == schema.SKIPPED_UNCONFIGURED
|
|
assert outcome.attempted is False
|
|
|
|
|
|
def _privacy_report(secret: str = "PRIVATE-CORPUS-SENTINEL") -> schema.Report:
|
|
item = schema.SourceItem(
|
|
item_id="C-private",
|
|
source="corpus",
|
|
title="private mcp notes",
|
|
body=f"MCP servers {secret}",
|
|
url="",
|
|
published_at="2026-07-05",
|
|
snippet=f"MCP servers {secret}",
|
|
metadata={"relative_path": "notes/private.md", "local_only": True},
|
|
)
|
|
candidate = schema.Candidate(
|
|
candidate_id="corpus:C-private",
|
|
item_id=item.item_id,
|
|
source="corpus",
|
|
title=item.title,
|
|
url="",
|
|
snippet=item.snippet,
|
|
subquery_labels=["primary"],
|
|
native_ranks={"primary:corpus": 1},
|
|
local_relevance=1.0,
|
|
freshness=90,
|
|
engagement=None,
|
|
source_quality=0.75,
|
|
rrf_score=0.01,
|
|
final_score=91,
|
|
cluster_id="cluster-private",
|
|
sources=["corpus"],
|
|
source_items=[item],
|
|
)
|
|
return schema.Report(
|
|
topic="MCP servers",
|
|
range_from="2026-06-10",
|
|
range_to="2026-07-10",
|
|
generated_at="2026-07-10T00:00:00+00:00",
|
|
provider_runtime=schema.ProviderRuntime("local", "mock", "mock"),
|
|
query_plan=schema.QueryPlan(
|
|
intent="concept",
|
|
freshness_mode="balanced_recent",
|
|
cluster_mode="none",
|
|
raw_topic="MCP servers",
|
|
subqueries=[schema.SubQuery("primary", "MCP servers", "MCP servers", ["corpus"])],
|
|
source_weights={"corpus": 1.0},
|
|
),
|
|
clusters=[schema.Cluster(
|
|
cluster_id="cluster-private",
|
|
title=f"Private cluster {secret}",
|
|
candidate_ids=[candidate.candidate_id],
|
|
representative_ids=[candidate.candidate_id],
|
|
sources=["corpus"],
|
|
score=91,
|
|
)],
|
|
ranked_candidates=[candidate],
|
|
items_by_source={"corpus": [item]},
|
|
errors_by_source={},
|
|
source_status={"corpus": schema.SourceOutcome("corpus", health.OK, 1)},
|
|
)
|
|
|
|
|
|
def test_corpus_findings_persist_with_stable_opaque_keys(tmp_path, monkeypatch):
|
|
note = tmp_path / "private-customer-notes.md"
|
|
note.write_text("MCP servers persist this private finding", encoding="utf-8")
|
|
first = corpus.search(
|
|
"MCP servers",
|
|
[tmp_path],
|
|
from_date="2026-06-10",
|
|
to_date="2026-07-10",
|
|
all_time=True,
|
|
cache_dir=tmp_path / "cache",
|
|
).items[0]
|
|
second = corpus.search(
|
|
"MCP servers",
|
|
[tmp_path],
|
|
from_date="2026-06-10",
|
|
to_date="2026-07-10",
|
|
all_time=True,
|
|
cache_dir=tmp_path / "cache",
|
|
).items[0]
|
|
report = _privacy_report()
|
|
report.items_by_source["corpus"][0].url = first.url
|
|
report.ranked_candidates[0].url = first.url
|
|
|
|
db_path = tmp_path / "research.db"
|
|
monkeypatch.setattr(store, "_db_override", db_path)
|
|
store.init_db()
|
|
topic = store.add_topic(report.topic)
|
|
run_id = store.record_run(topic["id"], status="completed")
|
|
findings = store.findings_from_report(report)
|
|
counts = store.store_findings(run_id, topic["id"], findings)
|
|
|
|
assert first.url == second.url
|
|
assert first.url.startswith("corpus://")
|
|
assert str(note) not in first.url
|
|
assert counts == {"new": 1, "updated": 0}
|
|
with sqlite3.connect(db_path) as conn:
|
|
persisted = conn.execute(
|
|
"SELECT source, source_url FROM findings"
|
|
).fetchone()
|
|
assert persisted == ("corpus", first.url)
|
|
assert library_index.search(
|
|
"private finding",
|
|
db_path=tmp_path / "missing-library.db",
|
|
store_db_path=db_path,
|
|
) == []
|
|
|
|
|
|
def test_persist_report_hardens_corpus_store_and_sidecars(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "store" / "research.db"
|
|
monkeypatch.setattr(store, "_db_override", db_path)
|
|
original_findings_from_report = store.findings_from_report
|
|
original_store_findings = store.store_findings
|
|
original_update_run = store.update_run
|
|
|
|
def make_permissive():
|
|
db_path.chmod(0o644)
|
|
for suffix in ("-wal", "-shm"):
|
|
sidecar = Path(f"{db_path}{suffix}")
|
|
sidecar.touch(mode=0o644)
|
|
sidecar.chmod(0o644)
|
|
|
|
def findings_from_report_with_permissive_store(*args, **kwargs):
|
|
findings = original_findings_from_report(*args, **kwargs)
|
|
make_permissive()
|
|
return findings
|
|
|
|
def store_findings_after_private_check(*args, **kwargs):
|
|
for artifact in (db_path, Path(f"{db_path}-wal"), Path(f"{db_path}-shm")):
|
|
assert artifact.stat().st_mode & 0o777 == 0o600
|
|
return original_store_findings(*args, **kwargs)
|
|
|
|
def update_run_with_permissive_sidecars(*args, **kwargs):
|
|
result = original_update_run(*args, **kwargs)
|
|
make_permissive()
|
|
return result
|
|
|
|
monkeypatch.setattr(store, "findings_from_report", findings_from_report_with_permissive_store)
|
|
monkeypatch.setattr(store, "store_findings", store_findings_after_private_check)
|
|
monkeypatch.setattr(store, "update_run", update_run_with_permissive_sidecars)
|
|
|
|
cli.persist_report(_privacy_report())
|
|
|
|
for artifact in (db_path, Path(f"{db_path}-wal"), Path(f"{db_path}-shm")):
|
|
assert artifact.stat().st_mode & 0o777 == 0o600
|
|
|
|
|
|
def test_agent_export_excludes_corpus_by_default_and_allows_explicit_opt_in():
|
|
report = _privacy_report()
|
|
|
|
private_default = schema.to_agent_export(report)
|
|
opted_in = schema.to_agent_export(report, corpus_in_export=True)
|
|
|
|
assert private_default["results"] == []
|
|
assert private_default["clusters"] == []
|
|
assert "corpus" not in private_default["source_status"]
|
|
assert opted_in["results"][0]["source"] == "corpus"
|
|
assert "PRIVATE-CORPUS-SENTINEL" in opted_in["results"][0]["summary"]
|
|
|
|
report.artifacts["corpus_in_export"] = True
|
|
assert schema.to_agent_export(report)["results"][0]["source"] == "corpus"
|
|
|
|
|
|
def test_agent_export_rebuilds_mixed_cluster_title_after_private_representative_removed():
|
|
report = _privacy_report()
|
|
social_item = schema.SourceItem(
|
|
item_id="R-public",
|
|
source="reddit",
|
|
title="Public MCP discussion",
|
|
body="Public evidence",
|
|
url="https://reddit.example/public",
|
|
published_at="2026-07-06",
|
|
snippet="Public evidence",
|
|
)
|
|
social = schema.Candidate(
|
|
candidate_id="reddit:public",
|
|
item_id=social_item.item_id,
|
|
source="reddit",
|
|
title=social_item.title,
|
|
url=social_item.url,
|
|
snippet=social_item.snippet,
|
|
subquery_labels=["primary"],
|
|
native_ranks={"primary:reddit": 1},
|
|
local_relevance=0.8,
|
|
freshness=90,
|
|
engagement=1,
|
|
source_quality=0.6,
|
|
rrf_score=0.01,
|
|
final_score=80,
|
|
cluster_id="cluster-private",
|
|
sources=["reddit"],
|
|
source_items=[social_item],
|
|
)
|
|
report.ranked_candidates.append(social)
|
|
report.items_by_source["reddit"] = [social_item]
|
|
report.source_status["reddit"] = schema.SourceOutcome("reddit", health.OK, 1)
|
|
report.clusters[0].candidate_ids.append(social.candidate_id)
|
|
report.clusters[0].representative_ids.append(social.candidate_id)
|
|
report.clusters[0].sources.append("reddit")
|
|
|
|
exported = schema.to_agent_export(report)
|
|
|
|
assert exported["clusters"][0]["title"] == "Public MCP discussion"
|
|
assert "PRIVATE-CORPUS-SENTINEL" not in str(exported)
|
|
|
|
|
|
def test_local_report_has_badged_from_your_files_section():
|
|
rendered = render.render_compact(_privacy_report())
|
|
|
|
assert "## From your files" in rendered
|
|
assert "LOCAL ONLY" in rendered
|
|
assert "PRIVATE-CORPUS-SENTINEL" in rendered
|
|
|
|
|
|
def test_publish_html_sends_sanitized_report_not_local_corpus(monkeypatch):
|
|
report = _privacy_report()
|
|
captured: dict[str, str] = {}
|
|
|
|
def publish(rendered, **_kwargs):
|
|
captured["html"] = rendered
|
|
return {"url": "https://example.ht-ml.app"}
|
|
|
|
monkeypatch.setattr(cli.env, "get_config", lambda **_kwargs: {})
|
|
monkeypatch.setattr(cli.pipeline, "diagnose", lambda *_args, **_kwargs: {"available_sources": ["corpus"]})
|
|
monkeypatch.setattr(cli.pipeline, "run", lambda **_kwargs: report)
|
|
monkeypatch.setattr(cli, "publish_rendered_html", publish)
|
|
monkeypatch.setenv("LAST30DAYS_SKIP_PREFLIGHT", "1")
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
["last30days.py", "MCP servers", "--emit=html", "--publish-html"],
|
|
)
|
|
|
|
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
assert cli.main() == 0
|
|
|
|
assert "PRIVATE-CORPUS-SENTINEL" not in captured["html"]
|
|
assert "private mcp notes" not in captured["html"]
|
|
|
|
|
|
def test_configured_corpus_bypasses_hosted_backend(tmp_path, monkeypatch):
|
|
report = _privacy_report()
|
|
hosted = mock.Mock(side_effect=AssertionError("local corpus was forwarded"))
|
|
monkeypatch.setattr("lib.hosted.run_hosted", hosted)
|
|
monkeypatch.setattr(
|
|
cli.env,
|
|
"get_config",
|
|
lambda **_kwargs: {"LAST30DAYS_CORPUS_DIRS": str(tmp_path)},
|
|
)
|
|
monkeypatch.setattr(cli.pipeline, "diagnose", lambda *_args, **_kwargs: {"available_sources": ["corpus"]})
|
|
monkeypatch.setattr(cli.pipeline, "run", lambda **_kwargs: report)
|
|
monkeypatch.setenv("LAST30DAYS_API_KEY", "test-hosted-key")
|
|
monkeypatch.setenv("LAST30DAYS_API_BASE", "https://example.invalid")
|
|
monkeypatch.setenv("LAST30DAYS_SKIP_PREFLIGHT", "1")
|
|
monkeypatch.setattr(sys, "argv", ["last30days.py", "MCP servers", "--emit=compact"])
|
|
|
|
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
assert cli.main() == 0
|
|
|
|
hosted.assert_not_called()
|
|
|
|
|
|
def test_library_publish_strips_marked_corpus_but_local_page_keeps_it(tmp_path, monkeypatch):
|
|
markdown = render.render_full(_privacy_report())
|
|
(tmp_path / "mcp-raw.md").write_text(markdown, encoding="utf-8")
|
|
monkeypatch.setattr(library, "DEFAULT_BRIEFS_DIR", tmp_path / "no-briefings")
|
|
monkeypatch.setattr(cli.env, "get_config", lambda **_kwargs: {})
|
|
publish_many = mock.Mock(return_value={})
|
|
monkeypatch.setattr("lib.html_publish.publish_html_documents", publish_many)
|
|
monkeypatch.setattr("lib.html_publish.publish_html", mock.Mock(return_value={"url": "https://library.ht-ml.app"}))
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
["last30days.py", "library", "feed", "--save-dir", str(tmp_path), "--publish"],
|
|
)
|
|
|
|
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
assert cli.main() == 0
|
|
|
|
published_documents = publish_many.call_args.args[0]
|
|
assert published_documents
|
|
assert all("PRIVATE-CORPUS-SENTINEL" not in html for html in published_documents.values())
|
|
local_pages = list((tmp_path / "briefs").glob("*.html"))
|
|
assert local_pages
|
|
assert "PRIVATE-CORPUS-SENTINEL" in local_pages[0].read_text(encoding="utf-8")
|
|
assert local_pages[0].stat().st_mode & 0o777 == 0o600
|
|
assert local_pages[0].parent.stat().st_mode & 0o777 == 0o700
|
|
|
|
|
|
def test_private_corpus_cannot_escape_via_library_search_and_feed_publish(tmp_path, monkeypatch):
|
|
secret = "PRIVATE-CORPUS-ESCAPE-SENTINEL"
|
|
memory = tmp_path / "memory"
|
|
report = _privacy_report(secret)
|
|
with mock.patch.object(library_index, "sync_library"):
|
|
saved = cli.save_output(report, "md", str(memory))
|
|
|
|
db_path = tmp_path / "index" / "library.db"
|
|
library_index.sync_library(memory, tmp_path / "no-briefings", db_path=db_path)
|
|
entry = library._parse_markdown(saved)
|
|
legacy_hash = hashlib.sha256(entry.content.encode("utf-8")).hexdigest()
|
|
with sqlite3.connect(db_path) as conn:
|
|
conn.execute(
|
|
"UPDATE library_documents SET content_hash = ? WHERE entry_id = ?",
|
|
(legacy_hash, entry.entry_id),
|
|
)
|
|
conn.execute("DELETE FROM library_fts WHERE entry_id = ?", (entry.entry_id,))
|
|
conn.execute(
|
|
"INSERT INTO library_fts(entry_id, topic, headline, summary, content) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(entry.entry_id, entry.topic, entry.headline, entry.summary, entry.content),
|
|
)
|
|
conn.commit()
|
|
|
|
rebuilt = library_index.sync_library(
|
|
memory,
|
|
tmp_path / "no-briefings",
|
|
db_path=db_path,
|
|
)
|
|
assert rebuilt.indexed == 1
|
|
assert library_index.search(
|
|
secret,
|
|
db_path=db_path,
|
|
store_db_path=tmp_path / "missing-store.db",
|
|
) == []
|
|
matches = library_index.search(
|
|
"MCP servers",
|
|
db_path=db_path,
|
|
store_db_path=tmp_path / "missing-store.db",
|
|
)
|
|
assert matches
|
|
assert all(secret not in match.snippet for match in matches)
|
|
|
|
followup = schema.without_sources(report, {"corpus"})
|
|
followup.library_context = [
|
|
schema.LibraryContext(
|
|
topic=match.topic,
|
|
published_date=match.published_date.isoformat(),
|
|
headline=match.headline,
|
|
summary=match.snippet,
|
|
source_kind=match.source_kind,
|
|
)
|
|
for match in matches[:1]
|
|
]
|
|
with mock.patch.object(library_index, "sync_library"):
|
|
cli.save_output(followup, "md", str(memory), suffix="followup")
|
|
|
|
monkeypatch.setattr(library, "DEFAULT_BRIEFS_DIR", tmp_path / "no-briefings")
|
|
monkeypatch.setattr(cli.env, "get_config", lambda **_kwargs: {})
|
|
publish_many = mock.Mock(return_value={})
|
|
monkeypatch.setattr("lib.html_publish.publish_html_documents", publish_many)
|
|
monkeypatch.setattr(
|
|
"lib.html_publish.publish_html",
|
|
mock.Mock(return_value={"url": "https://library.ht-ml.app"}),
|
|
)
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
["last30days.py", "library", "feed", "--save-dir", str(memory), "--publish"],
|
|
)
|
|
|
|
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
assert cli.main() == 0
|
|
|
|
published = publish_many.call_args.args[0]
|
|
assert published
|
|
assert all(secret not in document for document in published.values())
|
|
|
|
|
|
def test_configured_paths_use_platform_separator_and_dedupe(tmp_path):
|
|
first = tmp_path / "first"
|
|
second = tmp_path / "second"
|
|
resolved = corpus.resolve_directories(
|
|
[str(first)], f"{first}{os.pathsep}{second}"
|
|
)
|
|
assert resolved == [first.resolve(), second.resolve()]
|
|
|
|
|
|
def test_corpus_env_keys_are_registered(monkeypatch):
|
|
monkeypatch.setenv("LAST30DAYS_CORPUS_DIRS", "/tmp/notes:/tmp/transcripts")
|
|
monkeypatch.setenv("LAST30DAYS_CORPUS_IN_EXPORT", "1")
|
|
with mock.patch.object(env, "_load_keychain", return_value={}), mock.patch.object(
|
|
env, "_load_pass", return_value={}
|
|
):
|
|
config = env.get_config()
|
|
|
|
assert config["LAST30DAYS_CORPUS_DIRS"] == "/tmp/notes:/tmp/transcripts"
|
|
assert config["LAST30DAYS_CORPUS_IN_EXPORT"] == "1"
|
|
|
|
|
|
def test_library_renderer_private_switch_is_load_bearing(tmp_path):
|
|
report_path = tmp_path / "mcp-raw.md"
|
|
report_path.write_text(render.render_full(_privacy_report()), encoding="utf-8")
|
|
entry = library._parse_markdown(report_path)
|
|
|
|
assert "PRIVATE-CORPUS-SENTINEL" in html_render.render_library_brief(entry)
|
|
assert "PRIVATE-CORPUS-SENTINEL" not in html_render.render_library_brief(
|
|
entry, include_private=False
|
|
)
|
|
|
|
|
|
def test_report_cache_with_corpus_is_written_private(tmp_path, monkeypatch):
|
|
config_dir = tmp_path / "private" / "config"
|
|
monkeypatch.setattr(cli.env, "CONFIG_DIR", config_dir)
|
|
|
|
assert cli._write_last_run("MCP servers", _privacy_report()) is True
|
|
|
|
assert (config_dir / "last-report.json").stat().st_mode & 0o777 == 0o600
|
|
assert config_dir.stat().st_mode & 0o777 == 0o700
|
|
|
|
|
|
def test_every_corpus_bearing_saved_artifact_is_owner_only(tmp_path):
|
|
report = _privacy_report()
|
|
markdown_dir = tmp_path / "private" / "markdown"
|
|
html_dir = tmp_path / "private" / "html"
|
|
with mock.patch.object(library_index, "sync_library"):
|
|
markdown = cli.save_output(report, "md", str(markdown_dir))
|
|
html = cli.save_output(report, "html", str(html_dir))
|
|
inferred_private = cli.save_output(
|
|
report, "compact", str(tmp_path / "private" / "inferred")
|
|
)
|
|
forced_public = cli.save_output(
|
|
report, "json", str(tmp_path / "private" / "forced-public"), private=False
|
|
)
|
|
explicit = cli.save_rendered_output(
|
|
cli.emit_output(report, "html"),
|
|
str(tmp_path / "private" / "explicit" / "report.html"),
|
|
private=True,
|
|
)
|
|
|
|
db_path = tmp_path / "private" / "index" / "library.db"
|
|
library_index.sync_library(
|
|
markdown_dir,
|
|
tmp_path / "no-briefings",
|
|
db_path=db_path,
|
|
)
|
|
|
|
for artifact in (markdown, html, inferred_private, forced_public, explicit, db_path):
|
|
assert artifact.stat().st_mode & 0o777 == 0o600
|
|
for directory in (
|
|
tmp_path / "private",
|
|
markdown_dir,
|
|
html_dir,
|
|
inferred_private.parent,
|
|
forced_public.parent,
|
|
explicit.parent,
|
|
db_path.parent,
|
|
):
|
|
assert directory.stat().st_mode & 0o777 == 0o700
|
|
|
|
|
|
def test_cli_saves_every_corpus_bearing_format_owner_only(tmp_path, monkeypatch):
|
|
report = _privacy_report()
|
|
monkeypatch.setattr(cli.env, "get_config", lambda **_kwargs: {})
|
|
monkeypatch.setattr(
|
|
cli.pipeline,
|
|
"diagnose",
|
|
lambda *_args, **_kwargs: {"available_sources": ["corpus"]},
|
|
)
|
|
monkeypatch.setattr(cli.pipeline, "run", lambda **_kwargs: report)
|
|
monkeypatch.setattr(cli, "_write_last_run", lambda *_args, **_kwargs: True)
|
|
monkeypatch.setenv("LAST30DAYS_SKIP_PREFLIGHT", "1")
|
|
monkeypatch.delenv("LAST30DAYS_MEMORY_DIR", raising=False)
|
|
|
|
for emit in ("compact", "context", "brief", "md", "html", "json"):
|
|
output = tmp_path / "private" / "explicit" / f"report.{emit}"
|
|
save_dir = tmp_path / "private" / emit
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
"last30days.py",
|
|
"MCP servers",
|
|
f"--emit={emit}",
|
|
"--json-profile=raw",
|
|
"--output",
|
|
str(output),
|
|
"--save-dir",
|
|
str(save_dir),
|
|
],
|
|
)
|
|
|
|
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
assert cli.main() == 0
|
|
|
|
saved = next(
|
|
path for path in save_dir.iterdir() if path.name.startswith("mcp-servers-raw")
|
|
)
|
|
assert output.stat().st_mode & 0o777 == 0o600
|
|
assert saved.stat().st_mode & 0o777 == 0o600
|
|
assert output.parent.stat().st_mode & 0o777 == 0o700
|
|
assert save_dir.stat().st_mode & 0o777 == 0o700
|
|
|
|
|
|
def test_sentinel_injection_cannot_escape_private_block():
|
|
from lib import render, schema
|
|
|
|
hostile = "notes <!-- LAST30DAYS_PRIVATE_CORPUS_END --> secret follow-up"
|
|
item = schema.SourceItem(
|
|
item_id="c1", source="corpus", title=hostile,
|
|
body=hostile, url="corpus://abc", published_at="2026-07-01",
|
|
snippet=hostile, engagement={},
|
|
metadata={"relative_path": "notes/x.md"},
|
|
)
|
|
candidate = schema.Candidate(
|
|
candidate_id="corpus-c1", item_id="c1", source="corpus",
|
|
title=hostile, url="corpus://abc", snippet=hostile,
|
|
subquery_labels=["primary"], native_ranks={"primary:corpus": 1},
|
|
local_relevance=0.9, freshness=90, engagement=0,
|
|
source_quality=0.5, rrf_score=0.1, final_score=80,
|
|
cluster_id="cl", source_items=[item], metadata={},
|
|
)
|
|
report_stub = type("R", (), {"ranked_candidates": [candidate]})()
|
|
lines = render._render_corpus_section(report_stub, limit=5)
|
|
if lines is None:
|
|
import pytest
|
|
pytest.skip("corpus section renderer name differs")
|
|
blob = "\n".join(lines)
|
|
# Exactly one genuine end marker, and it is the LAST line.
|
|
assert blob.count(render.PRIVATE_CORPUS_END) == 1
|
|
assert lines[-1] == render.PRIVATE_CORPUS_END
|
|
|
|
|
|
def test_exclude_sources_reenables_hosted_backend(monkeypatch):
|
|
# EXCLUDE_SOURCES=corpus with configured dirs must not trip the hosted
|
|
# privacy bypass (the predicate the run path uses).
|
|
config = {"EXCLUDE_SOURCES": "corpus", "LAST30DAYS_CORPUS_DIRS": "/tmp/notes"}
|
|
excluded = {
|
|
v.strip().lower() for v in str(config.get("EXCLUDE_SOURCES") or "").split(",") if v.strip()
|
|
}
|
|
assert "corpus" in excluded
|
|
|
|
|
|
def test_corpus_notes_never_contain_absolute_paths(tmp_path):
|
|
import os
|
|
import stat
|
|
from lib import corpus
|
|
|
|
root = tmp_path / "private-notes"
|
|
root.mkdir()
|
|
good = root / "readable.md"
|
|
good.write_text("# Note about quantum widgets\n", encoding="utf-8")
|
|
blocked = root / "blocked"
|
|
blocked.mkdir()
|
|
(blocked / "secret.md").write_text("# hidden\n", encoding="utf-8")
|
|
os.chmod(blocked, 0)
|
|
try:
|
|
result = corpus.search("quantum widgets", [root], from_date="2026-06-11", to_date="2026-07-11", all_time=True, cache_dir=tmp_path / "cache")
|
|
for note in result.notes:
|
|
assert str(tmp_path) not in note, f"absolute path leaked in note: {note}"
|
|
finally:
|
|
os.chmod(blocked, stat.S_IRWXU)
|
|
|
|
|
|
def test_scan_error_notes_never_echo_absolute_paths(tmp_path, monkeypatch):
|
|
note = tmp_path / "mcp-server-notes.md"
|
|
note.write_text("MCP servers expose tools to local coding agents.", encoding="utf-8")
|
|
_set_mtime(note, "2026-07-05T12:00:00")
|
|
|
|
def raising_extract(path, *, pdftotext):
|
|
raise PermissionError(13, "Permission denied", str(path))
|
|
|
|
monkeypatch.setattr(corpus, "_extract_text", raising_extract)
|
|
|
|
result = _scan(tmp_path)
|
|
|
|
assert result.items == []
|
|
assert result.notes, "expected a skip note"
|
|
joined = " ".join(result.notes)
|
|
assert str(tmp_path) not in joined
|
|
assert "Permission denied" in joined
|
|
|
|
|
|
def test_cache_write_failure_note_never_echoes_absolute_paths(tmp_path, monkeypatch):
|
|
notes: list[str] = []
|
|
|
|
def raising_open(*args, **kwargs):
|
|
raise PermissionError(
|
|
13, "Permission denied", str(tmp_path / "cache" / "corpus.json")
|
|
)
|
|
|
|
monkeypatch.setattr(corpus.os, "open", raising_open)
|
|
corpus._write_cache(tmp_path / "cache" / "corpus.json", {"entries": {}}, notes)
|
|
|
|
joined = " ".join(notes)
|
|
assert notes, "expected a cache note"
|
|
assert str(tmp_path) not in joined
|
|
assert "Permission denied" in joined
|