613 lines
25 KiB
Python
613 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from collections.abc import Mapping
|
|
from contextlib import closing
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from mcp.shared.memory import create_connected_server_and_client_session
|
|
|
|
from docforge.adapter_contract import (
|
|
AdapterEdge,
|
|
AdapterManifest,
|
|
AdapterNode,
|
|
AdapterProject,
|
|
AdapterProjection,
|
|
AdapterProjectSettings,
|
|
AdapterSource,
|
|
AdapterSourceProjection,
|
|
ShadowArtifact,
|
|
compare_artifacts,
|
|
validate_projection,
|
|
)
|
|
from docforge.errors import DocForgeError
|
|
from docforge.index import ProjectIndex
|
|
from docforge.mcp_server import (
|
|
ALL_TOOLS,
|
|
READ_TOOLS,
|
|
create_project_server,
|
|
create_read_only_server,
|
|
)
|
|
from docforge.models import (
|
|
Edge,
|
|
LogicEdge,
|
|
LogicNode,
|
|
LogicProjection,
|
|
Node,
|
|
ProjectSnapshot,
|
|
ProposalWriter,
|
|
RenderConfig,
|
|
RenderView,
|
|
)
|
|
from docforge.visualization import VisualizationIndexSnapshot
|
|
|
|
|
|
class Loader:
|
|
def __init__(self, projection: AdapterProjection) -> None:
|
|
self.projection = projection
|
|
|
|
def load_projection(self) -> AdapterProjection:
|
|
return self.projection
|
|
|
|
|
|
class IncrementalLoader:
|
|
def __init__(self, root: Path) -> None:
|
|
self.root = root
|
|
self.sources = {
|
|
"guide.foundation": "Foundation content.",
|
|
"guide.workflow": "Workflow content.",
|
|
}
|
|
self.source_paths = {
|
|
"guide.foundation": "docs/foundation.md",
|
|
"guide.workflow": "docs/workflow.md",
|
|
}
|
|
self.workflow_dependencies = ("guide.foundation",)
|
|
self.extract_calls: list[str] = []
|
|
self.fail_source: str | None = None
|
|
|
|
@staticmethod
|
|
def _hash(value: str) -> str:
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
def load_manifest(self) -> AdapterManifest:
|
|
source_items = tuple(
|
|
AdapterSource(
|
|
source_id=source_id,
|
|
source_path=self.source_paths[source_id],
|
|
fingerprint=self._hash(content),
|
|
extractor_version="python-ast@1",
|
|
dependencies=(self.workflow_dependencies if source_id == "guide.workflow" else ()),
|
|
)
|
|
for source_id, content in sorted(self.sources.items())
|
|
)
|
|
digest = hashlib.sha256()
|
|
for source in source_items:
|
|
digest.update(source.source_id.encode())
|
|
digest.update(source.source_path.encode())
|
|
digest.update(source.fingerprint.encode())
|
|
digest.update(source.extractor_version.encode())
|
|
for dependency in source.dependencies:
|
|
digest.update(dependency.encode())
|
|
return AdapterManifest(
|
|
project_id="incremental-fixture",
|
|
title="Incremental fixture",
|
|
adapter_id="fixture-incremental",
|
|
adapter_version="1",
|
|
root=self.root,
|
|
revision=digest.hexdigest()[:12],
|
|
source_hash=digest.hexdigest(),
|
|
families=("guide",),
|
|
allowed_relations=("depends_on",),
|
|
sources=source_items,
|
|
estimated_nodes=10,
|
|
)
|
|
|
|
def extract_source(self, source: AdapterSource) -> AdapterSourceProjection:
|
|
self.extract_calls.append(source.source_id)
|
|
if self.fail_source == source.source_id:
|
|
raise RuntimeError("intentional extraction failure")
|
|
content = self.sources[source.source_id]
|
|
node = Node(
|
|
node_id=source.source_id,
|
|
title=source.source_id.rsplit(".", 1)[-1].title(),
|
|
family="guide",
|
|
authority="authoritative",
|
|
status="active",
|
|
tags=("guide",),
|
|
summary=f"{source.source_id} summary.",
|
|
content=content,
|
|
source_path=source.source_path,
|
|
source_anchor=None,
|
|
content_hash=self._hash(content),
|
|
)
|
|
edges = (
|
|
(AdapterEdge(Edge("guide.workflow", "depends_on", "guide.foundation")),)
|
|
if source.source_id == "guide.workflow"
|
|
else ()
|
|
)
|
|
logic = (
|
|
(
|
|
LogicProjection(
|
|
owner_node_id="guide.workflow",
|
|
source_id="guide.workflow",
|
|
nodes=(
|
|
LogicNode("entry", "entry", "Entry", None),
|
|
LogicNode("return", "return", "Return", None),
|
|
),
|
|
edges=(LogicEdge("entry", "return", "return", "RETURN", 0),),
|
|
),
|
|
)
|
|
if source.source_id == "guide.workflow"
|
|
else ()
|
|
)
|
|
return AdapterSourceProjection(
|
|
source_id=source.source_id,
|
|
fingerprint=source.fingerprint,
|
|
nodes=(AdapterNode(node),),
|
|
edges=edges,
|
|
logic=logic,
|
|
)
|
|
|
|
def load_projection(self) -> AdapterProjection:
|
|
manifest = self.load_manifest()
|
|
contributions = tuple(self.extract_source(source) for source in manifest.sources)
|
|
return AdapterProjection(
|
|
project_id=manifest.project_id,
|
|
title=manifest.title,
|
|
adapter_id=manifest.adapter_id,
|
|
adapter_version=manifest.adapter_version,
|
|
root=manifest.root,
|
|
revision=manifest.revision,
|
|
source_hash=manifest.source_hash,
|
|
nodes=tuple(
|
|
sorted(
|
|
(node for contribution in contributions for node in contribution.nodes),
|
|
key=lambda item: item.node.node_id,
|
|
)
|
|
),
|
|
edges=tuple(
|
|
sorted(
|
|
(edge for contribution in contributions for edge in contribution.edges),
|
|
key=lambda item: (
|
|
item.edge.source_id,
|
|
item.edge.relation,
|
|
item.edge.target_id,
|
|
),
|
|
)
|
|
),
|
|
)
|
|
|
|
|
|
class AdapterContractTests(unittest.TestCase):
|
|
def projection(self, root: Path) -> AdapterProjection:
|
|
foundation = Node(
|
|
node_id="guide.foundation",
|
|
title="Foundation",
|
|
family="guide",
|
|
authority="authoritative",
|
|
status="active",
|
|
tags=("guide",),
|
|
summary="The base contract.",
|
|
content="Foundation content.",
|
|
source_path="docs/foundation.md",
|
|
source_anchor=None,
|
|
content_hash="1" * 64,
|
|
)
|
|
workflow = Node(
|
|
node_id="guide.workflow",
|
|
title="Workflow",
|
|
family="guide",
|
|
authority="approved_plan",
|
|
status="planned",
|
|
tags=("guide", "workflow"),
|
|
summary="The editing workflow.",
|
|
content="Workflow content.",
|
|
source_path="docs/workflow.md",
|
|
source_anchor=None,
|
|
content_hash="2" * 64,
|
|
)
|
|
return AdapterProjection(
|
|
project_id="adapter-fixture",
|
|
title="Adapter fixture",
|
|
adapter_id="fixture-shadow",
|
|
adapter_version="1",
|
|
root=root,
|
|
revision="fixture-revision",
|
|
source_hash="3" * 64,
|
|
nodes=(
|
|
AdapterNode(foundation, (("acceptance", "proven"),)),
|
|
AdapterNode(workflow, (("acceptance", "pending"),)),
|
|
),
|
|
edges=(
|
|
AdapterEdge(
|
|
Edge("guide.workflow", "depends_on", "guide.foundation"),
|
|
(("source", "docs/workflow.md"),),
|
|
),
|
|
),
|
|
)
|
|
|
|
def test_adapter_projection_builds_and_checks_through_standard_index(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
projection = self.projection(root)
|
|
project = AdapterProject(Loader(projection), cache_root=root / ".cache" / "shadow")
|
|
index = ProjectIndex(project)
|
|
|
|
built = index.build()
|
|
checked = index.check()
|
|
|
|
self.assertEqual("fixture-shadow@1", built["adapter"])
|
|
self.assertEqual(2, checked["node_count"])
|
|
self.assertEqual("Workflow", index.get_node("guide.workflow")["node"]["title"])
|
|
self.assertEqual(projection.identity(), projection.identity())
|
|
|
|
def test_projection_rejects_unsorted_metadata_graph_and_identity_changes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
projection = self.projection(root)
|
|
invalid_metadata = replace(
|
|
projection,
|
|
nodes=(
|
|
replace(
|
|
projection.nodes[0],
|
|
metadata=(("z", "last"), ("a", "first")),
|
|
),
|
|
projection.nodes[1],
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "metadata keys"):
|
|
validate_projection(invalid_metadata)
|
|
|
|
invalid_source = replace(
|
|
projection,
|
|
nodes=(
|
|
replace(
|
|
projection.nodes[0],
|
|
node=replace(projection.nodes[0].node, source_path="../outside.md"),
|
|
),
|
|
projection.nodes[1],
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "source path"):
|
|
validate_projection(invalid_source)
|
|
|
|
broken = replace(
|
|
projection,
|
|
edges=(AdapterEdge(Edge("guide.workflow", "depends_on", "missing.node")),),
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "missing nodes"):
|
|
validate_projection(broken)
|
|
|
|
loader = Loader(projection)
|
|
project = AdapterProject(loader, cache_root=root / ".cache" / "shadow")
|
|
loader.projection = replace(projection, adapter_version="2")
|
|
with self.assertRaisesRegex(DocForgeError, "identity changed"):
|
|
project.load()
|
|
|
|
def test_adapter_cache_must_remain_inside_project(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
outside = root.parent / "outside-adapter-cache"
|
|
with self.assertRaisesRegex(DocForgeError, "confined"):
|
|
AdapterProject(Loader(self.projection(root)), cache_root=outside)
|
|
|
|
def test_incremental_adapter_reuses_sources_and_invalidates_reverse_dependencies(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
loader = IncrementalLoader(root)
|
|
project = AdapterProject(loader, cache_root=root / ".cache" / "incremental")
|
|
index = ProjectIndex(project)
|
|
|
|
first = index.build()
|
|
self.assertEqual(2, first["build"]["reparsed_sources"])
|
|
self.assertEqual(0, first["build"]["cache_hits"])
|
|
self.assertEqual(1, first["logic_projection_count"])
|
|
logic = index.get_logic("guide.workflow")
|
|
self.assertTrue(logic["available"])
|
|
self.assertEqual("guide.workflow", logic["projection"]["owner_node_id"])
|
|
self.assertEqual(2, len(logic["projection"]["nodes"]))
|
|
self.assertFalse(index.get_logic("guide.foundation")["available"])
|
|
visual_logic = VisualizationIndexSnapshot(index, index.check()).logic("guide.workflow")
|
|
self.assertTrue(visual_logic["available"])
|
|
self.assertEqual("entry", visual_logic["root"])
|
|
self.assertEqual("return", visual_logic["edges"][0]["relation"])
|
|
loader.extract_calls.clear()
|
|
cache_path = root / ".cache" / "incremental" / "extractions.json"
|
|
cache_modified = cache_path.stat().st_mtime_ns
|
|
|
|
second = index.build()
|
|
self.assertEqual([], loader.extract_calls)
|
|
self.assertEqual(0, second["build"]["reparsed_sources"])
|
|
self.assertEqual(2, second["build"]["cache_hits"])
|
|
self.assertEqual(cache_modified, cache_path.stat().st_mtime_ns)
|
|
|
|
loader.sources["guide.foundation"] = "Changed foundation."
|
|
loader.extract_calls.clear()
|
|
with self.assertRaisesRegex(DocForgeError, "does not match"):
|
|
index.check()
|
|
self.assertEqual([], loader.extract_calls)
|
|
changed = index.build()
|
|
self.assertEqual(
|
|
["guide.foundation", "guide.workflow"],
|
|
loader.extract_calls,
|
|
)
|
|
self.assertEqual(2, changed["build"]["invalidated_sources"])
|
|
self.assertEqual(
|
|
"Changed foundation.",
|
|
index.get_node("guide.foundation")["node"]["content"],
|
|
)
|
|
|
|
loader.source_paths["guide.workflow"] = "docs/workflow-renamed.md"
|
|
loader.workflow_dependencies = ()
|
|
loader.extract_calls.clear()
|
|
renamed = index.build()
|
|
self.assertEqual(["guide.workflow"], loader.extract_calls)
|
|
self.assertEqual(1, renamed["build"]["invalidated_sources"])
|
|
self.assertEqual(
|
|
"docs/workflow-renamed.md",
|
|
index.get_node("guide.workflow")["node"]["source_path"],
|
|
)
|
|
|
|
def test_fast_incremental_reads_reverify_a_changed_index_file(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
loader = IncrementalLoader(root)
|
|
project = AdapterProject(loader, cache_root=root / ".cache" / "incremental")
|
|
index = ProjectIndex(project)
|
|
index.build()
|
|
index.synchronize()
|
|
self.assertTrue(index.attestation_path.is_file())
|
|
fresh = ProjectIndex(project)
|
|
self.assertEqual(
|
|
"current",
|
|
fresh.synchronize()["synchronization"]["action"],
|
|
)
|
|
|
|
with closing(sqlite3.connect(index.path)) as connection:
|
|
connection.execute(
|
|
"UPDATE nodes SET content = ? WHERE node_id = ?",
|
|
("tampered", "guide.foundation"),
|
|
)
|
|
connection.commit()
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "rows do not match metadata"):
|
|
index.get_node("guide.foundation")
|
|
|
|
repaired = index.synchronize()
|
|
self.assertEqual("rebuilt", repaired["synchronization"]["action"])
|
|
self.assertEqual(
|
|
"Foundation content.",
|
|
index.get_node("guide.foundation")["node"]["content"],
|
|
)
|
|
|
|
def test_incremental_delete_failure_and_equivalence_are_safe(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
loader = IncrementalLoader(root)
|
|
project = AdapterProject(loader, cache_root=root / ".cache" / "incremental")
|
|
index = ProjectIndex(project)
|
|
index.build()
|
|
(root / ".cache" / "incremental" / "extractions.json").write_text(
|
|
"{broken", encoding="utf-8"
|
|
)
|
|
loader.extract_calls.clear()
|
|
cache_miss = index.build()
|
|
self.assertEqual(
|
|
["guide.foundation", "guide.workflow"],
|
|
loader.extract_calls,
|
|
)
|
|
self.assertEqual(2, cache_miss["build"]["reparsed_sources"])
|
|
original_index = index.path.read_bytes()
|
|
cache_path = root / ".cache" / "incremental" / "extractions.json"
|
|
original_cache = cache_path.read_bytes()
|
|
|
|
loader.sources["guide.foundation"] = "Broken extraction."
|
|
loader.fail_source = "guide.foundation"
|
|
with self.assertRaisesRegex(RuntimeError, "intentional"):
|
|
index.build()
|
|
self.assertEqual(original_index, index.path.read_bytes())
|
|
self.assertEqual(original_cache, cache_path.read_bytes())
|
|
|
|
loader.fail_source = None
|
|
index.build()
|
|
loader.sources.pop("guide.workflow")
|
|
deleted = index.build()
|
|
self.assertEqual(1, deleted["build"]["deleted_sources"])
|
|
self.assertEqual(1, deleted["node_count"])
|
|
self.assertEqual(0, deleted["edge_count"])
|
|
self.assertIsNone(project.logic_projection("guide.workflow"))
|
|
|
|
loader.extract_calls.clear()
|
|
equivalent = project.verify_incremental_equivalence()
|
|
self.assertEqual("ok", equivalent["status"])
|
|
self.assertEqual(1, equivalent["node_count"])
|
|
|
|
def test_artifact_comparison_is_complete_and_byte_exact(self) -> None:
|
|
reference = (
|
|
ShadowArtifact("manual", b"same"),
|
|
ShadowArtifact("timeline", b"old"),
|
|
)
|
|
exact = compare_artifacts(reference, reference)
|
|
self.assertEqual("ok", exact["status"])
|
|
self.assertEqual(2, exact["count"])
|
|
|
|
mismatch = compare_artifacts(
|
|
reference,
|
|
(
|
|
ShadowArtifact("timeline", b"new"),
|
|
ShadowArtifact("extra", b"extra"),
|
|
),
|
|
)
|
|
self.assertEqual("mismatch", mismatch["status"])
|
|
self.assertEqual(["manual"], mismatch["missing"])
|
|
self.assertEqual(["extra"], mismatch["unexpected"])
|
|
self.assertEqual(["timeline"], mismatch["changed"])
|
|
|
|
|
|
class AdapterReadOnlyMcpTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_adapter_project_exposes_only_read_tools_and_custom_context(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
fixture = AdapterContractTests()
|
|
project = AdapterProject(
|
|
Loader(fixture.projection(root)),
|
|
cache_root=root / ".cache" / "adapter-read-only",
|
|
)
|
|
index = ProjectIndex(project)
|
|
index.build()
|
|
calls: list[tuple[str, int | None]] = []
|
|
|
|
def context_provider(
|
|
current: ProjectIndex, profile: str, budget: int | None
|
|
) -> dict[str, object]:
|
|
checked = current.check()
|
|
calls.append((profile, budget))
|
|
return {
|
|
"status": "ok",
|
|
"project_id": checked["project_id"],
|
|
"project_root_fingerprint": checked["project_root_fingerprint"],
|
|
"revision": checked["revision"],
|
|
"source_hash": checked["source_hash"],
|
|
"adapter": checked["adapter"],
|
|
"profile": profile,
|
|
"budget": budget,
|
|
"estimated_tokens": 1,
|
|
"entries": [],
|
|
"omissions": [],
|
|
}
|
|
|
|
server = create_read_only_server(project, context_provider=context_provider)
|
|
async with create_connected_server_and_client_session(
|
|
server, raise_exceptions=True
|
|
) as session:
|
|
tools = await session.list_tools()
|
|
info = await session.call_tool("docforge_project_info", {})
|
|
contract = await session.call_tool("docforge_get_contract", {})
|
|
context = await session.call_tool(
|
|
"docforge_get_context", {"profile": "fixture", "budget": 321}
|
|
)
|
|
|
|
self.assertEqual(READ_TOOLS, tuple(tool.name for tool in tools.tools))
|
|
self.assertEqual("adapter-fixture", info.structuredContent["project_id"])
|
|
self.assertEqual(list(READ_TOOLS), contract.structuredContent["allowed_tools"])
|
|
self.assertIn(
|
|
"isolated_changeset_writes",
|
|
contract.structuredContent["excluded_operations"],
|
|
)
|
|
self.assertFalse(contract.structuredContent["proposal_access"]["enabled"])
|
|
self.assertFalse(contract.structuredContent["isolated_changeset_writes_allowed"])
|
|
self.assertEqual("fixture", context.structuredContent["profile"])
|
|
self.assertEqual([("fixture", 321)], calls)
|
|
self.assertFalse(project.descriptor.changeset_root.exists())
|
|
|
|
async def test_adapter_project_proposals_require_explicit_policy_and_stay_isolated(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
(root / ".docforge").mkdir()
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
descriptor.write_text("adapter fixture\n", encoding="utf-8")
|
|
content_root = root / "docs"
|
|
content_root.mkdir()
|
|
sources = (content_root / "foundation.md", content_root / "workflow.md")
|
|
for source in sources:
|
|
source.write_text(f"canonical {source.stem}\n", encoding="utf-8")
|
|
template_root = root / "templates"
|
|
template_root.mkdir()
|
|
template = template_root / "review.html"
|
|
template.write_text(
|
|
"<html><head><title>{{ docforge_title }}</title></head>"
|
|
"<body>{{ docforge_content }}</body></html>\n",
|
|
encoding="utf-8",
|
|
)
|
|
validation_calls: list[tuple[Mapping[str, object], ...]] = []
|
|
|
|
def validate_proposal(
|
|
base: ProjectSnapshot,
|
|
projected: ProjectSnapshot,
|
|
operations: tuple[Mapping[str, object], ...],
|
|
) -> None:
|
|
self.assertEqual(base.edges, projected.edges)
|
|
validation_calls.append(operations)
|
|
|
|
settings = AdapterProjectSettings(
|
|
descriptor_path=descriptor,
|
|
content_roots=(content_root,),
|
|
canonical_sources=sources,
|
|
changeset_root=root / ".cache" / "changesets",
|
|
proposal_writers=(ProposalWriter("fixture-writer", ("guide",), ("update",)),),
|
|
render=RenderConfig(
|
|
template_root=template_root,
|
|
preview_root=root / ".cache" / "previews",
|
|
views=(
|
|
RenderView(
|
|
"review",
|
|
"generic_html",
|
|
template,
|
|
root / ".cache" / "rendered" / "review.html",
|
|
"Adapter review",
|
|
("guide",),
|
|
),
|
|
),
|
|
),
|
|
proposal_validator=validate_proposal,
|
|
)
|
|
project = AdapterProject(
|
|
Loader(AdapterContractTests().projection(root)),
|
|
cache_root=root / ".cache" / "adapter",
|
|
settings=settings,
|
|
)
|
|
ProjectIndex(project).build()
|
|
before = {source: source.read_bytes() for source in sources}
|
|
server = create_project_server(project, proposal_writer="fixture-writer")
|
|
async with create_connected_server_and_client_session(
|
|
server, raise_exceptions=True
|
|
) as session:
|
|
tools = await session.list_tools()
|
|
created = await session.call_tool(
|
|
"docforge_create_changeset", {"changeset_id": "adapter-update"}
|
|
)
|
|
node = await session.call_tool("docforge_get_node", {"node_id": "guide.workflow"})
|
|
proposed = await session.call_tool(
|
|
"docforge_propose_node_update",
|
|
{
|
|
"changeset_id": "adapter-update",
|
|
"expected_changeset_hash": created.structuredContent["changeset_hash"],
|
|
"node_id": "guide.workflow",
|
|
"expected_content_hash": node.structuredContent["node"]["content_hash"],
|
|
"metadata": None,
|
|
"content": "Updated workflow content.",
|
|
"relationship_changes": [],
|
|
"rationale": "Prove adapter proposal policy.",
|
|
},
|
|
)
|
|
validated = await session.call_tool(
|
|
"docforge_validate_changeset",
|
|
{"changeset_id": "adapter-update"},
|
|
)
|
|
diff = await session.call_tool(
|
|
"docforge_get_changeset_diff",
|
|
{"changeset_id": "adapter-update"},
|
|
)
|
|
preview = await session.call_tool(
|
|
"docforge_preview_changeset",
|
|
{"changeset_id": "adapter-update", "view_id": "review"},
|
|
)
|
|
|
|
self.assertEqual(ALL_TOOLS, tuple(tool.name for tool in tools.tools))
|
|
self.assertEqual("ok", proposed.structuredContent["status"])
|
|
self.assertTrue(validated.structuredContent["valid"])
|
|
self.assertEqual("update", diff.structuredContent["changes"][0]["operation"])
|
|
preview_path = root / preview.structuredContent["preview"]["path"]
|
|
self.assertTrue(preview_path.is_file())
|
|
self.assertTrue(validation_calls)
|
|
self.assertEqual(before, {source: source.read_bytes() for source in sources})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|