from __future__ import annotations from collections.abc import Mapping from dataclasses import asdict, replace import hashlib import json import random import pytest from glyph_ascii_v1 import ( GLYPH_ASCII_V1_SYMBOLS, GlyphAsciiV1AtomProof, GlyphAsciiV1Proof, ) from glyph_hybrid_plan import ( GLYPH_HYBRID_ASSET_AUDIO_PLACEHOLDER_SAMPLES_PER_ATOM, GLYPH_HYBRID_BOUNDARY_ID, GLYPH_HYBRID_MAX_ASSET_ATOMS, GLYPH_HYBRID_MAX_ASSET_CANONICAL_UNITS, GLYPH_HYBRID_MAX_IDENTIFIERS, GLYPH_HYBRID_MAX_MODEL_GENERATED_CHUNKS, GLYPH_HYBRID_MAX_MODEL_GENERATED_TEXT_UNITS, GLYPH_HYBRID_MAX_RAW_CHARS, GLYPH_HYBRID_NORMALIZATION_ID, GLYPH_HYBRID_PROFILE_ID, GlyphHybridPlanError, HybridRenderPlan, HybridSegmentSpec, build_glyph_hybrid_plan, glyph_hybrid_plan_sha256, inverse_glyph_hybrid_plan, ) from production import ( count_speech_units, normalize_spoken_forms, plan_generation_chunks, ) def _asset_manifest(prefix: str = "test") -> dict[str, str]: return { symbol.asset_id: hashlib.sha256( f"{prefix}:{symbol.asset_id}".encode("ascii") ).hexdigest() for symbol in GLYPH_ASCII_V1_SYMBOLS } ASSET_MANIFEST = _asset_manifest() def _build(raw: str) -> HybridRenderPlan | None: return build_glyph_hybrid_plan( raw, asset_entry_sha256_by_asset_id=ASSET_MANIFEST, ) def _require_plan(raw: str) -> HybridRenderPlan: plan = _build(raw) assert plan is not None return plan def _inverse( plan: HybridRenderPlan, raw: str, *, manifest: dict[str, str] | None = None, ) -> str: return inverse_glyph_hybrid_plan( plan, asset_entry_sha256_by_asset_id=( ASSET_MANIFEST if manifest is None else manifest ), expected_raw_request=raw, ) @pytest.mark.parametrize( "raw", ( "請寄mead@forest.org確認。", "客服信箱owl@island.net，下載頁面http://forest.org/release。", ( "第一個是owl@island.net，第二個網址" "https://audiocoast.org/tour，完成。" ), ( "若要修改設定，請寄信到mead.voice@forest-mail.org，" "或開啟https://audio-coast.org/release_notes。" ), "，mead@forest.org。", "mead@forest.org", ), ) def test_valid_development_style_requests_have_exact_inverse_and_frozen_headers(raw): plan = _require_plan(raw) assert _inverse(plan, raw) == raw assert plan.profile_id == GLYPH_HYBRID_PROFILE_ID assert plan.normalization_id == GLYPH_HYBRID_NORMALIZATION_ID assert plan.boundary_contract_id == GLYPH_HYBRID_BOUNDARY_ID assert plan.raw_input_sha256 == hashlib.sha256(raw.encode("utf-8")).hexdigest() assert plan.canonical_target_sha256 == hashlib.sha256( plan.canonical_target.encode("utf-8") ).hexdigest() assert plan.plan_sha256 == glyph_hybrid_plan_sha256(plan) assert plan.canonical_target == "".join( segment.canonical_text for segment in plan.segments ) assert plan.canonical_target == plan.canonical_target.strip() assert " " not in plan.canonical_target def test_multi_identifier_ranges_and_core_renderings_are_absolute_and_disjoint(): raw = ( "先寄owl@island.net，再看https://audiocoast.org/tour，" "最後通知mead@forest.org。" ) plan = _require_plan(raw) assert len(plan.identifier_proofs) == 3 previous_raw_end = -1 previous_canonical_end = -1 for proof in plan.identifier_proofs: assert previous_raw_end < proof.absolute_raw_start < proof.absolute_raw_end assert ( previous_canonical_end < proof.canonical_start < proof.canonical_end ) assert plan.canonical_target[ proof.canonical_start : proof.canonical_end ] == " ".join(atom.canonical_token for atom in proof.atoms) previous_raw_end = proof.absolute_raw_end previous_canonical_end = proof.canonical_end raw_coverage = "".join( segment.raw_text for segment in plan.segments if segment.kind != "separator" ) assert raw_coverage == raw assert _inverse(plan, raw) == raw def test_boundary_contract_is_one_explicit_zero_audio_space_between_payloads(): raw = "開頭mead@forest.org，接著https://audiocoast.org/tour結束。" plan = _require_plan(raw) nonempty_payload_indices = [ index for index, segment in enumerate(plan.segments) if segment.kind != "separator" and segment.canonical_text ] assert len(nonempty_payload_indices) > 2 for left, right in zip( nonempty_payload_indices, nonempty_payload_indices[1:], ): assert right == left + 2 separator = plan.segments[left + 1] assert separator.kind == "separator" assert separator.audio_source == "zero_audio" assert separator.raw_text == "" assert separator.absolute_raw_start == separator.absolute_raw_end assert separator.canonical_text == " " assert all( segment.canonical_text != " " for segment in plan.segments if segment.kind != "separator" ) def test_punctuation_only_carriers_are_controls_and_never_model_calls(): raw = "，mead@forest.org。" plan = _require_plan(raw) controls = [ segment for segment in plan.segments if segment.kind == "control" ] assert [segment.raw_text for segment in controls] == ["，", "。"] assert all(segment.audio_source == "zero_audio" for segment in controls) assert not any(segment.kind == "carrier" for segment in plan.segments) assert plan.reservation.model_generated_chunks == 0 assert plan.reservation.model_generated_text_units == 0 assert _inverse(plan, raw) == raw def test_carrier_uses_existing_zh_tw_normalization_without_network_content(): raw = "版本v1.2.3，寄到mead@forest.org，日期是2026/07/18。" plan = _require_plan(raw) carriers = [ segment for segment in plan.segments if segment.kind == "carrier" ] assert [segment.canonical_text for segment in carriers] == [ normalize_spoken_forms(segment.raw_text, locale="zh-TW") for segment in carriers ] assert all(count_speech_units(segment.canonical_text) > 0 for segment in carriers) assert plan.reservation.model_generated_chunks == len(carriers) assert plan.reservation.model_generated_text_units == sum( count_speech_units(segment.canonical_text) for segment in carriers ) def test_non_network_ascii_code_and_semver_remain_ordinary_carrier_content(): raw = "版本v1.2.3與程式C++保持原讀法，聯絡mead@forest.org。" plan = _require_plan(raw) carrier_raw = "".join( segment.raw_text for segment in plan.segments if segment.kind in {"carrier", "control"} ) assert "v1.2.3" in carrier_raw assert "C++" in carrier_raw assert len(plan.identifier_proofs) == 1 assert _inverse(plan, raw) == raw def _alpha(rng: random.Random, minimum: int = 1, maximum: int = 7) -> str: return "".join( rng.choice("abcdefghijklmnopqrstuvwxyz") for _ in range(rng.randint(minimum, maximum)) ) def _email(rng: random.Random) -> str: local = _alpha(rng) if rng.randrange(2): local += rng.choice("._-") + _alpha(rng) return f"{local}@{_alpha(rng)}.{_alpha(rng, 1, 3)}" def _url(rng: random.Random) -> str: url = f"{rng.choice(('http', 'https'))}://{_alpha(rng)}.{_alpha(rng, 1, 3)}" if rng.randrange(2): url += "/" + _alpha(rng) return url def test_seeded_property_valid_multi_identifier_requests_roundtrip_exactly(): rng = random.Random(0xB10E_5A11) carriers = ("請聯絡", "，再查看", "，確認完成。") for _ in range(300): first = _email(rng) second = _url(rng) raw = f"{carriers[0]}{first}{carriers[1]}{second}{carriers[2]}" plan = _require_plan(raw) assert _inverse(plan, raw) == raw assert len(plan.identifier_proofs) == 2 assert plan.reservation.identifiers == 2 assert plan.reservation.asset_atoms == len(first) + len(second) assert plan.reservation.asset_audio_placeholder_samples == ( plan.reservation.asset_atoms * GLYPH_HYBRID_ASSET_AUDIO_PLACEHOLDER_SAMPLES_PER_ATOM ) assert plan.reservation.model_generated_chunks <= ( GLYPH_HYBRID_MAX_MODEL_GENERATED_CHUNKS ) assert plan.reservation.model_generated_text_units <= ( GLYPH_HYBRID_MAX_MODEL_GENERATED_TEXT_UNITS ) @pytest.mark.parametrize( "raw", ( # Whitespace and ASCII punctuation are part of the maximal run; the # planner may not strip them to rescue a substring. "聯絡 mead@forest.org。", "聯絡mead@forest.org ", "聯絡mead@forest.org.", "聯絡mead@forest.org,", "聯絡mead@forest.org!", "請看 https://forest.org/path。", "請看:https://forest.org/path。", # Native lowercase and strict path/domain grammar are mandatory. "聯絡Mead@forest.org。", "聯絡mead@Forest.org。", "請看HTTP://forest.org/path。", "請看https://forest.org/Path。", "聯絡mead1@forest.org。", "聯絡mead@forest2.org。", "請看https://forest.org/path2。", "請看https://forest.org/path?query。", "請看https://forest.org/path#fragment。", "請看https://forest.org:443/path。", "請看https://user@forest.org/path。", "請看https://user:pass@forest.org/path。", "請看https://forest.org/。", # Unicode/confusable/bidi/control data may not split a run and lend a # valid-looking suffix to the strict renderer. "聯絡méad@forest.org。", "聯絡ｍead@forest.org。", "聯絡mead@forest.org\u202e。", "聯絡mead@forest.org\u2066。", "聯絡mead@forest.org\x00。", "聯絡mead@forest.org\n。", "聯絡mead@forest.org\t。", "聯絡梅@forest.org。", "請看https://forest.org/路徑。", # Other legacy network forms in a carrier make activation partial. "聯絡mead@forest.org，備援www.forest.org。", "聯絡mead@forest.org，備援ftp://forest.org/path。", ), ) def test_malformed_ambiguous_or_partially_eligible_requests_fail_closed(raw): assert _build(raw) is None @pytest.mark.parametrize( "raw", ( ( "聯絡mead@forest.org，備援MEAD@forest.org。" ), ( "聯絡mead@forest.org，再看https://forest.org/path?query。" ), ( "先看https://forest.org/path，再寄mead1@forest.org。" ), ), ) def test_one_invalid_network_run_rejects_the_entire_profile_without_partial_plan(raw): assert _build(raw) is None @pytest.mark.parametrize( "malformed_run", ( "foohttps://", "9http://", "fooHTTPS://", "fooftp://", "foowww.", "foo://", "@", "https://", ), ) def test_marker_anywhere_in_a_maximal_ascii_run_blocks_partial_activation( malformed_run, ): raw = f"聯絡mead@forest.org，異常值{malformed_run}，結束。" assert _build(raw) is None @pytest.mark.parametrize( "raw", ( "", "今天測試一般句子，保持原有規劃。", "forest.org只是普通文字。", "網址可能是WWW.FOREST.ORG但不啟用。", "大寫HTTPS://FOREST.ORG不會被修正。", "電子郵件這四個字不是地址。", ), ) def test_no_strict_network_run_never_activates_the_profile(raw): assert _build(raw) is None def test_frozen_reservations_are_computed_before_any_runtime_work(): raw = "請寄mead@forest.org，再看https://audiocoast.org/tour。" plan = _require_plan(raw) assert plan.reservation.raw_chars == len(raw) assert plan.reservation.identifiers == len(plan.identifier_proofs) assert plan.reservation.asset_atoms == sum( len(proof.atoms) for proof in plan.identifier_proofs ) assert plan.reservation.asset_canonical_units == sum( count_speech_units(atom.canonical_token) for proof in plan.identifier_proofs for atom in proof.atoms ) assert plan.reservation.asset_atoms <= GLYPH_HYBRID_MAX_ASSET_ATOMS assert ( plan.reservation.asset_canonical_units <= GLYPH_HYBRID_MAX_ASSET_CANONICAL_UNITS ) def test_raw_identifier_count_atom_and_canonical_caps_fail_closed(): overlong = "前" * GLYPH_HYBRID_MAX_RAW_CHARS + "a@b.c" too_many_identifiers = "，".join( f"{chr(ord('a') + index)}@b.c" for index in range(GLYPH_HYBRID_MAX_IDENTIFIERS + 1) ) # 129 raw glyphs; each identifier itself remains under the core cap. too_many_atoms = "a" * 125 + "@a.b" assert len(too_many_atoms) == GLYPH_HYBRID_MAX_ASSET_ATOMS + 1 # 59 glyphs but >512 speech units because z -> "第二十六個英文字母". too_many_canonical_units = "z" * 55 + "@a.b" assert len(too_many_canonical_units) < GLYPH_HYBRID_MAX_ASSET_ATOMS assert _build(overlong) is None assert _build(too_many_identifiers) is None assert _build(too_many_atoms) is None assert _build(too_many_canonical_units) is None def _plan_with_segments( plan: HybridRenderPlan, segments: tuple[HybridSegmentSpec, ...] | list[HybridSegmentSpec], ) -> HybridRenderPlan: return replace(plan, segments=segments) # type: ignore[arg-type] def test_segment_reorder_omit_duplicate_borrow_substitute_and_gap_are_rejected(): raw = "請寄mead@forest.org確認。" plan = _require_plan(raw) donor = _require_plan("請寄owl@island.net確認。") segments = plan.segments first_asset = next( index for index, segment in enumerate(segments) if segment.kind == "asset_atom" ) donor_asset = next( segment for segment in donor.segments if segment.kind == "asset_atom" ) substituted = list(segments) substituted[first_asset] = replace( segments[first_asset], raw_text="z", raw_sha256=hashlib.sha256(b"z").hexdigest(), ) borrowed = list(segments) borrowed[first_asset] = donor_asset gapped = list(segments) gapped[first_asset] = replace( segments[first_asset], absolute_raw_start=segments[first_asset].absolute_raw_start + 1, ) overlapped = list(segments) overlapped[first_asset] = replace( segments[first_asset], absolute_raw_start=segments[first_asset].absolute_raw_start - 1, ) forged = ( _plan_with_segments(plan, (segments[1], segments[0], *segments[2:])), _plan_with_segments( plan, (*segments[:first_asset], *segments[first_asset + 1 :]), ), _plan_with_segments( plan, (*segments[:first_asset], segments[first_asset], *segments[first_asset:]), ), _plan_with_segments(plan, tuple(borrowed)), _plan_with_segments(plan, tuple(substituted)), _plan_with_segments(plan, tuple(gapped)), _plan_with_segments(plan, tuple(overlapped)), ) for candidate in forged: with pytest.raises(GlyphHybridPlanError): _inverse(candidate, raw) @pytest.mark.parametrize( "mutate", ( lambda segment: replace(segment, ordinal=segment.ordinal + 1), lambda segment: replace(segment, kind="carrier"), lambda segment: replace(segment, audio_source="model"), lambda segment: replace( segment, absolute_raw_end=segment.absolute_raw_end + 1, ), lambda segment: replace( segment, canonical_start=segment.canonical_start + 1, ), lambda segment: replace( segment, canonical_end=segment.canonical_end + 1, ), lambda segment: replace( segment, canonical_text="第二十六個英文字母" ), lambda segment: replace(segment, raw_sha256="0" * 64), lambda segment: replace(segment, canonical_sha256="0" * 64), lambda segment: replace(segment, identifier_ordinal=99), lambda segment: replace(segment, atom_ordinal=99), lambda segment: replace(segment, asset_id="letter-z"), lambda segment: replace(segment, manifest_entry_sha256="0" * 64), ), ) def test_asset_segment_field_provenance_tampering_is_rejected(mutate): raw = "請寄mead@forest.org。" plan = _require_plan(raw) segments = list(plan.segments) index = next( index for index, segment in enumerate(segments) if segment.kind == "asset_atom" ) segments[index] = mutate(segments[index]) with pytest.raises(GlyphHybridPlanError): _inverse(_plan_with_segments(plan, tuple(segments)), raw) def test_identifier_atom_and_whole_proof_tampering_is_rejected(): raw = "請寄mead@forest.org。" plan = _require_plan(raw) proof = plan.identifier_proofs[0] atoms = list(proof.atoms) atom: GlyphAsciiV1AtomProof = atoms[0] atoms[0] = replace(atom, raw_byte=ord("z")) forged_proof: GlyphAsciiV1Proof = replace(proof, atoms=tuple(atoms)) forged = replace(plan, identifier_proofs=(forged_proof,)) with pytest.raises(GlyphHybridPlanError): _inverse(forged, raw) @pytest.mark.parametrize( "mutate", ( lambda plan: replace(plan, profile_id="glyph_hybrid_request_v2"), lambda plan: replace(plan, grammar_id="glyph_ascii_v2"), lambda plan: replace(plan, normalization_id="different"), lambda plan: replace(plan, boundary_contract_id="different"), lambda plan: replace(plan, raw_input_sha256="0" * 64), lambda plan: replace(plan, canonical_target=plan.canonical_target + "X"), lambda plan: replace(plan, canonical_target_sha256="0" * 64), lambda plan: replace(plan, asset_manifest_logical_sha256="0" * 64), lambda plan: replace(plan, plan_sha256="0" * 64), lambda plan: replace( plan, reservation=replace(plan.reservation, asset_atoms=0), ), lambda plan: replace(plan, segments=list(plan.segments)), lambda plan: replace(plan, identifier_proofs=list(plan.identifier_proofs)), ), ) def test_header_hash_type_and_reservation_tampering_is_rejected(mutate): raw = "請寄mead@forest.org。" plan = _require_plan(raw) with pytest.raises(GlyphHybridPlanError): _inverse(mutate(plan), raw) def test_external_raw_request_commitment_rejects_a_different_valid_whole_plan(): raw = "請寄mead@forest.org。" other_raw = "請寄owl@island.net。" other_plan = _require_plan(other_raw) with pytest.raises(GlyphHybridPlanError): _inverse(other_plan, raw) def test_fresh_rehash_cannot_make_a_changed_reservation_the_expected_plan(): raw = "請寄mead@forest.org。" plan = _require_plan(raw) forged = replace( plan, reservation=replace( plan.reservation, model_generated_text_units=( plan.reservation.model_generated_text_units + 1 ), ), plan_sha256="", ) forged = replace( forged, plan_sha256=glyph_hybrid_plan_sha256(forged), ) with pytest.raises(GlyphHybridPlanError, match="unique expected plan"): _inverse(forged, raw) def test_manifest_substitution_and_invalid_manifest_configuration_are_rejected(): raw = "請寄mead@forest.org。" plan = _require_plan(raw) changed = dict(ASSET_MANIFEST) changed[next(iter(changed))] = hashlib.sha256(b"changed").hexdigest() missing = dict(ASSET_MANIFEST) missing.pop(next(iter(missing))) with pytest.raises(GlyphHybridPlanError): _inverse(plan, raw, manifest=changed) with pytest.raises(GlyphHybridPlanError): build_glyph_hybrid_plan( raw, asset_entry_sha256_by_asset_id=missing, ) class _CyclingManifest(Mapping[str, str]): """Return the frozen value once, then a changed value on later reads.""" def __init__(self) -> None: self.accesses = {asset_id: 0 for asset_id in ASSET_MANIFEST} def __iter__(self): return iter(ASSET_MANIFEST) def __len__(self) -> int: return len(ASSET_MANIFEST) def __getitem__(self, asset_id: str) -> str: count = self.accesses[asset_id] self.accesses[asset_id] = count + 1 if count == 0: return ASSET_MANIFEST[asset_id] return hashlib.sha256(f"changed:{asset_id}".encode("ascii")).hexdigest() def test_build_materializes_a_changing_manifest_exactly_once(): raw = ( "先寄mead@forest.org，再看https://audiocoast.org/tour，" "確認完成。" ) cycling = _CyclingManifest() plan = build_glyph_hybrid_plan( raw, asset_entry_sha256_by_asset_id=cycling, ) assert plan == _require_plan(raw) assert set(cycling.accesses.values()) == {1} assert all( proof.asset_manifest_sha256 == plan.asset_manifest_logical_sha256 for proof in plan.identifier_proofs ) def test_inverse_materializes_a_changing_manifest_exactly_once(): raw = ( "先寄mead@forest.org，再看https://audiocoast.org/tour，" "確認完成。" ) plan = _require_plan(raw) cycling = _CyclingManifest() reconstructed = inverse_glyph_hybrid_plan( plan, asset_entry_sha256_by_asset_id=cycling, expected_raw_request=raw, ) assert reconstructed == raw assert set(cycling.accesses.values()) == {1} def test_identifier_proof_manifest_digest_must_equal_plan_header(): raw = "請寄mead@forest.org。" plan = _require_plan(raw) proof = replace(plan.identifier_proofs[0], asset_manifest_sha256="0" * 64) forged = replace( plan, identifier_proofs=(proof,), plan_sha256="", ) forged = replace(forged, plan_sha256=glyph_hybrid_plan_sha256(forged)) with pytest.raises( GlyphHybridPlanError, match="identifier proof manifest does not match plan header", ): _inverse(forged, raw) def test_non_string_inputs_and_plan_subclasses_are_rejected(): class ForgedPlan(HybridRenderPlan): pass raw = "請寄mead@forest.org。" plan = _require_plan(raw) forged = ForgedPlan(**plan.__dict__) with pytest.raises(GlyphHybridPlanError): build_glyph_hybrid_plan( # type: ignore[arg-type] b"mead@forest.org", asset_entry_sha256_by_asset_id=ASSET_MANIFEST, ) with pytest.raises(GlyphHybridPlanError): _inverse(forged, raw) with pytest.raises(GlyphHybridPlanError): inverse_glyph_hybrid_plan( plan, asset_entry_sha256_by_asset_id=ASSET_MANIFEST, expected_raw_request=b"mead@forest.org", # type: ignore[arg-type] ) def test_no_network_legacy_normalization_and_plan_remain_byte_for_byte_frozen(): samples = ( "今天測試一般句子，保持原有規劃。", "日期是2026/07/18，數值為12.5%。", "型號RTX 5090將沿用舊前端。", ) output = [] for raw in samples: assert _build(raw) is None normalized = normalize_spoken_forms(raw, locale="zh-TW") output.append( { "raw": raw, "normalized": normalized, "plan": [ asdict(spec) for spec in plan_generation_chunks(raw, normalized) ], } ) serialized = json.dumps( output, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") assert len(serialized) == 1_128 assert hashlib.sha256(serialized).hexdigest() == ( "d4986f95a0912bc76332a8f0215a22b0359c1fe8ed90733bea99df39535d1759" )