184 lines
6.1 KiB
Python
184 lines
6.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
from datetime import UTC, datetime
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from govoplan_fit_connect.backend.inbound import (
|
|
FIT_CONNECT_ACCEPT_EVENT,
|
|
FIT_CONNECT_REJECT_EVENT,
|
|
FitConnectAttachmentEvidence,
|
|
FitConnectInboundError,
|
|
FitConnectProfile,
|
|
FitConnectSubmission,
|
|
FitConnectSubmissionEvidence,
|
|
FitConnectVerification,
|
|
build_acknowledgement_plan,
|
|
create_ingress_receipt,
|
|
)
|
|
from govoplan_fit_connect.backend.manifest import get_manifest
|
|
|
|
|
|
DESTINATION_ID = "736c4581-da80-4710-9384-d19ebe1ff2bc"
|
|
SUBMISSION_ID = "c82585d8-e49d-4e27-abd6-be3dbf61a76f"
|
|
|
|
|
|
def _profile() -> FitConnectProfile:
|
|
return FitConnectProfile(
|
|
profile_id="fit-connect-test-binding",
|
|
destination_id=DESTINATION_ID,
|
|
submission_api_version="configured-exact-version",
|
|
metadata_schema_version="configured-exact-version",
|
|
profile_revision="revision-4",
|
|
connection_ref="core-credential:fit-connect-client",
|
|
decryption_key_ref="core-key:fit-connect-decryption",
|
|
event_signing_key_ref="core-key:fit-connect-event-signing",
|
|
)
|
|
|
|
|
|
def _submission() -> FitConnectSubmission:
|
|
return FitConnectSubmission(
|
|
destination_id=DESTINATION_ID,
|
|
submission_id=SUBMISSION_ID,
|
|
transaction_reference="case:transaction-7",
|
|
public_service_identifier="urn:de:fim:leika:leistung:99102013104000",
|
|
region="DE094750156156",
|
|
received_at=datetime(2026, 8, 23, 12, tzinfo=UTC),
|
|
evidence=FitConnectSubmissionEvidence(
|
|
metadata_sha256="a" * 64,
|
|
data_sha256="b" * 64,
|
|
metadata_authentication_tag="metadata-tag",
|
|
data_authentication_tag="data-tag",
|
|
attachments=(
|
|
FitConnectAttachmentEvidence(
|
|
attachment_id="19ffb5ed-0a9d-02b2-9bfb-e271a8474c61",
|
|
content_sha256="c" * 64,
|
|
authentication_tag="attachment-tag",
|
|
),
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _verification(**changes) -> FitConnectVerification:
|
|
values = {
|
|
"downloaded_complete": True,
|
|
"decryption_succeeded": True,
|
|
"metadata_schema_valid": True,
|
|
"data_schema_valid": True,
|
|
"authentication_tags_verified": True,
|
|
"verified_at": datetime(2026, 8, 23, 12, 5, tzinfo=UTC),
|
|
"durable_handoff_reference": "cases:case-4:submission-1",
|
|
"durable_handoff_sha256": "d" * 64,
|
|
}
|
|
values.update(changes)
|
|
return FitConnectVerification(**values)
|
|
|
|
|
|
def test_ingress_receipt_is_exact_and_claims_no_acknowledgement() -> None:
|
|
receipt = create_ingress_receipt(_profile(), _submission())
|
|
|
|
assert receipt.acknowledged is False
|
|
assert receipt.business_accepted is False
|
|
assert len(receipt.receipt_sha256) == 64
|
|
|
|
|
|
def test_complete_durable_verification_builds_effect_free_accept_plan() -> None:
|
|
profile = _profile()
|
|
submission = _submission()
|
|
receipt = create_ingress_receipt(profile, submission)
|
|
|
|
plan = build_acknowledgement_plan(
|
|
profile,
|
|
submission,
|
|
receipt,
|
|
_verification(),
|
|
)
|
|
|
|
assert plan.disposition == "accept"
|
|
assert plan.event_type == FIT_CONNECT_ACCEPT_EVENT
|
|
assert plan.dispatch_allowed is False
|
|
assert plan.technical_receipt_only is True
|
|
assert plan.business_accepted is False
|
|
payload = json.loads(plan.event_request_json)
|
|
assert payload["authentication_tags"]["metadata"] == "metadata-tag"
|
|
assert payload["durable_handoff_sha256"] == "d" * 64
|
|
|
|
|
|
def test_complete_verification_without_durable_handoff_defers() -> None:
|
|
profile = _profile()
|
|
submission = _submission()
|
|
receipt = create_ingress_receipt(profile, submission)
|
|
verification = _verification(
|
|
durable_handoff_reference=None,
|
|
durable_handoff_sha256=None,
|
|
)
|
|
|
|
plan = build_acknowledgement_plan(profile, submission, receipt, verification)
|
|
|
|
assert plan.disposition == "defer"
|
|
assert plan.event_type is None
|
|
assert plan.event_request_json is None
|
|
|
|
|
|
def test_incomplete_download_cannot_be_turned_into_rejection() -> None:
|
|
profile = _profile()
|
|
submission = _submission()
|
|
receipt = create_ingress_receipt(profile, submission)
|
|
verification = _verification(
|
|
downloaded_complete=False,
|
|
durable_handoff_reference=None,
|
|
durable_handoff_sha256=None,
|
|
failure_disposition="reject",
|
|
problem_codes=("download-incomplete",),
|
|
)
|
|
|
|
plan = build_acknowledgement_plan(profile, submission, receipt, verification)
|
|
|
|
assert plan.disposition == "defer"
|
|
|
|
|
|
def test_explicit_permanent_technical_failure_builds_reject_plan() -> None:
|
|
profile = _profile()
|
|
submission = _submission()
|
|
receipt = create_ingress_receipt(profile, submission)
|
|
verification = _verification(
|
|
authentication_tags_verified=False,
|
|
durable_handoff_reference=None,
|
|
durable_handoff_sha256=None,
|
|
failure_disposition="reject",
|
|
problem_codes=("authentication-tags-invalid",),
|
|
)
|
|
|
|
plan = build_acknowledgement_plan(profile, submission, receipt, verification)
|
|
|
|
assert plan.disposition == "reject"
|
|
assert plan.event_type == FIT_CONNECT_REJECT_EVENT
|
|
assert json.loads(plan.event_request_json)["problem_codes"] == [
|
|
"authentication-tags-invalid"
|
|
]
|
|
|
|
|
|
def test_receipt_from_another_submission_is_rejected() -> None:
|
|
profile = _profile()
|
|
submission = _submission()
|
|
receipt = create_ingress_receipt(profile, submission)
|
|
wrong_receipt = replace(
|
|
receipt,
|
|
submission_id="02bf1d9f-282d-4abf-810a-c4104baf0afe",
|
|
)
|
|
|
|
with pytest.raises(FitConnectInboundError, match="receipt submission"):
|
|
build_acknowledgement_plan(profile, submission, wrong_receipt, _verification())
|
|
|
|
|
|
def test_manifest_declares_no_active_target_and_has_german_documentation() -> None:
|
|
manifest = get_manifest()
|
|
assert manifest.version == "0.1.20"
|
|
assert manifest.external_providers[0].id == "fit_connect.submission_api"
|
|
assert manifest.architecture is not None
|
|
assert manifest.architecture.target_tested_providers == ()
|
|
assert manifest.documentation[0].translations["de"]["title"]
|