830 lines
33 KiB
Python
830 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import unittest
|
|
from base64 import b64decode, urlsafe_b64encode
|
|
from dataclasses import dataclass
|
|
from typing import Any, cast
|
|
|
|
from services.runtime.ingestion_auth import ActorContext
|
|
from services.runtime.ingestion_files import FileIngestionError, FilePolicy, ProcessedSource
|
|
from services.runtime.ingestion_hub_client import HubClient
|
|
from services.runtime.ingestion_workflow import IngestionWorkflow
|
|
|
|
PROJECT_ID = "11111111-1111-4111-8111-111111111111"
|
|
PROPOSAL_ID = "22222222-2222-4222-8222-222222222222"
|
|
OTHER_PAGE_ID = "33333333-3333-4333-8333-333333333333"
|
|
PROJECT_PAGE_ID = "44444444-4444-4444-8444-444444444444"
|
|
|
|
|
|
def _git_blob_sha(text: str, *, sha256: bool = False) -> str:
|
|
raw = text.encode("utf-8")
|
|
blob = f"blob {len(raw)}\0".encode("ascii") + raw
|
|
return (
|
|
hashlib.sha256(blob).hexdigest()
|
|
if sha256
|
|
else hashlib.sha1(blob, usedforsecurity=False).hexdigest()
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _Reply:
|
|
data: dict[str, Any]
|
|
etag: str | None = None
|
|
status_code: int = 200
|
|
correlation_id: str = "trace-1"
|
|
|
|
@property
|
|
def id(self) -> str | None:
|
|
value = self.data.get("id")
|
|
return str(value) if value is not None else None
|
|
|
|
|
|
class _Hub:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, str, dict[str, Any]]] = []
|
|
self.uploads: list[tuple[str, Any, str | None, str, str | None]] = []
|
|
self.etag = '"v1"'
|
|
self.detail: dict[str, Any] = {
|
|
"id": PROPOSAL_ID,
|
|
"correlationId": "trace-1",
|
|
"status": "Received",
|
|
"manifest": json.dumps({"projectId": PROJECT_ID, "userManifest": "Quarterly runbook"}),
|
|
"destination": {
|
|
"pageId": None,
|
|
"parentPageId": None,
|
|
"title": "Quarterly restart guide",
|
|
},
|
|
"metadata": {"documentType": "Runbook", "documentStatus": "Draft"},
|
|
"analysis": None,
|
|
"reviewerComment": None,
|
|
"retryCount": 0,
|
|
"maxPublishRetries": 3,
|
|
"isRetryEligible": False,
|
|
"sources": [],
|
|
"citations": [],
|
|
}
|
|
|
|
async def request_json(self, method: str, path: str, **kwargs: Any) -> _Reply:
|
|
self.calls.append((method, path, kwargs))
|
|
if method == "POST" and path == "/api/v1/ingestion-proposals":
|
|
self.detail["manifest"] = kwargs["json_body"]["manifest"]
|
|
return _Reply({"id": PROPOSAL_ID, "status": "Received"}, etag='"v1"', status_code=201)
|
|
if method == "GET" and path in {
|
|
"/api/v1/pages/roots?limit=2",
|
|
"/api/v1/pages/roots?cursor=roots-next&limit=2",
|
|
}:
|
|
return _Reply(
|
|
{
|
|
"items": [
|
|
{
|
|
"id": PROJECT_PAGE_ID,
|
|
"title": "Course",
|
|
"slug": "course",
|
|
"documentType": "Course",
|
|
"documentStatus": "Published",
|
|
"hasChildren": True,
|
|
}
|
|
],
|
|
"nextCursor": "roots-next" if "cursor=" not in path else None,
|
|
}
|
|
)
|
|
if method == "GET" and path == f"/api/v1/pages/{PROJECT_PAGE_ID}/children?limit=2":
|
|
return _Reply(
|
|
{
|
|
"items": [
|
|
{
|
|
"id": OTHER_PAGE_ID,
|
|
"parentPageId": PROJECT_PAGE_ID,
|
|
"title": "Module",
|
|
"slug": "module",
|
|
"documentType": "Module",
|
|
"documentStatus": "Published",
|
|
"hasChildren": False,
|
|
}
|
|
],
|
|
"nextCursor": None,
|
|
}
|
|
)
|
|
if method == "GET" and path == f"/api/v1/pages/{PROJECT_PAGE_ID}":
|
|
return _Reply(
|
|
{
|
|
"id": PROJECT_PAGE_ID,
|
|
"title": "Course",
|
|
"slug": "course",
|
|
"markdown": "# Course\n" + ("A" * 40),
|
|
"documentType": "Course",
|
|
"documentStatus": "Published",
|
|
"parentPageId": None,
|
|
"breadcrumb": [{"id": PROJECT_PAGE_ID, "title": "Course"}],
|
|
},
|
|
etag='"page-v7"',
|
|
)
|
|
if method == "GET" and path == "/api/v1/search?q=architecture&limit=10":
|
|
return _Reply(
|
|
{
|
|
"items": [
|
|
{
|
|
"id": PROJECT_PAGE_ID,
|
|
"title": "Project guide",
|
|
"slug": "guide",
|
|
"snippet": "Allowed",
|
|
},
|
|
{
|
|
"id": OTHER_PAGE_ID,
|
|
"title": "Other project",
|
|
"slug": "other",
|
|
"snippet": "Must be filtered",
|
|
},
|
|
],
|
|
"nextCursor": None,
|
|
}
|
|
)
|
|
if (
|
|
method == "GET"
|
|
and path
|
|
== f"/api/v1/projects/{PROJECT_ID}/context?updates=10&decisions=10&documents=20"
|
|
):
|
|
return _Reply(
|
|
{
|
|
"project": {"id": PROJECT_ID, "name": "Project", "summary": "Scoped"},
|
|
"citations": [
|
|
{
|
|
"kind": "page",
|
|
"id": PROJECT_PAGE_ID,
|
|
"revision": 7,
|
|
"title": "Project guide",
|
|
"canonicalPath": f"/knowledge/{PROJECT_PAGE_ID}/guide",
|
|
}
|
|
],
|
|
"truncation": {"truncated": False, "omittedCounts": {}},
|
|
}
|
|
)
|
|
if method == "GET" and path == f"/api/v1/ingestion-proposals/{PROPOSAL_ID}":
|
|
return _Reply(dict(self.detail), etag=self.etag)
|
|
if method == "PATCH" and path == f"/api/v1/ingestion-proposals/{PROPOSAL_ID}":
|
|
self.detail["analysis"] = kwargs["json_body"]["analysis"]
|
|
self.detail["status"] = "ReadyForReview"
|
|
self.etag = '"v3"'
|
|
return _Reply(dict(self.detail), etag=self.etag)
|
|
raise AssertionError(f"unexpected Hub request: {method} {path}")
|
|
|
|
async def upload_source(self, proposal_id: str, source: Any, **kwargs: Any) -> _Reply:
|
|
self.uploads.append(
|
|
(
|
|
proposal_id,
|
|
source,
|
|
kwargs.get("etag"),
|
|
kwargs["idempotency_key"],
|
|
kwargs["project_id"],
|
|
)
|
|
)
|
|
self.detail["sources"].append(
|
|
{
|
|
"fileName": source.file_name,
|
|
"mimeType": source.content_type,
|
|
"byteSize": len(source.content),
|
|
"digestSha256": "a" * 64,
|
|
"scanResult": "Clean",
|
|
}
|
|
)
|
|
self.etag = f'"v{len(self.uploads) + 1}"'
|
|
return _Reply(dict(self.detail), etag=self.etag)
|
|
|
|
|
|
class _ActorAuthorizer:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[Any, str, str | None]] = []
|
|
|
|
def __call__(self, actor: ActorContext, *, capability: str, project: str | None = None) -> None:
|
|
self.calls.append((actor, capability, project))
|
|
if capability not in actor.scopes or (
|
|
project is not None and project not in actor.projects
|
|
):
|
|
raise PermissionError("actor not authorized")
|
|
|
|
|
|
class IngestionWorkflowTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.actor = ActorContext(
|
|
subject="user-1",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"knowledge:read", "ingestion:write"}),
|
|
correlation_id="trace-1",
|
|
)
|
|
self.hub = _Hub()
|
|
self.authorizer = _ActorAuthorizer()
|
|
self.processed = (
|
|
type(
|
|
"ProcessedSourceFixture",
|
|
(),
|
|
{
|
|
"file_id": "file_1",
|
|
"file_name": "runbook.txt",
|
|
"mime_type": "text/plain",
|
|
"sha256": "a" * 64,
|
|
"text": "restart and verify",
|
|
"byte_size": 20,
|
|
"content": b"restart and verify\n",
|
|
},
|
|
)(),
|
|
)
|
|
|
|
async def acquire(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
|
|
self.assertIs(actor, self.actor)
|
|
self.assertIsNotNone(policy)
|
|
self.assertEqual(files[0]["file_id"], "file_1")
|
|
return self.processed
|
|
|
|
self.workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
file_acquirer=acquire,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
|
|
async def scan_text_source(text: str, file_name: str, *, policy: Any) -> Any:
|
|
raw = text.encode("utf-8")
|
|
return ProcessedSource(
|
|
"repo-source",
|
|
file_name,
|
|
"text/plain",
|
|
hashlib.sha256(raw).hexdigest(),
|
|
text,
|
|
len(raw),
|
|
raw,
|
|
)
|
|
|
|
self.text_source_scanner = scan_text_source
|
|
|
|
async def test_analysis_upload_and_submit_use_canonical_scoped_review_flow(self) -> None:
|
|
files = [
|
|
{
|
|
"download_url": "https://files.openai.com/temporary/secret-url",
|
|
"file_id": "file_1",
|
|
"mime_type": "text/plain",
|
|
"file_name": "runbook.txt",
|
|
}
|
|
]
|
|
publication = {
|
|
"title": "Quarterly restart guide",
|
|
"markdown": "# Restart\nVerify service health.",
|
|
"document_type": "Runbook",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
"parent_page_id": None,
|
|
"target_page_id": None,
|
|
"expected_target_etag": None,
|
|
}
|
|
|
|
created = await self.workflow.analyze_knowledge_files(
|
|
project_id=PROJECT_ID,
|
|
files=files,
|
|
manifest="User requested a restart guide.",
|
|
publication=publication,
|
|
actor=self.actor,
|
|
)
|
|
completed = await self.workflow.submit_ingestion_proposal(
|
|
project_id=PROJECT_ID,
|
|
proposal_id=PROPOSAL_ID,
|
|
analysis="Source checked; this is a machine proposal for review.",
|
|
actor=self.actor,
|
|
)
|
|
|
|
self.assertEqual(created["proposalId"], PROPOSAL_ID)
|
|
self.assertEqual(created["status"], "Received")
|
|
self.assertEqual(len(self.hub.uploads), 1)
|
|
uploaded_proposal, source, upload_etag, _, upload_project = self.hub.uploads[0]
|
|
self.assertEqual(uploaded_proposal, PROPOSAL_ID)
|
|
self.assertEqual(source.content, b"restart and verify\n")
|
|
self.assertEqual(upload_etag, '"v1"')
|
|
self.assertEqual(upload_project, PROJECT_ID)
|
|
|
|
create_method, create_path, create_kwargs = self.hub.calls[0]
|
|
self.assertEqual((create_method, create_path), ("POST", "/api/v1/ingestion-proposals"))
|
|
self.assertEqual(create_kwargs["project_id"], PROJECT_ID)
|
|
manifest = json.loads(create_kwargs["json_body"]["manifest"])
|
|
self.assertEqual(manifest["projectId"], PROJECT_ID)
|
|
self.assertNotIn("download_url", create_kwargs["json_body"]["manifest"])
|
|
self.assertEqual(
|
|
create_kwargs["json_body"]["publication"]["mutationKind"],
|
|
"CreatePage",
|
|
)
|
|
|
|
patch = next(call for call in self.hub.calls if call[0] == "PATCH")
|
|
self.assertEqual(patch[1], f"/api/v1/ingestion-proposals/{PROPOSAL_ID}")
|
|
self.assertEqual(
|
|
patch[2]["json_body"],
|
|
{"analysis": "Source checked; this is a machine proposal for review."},
|
|
)
|
|
self.assertEqual(patch[2]["etag"], '"v2"')
|
|
self.assertEqual(patch[2]["project_id"], PROJECT_ID)
|
|
self.assertEqual(completed["status"], "ReadyForReview")
|
|
self.assertNotIn("restart and verify", json.dumps(created))
|
|
self.assertNotIn("secret-url", json.dumps(created))
|
|
self.assertFalse(
|
|
any(
|
|
path.endswith(("/submit", "/publish", "/approve", "/reject"))
|
|
for _, path, _ in self.hub.calls
|
|
)
|
|
)
|
|
|
|
async def test_search_filters_workspace_results_to_project_citations(self) -> None:
|
|
result = await self.workflow.search_knowledge(
|
|
project_id=PROJECT_ID,
|
|
query="architecture",
|
|
limit=10,
|
|
actor=self.actor,
|
|
)
|
|
self.assertEqual([item["id"] for item in result["results"]], [PROJECT_PAGE_ID])
|
|
self.assertEqual(result["results"][0]["citation"]["revision"], 7)
|
|
self.assertEqual(
|
|
[call[1] for call in self.hub.calls],
|
|
[
|
|
"/api/v1/search?q=architecture&limit=10",
|
|
f"/api/v1/projects/{PROJECT_ID}/context?updates=10&decisions=10&documents=20",
|
|
],
|
|
)
|
|
|
|
async def test_projectless_workspace_flow_is_opted_in_and_keeps_project_context_separate(
|
|
self,
|
|
) -> None:
|
|
trusted = ActorContext(
|
|
subject="trusted-client",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"knowledge:read", "ingestion:write"}),
|
|
correlation_id="trace-1",
|
|
connector_project="knowledge-hub",
|
|
connector_environment="prod",
|
|
)
|
|
|
|
async def acquire_workspace(
|
|
files: Any, *, actor: ActorContext, policy: Any
|
|
) -> tuple[Any, ...]:
|
|
self.assertIs(actor, trusted)
|
|
return self.processed
|
|
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
file_acquirer=acquire_workspace,
|
|
workspace_read_enabled=True,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
with self.assertRaises(PermissionError):
|
|
await self.workflow.search_knowledge(
|
|
project_id=None, query="architecture", actor=trusted
|
|
)
|
|
self.assertEqual(self.hub.calls, [])
|
|
search = await workflow.search_knowledge(
|
|
project_id=None, query="architecture", actor=trusted
|
|
)
|
|
self.assertEqual(
|
|
[item["id"] for item in search["results"]], [PROJECT_PAGE_ID, OTHER_PAGE_ID]
|
|
)
|
|
self.assertEqual(
|
|
[call[1] for call in self.hub.calls], ["/api/v1/search?q=architecture&limit=10"]
|
|
)
|
|
self.assertIsNone(search["projectId"])
|
|
|
|
files = [{"download_url": "https://files.openai.com/file", "file_id": "file_1"}]
|
|
extracted = await workflow.extract_knowledge_files(
|
|
project_id=None, files=files, actor=trusted
|
|
)
|
|
self.assertEqual(extracted["page"]["text"], "restart and verify")
|
|
self.assertEqual(len(self.hub.calls), 1)
|
|
created = await workflow.analyze_knowledge_files(
|
|
project_id=None,
|
|
files=files,
|
|
manifest="Review the course content.",
|
|
publication={
|
|
"title": "Course guide",
|
|
"markdown": "# Course guide",
|
|
"document_type": "Course",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
"parent_page_id": None,
|
|
},
|
|
actor=trusted,
|
|
)
|
|
self.assertEqual(created["proposalId"], PROPOSAL_ID)
|
|
manifest = json.loads(self.hub.detail["manifest"])
|
|
self.assertEqual(manifest["workspace"], "knowledge-hub")
|
|
self.assertNotIn("projectId", manifest)
|
|
self.assertIsNone(self.hub.uploads[0][-1])
|
|
submitted = await workflow.submit_ingestion_proposal(
|
|
project_id=None,
|
|
proposal_id=PROPOSAL_ID,
|
|
analysis="Ready for human review.",
|
|
actor=trusted,
|
|
)
|
|
status = await workflow.get_ingestion_status(
|
|
project_id=None, proposal_id=PROPOSAL_ID, actor=trusted
|
|
)
|
|
self.assertEqual(submitted["status"], "ReadyForReview")
|
|
self.assertEqual(status["status"], "ReadyForReview")
|
|
|
|
wrong = ActorContext(
|
|
subject="untrusted-client",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"knowledge:read", "ingestion:write"}),
|
|
correlation_id="trace-2",
|
|
)
|
|
calls_before = len(self.hub.calls)
|
|
with self.assertRaises(PermissionError):
|
|
await workflow.get_ingestion_status(
|
|
project_id=None, proposal_id=PROPOSAL_ID, actor=wrong
|
|
)
|
|
with self.assertRaises(PermissionError):
|
|
await workflow.search_knowledge(project_id=OTHER_PAGE_ID, query="course", actor=trusted)
|
|
self.assertEqual(len(self.hub.calls), calls_before)
|
|
|
|
async def test_project_scope_denial_happens_before_any_hub_request(self) -> None:
|
|
actor = ActorContext(
|
|
subject="user-2",
|
|
projects=frozenset(),
|
|
scopes=frozenset({"knowledge:read"}),
|
|
correlation_id="trace-2",
|
|
)
|
|
|
|
with self.assertRaises(PermissionError):
|
|
await self.workflow.get_knowledge_context(project_id=PROJECT_ID, actor=actor)
|
|
|
|
self.assertEqual(self.hub.calls, [])
|
|
|
|
async def test_rejected_file_creates_no_proposal_shell(self) -> None:
|
|
async def reject_files(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
|
|
raise ValueError("private download URL and extracted body must never be returned")
|
|
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
file_acquirer=reject_files,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
await workflow.analyze_knowledge_files(
|
|
project_id=PROJECT_ID,
|
|
files=[{"download_url": "https://invalid.test/file", "file_id": "file_1"}],
|
|
manifest="Fixture",
|
|
publication={
|
|
"title": "Guide",
|
|
"markdown": "# Guide",
|
|
"document_type": "Runbook",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
},
|
|
actor=self.actor,
|
|
)
|
|
self.assertEqual(self.hub.calls, [])
|
|
|
|
async def test_no_extractable_text_cannot_create_or_complete_proposal(self) -> None:
|
|
async def image_only(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
|
|
source = self.processed[0]
|
|
return (
|
|
ProcessedSource(
|
|
source.file_id,
|
|
"image-only.pdf",
|
|
"application/pdf",
|
|
source.sha256,
|
|
"",
|
|
source.byte_size,
|
|
source.content,
|
|
),
|
|
)
|
|
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
file_acquirer=image_only,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
files = [{"download_url": "https://files.openai.com/image-only.pdf", "file_id": "file_1"}]
|
|
with self.assertRaises(FileIngestionError):
|
|
await workflow.extract_knowledge_files(
|
|
project_id=PROJECT_ID, files=files, actor=self.actor
|
|
)
|
|
with self.assertRaises(FileIngestionError):
|
|
await workflow.analyze_knowledge_files(
|
|
project_id=PROJECT_ID,
|
|
files=files,
|
|
manifest="Image only",
|
|
publication={
|
|
"title": "Image",
|
|
"markdown": "# Image",
|
|
"document_type": "Course",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
"parent_page_id": None,
|
|
},
|
|
actor=self.actor,
|
|
)
|
|
self.assertEqual(self.hub.calls, [])
|
|
self.assertEqual(self.hub.uploads, [])
|
|
|
|
async def test_attachment_extraction_returns_real_text_with_per_file_coverage_and_no_proposal(
|
|
self,
|
|
) -> None:
|
|
text = "DISTINCTIVE-BEGIN\n" + ("line-knowledge-content\n" * 900) + "DISTINCTIVE-END"
|
|
processed = (
|
|
type(
|
|
"ProcessedSourceFixture",
|
|
(),
|
|
{
|
|
"file_id": "file_1",
|
|
"file_name": "guide.txt",
|
|
"mime_type": "text/plain",
|
|
"sha256": "b" * 64,
|
|
"text": text,
|
|
"byte_size": len(text.encode()),
|
|
"content": text.encode(),
|
|
},
|
|
)(),
|
|
)
|
|
|
|
async def acquire(files: Any, *, actor: ActorContext, policy: Any) -> tuple[Any, ...]:
|
|
return processed
|
|
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
file_acquirer=acquire,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
file_input = [{"download_url": "https://files.openai.com/file", "file_id": "file_1"}]
|
|
first = await workflow.extract_knowledge_files(
|
|
project_id=PROJECT_ID, files=file_input, actor=self.actor
|
|
)
|
|
|
|
self.assertIn("DISTINCTIVE-BEGIN", first["page"]["text"])
|
|
self.assertEqual(first["page"]["sha256"], "b" * 64)
|
|
self.assertEqual(first["page"]["startOffset"], 0)
|
|
self.assertFalse(first["page"]["complete"])
|
|
self.assertEqual(first["fileCoverage"][0]["coveredChars"], len(first["page"]["text"]))
|
|
self.assertIsNotNone(first["nextCursor"])
|
|
cursor = first["nextCursor"]
|
|
encoded = cursor[3:]
|
|
raw = b64decode(encoded + "=" * (-len(encoded) % 4), altchars=b"-_", validate=True)
|
|
payload = json.loads(raw[:-32])
|
|
payload["offset"] = len(text) - 1
|
|
forged = "v2:" + urlsafe_b64encode(
|
|
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + raw[-32:]
|
|
).decode().rstrip("=")
|
|
with self.assertRaises(ValueError):
|
|
await workflow.extract_knowledge_files(
|
|
project_id=PROJECT_ID, files=file_input, page_cursor=forged, actor=self.actor
|
|
)
|
|
other_actor = ActorContext(
|
|
subject="other-user",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"ingestion:write", "knowledge:read"}),
|
|
correlation_id="trace-other",
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
await workflow.extract_knowledge_files(
|
|
project_id=PROJECT_ID, files=file_input, page_cursor=cursor, actor=other_actor
|
|
)
|
|
second = await workflow.extract_knowledge_files(
|
|
project_id=PROJECT_ID,
|
|
files=file_input,
|
|
page_cursor=first["nextCursor"],
|
|
actor=self.actor,
|
|
)
|
|
|
|
self.assertIn("DISTINCTIVE-END", second["page"]["text"])
|
|
self.assertTrue(second["page"]["complete"])
|
|
self.assertTrue(second["allFilesComplete"])
|
|
self.assertIsNone(second["nextCursor"])
|
|
self.assertEqual(self.hub.calls, [])
|
|
self.assertEqual(self.hub.uploads, [])
|
|
|
|
async def test_workspace_read_is_default_off_and_checks_exact_trusted_identity(self) -> None:
|
|
with self.assertRaises(PermissionError):
|
|
await self.workflow.get_workspace_roots(actor=self.actor, limit=2)
|
|
self.assertEqual(self.hub.calls, [])
|
|
|
|
workspace_actor = ActorContext(
|
|
subject="trusted-client",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"knowledge:read", "ingestion:write"}),
|
|
correlation_id="trace-1",
|
|
connector_project="knowledge-hub",
|
|
connector_environment="prod",
|
|
)
|
|
enabled = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
workspace_read_enabled=True,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
roots = await enabled.get_workspace_roots(actor=workspace_actor, limit=2)
|
|
children = await enabled.get_workspace_children(
|
|
parent_page_id=PROJECT_PAGE_ID, actor=workspace_actor, limit=2
|
|
)
|
|
page = await enabled.get_workspace_page(
|
|
page_id=PROJECT_PAGE_ID, actor=workspace_actor, max_chars=10
|
|
)
|
|
|
|
self.assertEqual(roots["items"][0]["title"], "Course")
|
|
self.assertEqual(roots["nextCursor"], "roots-next")
|
|
self.assertEqual(children["items"][0]["parentPageId"], PROJECT_PAGE_ID)
|
|
self.assertEqual(page["etag"], '"page-v7"')
|
|
self.assertEqual(page["markdown"], "# Course\nA")
|
|
self.assertEqual(page["nextOffset"], 10)
|
|
self.assertFalse(page["complete"])
|
|
|
|
for marker in ("other-hub", "staging"):
|
|
wrong_actor = ActorContext(
|
|
subject="trusted-client",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"knowledge:read"}),
|
|
correlation_id="trace-1",
|
|
connector_project=marker,
|
|
connector_environment="prod" if marker == "other-hub" else "staging",
|
|
)
|
|
with self.subTest(marker=marker), self.assertRaises(PermissionError):
|
|
await enabled.get_workspace_roots(actor=wrong_actor, limit=2)
|
|
|
|
async def test_workspace_page_pagination_requires_a_stable_etag_and_update_uses_it(
|
|
self,
|
|
) -> None:
|
|
workspace_actor = ActorContext(
|
|
subject="trusted-client",
|
|
projects=frozenset({PROJECT_ID}),
|
|
scopes=frozenset({"knowledge:read", "ingestion:write"}),
|
|
correlation_id="trace-1",
|
|
connector_project="knowledge-hub",
|
|
connector_environment="prod",
|
|
)
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
workspace_read_enabled=True,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
await workflow.get_workspace_page(
|
|
page_id=PROJECT_PAGE_ID,
|
|
actor=workspace_actor,
|
|
offset_chars=10,
|
|
max_chars=10,
|
|
)
|
|
page = await workflow.get_workspace_page(
|
|
page_id=PROJECT_PAGE_ID,
|
|
actor=workspace_actor,
|
|
offset_chars=10,
|
|
max_chars=10,
|
|
expected_etag='"page-v7"',
|
|
)
|
|
self.assertEqual(page["markdown"], "AAAAAAAAAA")
|
|
self.assertEqual(page["etag"], '"page-v7"')
|
|
|
|
update = self.workflow._validate_publication(
|
|
{
|
|
"title": "Course",
|
|
"markdown": "# Revised course",
|
|
"document_type": "Course",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "UpdatePage",
|
|
"target_page_id": PROJECT_PAGE_ID,
|
|
"expected_target_etag": page["etag"],
|
|
}
|
|
)
|
|
self.assertEqual(update["expectedTargetETag"], page["etag"])
|
|
|
|
async def test_text_source_proposal_keeps_caller_provenance_and_full_publication(self) -> None:
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
text_source_scanner=self.text_source_scanner,
|
|
)
|
|
result = await workflow.create_text_source_proposal(
|
|
project_id=PROJECT_ID,
|
|
source_file_name="guide.md",
|
|
source_text="# Source guide\nVerified instructions.",
|
|
source_complete=True,
|
|
source_uri="itworx-git://ludarium/app/docs/guide.md",
|
|
source_ref="main",
|
|
source_commit="a" * 40,
|
|
source_sha=_git_blob_sha("# Source guide\nVerified instructions."),
|
|
manifest="Propose a reviewed guide.",
|
|
publication={
|
|
"title": "Reviewed guide",
|
|
"markdown": "# Reviewed guide\nCleaned and current.",
|
|
"document_type": "Runbook",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
"parent_page_id": None,
|
|
},
|
|
actor=self.actor,
|
|
)
|
|
|
|
self.assertEqual(result["proposalId"], PROPOSAL_ID)
|
|
create = next(call for call in self.hub.calls if call[0] == "POST")
|
|
body = create[2]["json_body"]
|
|
self.assertEqual(body["publication"]["title"], "Reviewed guide")
|
|
manifest = json.loads(body["manifest"])
|
|
self.assertEqual(
|
|
manifest["sourceProvenance"][0]["sourceUri"], "itworx-git://ludarium/app/docs/guide.md"
|
|
)
|
|
self.assertEqual(
|
|
manifest["sourceProvenance"][0]["sourceSha"],
|
|
_git_blob_sha("# Source guide\nVerified instructions."),
|
|
)
|
|
self.assertFalse(manifest["sourceProvenance"][0]["verified"])
|
|
self.assertEqual(self.hub.uploads[0][1].content, b"# Source guide\nVerified instructions.")
|
|
|
|
async def test_projectless_text_source_proposal_requires_trusted_workspace(self) -> None:
|
|
trusted = ActorContext(
|
|
subject="trusted-client",
|
|
projects=frozenset(),
|
|
scopes=frozenset({"knowledge:read", "ingestion:write"}),
|
|
correlation_id="trace-workspace",
|
|
connector_project="knowledge-hub",
|
|
connector_environment="prod",
|
|
)
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
text_source_scanner=self.text_source_scanner,
|
|
workspace_read_enabled=True,
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
)
|
|
result = await workflow.create_text_source_proposal(
|
|
project_id=None,
|
|
source_file_name="guide.md",
|
|
source_text="# Complete source",
|
|
source_complete=True,
|
|
source_uri="itworx-git://school/courses/docs/guide.md",
|
|
source_ref="main",
|
|
source_commit="a" * 40,
|
|
source_sha=_git_blob_sha("# Complete source", sha256=True),
|
|
manifest="Course source",
|
|
publication={
|
|
"title": "Guide",
|
|
"markdown": "# Guide",
|
|
"document_type": "Course",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
"parent_page_id": None,
|
|
},
|
|
actor=trusted,
|
|
)
|
|
self.assertEqual(result["proposalId"], PROPOSAL_ID)
|
|
manifest = json.loads(self.hub.detail["manifest"])
|
|
self.assertEqual(manifest["workspace"], "knowledge-hub")
|
|
self.assertNotIn("projectId", manifest)
|
|
self.assertFalse(manifest["sourceProvenance"][0]["verified"])
|
|
self.assertIsNone(self.hub.uploads[0][-1])
|
|
|
|
async def test_text_source_rejects_incomplete_or_invalid_provenance_before_proposal(
|
|
self,
|
|
) -> None:
|
|
workflow = IngestionWorkflow(
|
|
cast(HubClient, self.hub),
|
|
file_policy=cast(FilePolicy, object()),
|
|
actor_authorizer=self.authorizer,
|
|
cursor_signing_key=b"c" * 32,
|
|
text_source_scanner=self.text_source_scanner,
|
|
)
|
|
arguments = {
|
|
"project_id": PROJECT_ID,
|
|
"source_file_name": "guide.md",
|
|
"source_text": "# Source guide",
|
|
"source_complete": False,
|
|
"source_uri": "itworx-git://ludarium/app/docs/guide.md",
|
|
"source_ref": "main",
|
|
"source_commit": "a" * 40,
|
|
"source_sha": "b" * 40,
|
|
"manifest": "Fixture",
|
|
"publication": {
|
|
"title": "Guide",
|
|
"markdown": "# Guide",
|
|
"document_type": "Runbook",
|
|
"document_status": "Draft",
|
|
"mutation_kind": "CreatePage",
|
|
"parent_page_id": None,
|
|
},
|
|
"actor": self.actor,
|
|
}
|
|
with self.assertRaises(ValueError):
|
|
await workflow.create_text_source_proposal(**arguments)
|
|
self.assertEqual(self.hub.calls, [])
|
|
arguments["source_complete"] = True
|
|
with self.assertRaisesRegex(ValueError, "Git blob SHA"):
|
|
await workflow.create_text_source_proposal(**arguments)
|
|
self.assertEqual(self.hub.calls, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|