122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from govoplan_core.core.records import (
|
|
RecordArchiveTransferRequest,
|
|
RecordTransferPackage,
|
|
)
|
|
from govoplan_core.security.http_fetch import HttpFetchResponse
|
|
from govoplan_dms.backend.dvelop_d3 import (
|
|
DVELOP_D3_ARCHIVE_PROFILE,
|
|
DvelopD3Client,
|
|
DvelopD3Profile,
|
|
UnconfiguredDvelopD3ArchiveProvider,
|
|
)
|
|
from govoplan_dms.backend.manifest import get_manifest
|
|
|
|
|
|
def _profile() -> DvelopD3Profile:
|
|
return DvelopD3Profile(
|
|
api_base_url="https://d3.example.test/api",
|
|
repository_id="repo-1",
|
|
origin="https://govoplan.example.test",
|
|
source_category="govoplan-record",
|
|
source_id="tenant-a",
|
|
mapping_revision="mapping-7",
|
|
credential_ref="core-credential:d3-tenant-a",
|
|
)
|
|
|
|
|
|
def test_preflight_is_digest_bound_and_does_not_claim_dispatch_readiness() -> None:
|
|
calls: list[tuple[str, MappingLike]] = []
|
|
|
|
def transport(url, *, method, headers, body):
|
|
calls.append((url, dict(headers)))
|
|
assert method == "GET"
|
|
assert body is None
|
|
return HttpFetchResponse(
|
|
status=200,
|
|
headers={"Content-Type": "application/hal+json"},
|
|
body=json.dumps({"_links": {"self": {"href": url}}}).encode(),
|
|
)
|
|
|
|
result = DvelopD3Client(_profile(), transport=transport).preflight()
|
|
|
|
assert result.ready_for_mapping_test is True
|
|
assert result.dispatch_ready is False
|
|
assert len(calls) == 3
|
|
assert all(headers["Origin"] == "https://govoplan.example.test" for _, headers in calls)
|
|
assert all("Authorization" not in headers for _, headers in calls)
|
|
|
|
|
|
def test_store_plan_binds_exact_record_package_without_effect() -> None:
|
|
package = RecordTransferPackage(
|
|
tenant_id="tenant-a",
|
|
package_id="package-1",
|
|
record_id="record-1",
|
|
record_revision=3,
|
|
profile=DVELOP_D3_ARCHIVE_PROFILE,
|
|
manifest_sha256="a" * 64,
|
|
manifest={"version": 1},
|
|
)
|
|
request = RecordArchiveTransferRequest(
|
|
package=package,
|
|
purpose="approved transfer",
|
|
idempotency_key="transfer-1",
|
|
)
|
|
|
|
plan = DvelopD3Client(_profile(), transport=lambda *args, **kwargs: None).build_record_store_plan(
|
|
request,
|
|
content_location_uri="https://files.example.test/content/package-1",
|
|
)
|
|
|
|
assert plan.body["sourceProperties"]["govoplanManifestSha256"] == "a" * 64
|
|
assert plan.body["sourceProperties"]["govoplanMappingRevision"] == "mapping-7"
|
|
canonical = json.dumps(plan.body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
|
|
assert plan.body_sha256 == hashlib.sha256(canonical).hexdigest()
|
|
|
|
|
|
def test_unconfigured_provider_fails_closed() -> None:
|
|
provider = UnconfiguredDvelopD3ArchiveProvider()
|
|
assert provider.state().healthy is False
|
|
with pytest.raises(Exception, match="dispatch is disabled"):
|
|
provider.dispatch(None, None, request=None) # type: ignore[arg-type]
|
|
|
|
|
|
def test_profile_rejects_origin_paths_and_embedded_credentials() -> None:
|
|
with pytest.raises(ValueError, match="Origin"):
|
|
DvelopD3Profile(
|
|
api_base_url="https://d3.example.test/api",
|
|
repository_id="repo",
|
|
origin="https://govoplan.example.test/path",
|
|
source_category="category",
|
|
source_id="source",
|
|
mapping_revision="one",
|
|
credential_ref="credential",
|
|
)
|
|
with pytest.raises(ValueError, match="credentials"):
|
|
DvelopD3Profile(
|
|
api_base_url="https://user:secret@d3.example.test/api",
|
|
repository_id="repo",
|
|
origin="https://govoplan.example.test",
|
|
source_category="category",
|
|
source_id="source",
|
|
mapping_revision="one",
|
|
credential_ref="credential",
|
|
)
|
|
|
|
|
|
def test_manifest_declares_d3_without_claiming_target_test() -> None:
|
|
manifest = get_manifest()
|
|
assert manifest.version == "0.1.19"
|
|
assert manifest.external_providers[0].id == "dms.dvelop_d3"
|
|
assert manifest.architecture is not None
|
|
assert manifest.architecture.target_tested_providers == ()
|
|
|
|
|
|
MappingLike = dict[str, str]
|